Skip to content

fix(web): handle BrokenPipeError in SessionProcess.send_message #2324

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Ricardo-M-L wants to merge 1 commit into
base: main
Choose a base branch
from
+18 −2
Open

fix(web): handle BrokenPipeError in SessionProcess.send_message #2324

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 18 additions & 2 deletions src/kimi_cli/web/runner/process.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -661,8 +661,24 @@ 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.
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}",
)
Comment on lines +677 to +681
Copy link
Copy Markdown

@chatgpt-codex-connector chatgpt-codex-connector Bot May 19, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Clear in-flight prompts before emitting stdin_broken

When the broken stdin write is for a JSONRPCPromptMessage, the prompt id was already added to _in_flight_prompt_ids, but this new error path emits the stdin_broken status without clearing it. That makes the session report an error while is_busy remains true, so clients reacting immediately to the status can still be rejected by paths such as get_editable_session()'s busy check until _read_loop later catches up; the existing EOF/error paths explicitly clear in-flight ids before broadcasting for this reason. Clear _in_flight_prompt_ids before this _emit_status call.

Useful? React with 👍 / 👎.

Comment on lines +667 to +681
Copy link
Copy Markdown
Contributor

@devin-ai-integration devin-ai-integration Bot May 19, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 In-flight prompt ID not cleaned up on BrokenPipeError leaves session stuck in 'busy' state

When send_message handles a JSONRPCPromptMessage, it adds the prompt ID to _in_flight_prompt_ids at line 647 and emits a "busy" status at line 649 before the stdin write attempt. If the write then fails with BrokenPipeError, the exception handler (lines 667-681) emits an "error" status but never removes the prompt ID from _in_flight_prompt_ids. This leaves is_busy returning True even though the prompt was never delivered to the subprocess.

The comment says _read_loop will eventually clean up, but there's a race window where _read_loop may have already completed its EOF handling (and called clear()) before the ID was added at line 647. In that case, no one clears the orphaned ID until the next start() call or the error-state recovery path in sessions.py:1011-1016. During this window, the session is simultaneously in "error" and "busy" states, and non-prompt messages (like cancel) that check is_busy will behave incorrectly.

Suggested change
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.
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}",
)
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.
logger.warning(
f"send_message: subprocess stdin {e.__class__.__name__}; "
f"process likely exited (returncode={process.returncode})"
)
self._in_flight_prompt_ids.clear()
await self._emit_status(
"error",
reason="stdin_broken",
detail=f"{e.__class__.__name__}: {e}",
)
Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.



class KimiCLIRunner:
Expand Down
Loading