From d26405a3024faa53a1f00f3a398365c1e90d37c1 Mon Sep 17 00:00:00 2001
From: yuj <yuj@ztjzsoft.com>
Date: Tue, 19 May 2026 12:10:05 +0800
Subject: [PATCH] fix(web): handle BrokenPipeError in
 SessionProcess.send_message

If the worker subprocess dies between the start() check and the actual
stdin write, process.stdin.write/drain raises BrokenPipeError (or
ConnectionResetError). Previously this propagated raw to the caller
(FastAPI / websocket handler); now we log it and emit an
"error"/"stdin_broken" status so attached clients see the failure
synchronously.

Also drop the prompt's id from _in_flight_prompt_ids on this path: for a
JSONRPCPromptMessage the id was registered (and "busy" emitted) before the
write, so without cleanup a failed write would leave the session wedged in
is_busy forever. (Flagged in review by Codex and Devin.)

Adds a regression test asserting is_busy is cleared and status is
error/stdin_broken after a broken-pipe write.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 src/kimi_cli/web/runner/process.py       | 26 ++++++++++++-
 tests/web/test_session_error_recovery.py | 49 ++++++++++++++++++++++++
 2 files changed, 73 insertions(+), 2 deletions(-)

diff --git a/src/kimi_cli/web/runner/process.py b/src/kimi_cli/web/runner/process.py
index 16f60a8950..ae1da8aac6 100644
--- a/src/kimi_cli/web/runner/process.py
+++ b/src/kimi_cli/web/runner/process.py
@@ -661,8 +661,30 @@ async def send_message(self, message: str) -> None:
             logger.error(f"{e.__class__.__name__} {e}: Invalid JSONRPC in message: {message}")
             return
 
-        process.stdin.write((message + "\n").encode("utf-8"))
-        await process.stdin.drain()
+        try:
+            process.stdin.write((message + "\n").encode("utf-8"))
+            await process.stdin.drain()
+        except (BrokenPipeError, ConnectionResetError) as e:
+            # Subprocess died between our `start()` check above and the actual write.
+            # `_read_loop` will eventually observe the exit and emit "stopped" /
+            # "crashed", but right now the caller (FastAPI / websocket handler) would
+            # otherwise see a raw exception propagate to the response. Emit an error
+            # status so any attached websocket clients see the failure synchronously.
+            #
+            # If this was a prompt, its id was already registered in
+            # `_in_flight_prompt_ids` above; drop it so the failed turn doesn't
+            # leave the session wedged in `is_busy` forever.
+            if isinstance(in_message, JSONRPCPromptMessage):
+                self._in_flight_prompt_ids.discard(in_message.id)
+            logger.warning(
+                f"send_message: subprocess stdin {e.__class__.__name__}; "
+                f"process likely exited (returncode={process.returncode})"
+            )
+            await self._emit_status(
+                "error",
+                reason="stdin_broken",
+                detail=f"{e.__class__.__name__}: {e}",
+            )
 
 
 class KimiCLIRunner:
diff --git a/tests/web/test_session_error_recovery.py b/tests/web/test_session_error_recovery.py
index 191a100b7c..84d252f0c0 100644
--- a/tests/web/test_session_error_recovery.py
+++ b/tests/web/test_session_error_recovery.py
@@ -162,3 +162,52 @@ def test_session_in_error_state_clears_stale_ids_on_new_prompt() -> None:
         sp.clear_in_flight()
 
     assert sp.is_busy is False
+
+
+# ---------------------------------------------------------------------------
+# Tests: send_message broken stdin clears in-flight
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_send_message_broken_pipe_clears_in_flight() -> None:
+    """A BrokenPipeError while writing a prompt to stdin must drop the prompt's
+    in-flight id, so the failed turn doesn't leave the session stuck in 'busy'.
+    """
+    sp = SessionProcess(uuid4())
+
+    # Don't spawn a real subprocess.
+    async def noop_start() -> None:
+        return None
+
+    sp.start = noop_start  # type: ignore[assignment]
+
+    # stdin whose drain() reports the worker is gone.
+    async def broken_drain() -> None:
+        raise BrokenPipeError("worker gone")
+
+    mock_stdin = MagicMock()
+    mock_stdin.drain = broken_drain
+
+    mock_process = MagicMock()
+    mock_process.stdin = mock_stdin
+    mock_process.returncode = 1
+    sp._process = mock_process
+
+    # Keep the message untouched and broadcasts inert.
+    async def passthrough_handle(in_message: object) -> None:
+        return None
+
+    async def noop_broadcast(msg: str) -> None:
+        return None
+
+    sp._handle_in_message = passthrough_handle  # type: ignore[assignment]
+    sp._broadcast = noop_broadcast  # type: ignore[assignment]
+
+    prompt = '{"jsonrpc":"2.0","method":"prompt","id":"p1","params":{"user_input":"hi"}}'
+    await sp.send_message(prompt)
+
+    assert "p1" not in sp._in_flight_prompt_ids
+    assert sp.is_busy is False
+    assert sp.status.state == "error"
+    assert sp.status.reason == "stdin_broken"