diff --git a/src/claude_code_api/__init__.py b/src/claude_code_api/__init__.py index 675037e..243dd9c 100644 --- a/src/claude_code_api/__init__.py +++ b/src/claude_code_api/__init__.py @@ -43,6 +43,7 @@ from claude_code_api.models import ( is_valid_model, ) from claude_code_api.normalizer import normalize +from claude_code_api.pty import kill_orphaned_processes from claude_code_api.turn import TurnManager __version__ = "0.1.0" @@ -76,6 +77,7 @@ __all__ = [ "UserMessage", "classify_pty_failure", "is_valid_model", + "kill_orphaned_processes", "normalize", "synthesize_turn_messages", ] diff --git a/src/claude_code_api/backend.py b/src/claude_code_api/backend.py index c66d902..a125518 100644 --- a/src/claude_code_api/backend.py +++ b/src/claude_code_api/backend.py @@ -22,6 +22,7 @@ import json import logging import os import tempfile +import time import uuid from collections.abc import AsyncIterator, Callable, Iterable, Mapping from dataclasses import dataclass, field @@ -55,6 +56,10 @@ HistoryInjectionMode = Literal["native_jsonl", "concat_message"] ParseErrorCallback = Callable[[MessageParseError, dict[str, Any]], None] +# How often the reaper wakes up to look for idle sessions. Far shorter +# than any sane TTL — the cost is one dict scan. +_REAP_INTERVAL = 60.0 + @dataclass(frozen=True) class BackendOptions: @@ -88,6 +93,19 @@ class BackendOptions: file_wait_timeout: float = 30.0 turn_duration_timeout: float = 5.0 + idle_session_ttl: float = 1800.0 + """Seconds a pooled session may sit unused before it is terminated. + + The pool is keyed by history fingerprint, so a conversation that + forks (edited transcript, a reply that never made it back to the + client) leaves its previous session behind under a key nothing will + ever look up again. Without a TTL those sessions accumulate one live + ``claude`` each — on a 4 GB host a handful is the whole machine. + + There is deliberately no count cap: a burst of genuinely concurrent + conversations should all stay warm. Set to ``0`` to disable reaping. + """ + @dataclass class _LiveSession: @@ -101,6 +119,25 @@ class _LiveSession: def session_id(self) -> str: return self.pty.session_id + @property + def idle_seconds(self) -> float: + """Seconds since the last prompt, or 0 for a PTY that can't say. + + Both accessors tolerate a PTY without the timestamps: the + ``_session_factory`` seam exists so callers can inject minimal + fakes, and session bookkeeping must not widen that contract. A + fake reporting 0 is simply never idle enough to reap. + """ + last = getattr(self.pty, "last_activity_at", None) + if last is None: + return 0.0 + return max(0.0, time.time() - last) + + def touch(self) -> None: + touch = getattr(self.pty, "touch", None) + if touch is not None: + touch() + async def aclose(self) -> None: await self.tm.aclose() @@ -141,6 +178,7 @@ class ClaudeCodeBackend: self._session_factory = _session_factory self._closed = False self._lock = asyncio.Lock() + self._reaper: asyncio.Task[None] | None = None @property def options(self) -> BackendOptions: @@ -205,6 +243,7 @@ class ClaudeCodeBackend: session: _LiveSession if prior and fp_prior in self._sessions: session = self._sessions.pop(fp_prior) + session.touch() send_text = last_text _log.info( "complete: POOL HIT fp=%s -> reusing session_id=%s", @@ -212,10 +251,22 @@ class ClaudeCodeBackend: session.session_id, ) else: + # A miss means the caller's history no longer matches any + # session we're holding — almost always because the caller + # rewrote its own transcript, not because this is a new + # conversation. Log the pool's shape so that's diagnosable + # from the outside instead of by attaching a debugger. _log.info( - "complete: POOL MISS fp=%s (prior=%d msgs) -> spawning new session", + "complete: POOL MISS fp=%s (prior=%d msgs) -> spawning new " + "session; pooled=%d [%s]", fp_prior[:12], len(prior), + len(self._sessions), + ", ".join( + f"{fp[:12]}:{s.session_id[:8]}:idle{s.idle_seconds:.0f}s" + for fp, s in self._sessions.items() + ) + or "empty", ) session = await self._create_session(prior) if prior and self._opts.history_injection_mode == "concat_message": @@ -276,12 +327,70 @@ class ClaudeCodeBackend: new_fp[:12], ) + async def _reap_idle_sessions(self) -> None: + """Terminate pooled sessions that have sat unused past the TTL. + + Only ``_sessions`` is eligible — a session in ``_active`` is + mid-turn by definition. Runs under the same lock as ``complete`` + so a session can't be reaped between fingerprint lookup and use. + """ + ttl = self._opts.idle_session_ttl + if ttl <= 0: + return + async with self._lock: + stale = [ + (fp, s) for fp, s in self._sessions.items() if s.idle_seconds > ttl + ] + for fp, _ in stale: + del self._sessions[fp] + for fp, s in stale: + _log.info( + "reaper: terminating session_id=%s (idle %.0fs > ttl %.0fs, fp=%s)", + s.session_id, + s.idle_seconds, + ttl, + fp[:12], + ) + with contextlib.suppress(Exception): + await s.aclose() + if stale: + _log.info( + "reaper: reaped %d session(s), %d remain pooled", + len(stale), + len(self._sessions), + ) + + async def _reaper_loop(self) -> None: + while not self._closed: + await asyncio.sleep(_REAP_INTERVAL) + try: + await self._reap_idle_sessions() + except asyncio.CancelledError: + raise + except Exception: + _log.exception("reaper: pass failed — continuing") + async def aclose(self) -> None: - """Shut down all live sessions; remove the temp mcp-config file.""" + """Shut down every live session; remove the temp mcp-config file. + + ``_active`` is drained alongside ``_sessions``: a turn in flight + during shutdown used to leave its ``claude`` running forever, + because ptyprocess puts each child in its own session (setsid), + so it never sees the Ctrl-C that killed the gateway. + """ self._closed = True - sessions = list(self._sessions.values()) + reaper = self._reaper + self._reaper = None + if reaper is not None and not reaper.done(): + reaper.cancel() + with contextlib.suppress(asyncio.CancelledError, Exception): + await reaper + sessions = {id(s): s for s in self._sessions.values()} + sessions.update({id(s): s for s in self._active.values()}) self._sessions.clear() - for s in sessions: + self._active.clear() + _log.info("aclose: terminating %d live session(s)", len(sessions)) + for s in sessions.values(): with contextlib.suppress(Exception): await s.aclose() if self._mcp_config_path is not None: @@ -290,6 +399,10 @@ class ClaudeCodeBackend: self._mcp_config_path = None async def __aenter__(self) -> Self: + if self._reaper is None and self._opts.idle_session_ttl > 0: + self._reaper = asyncio.create_task( + self._reaper_loop(), name="claude-session-reaper" + ) return self async def __aexit__(self, _exc_type: object, _exc: object, _tb: object) -> None: diff --git a/src/claude_code_api/pty.py b/src/claude_code_api/pty.py index d6cbed8..010841f 100644 --- a/src/claude_code_api/pty.py +++ b/src/claude_code_api/pty.py @@ -13,6 +13,7 @@ This module knows nothing about turns, JSONL, or event normalization. from __future__ import annotations import asyncio +import atexit import contextlib import errno import logging @@ -21,6 +22,7 @@ import pathlib import select import signal import threading +import time import uuid from collections.abc import Callable, Iterable, Mapping from dataclasses import dataclass, field @@ -34,6 +36,56 @@ _log = logging.getLogger("claude_code_api.pty") PtyOutputCallback = Callable[[bytes], None] +# ptyprocess spawns via ``pty.fork()``, which calls ``setsid()`` in the +# child — every claude lands in its own session and process group, with +# pgid == pid. Two consequences: +# +# * a Ctrl-C in the gateway's terminal goes to the *gateway's* foreground +# process group and never reaches claude, so orphans survive unless we +# kill them explicitly; +# * killing the group (rather than just the pid) also takes down whatever +# claude spawned — MCP stdio servers, ripgrep, node workers. +# +# We keep the pgids of every live process here so an ``atexit`` hook can +# sweep anything a disorderly shutdown left behind. Nothing saves us from +# SIGKILL on the gateway itself; everything short of that is covered. +_LIVE_PGIDS: set[int] = set() +_LIVE_PGIDS_LOCK = threading.Lock() + + +def _register_live(pgid: int) -> None: + with _LIVE_PGIDS_LOCK: + _LIVE_PGIDS.add(pgid) + + +def _unregister_live(pgid: int) -> None: + with _LIVE_PGIDS_LOCK: + _LIVE_PGIDS.discard(pgid) + + +def kill_orphaned_processes() -> int: + """SIGKILL every process group we spawned and haven't reaped. + + Registered as an ``atexit`` hook, and safe to call directly (e.g. + from a SIGTERM handler). Returns the number of groups signalled. + """ + with _LIVE_PGIDS_LOCK: + pgids = sorted(_LIVE_PGIDS) + _LIVE_PGIDS.clear() + killed = 0 + for pgid in pgids: + try: + os.killpg(pgid, signal.SIGKILL) + except (OSError, ProcessLookupError): + continue + killed += 1 + if killed: + _log.warning("kill_orphaned_processes: SIGKILLed %d orphan group(s)", killed) + return killed + + +atexit.register(kill_orphaned_processes) + _PROVIDER_ENV_VARS: tuple[str, ...] = ( "ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN", @@ -64,21 +116,13 @@ _SNAPSHOT_INTERVAL = 10.0 # the paste handler is still scheduling when Enter arrives. _SUBMIT_KEY_DELAY = 0.25 -# After sending Enter we poll the PTY output for the paste-indicator -# (claude's TUI renders ``[Pasted text #...`` while the paste is still -# buffered in the input box). If it's still there after this window, -# Enter never registered — we retry up to ``_SUBMIT_RETRIES`` times. -_SUBMIT_VERIFY_WINDOW = 1.5 -_SUBMIT_VERIFY_POLL = 0.1 -_SUBMIT_RETRIES = 3 -_PASTE_INDICATOR = b"[Pasted text" - -# Size of the most-recent slice of PTY output considered "the current -# screen render" when verifying a submit. The TUI emits a full clear + -# redraw per state change; the latest redraw fits comfortably in this -# window, while earlier intermediate frames (which may still show the -# paste indicator) live further back in the same buffer. -_TUI_TAIL_WINDOW = 8192 +# Submit verification does NOT live here — it lives in ``TurnManager``, +# which can see the session JSONL. The PTY layer has no trustworthy +# signal: claude's TUI renders ``[Pasted text #N +M lines]`` only for +# long/multi-line pastes, so a short prompt produces no indicator at all +# and any "indicator is gone ⇒ submitted" check passes vacuously. The +# JSONL user record is ground truth, so ``send_submit_key()`` is exposed +# for the turn layer to re-press Enter when that record doesn't appear. @dataclass(frozen=True) @@ -236,11 +280,34 @@ class PtyClaudeProcess: self._output_lock = threading.Lock() self._output_buffer = bytearray() self._snapshot_task: asyncio.Task[None] | None = None + self._created_at = time.time() + self._last_activity_at = self._created_at @property def session_id(self) -> str: return self._session_id + @property + def created_at(self) -> float: + """Unix timestamp of construction (not of ``start()``).""" + return self._created_at + + @property + def last_activity_at(self) -> float: + """Unix timestamp of the last prompt written to this PTY. + + Drives idle eviction in ``ClaudeCodeBackend`` and the + most-recent-first ordering in the admin PTY list. Deliberately + tracks *prompts*, not drain-thread output: a session rendering a + spinner is not being used, and claude redraws its TUI clock + forever, which would make every session look permanently busy. + """ + return self._last_activity_at + + def touch(self) -> None: + """Mark this session as used right now.""" + self._last_activity_at = time.time() + @property def argv(self) -> list[str]: return list(self._argv) @@ -318,6 +385,7 @@ class PtyClaudeProcess: self._session_id, self._pty.pid, ) + _register_live(self._pgid()) self._drain_stop.clear() self._drain_thread = threading.Thread( target=self._drain_loop, @@ -366,35 +434,6 @@ class PtyClaudeProcess: if overflow > 0: del self._output_buffer[:overflow] - async def _wait_for_submit(self, pre_enter_len: int) -> bool: - """Poll the PTY output to confirm Enter actually submitted the paste. - - Claude's TUI renders ``[Pasted text #N +M lines]`` in the input - box while the paste is still buffered (unsubmitted). After Enter - registers, the input clears and the TUI redraws — the next - screen render no longer contains the indicator. - - We compare bytes written *after* the Enter write (``pre_enter_len`` - was snapshotted just before). Within those new bytes, only the - most recent ``_TUI_TAIL_WINDOW`` matter — that's the current - rendered screen; earlier bytes are intermediate redraws (one of - which would still show the indicator). - - Returns True once the indicator is absent from the recent tail, - False on timeout. - """ - deadline = asyncio.get_running_loop().time() + _SUBMIT_VERIFY_WINDOW - while asyncio.get_running_loop().time() < deadline: - buf = self.captured_output() - new_bytes = buf[pre_enter_len:] - tail = new_bytes[-_TUI_TAIL_WINDOW:] - # Only check once the TUI has produced at least one redraw - # after our Enter (otherwise tail is empty / pre-render). - if tail and _PASTE_INDICATOR not in tail: - return True - await asyncio.sleep(_SUBMIT_VERIFY_POLL) - return False - async def _snapshot_loop(self, snapshot_dir: pathlib.Path) -> None: """Periodically dump the captured PTY buffer to a file. @@ -573,6 +612,11 @@ class PtyClaudeProcess: 3. A separate write sends the `\r` as a fresh keystroke that triggers Submit. + That Enter is best-effort: the TUI can still swallow it under + load, and this layer cannot tell. `TurnManager` confirms the + submit against the session JSONL and calls `send_submit_key()` + to re-press Enter when nothing shows up. + Callers that need raw byte streaming (e.g. arrow keys, individual keypresses) pass `newline=False` and write the framing themselves. """ @@ -581,6 +625,7 @@ class PtyClaudeProcess: raise RuntimeError(msg) payload = data.encode("utf-8") if isinstance(data, str) else bytes(data) pty = self._pty + self.touch() if not newline: _log.info( "write: session_id=%s RAW %d bytes (newline=False)", @@ -599,45 +644,31 @@ class PtyClaudeProcess: ) n1 = await asyncio.to_thread(pty.write, paste_chunk) await asyncio.sleep(_SUBMIT_KEY_DELAY) - n2 = 0 - for attempt in range(1, _SUBMIT_RETRIES + 1): - # Snapshot just before each Enter write so the verifier only - # looks at bytes the TUI produces in response to *this* - # Enter, not earlier renders (which still show the paste - # indicator from when the paste was first buffered). - pre_enter_len = len(self.captured_output()) - n2 += await asyncio.to_thread(pty.write, b"\r") - submitted = await self._wait_for_submit(pre_enter_len) - if submitted: - if attempt > 1: - _log.info( - "write: session_id=%s SUBMIT registered after %d Enter(s)", - self._session_id, - attempt, - ) - break - _log.warning( - "write: session_id=%s Enter #%d did not clear paste indicator " - "(buf=%d bytes) — retrying", - self._session_id, - attempt, - len(self.captured_output()), - ) - else: - _log.error( - "write: session_id=%s SUBMIT never registered after %d " - "Enter attempts — paste likely stuck in input box", - self._session_id, - _SUBMIT_RETRIES, - ) + n2 = await asyncio.to_thread(pty.write, b"\r") _log.info( - "write: session_id=%s SUBMIT done (paste=%d Enter=%d)", + "write: session_id=%s paste=%d bytes + Enter=%d bytes written " + "(submit unconfirmed until JSONL moves)", self._session_id, n1, n2, ) return n1 + n2 + async def send_submit_key(self) -> int: + r"""Press Enter again, on its own, to submit a stuck paste. + + Used by `TurnManager` when the JSONL shows no sign that the + previous Enter registered. Writing a bare `\r` is safe to repeat: + if the input box is already empty, claude treats it as a no-op. + """ + if self._pty is None: + msg = "PtyClaudeProcess not started" + raise RuntimeError(msg) + _log.warning( + "send_submit_key: session_id=%s re-pressing Enter", self._session_id + ) + return await asyncio.to_thread(self._pty.write, b"\r") + async def send_control(self, char: str) -> None: """Send a control character (e.g. 'c' for Ctrl-C, 'd' for Ctrl-D).""" if self._pty is None: @@ -653,20 +684,50 @@ class PtyClaudeProcess: pty = self._pty return await asyncio.to_thread(pty.wait) + def _pgid(self) -> int: + """Process-group id of the child. + + ``pty.fork()`` makes the child a session leader, so its pgid + equals its pid; we still ask the kernel rather than assume it. + """ + pty = self._pty + if pty is None: + msg = "PtyClaudeProcess not started" + raise RuntimeError(msg) + try: + return os.getpgid(pty.pid) + except OSError: + return pty.pid + + def _signal_group(self, sig: int) -> None: + """Signal claude *and everything it spawned*. + + Signalling only the pid leaves claude's own children behind — + MCP stdio servers, node workers, long-running Bash tools — which + is how a "terminated" session keeps holding hundreds of MB. + Falls back to a plain pid kill if the group is already gone. + """ + pty = self._pty + if pty is None: + return + try: + os.killpg(self._pgid(), sig) + except (OSError, ProcessLookupError): + with contextlib.suppress(OSError): + pty.kill(sig) + async def terminate(self, *, grace: float = 5.0) -> int | None: """SIGTERM → wait up to `grace` seconds → SIGKILL ladder.""" if self._pty is None: return None pty = self._pty if pty.isalive(): - with contextlib.suppress(OSError): - pty.kill(signal.SIGTERM) + self._signal_group(signal.SIGTERM) deadline = asyncio.get_running_loop().time() + grace while pty.isalive() and asyncio.get_running_loop().time() < deadline: await asyncio.sleep(0.05) if pty.isalive(): - with contextlib.suppress(OSError): - pty.kill(signal.SIGKILL) + self._signal_group(signal.SIGKILL) return await self._reap() async def kill(self) -> int | None: @@ -675,14 +736,15 @@ class PtyClaudeProcess: return None pty = self._pty if pty.isalive(): - with contextlib.suppress(OSError): - pty.kill(signal.SIGKILL) + self._signal_group(signal.SIGKILL) return await self._reap() async def _reap(self) -> int | None: pty = self._pty if pty is None: return None + with contextlib.suppress(RuntimeError): + _unregister_live(self._pgid()) exit_status = await asyncio.to_thread(pty.wait) self._drain_stop.set() thread = self._drain_thread @@ -720,4 +782,5 @@ __all__: Iterable[str] = ( "PtyProcessOptions", "build_argv", "build_env", + "kill_orphaned_processes", ) diff --git a/src/claude_code_api/turn.py b/src/claude_code_api/turn.py index b071255..d28562a 100644 --- a/src/claude_code_api/turn.py +++ b/src/claude_code_api/turn.py @@ -64,6 +64,20 @@ _DEFAULT_STARTUP_DELAY = 10.0 _TUI_READY_MARKER: bytes = b"\x1b[?2004h" _TUI_QUIET_PERIOD: float = 1.5 +# Submitting a prompt is not reliably observable from the PTY: claude's +# TUI only renders a ``[Pasted text #N]`` placeholder for long/multi-line +# pastes, so short prompts leave no trace to check against and the Enter +# that follows a paste can be swallowed by the Ink stdin reader under +# load. The session JSONL *is* ground truth — claude appends a record as +# soon as a prompt is accepted. So we wait for the file to move, and +# re-press Enter if it doesn't. +# +# The window has to clear claude's own start-of-turn latency on a loaded +# Pi (spinner up, context assembled) without stranding the user behind a +# genuinely stuck input box for a minute. +_SUBMIT_CONFIRM_WINDOW: float = 12.0 +_SUBMIT_RETRIES: int = 3 + ParseErrorCallback = Callable[[MessageParseError, JsonlRecord], None] @@ -257,38 +271,14 @@ class TurnManager: fallback_message="claude process not accepting input" ) from exc - if not self._watcher.path.exists(): - _log.info( - "send_user_message: session_id=%s waiting for JSONL %s (timeout=%.1fs)", - sid, - self._watcher.path, - self._file_wait_timeout or 0.0, - ) - try: - await self._watcher.wait_for_file(timeout=self._file_wait_timeout) - _log.info( - "send_user_message: session_id=%s JSONL appeared", sid - ) - except TimeoutError as exc: - _log.error( - "send_user_message: session_id=%s JSONL DID NOT APPEAR within %.1fs at %s", - sid, - self._file_wait_timeout or 0.0, - self._watcher.path, - ) - raise self._classify_pty_failure( - fallback_cls=SessionError, - fallback_message=( - f"JSONL file did not appear within " - f"{self._file_wait_timeout}s: {self._watcher.path}" - ), - ) from exc - else: - _log.info( - "send_user_message: session_id=%s JSONL already exists at %s (resume mode)", - sid, - self._watcher.path, - ) + _log.info( + "send_user_message: session_id=%s JSONL %s %s", + sid, + self._watcher.path, + "exists (resume/reuse)" + if self._watcher.path.exists() + else "not created yet (fresh session)", + ) terminal_assistant: AssistantMessage | None = None terminal_seen_at: float | None = None @@ -300,10 +290,81 @@ class TurnManager: n_events_yielded = 0 last_progress_log = t_loop_start + # Until the first record of this turn lands we cannot tell + # "claude is thinking" from "the Enter never registered", so + # we re-press Enter on a timer instead of hanging forever. + submit_confirmed = False + submit_attempts = 1 + submit_deadline = t_loop_start + _SUBMIT_CONFIRM_WINDOW + while True: records = await self._watcher.read_once() + if records and not submit_confirmed: + submit_confirmed = True + if submit_attempts > 1: + _log.info( + "send_user_message: session_id=%s SUBMIT confirmed " + "after %d Enter(s), %.1fs", + sid, + submit_attempts, + loop.time() - t_loop_start, + ) if not records: now = loop.time() + if ( + not submit_confirmed + and self._file_wait_timeout is not None + and now - t_loop_start > self._file_wait_timeout + and not self._watcher.path.exists() + ): + # The file never even got created. That is not a + # swallowed Enter — it is claude failing to open a + # session at all (auth / config), and no number of + # extra Enters will fix it. + _log.error( + "send_user_message: session_id=%s JSONL DID NOT APPEAR " + "within %.1fs at %s", + sid, + self._file_wait_timeout, + self._watcher.path, + ) + raise self._classify_pty_failure( + fallback_cls=SessionError, + fallback_message=( + f"JSONL file did not appear within " + f"{self._file_wait_timeout}s: {self._watcher.path}" + ), + ) + if not submit_confirmed and now >= submit_deadline: + if submit_attempts > _SUBMIT_RETRIES: + _log.error( + "send_user_message: session_id=%s prompt NEVER " + "submitted — %d Enter(s) over %.1fs left the JSONL " + "at %s untouched", + sid, + submit_attempts, + now - t_loop_start, + self._watcher.path, + ) + raise self._classify_pty_failure( + fallback_cls=SessionError, + fallback_message=( + f"prompt was never submitted: no JSONL record " + f"after {submit_attempts} Enter attempts over " + f"{now - t_loop_start:.0f}s ({self._watcher.path})" + ), + ) + _log.warning( + "send_user_message: session_id=%s no JSONL record %.1fs " + "after Enter #%d — paste likely stuck in the input box, " + "re-pressing Enter", + sid, + _SUBMIT_CONFIRM_WINDOW, + submit_attempts, + ) + await self._send_submit_key() + submit_attempts += 1 + submit_deadline = now + _SUBMIT_CONFIRM_WINDOW # Heartbeat log every 10s so a hung wait is visible. if now - last_progress_log >= 10.0: _log.info( @@ -416,6 +477,14 @@ class TurnManager: finally: self._turn_in_progress = False + async def _send_submit_key(self) -> None: + """Re-press Enter, tolerating PTY fakes that lack the method.""" + send = getattr(self._pty, "send_submit_key", None) + if send is None: + return + with contextlib.suppress(OSError): + await send() + def _pty_is_alive(self) -> bool: """Best-effort liveness check. Test fakes may lack `is_alive()`.""" is_alive = getattr(self._pty, "is_alive", None)