fix(web): handle BrokenPipeError in SessionProcess.send_message

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:

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"