From 7af87f773dc23a04a22790d302f23efc482aa2f1 Mon Sep 17 00:00:00 2001 From: can1357 Date: Fri, 15 May 2026 05:45:48 +0200 Subject: [PATCH] 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. --- .../test/deepseek-reasoning-content.test.ts | 6 +- .../ai/test/openai-completions-compat.test.ts | 64 +++++++++++++-- python/omp-rpc/src/omp_rpc/client.py | 13 ++- python/omp-rpc/tests/test_client.py | 82 +++++++++++++++++++ 4 files changed, 153 insertions(+), 12 deletions(-) diff --git a/packages/ai/test/deepseek-reasoning-content.test.ts b/packages/ai/test/deepseek-reasoning-content.test.ts index 8edc31e63..c4899f9f3 100644 --- a/packages/ai/test/deepseek-reasoning-content.test.ts +++ b/packages/ai/test/deepseek-reasoning-content.test.ts @@ -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); diff --git a/packages/ai/test/openai-completions-compat.test.ts b/packages/ai/test/openai-completions-compat.test.ts index 55bd3cb97..1ee1f7de4 100644 --- a/packages/ai/test/openai-completions-compat.test.ts +++ b/packages/ai/test/openai-completions-compat.test.ts @@ -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); }); diff --git a/python/omp-rpc/src/omp_rpc/client.py b/python/omp-rpc/src/omp_rpc/client.py index e1906465c..04e7e5b8f 100644 --- a/python/omp-rpc/src/omp_rpc/client.py +++ b/python/omp-rpc/src/omp_rpc/client.py @@ -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: diff --git a/python/omp-rpc/tests/test_client.py b/python/omp-rpc/tests/test_client.py index d1113dc63..d46f005fc 100644 --- a/python/omp-rpc/tests/test_client.py +++ b/python/omp-rpc/tests/test_client.py @@ -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()