fix(pty,turn,backend): verify submit via jsonl, killpg orphans, idle session ttl

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",
 ]

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:

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
+# Submit verification does NOT live here — it lives in ``TurnManager``,
-# (claude's TUI renders ``[Pasted text #...`` while the paste is still
+# which can see the session JSONL. The PTY layer has no trustworthy
-# buffered in the input box). If it's still there after this window,
+# signal: claude's TUI renders ``[Pasted text #N +M lines]`` only for
-# Enter never registered — we retry up to ``_SUBMIT_RETRIES`` times.
+# long/multi-line pastes, so a short prompt produces no indicator at all
-_SUBMIT_VERIFY_WINDOW = 1.5
+# and any "indicator is gone ⇒ submitted" check passes vacuously. The
-_SUBMIT_VERIFY_POLL = 0.1
+# JSONL user record is ground truth, so ``send_submit_key()`` is exposed
-_SUBMIT_RETRIES = 3
+# for the turn layer to re-press Enter when that record doesn't appear.
-_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
 
 
 @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
+        n2 = await asyncio.to_thread(pty.write, b"\r")
-        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)",
+            "write: session_id=%s paste=%d bytes + Enter=%d bytes written "
-                        self._session_id,
+            "(submit unconfirmed until JSONL moves)",
-                        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,
-            )
-        _log.info(
-            "write: session_id=%s SUBMIT done (paste=%d Enter=%d)",
             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):
+            self._signal_group(signal.SIGTERM)
-                pty.kill(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):
+                self._signal_group(signal.SIGKILL)
-                    pty.kill(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):
+            self._signal_group(signal.SIGKILL)
-                pty.kill(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",
 )

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,37 +271,13 @@ 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)",
+                "send_user_message: session_id=%s JSONL %s %s",
-                    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,
+                "exists (resume/reuse)"
+                if self._watcher.path.exists()
+                else "not created yet (fresh session)",
             )
 
             terminal_assistant: AssistantMessage | 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)