fix(stream): don't strand a turn on shutdown, and say goodbye over stdin
The EOF sentinel was pushed with `await queue.put()` from a `finally` that a shutdown-time cancellation may have entered. An unbounded queue never blocks there, so it happened to work, but one suspension point inside that `finally` would swallow the sentinel and park a turn generator on `next_record()` forever. `put_nowait` can't. `terminate()` now closes stdin before signalling: EOF is the protocol's own goodbye, so a healthy claude exits on its own and never sees SIGTERM.
This commit is contained in:
@@ -377,7 +377,12 @@ class StreamClaudeProcess:
|
|||||||
await self._queue.put(record)
|
await self._queue.put(record)
|
||||||
finally:
|
finally:
|
||||||
self._eof = True
|
self._eof = True
|
||||||
await self._queue.put(None)
|
# `put_nowait`, not `await put`: this runs in a `finally`
|
||||||
|
# that a shutdown-time cancellation may have entered, and an
|
||||||
|
# unbounded queue never blocks anyway. Awaiting here would
|
||||||
|
# risk swallowing the sentinel and leaving a turn generator
|
||||||
|
# parked on `next_record()` forever.
|
||||||
|
self._queue.put_nowait(None)
|
||||||
|
|
||||||
async def _read_stderr(self) -> None:
|
async def _read_stderr(self) -> None:
|
||||||
proc = self._proc
|
proc = self._proc
|
||||||
@@ -456,11 +461,19 @@ class StreamClaudeProcess:
|
|||||||
# ---- shutdown -----------------------------------------------------
|
# ---- shutdown -----------------------------------------------------
|
||||||
|
|
||||||
async def terminate(self, *, grace: float = 5.0) -> int | None:
|
async def terminate(self, *, grace: float = 5.0) -> int | None:
|
||||||
"""SIGTERM → wait up to `grace` seconds → SIGKILL ladder."""
|
"""Close stdin → SIGTERM → wait up to `grace` → SIGKILL ladder.
|
||||||
|
|
||||||
|
Closing stdin first is the protocol's own goodbye: headless
|
||||||
|
claude exits on EOF, so a healthy process usually never sees the
|
||||||
|
signals at all.
|
||||||
|
"""
|
||||||
proc = self._proc
|
proc = self._proc
|
||||||
if proc is None:
|
if proc is None:
|
||||||
return None
|
return None
|
||||||
if proc.returncode is None:
|
if proc.returncode is None:
|
||||||
|
with contextlib.suppress(Exception):
|
||||||
|
if proc.stdin is not None and not proc.stdin.is_closing():
|
||||||
|
proc.stdin.close()
|
||||||
signal_group(proc.pid, signal.SIGTERM)
|
signal_group(proc.pid, signal.SIGTERM)
|
||||||
with contextlib.suppress(TimeoutError):
|
with contextlib.suppress(TimeoutError):
|
||||||
await asyncio.wait_for(proc.wait(), timeout=grace)
|
await asyncio.wait_for(proc.wait(), timeout=grace)
|
||||||
|
|||||||
Reference in New Issue
Block a user