fix(python/omp-rpc): marked RPC client as closed during stop to unblock prompt waits
- Stop now marks the client as closed via `_mark_closed` with `RpcProcessExitError`, so `_wait_for_agent_end` wakes immediately instead of waiting for a timeout. - Added a regression test using a hanging server subprocess to verify `stop()` unblocks `prompt_and_wait` promptly and raises `RpcProcessExitError`. - Updated Kimi compatibility tests to distinguish Moonshot-hosted models from OpenCode-hosted ones for `reasoning_content` and assistant-content tool-call behavior.
This commit is contained in:
@@ -495,9 +495,9 @@ describe("DeepSeek reasoning_content tool-call replay", () => {
|
||||
const model: Model<"openai-completions"> = {
|
||||
...getBundledModel("openai", "gpt-4o-mini"),
|
||||
api: "openai-completions",
|
||||
provider: "opencode-go",
|
||||
baseUrl: "https://opencode.ai/zen/go/v1",
|
||||
id: "moonshotai/kimi-k2.5",
|
||||
provider: "moonshot",
|
||||
baseUrl: "https://api.moonshot.ai/v1",
|
||||
id: "kimi-k2.5",
|
||||
reasoning: true,
|
||||
};
|
||||
const compat = detectCompat(model);
|
||||
|
||||
@@ -482,19 +482,68 @@ describe("kimi model detection via detectCompat", () => {
|
||||
};
|
||||
}
|
||||
|
||||
it("requires reasoning_content for tool calls on kimi-k2.5 (opencode-go)", () => {
|
||||
function kimiMoonshotModel(id: string): Model<"openai-completions"> {
|
||||
return {
|
||||
...getBundledModel("openai", "gpt-4o-mini"),
|
||||
api: "openai-completions",
|
||||
provider: "moonshot",
|
||||
baseUrl: "https://api.moonshot.ai/v1",
|
||||
id,
|
||||
reasoning: true,
|
||||
};
|
||||
}
|
||||
|
||||
// Regression for #1071: OpenCode-Go/Zen handle reasoning content server-side
|
||||
// and reject client-supplied `reasoning_content` ("Extra inputs are not
|
||||
// permitted"). Kimi on opencode-* MUST NOT have reasoning_content injected,
|
||||
// even though it's still recognized as a Kimi model for other quirks.
|
||||
it("does not require reasoning_content for tool calls on kimi-k2.5 (opencode-go)", () => {
|
||||
const compat = detectCompat(kimiOpenCodeModel("kimi-k2.5"));
|
||||
expect(compat.requiresReasoningContentForToolCalls).toBe(true);
|
||||
expect(compat.requiresReasoningContentForToolCalls).toBe(false);
|
||||
// Kimi-specific quirks still apply even on opencode hosts.
|
||||
expect(compat.requiresAssistantContentForToolCalls).toBe(true);
|
||||
});
|
||||
|
||||
it("injects reasoning_content placeholder when assistant with tool calls has no reasoning field", () => {
|
||||
it("does not inject reasoning_content placeholder for kimi on opencode-go", () => {
|
||||
const model = kimiOpenCodeModel("kimi-k2.5");
|
||||
const compat = detectCompat(model);
|
||||
const toolCallMessage: AssistantMessage = {
|
||||
role: "assistant",
|
||||
content: [
|
||||
// Thinking returned as plain text (as kimi-k2.5 on opencode-go does)
|
||||
{ type: "text", text: "Let me research this." },
|
||||
{
|
||||
type: "toolCall",
|
||||
id: "call_abc123",
|
||||
name: "web_search",
|
||||
arguments: { query: "beads gastownhall" },
|
||||
},
|
||||
],
|
||||
api: model.api,
|
||||
provider: model.provider,
|
||||
model: model.id,
|
||||
usage: {
|
||||
input: 0,
|
||||
output: 0,
|
||||
cacheRead: 0,
|
||||
cacheWrite: 0,
|
||||
totalTokens: 0,
|
||||
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
|
||||
},
|
||||
stopReason: "toolUse",
|
||||
timestamp: Date.now(),
|
||||
};
|
||||
const messages = convertMessages(model, { messages: [toolCallMessage] }, compat);
|
||||
const assistant = messages.find(m => m.role === "assistant");
|
||||
expect(assistant).toBeDefined();
|
||||
expect(Reflect.get(assistant as object, "reasoning_content")).toBeUndefined();
|
||||
});
|
||||
|
||||
it("injects reasoning_content placeholder when kimi-on-moonshot has tool calls without reasoning field", () => {
|
||||
const model = kimiMoonshotModel("kimi-k2.5");
|
||||
const compat = detectCompat(model);
|
||||
const toolCallMessage: AssistantMessage = {
|
||||
role: "assistant",
|
||||
content: [
|
||||
{ type: "text", text: "Let me research this." },
|
||||
{
|
||||
type: "toolCall",
|
||||
@@ -579,10 +628,15 @@ describe("kimi model detection via detectCompat", () => {
|
||||
};
|
||||
const compat = detectCompat(model);
|
||||
expect(compat.requiresReasoningContentForToolCalls).toBe(false);
|
||||
expect(compat.requiresAssistantContentForToolCalls).toBe(false);
|
||||
});
|
||||
|
||||
// `requiresAssistantContentForToolCalls` keys directly off isKimiModel and
|
||||
// is provider-agnostic, so it's the cleanest signal that the id-pattern
|
||||
// match recognizes every Kimi variant.
|
||||
it.each(["kimi-k2.5", "kimi-k1.5", "kimi-k2-5"])("matches kimi model id: %s", id => {
|
||||
const compat = detectCompat(kimiOpenCodeModel(id));
|
||||
const compat = detectCompat(kimiMoonshotModel(id));
|
||||
expect(compat.requiresAssistantContentForToolCalls).toBe(true);
|
||||
expect(compat.requiresReasoningContentForToolCalls).toBe(true);
|
||||
});
|
||||
|
||||
|
||||
@@ -484,13 +484,18 @@ class RpcClient:
|
||||
process.stderr.close()
|
||||
except OSError:
|
||||
pass
|
||||
self._fail_pending(RpcProcessExitError("RPC process stopped"))
|
||||
# Mark the client closed so any thread blocked in
|
||||
# `_wait_for_agent_end` raises `RpcProcessExitError` instead of
|
||||
# waiting for its request timeout. The stdout reader loop would
|
||||
# normally do this when it observes the closed pipe, but it
|
||||
# guards on `if not self._stopping:` — which is True by the time
|
||||
# we get here — and so skips it. Calling `_mark_closed` directly
|
||||
# closes the gap. It is idempotent: a second call (e.g. from the
|
||||
# reader's exception path) returns early.
|
||||
self._mark_closed(RpcProcessExitError("RPC process stopped"))
|
||||
self._pending_host_tool_calls.clear()
|
||||
self._pending_host_uri_requests.clear()
|
||||
self._process = None
|
||||
self._ready.set()
|
||||
with self._event_condition:
|
||||
self._event_condition.notify_all()
|
||||
if self._stdout_thread is not None:
|
||||
self._stdout_thread.join(timeout=1.0)
|
||||
if self._stderr_thread is not None:
|
||||
|
||||
@@ -986,5 +986,87 @@ class RpcClientTests(unittest.TestCase):
|
||||
self.assertIn("max_event_history", str(ctx.exception))
|
||||
|
||||
|
||||
|
||||
HANGING_SERVER = textwrap.dedent(
|
||||
"""
|
||||
import json
|
||||
import sys
|
||||
|
||||
print(json.dumps({"type": "ready"}), flush=True)
|
||||
# Read one line (the prompt) and acknowledge it, then never emit agent_end.
|
||||
# The client's prompt_and_wait should sit in _wait_for_agent_end forever
|
||||
# unless stop() unblocks it.
|
||||
line = sys.stdin.readline()
|
||||
if line:
|
||||
command = json.loads(line)
|
||||
if command.get("type") == "prompt":
|
||||
print(
|
||||
json.dumps(
|
||||
{
|
||||
"id": command["id"],
|
||||
"type": "response",
|
||||
"command": "prompt",
|
||||
"success": True,
|
||||
}
|
||||
),
|
||||
flush=True,
|
||||
)
|
||||
# Block forever on stdin so the subprocess does not exit on its own.
|
||||
sys.stdin.read()
|
||||
"""
|
||||
)
|
||||
|
||||
|
||||
class StopUnblocksPromptAndWaitTests(unittest.TestCase):
|
||||
"""Regression: stop() must wake `_wait_for_agent_end` immediately.
|
||||
|
||||
Previously, the stdout reader's "if not self._stopping:" guard caused
|
||||
`_mark_closed` to be skipped after stop(), so `_closed_error` stayed
|
||||
`None` and `_wait_for_agent_end` blocked on its condition variable until
|
||||
the prompt timeout. The fix sets `_closed_error` from `stop()` itself.
|
||||
"""
|
||||
|
||||
def test_stop_during_prompt_unblocks_waiter(self) -> None:
|
||||
from omp_rpc import RpcProcessExitError
|
||||
|
||||
client = RpcClient(
|
||||
command=[sys.executable, "-u", "-c", HANGING_SERVER],
|
||||
startup_timeout=2.0,
|
||||
request_timeout=2.0,
|
||||
)
|
||||
client.start()
|
||||
try:
|
||||
errors: list[BaseException] = []
|
||||
|
||||
def run_prompt() -> None:
|
||||
try:
|
||||
# 30s is more than enough to let stop() race in; if the
|
||||
# bug regresses, the worker hangs the full 30s.
|
||||
client.prompt_and_wait("hang", timeout=30.0)
|
||||
except BaseException as exc:
|
||||
errors.append(exc)
|
||||
|
||||
thread = threading.Thread(target=run_prompt)
|
||||
thread.start()
|
||||
|
||||
# Wait until the prompt is in flight.
|
||||
deadline = time.time() + 2.0
|
||||
while client._prompt_lifecycle.active_operation != "prompt_and_wait" and time.time() < deadline:
|
||||
time.sleep(0.01)
|
||||
self.assertEqual(client._prompt_lifecycle.active_operation, "prompt_and_wait")
|
||||
|
||||
t0 = time.time()
|
||||
client.stop()
|
||||
thread.join(timeout=2.0)
|
||||
elapsed = time.time() - t0
|
||||
|
||||
self.assertFalse(thread.is_alive(), "prompt_and_wait did not return after stop()")
|
||||
self.assertLess(elapsed, 2.0, f"stop() took {elapsed:.2f}s to unblock prompt_and_wait")
|
||||
self.assertEqual(len(errors), 1)
|
||||
self.assertIsInstance(errors[0], RpcProcessExitError)
|
||||
finally:
|
||||
# stop() is idempotent; safe to call again on cleanup paths.
|
||||
client.stop()
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user