Merge remote-tracking branch 'origin/master' into franksong2702/live-worklog-settle-no-blank

CHANGELOG.md

@@ -25,6 +25,8 @@
 
 ### Fixed
 
+- **Abandoned embedded terminals are now reaped instead of leaking a running shell forever (#4633 backend half).** A WebUI terminal whose viewer dropped its output stream without an explicit close (tab close, crash, network drop) previously kept its shell process and pty fd alive indefinitely — there is no `PDEATHSIG`, so nothing tore it down until the terminal cap happened to evict it. A background reaper now closes any terminal that has been unwatched (zero attached output streams) for longer than a 15-minute grace, and sweeps dead-process terminals immediately; the grace deliberately spans a tab refresh or brief network drop so a genuine reconnect keeps the session. The output-stream attach is now atomic against the reaper (a single registry-locked lookup+subscribe), so a viewer can never end up subscribed to a terminal the reaper is tearing down, and a race resolves to exactly one winner — attach keeps the terminal, or the reaper wins and the stream cleanly reports "terminal not running." A related leak is closed too: if writing the SSE response headers fails after attaching (client already gone), the subscription is now always unsubscribed instead of pinning the terminal un-reapable. Thanks @ai-ag2026. (#5844, #4633)
+
 - **Typing in the composer no longer causes a layout round-trip on every keystroke.** A single-row composer append now skips rewriting the textarea height when the height can't change, removing a per-keystroke reflow that could disturb caret stability and assistive-tech focus. An oversized composer, a text replacement, and session/draft restore still fully remeasure so the box shrinks back correctly, and the minimum-height check parses only genuine pixel values (a percentage/`auto`/`calc()` min-height fails closed to the full resize). Thanks @sheldon-im. (#6349)
 
 - **Gateway approval cards are now routed by run ID, so approvals and Stop always act on the right run (#6008).** A gateway approval card that arrived without an `approval_id` couldn't be acted on, and with multiple concurrent runs an approval or a Stop could be applied to the wrong run. Approvals are now owned per-`(session, run)`: identity-less and identity-bearing approvals route through the same single-flight owner, concurrent duplicate submits can't double-relay, different runs don't block each other, and a failed relay releases ownership so a retry can proceed. The composer/sidebar Stop now waits boundedly for an acknowledged exact-run gateway stop and surfaces a persistent, actionable "Cancel failed." error toast (with Copy and Dismiss, localized) instead of silently settling the UI when the stop fails. When multiple approvals are queued the card shows a "1 of N pending" counter. Thanks @rodboev. (#6040, #6008)

api/routes.py

@@ -17842,18 +17842,7 @@ def _handle_terminal_output(handler, parsed):
     sid = qs.get("session_id", [""])[0]
     if not sid:
         return bad(handler, "session_id required")
-    from api.terminal import get_terminal
-    term = get_terminal(sid)
-    if term is None:
-        return j(handler, {"error": "terminal not running"}, status=404)
-
-    handler.send_response(200)
-    handler.send_header("Content-Type", "text/event-stream; charset=utf-8")
-    handler.send_header("Cache-Control", "no-cache")
-    handler.send_header("X-Accel-Buffering", "no")
-    handler.send_header("Connection", "close")
-    end_sse_headers(handler)
-    _sse_set_write_deadline(handler)  # Defect A: slow tab can't pin this thread
+    from api.terminal import attach_terminal
     # EventSource automatically returns the last received SSE id on transport
     # reconnect. Seed only newer backlog entries in that case so already-rendered
     # terminal bytes (including ANSI cursor controls) are not written twice. A
@@ -17865,8 +17854,32 @@ def _handle_terminal_output(handler, parsed):
             after_seq = max(0, int(last_event_id))
         except ValueError:
             pass
-    output = term.subscribe(after_seq=after_seq)
+    # Look up and subscribe in one atomic step. A separate `get_terminal()` then
+    # `term.subscribe()` leaves a window in which the idle reaper can claim and
+    # tear the terminal down, leaving this stream attached to a corpse after we
+    # already committed a 200. Attaching atomically means we either hold a live
+    # viewer (which makes the terminal un-reapable) or learn it is gone in time
+    # to answer 404.
+    attached = attach_terminal(sid, after_seq=after_seq)
+    if attached is None:
+        return j(handler, {"error": "terminal not running"}, status=404)
+    term, output = attached
+
+    # The subscription is live from here on, so EVERY exit path — including a
+    # failure while writing the response headers — must unsubscribe. Writing
+    # headers to a client that already dropped raises BrokenPipeError, and if
+    # that escaped before the try block the queue would stay in
+    # `_subscribers` forever, pinning `unwatched_since` at None and making the
+    # terminal permanently unreapable: the exact fd/thread leak this reaper
+    # exists to prevent. Hence the try starts immediately after the attach.
     try:
+        handler.send_response(200)
+        handler.send_header("Content-Type", "text/event-stream; charset=utf-8")
+        handler.send_header("Cache-Control", "no-cache")
+        handler.send_header("X-Accel-Buffering", "no")
+        handler.send_header("Connection", "close")
+        end_sse_headers(handler)
+        _sse_set_write_deadline(handler)  # Defect A: slow tab can't pin this thread
         while True:
             try:
                 event_seq, event, data = output.get(timeout=_SSE_HEARTBEAT_INTERVAL_SECONDS)

api/terminal.py

@@ -93,6 +93,14 @@ class TerminalSession:
     # at its prompt has neither, so it sorts oldest and is evicted before an
     # actively used one.
     last_activity: float = field(default_factory=time.time)
+    # Wall-clock of when the terminal last had zero attached viewers, or None
+    # while at least one is attached. The reaper closes a terminal that has been
+    # unwatched for longer than the idle grace: a client that drops its output
+    # stream without POSTing /api/terminal/close (tab close, crash, network drop)
+    # otherwise leaves the shell running forever (no PDEATHSIG). A terminal is
+    # born unwatched, so a spawn nobody ever attaches to is reaped too. The grace
+    # spans transient reconnects (a tab refresh re-attaches and clears it).
+    unwatched_since: float | None = field(default_factory=time.time)
 
     def is_alive(self) -> bool:
         return not self.closed.is_set() and self.proc.poll() is None
@@ -111,6 +119,7 @@ class TerminalSession:
                 if after_seq is None or item[0] > after_seq:
                     q.put_nowait(item)
             self._subscribers.append(q)
+            self.unwatched_since = None  # a viewer is attached
         return q
 
     def unsubscribe(self, q: queue.Queue) -> None:
@@ -119,6 +128,8 @@ class TerminalSession:
                 self._subscribers.remove(q)
             except ValueError:
                 pass
+            if not self._subscribers:
+                self.unwatched_since = time.time()
 
     def put_output(self, event: str, payload: dict) -> None:
         self.last_activity = time.time()
@@ -163,6 +174,19 @@ _spawn_supervisor_thread: threading.Thread | None = None
 _terminal_descendant_reaper_lock = threading.Lock()
 _TERMINAL_DESCENDANT_REAPER_LIMIT = 64
 
+# Idle-terminal reaper: proactively close terminals whose viewers have all gone
+# away, instead of leaving an abandoned shell running until the cap evicts it.
+# A terminal unwatched (zero attached output streams) for longer than the grace
+# is closed; the grace spans a tab refresh / brief network drop so a real
+# reconnect keeps the session. Dead-process terminals are swept too as a
+# belt-and-suspenders for the reader-loop retire.
+_TERMINAL_IDLE_GRACE_SECONDS = 900  # 15 min unwatched -> reap
+_TERMINAL_REAPER_INTERVAL_SECONDS = 60
+_terminal_reaper_started = False
+_terminal_reaper_lock = threading.Lock()
+_terminal_reaper_thread: threading.Thread | None = None
+_terminal_reaper_stop = threading.Event()
+
 
 @dataclass
 class _SpawnRequest:
@@ -414,6 +438,101 @@ def _enforce_terminal_cap(*, exclude_sid: str | None = None) -> None:
         close_terminal(victim_sid, expected=victim_term)
 
 
+def _terminals_to_reap(now: float) -> list[tuple[str, TerminalSession]]:
+    """Return (sid, term) pairs the reaper should close: a dead process, or a
+    terminal unwatched for longer than the idle grace. Pure/snapshotted under
+    the lock so it can be unit-tested without threads."""
+    victims = []
+    with _LOCK:
+        for sid, term in _TERMINALS.items():
+            if not term.is_alive():
+                victims.append((sid, term))
+                continue
+            unwatched = term.unwatched_since
+            if unwatched is not None and (now - unwatched) >= _TERMINAL_IDLE_GRACE_SECONDS:
+                victims.append((sid, term))
+    return victims
+
+
+def _claim_reap_victim(sid: str, term: TerminalSession, now: float) -> TerminalSession | None:
+    """Atomically remove *term* from the registry, or refuse.
+
+    ``_terminals_to_reap`` snapshots victims and then releases ``_LOCK``, so by
+    the time we get here a viewer may have reconnected: ``subscribe()`` appends
+    to ``_subscribers`` and clears ``unwatched_since`` under the terminal's
+    ``_sub_lock``, which the selection pass never held. Object identity alone —
+    what ``close_terminal(expected=…)`` checks — is still true in that case, so
+    the reaper would kill a terminal that now has a live viewer.
+
+    The claim therefore re-establishes the *whole* selection predicate while
+    holding both locks, and takes the entry out of ``_TERMINALS`` in the same
+    critical section. A concurrent ``attach_terminal()`` acquires the same two
+    locks in the same order, so exactly one of the two wins: either the viewer
+    is attached (and we refuse) or the entry is already gone (and the attach
+    reports "not running").
+
+    Returns the claimed terminal — the caller owns its teardown — or ``None``.
+    """
+    with _LOCK:
+        if _TERMINALS.get(sid) is not term:
+            return None
+        # A dead process is reaped unconditionally: it cannot come back to life,
+        # and an attached viewer only means someone is watching a corpse.
+        if term.is_alive():
+            with term._sub_lock:
+                if term._subscribers:
+                    return None
+                unwatched = term.unwatched_since
+                if unwatched is None:
+                    return None
+                if (now - unwatched) < _TERMINAL_IDLE_GRACE_SECONDS:
+                    return None
+        del _TERMINALS[sid]
+    return term
+
+
+def _reap_idle_terminals(now: float) -> int:
+    """Close every terminal selected by ``_terminals_to_reap`` that is *still*
+    idle when claimed. Returns the count closed."""
+    reaped = 0
+    for sid, term in _terminals_to_reap(now):
+        claimed = _claim_reap_victim(sid, term, now)
+        if claimed is None:
+            continue
+        # Process/fd teardown runs after both locks are released: killpg + wait
+        # can take seconds and must not block attaches or spawns.
+        _teardown_terminal(claimed)
+        reaped += 1
+    return reaped
+
+
+def _terminal_reaper_loop() -> None:
+    while not _terminal_reaper_stop.wait(_TERMINAL_REAPER_INTERVAL_SECONDS):
+        try:
+            # Wall-clock, consistent with unwatched_since / last_activity.
+            _reap_idle_terminals(time.time())
+        except Exception:
+            # Never let a transient error kill the reaper thread.
+            pass
+
+
+def _ensure_terminal_reaper() -> None:
+    global _terminal_reaper_started, _terminal_reaper_thread
+    if not _TERMINAL_SUPPORTED:
+        return
+    with _terminal_reaper_lock:
+        if (
+            _terminal_reaper_started
+            and _terminal_reaper_thread is not None
+            and getattr(_terminal_reaper_thread, "is_alive", lambda: False)()
+        ):
+            return
+        thread = threading.Thread(target=_terminal_reaper_loop, daemon=True)
+        thread.start()
+        _terminal_reaper_thread = thread
+        _terminal_reaper_started = True
+
+
 def start_terminal(session_id: str, workspace: Path, rows: int = 24, cols: int = 80, restart: bool = False) -> TerminalSession:
     """Start or return the embedded terminal for a WebUI session."""
     if not _TERMINAL_SUPPORTED:
@@ -475,6 +594,7 @@ def start_terminal(session_id: str, workspace: Path, rows: int = 24, cols: int =
             }
         )
         _ensure_spawn_supervisor()
+        _ensure_terminal_reaper()
         _spawn_queue.put(request)
         try:
             if not request.done.wait(timeout=5.0):
@@ -522,6 +642,42 @@ def get_terminal(session_id: str) -> TerminalSession | None:
         return term
 
 
+def attach_terminal(
+    session_id: str, after_seq: int | None = None
+) -> tuple[TerminalSession, queue.Queue] | None:
+    """Attach a viewer atomically against the idle reaper.
+
+    ``get_terminal()`` followed by ``term.subscribe()`` leaves a window: the
+    reaper can claim and tear the terminal down in between, so the viewer ends
+    up subscribed to a corpse and the caller reports a live stream that will
+    never produce output. Doing the lookup and the subscribe inside the same
+    ``_LOCK`` section — the same lock, in the same order, that
+    ``_claim_reap_victim`` takes — makes the two mutually exclusive:
+
+    * attach wins → ``_subscribers`` is non-empty and ``unwatched_since`` is
+      ``None`` before the reaper can revalidate, so the reap is refused;
+    * reap wins → the entry is already out of ``_TERMINALS``, so this returns
+      ``None`` and the route answers "terminal not running" instead of hanging.
+
+    Registration is the authority, deliberately: both the reaper and
+    ``close_terminal()`` remove the entry *before* tearing the terminal down, so
+    "still in ``_TERMINALS``" is exactly the condition that cannot race. A
+    terminal that is registered but already flagged ``closed`` (its shell exited
+    and the reader loop has not retired it yet) still attaches, so the viewer
+    receives the ``terminal_closed`` event instead of a bare 404.
+
+    Returns ``(term, queue)`` or ``None``.
+    """
+    if not _TERMINAL_SUPPORTED:
+        return None
+    sid = str(session_id or "")
+    with _LOCK:
+        term = _TERMINALS.get(sid)
+        if term is None:
+            return None
+        return term, term.subscribe(after_seq=after_seq)
+
+
 def write_terminal(session_id: str, data: str) -> None:
     if not _TERMINAL_SUPPORTED:
         raise NotImplementedError("Embedded terminal is not supported on Windows")
@@ -565,6 +721,18 @@ def close_terminal(session_id: str, *, expected: TerminalSession | None = None)
         term = _TERMINALS.pop(sid, None)
     if not term:
         return False
+    _teardown_terminal(term)
+    return True
+
+
+def _teardown_terminal(term: TerminalSession) -> None:
+    """Kill the shell, close the pty master fd and reap descendants.
+
+    Split out of ``close_terminal`` so a caller that has already claimed the
+    registry entry (the idle reaper) can run the teardown without re-entering
+    the registry lookup — and, importantly, without holding any lock while
+    ``killpg``/``wait`` block for up to ~2.5s.
+    """
     term.closed.set()
     try:
         if term.proc.poll() is None:
@@ -592,7 +760,6 @@ def close_terminal(session_id: str, *, expected: TerminalSession | None = None)
             except OSError:
                 pass
         _reap_terminal_descendants(term.proc.pid)
-    return True
 
 
 def close_all_terminals() -> None:

tests/test_terminal_idle_reaper.py

@@ -0,0 +1,392 @@
+"""Regression tests — the idle-terminal reaper closes abandoned shells (#4633).
+
+The fd-leak fix retires a terminal whose shell EXITS, and the cap bounds the
+worst case, but an interactive shell abandoned by its client (tab closed without
+POSTing /api/terminal/close, browser crash, network drop) keeps running because
+there is deliberately no PDEATHSIG. The reaper closes a terminal that has had
+zero attached output streams for longer than the idle grace, plus any dead-proc
+terminal as a belt-and-suspenders. A tab refresh / brief drop re-attaches within
+the grace and keeps the session.
+"""
+import os
+import threading
+import time
+
+import pytest
+
+if os.name != "posix":
+    pytest.skip("terminal tests require POSIX terminal support", allow_module_level=True)
+
+import api.terminal as terminal
+
+
+class _FakeProc:
+    def __init__(self, pid=515151, alive=True):
+        self.pid = pid
+        self._alive = alive
+
+    def poll(self):
+        return None if self._alive else 0
+
+    def wait(self, timeout=None):
+        return 0
+
+
+def _make_registered_term(monkeypatch, sid, *, alive=True, unwatched_since=None):
+    r, w = os.pipe()
+    os.close(w)
+    term = terminal.TerminalSession(
+        session_id=sid, workspace="/tmp", proc=_FakeProc(alive=alive), master_fd=r
+    )
+    term.unwatched_since = unwatched_since
+    with terminal._LOCK:
+        terminal._TERMINALS[sid] = term
+    return term
+
+
+@pytest.fixture(autouse=True)
+def _clean(monkeypatch):
+    monkeypatch.setattr(terminal.os, "killpg", lambda *a, **k: None)
+    yield
+    with terminal._LOCK:
+        sids = list(terminal._TERMINALS)
+    for sid in sids:
+        try:
+            terminal.close_terminal(sid)
+        except Exception:
+            pass
+
+
+# ── unwatched_since tracking via subscribe/unsubscribe ───────────────────────
+
+def test_unwatched_since_tracks_viewer_attachment():
+    term = _make_registered_term_local()
+    # Born unwatched.
+    assert term.unwatched_since is not None
+
+    a = term.subscribe()
+    assert term.unwatched_since is None  # a viewer attached
+
+    b = term.subscribe()
+    term.unsubscribe(a)
+    assert term.unwatched_since is None  # b still attached
+
+    term.unsubscribe(b)
+    assert term.unwatched_since is not None  # last viewer left
+
+
+def _make_registered_term_local():
+    class _P:
+        pid = 1
+
+        def poll(self):
+            return None
+
+    return terminal.TerminalSession(session_id="x", workspace="/tmp", proc=_P(), master_fd=-1)
+
+
+# ── _terminals_to_reap selection ─────────────────────────────────────────────
+
+def test_reaps_unwatched_beyond_grace(monkeypatch):
+    now = 10_000.0
+    grace = terminal._TERMINAL_IDLE_GRACE_SECONDS
+    _make_registered_term(monkeypatch, "old", alive=True, unwatched_since=now - grace - 1)
+    victims = {sid for sid, _ in terminal._terminals_to_reap(now)}
+    assert "old" in victims
+
+
+def test_keeps_watched_terminal(monkeypatch):
+    now = 10_000.0
+    # unwatched_since None => a viewer is attached => never reaped, however old.
+    _make_registered_term(monkeypatch, "watched", alive=True, unwatched_since=None)
+    victims = {sid for sid, _ in terminal._terminals_to_reap(now)}
+    assert "watched" not in victims
+
+
+def test_keeps_recently_unwatched_within_grace(monkeypatch):
+    now = 10_000.0
+    # Detached 5s ago — within grace, so a reconnecting tab is not killed.
+    _make_registered_term(monkeypatch, "reconnecting", alive=True, unwatched_since=now - 5)
+    victims = {sid for sid, _ in terminal._terminals_to_reap(now)}
+    assert "reconnecting" not in victims
+
+
+def test_reaps_dead_process_regardless(monkeypatch):
+    now = 10_000.0
+    # Dead proc but "watched" — still swept (belt-and-suspenders).
+    _make_registered_term(monkeypatch, "dead", alive=False, unwatched_since=None)
+    victims = {sid for sid, _ in terminal._terminals_to_reap(now)}
+    assert "dead" in victims
+
+
+# ── _reap_idle_terminals effect ──────────────────────────────────────────────
+
+def test_reap_closes_and_removes(monkeypatch):
+    now = 10_000.0
+    grace = terminal._TERMINAL_IDLE_GRACE_SECONDS
+    term = _make_registered_term(monkeypatch, "reapme", alive=True, unwatched_since=now - grace - 1)
+    fd = term.master_fd
+
+    n = terminal._reap_idle_terminals(now)
+
+    assert n == 1
+    with terminal._LOCK:
+        assert "reapme" not in terminal._TERMINALS
+    with pytest.raises(OSError):
+        os.fstat(fd)  # fd closed
+
+
+def test_reaper_thread_starts_idempotently(monkeypatch):
+    # Reset reaper state so the ensure actually starts a (dummy) thread.
+    monkeypatch.setattr(terminal, "_terminal_reaper_started", False)
+    monkeypatch.setattr(terminal, "_terminal_reaper_thread", None)
+    started = []
+
+    class _DummyThread:
+        def __init__(self, *a, **k):
+            self._k = k
+
+        def start(self):
+            started.append(1)
+
+        def is_alive(self):
+            return True
+
+    monkeypatch.setattr(terminal.threading, "Thread", _DummyThread)
+    terminal._ensure_terminal_reaper()
+    terminal._ensure_terminal_reaper()  # second call is a no-op (already alive)
+    assert started == [1]
+
+
+# ── Selection → reconnect → close TOCTOU ────────────────────────────────────
+# `_terminals_to_reap()` snapshots victims under `_LOCK` and then RELEASES it.
+# `subscribe()` clears `unwatched_since` under the terminal's own `_sub_lock`,
+# which the selection pass never held, so a viewer can attach in that gap. The
+# old close path re-checked object identity only — still true — and killed a
+# terminal that now had a live viewer. The claim must re-establish the whole
+# selection predicate under both locks.
+
+
+def _expired_since():
+    """A wall-clock stamp old enough to select a terminal for reaping."""
+    return time.time() - terminal._TERMINAL_IDLE_GRACE_SECONDS - 5
+
+
+def test_reconnect_between_selection_and_close_keeps_the_terminal(monkeypatch):
+    """Attach a viewer after selection but before the claim: no reap.
+
+    Drives the claim with the *stale snapshot* the selection pass produced.
+    Re-running the full `_reap_idle_terminals()` here would not pin anything:
+    its own second selection pass no longer sees the terminal as idle, so the
+    test would pass even with an identity-only claim — the exact defect.
+    """
+    sid = "toctou-reconnect"
+    now = time.time()
+    term = _make_registered_term(monkeypatch, sid, unwatched_since=_expired_since())
+
+    victims = terminal._terminals_to_reap(now)
+    assert [s for s, _ in victims] == [sid], "the terminal was not even selected"
+
+    # The gap: a reconnecting EventSource attaches through the production path.
+    attached = terminal.attach_terminal(sid)
+    assert attached is not None
+    _term, output = attached
+    assert term.unwatched_since is None
+
+    # The claim runs against the snapshot taken BEFORE the reconnect.
+    for victim_sid, victim_term in victims:
+        assert terminal._claim_reap_victim(victim_sid, victim_term, now) is None
+
+    with terminal._LOCK:
+        assert terminal._TERMINALS.get(sid) is term, "a watched terminal was reaped"
+    assert not term.closed.is_set()
+
+    # And it still receives output on that viewer's queue.
+    term.put_output("terminal_output", {"data": "hello"})
+    assert output.get(timeout=1)[1] == "terminal_output"
+
+
+def test_reconnect_racing_the_claim_under_a_barrier_keeps_the_terminal(monkeypatch):
+    """Same race, driven deterministically across two threads.
+
+    The reaper is paused *after* it has selected its victim and is about to
+    claim it; the viewer attaches through the production path in that window;
+    then the reaper resumes.
+    """
+    sid = "toctou-barrier"
+    term = _make_registered_term(monkeypatch, sid, unwatched_since=_expired_since())
+
+    selected = threading.Event()
+    attached_done = threading.Event()
+    real_select = terminal._terminals_to_reap
+
+    def paused_select(now):
+        victims = real_select(now)
+        selected.set()
+        # Hold the reaper here until the viewer is attached.
+        assert attached_done.wait(timeout=5)
+        return victims
+
+    monkeypatch.setattr(terminal, "_terminals_to_reap", paused_select)
+
+    result = {}
+
+    def reap():
+        result["reaped"] = terminal._reap_idle_terminals(time.time())
+
+    reaper = threading.Thread(target=reap)
+    reaper.start()
+    try:
+        assert selected.wait(timeout=5)
+        attached = terminal.attach_terminal(sid)
+        assert attached is not None, "attach lost to a reaper that had not claimed yet"
+        _term, output = attached
+    finally:
+        attached_done.set()
+        reaper.join(timeout=10)
+    assert not reaper.is_alive()
+
+    assert result["reaped"] == 0
+    with terminal._LOCK:
+        assert terminal._TERMINALS.get(sid) is term
+    assert not term.closed.is_set()
+    term.put_output("terminal_output", {"data": "still alive"})
+    assert output.get(timeout=1)[1] == "terminal_output"
+
+
+def test_reap_that_claims_first_makes_attach_report_not_running(monkeypatch):
+    """The inverse: once the reaper has claimed, an attach must fail cleanly.
+
+    It must not hand back a queue for a terminal that is being torn down —
+    the caller would commit a 200 and stream from a corpse forever.
+    """
+    sid = "toctou-reap-wins"
+    term = _make_registered_term(monkeypatch, sid, unwatched_since=_expired_since())
+
+    assert terminal._reap_idle_terminals(time.time()) == 1
+    assert term.closed.is_set()
+    with terminal._LOCK:
+        assert sid not in terminal._TERMINALS
+
+    assert terminal.attach_terminal(sid) is None
+
+
+def test_claim_refuses_a_terminal_whose_grace_was_refreshed(monkeypatch):
+    """`unwatched_since` moved forward after selection → not idle any more."""
+    sid = "toctou-refreshed"
+    term = _make_registered_term(monkeypatch, sid, unwatched_since=_expired_since())
+    now = time.time()
+    assert [s for s, _ in terminal._terminals_to_reap(now)] == [sid]
+
+    # A viewer attached and detached again inside the gap: no subscribers, but
+    # the grace restarted.
+    term.unwatched_since = time.time()
+
+    assert terminal._claim_reap_victim(sid, term, now) is None
+    with terminal._LOCK:
+        assert terminal._TERMINALS.get(sid) is term
+
+
+def test_claim_refuses_a_replaced_registry_entry(monkeypatch):
+    """A restart replaced the sid: the old snapshot must not kill the new one."""
+    sid = "toctou-replaced"
+    old = _make_registered_term(monkeypatch, sid, unwatched_since=_expired_since())
+    now = time.time()
+    assert [s for s, _ in terminal._terminals_to_reap(now)] == [sid]
+
+    replacement = _make_registered_term(monkeypatch, sid, unwatched_since=None)
+    assert terminal._claim_reap_victim(sid, old, now) is None
+    with terminal._LOCK:
+        assert terminal._TERMINALS.get(sid) is replacement
+
+
+def test_dead_terminal_is_reaped_even_with_an_attached_viewer(monkeypatch):
+    """A viewer watching a corpse must not keep the entry alive forever."""
+    sid = "toctou-dead"
+    term = _make_registered_term(monkeypatch, sid, alive=False, unwatched_since=None)
+    term.subscribe()
+
+    assert terminal._reap_idle_terminals(time.time()) == 1
+    with terminal._LOCK:
+        assert sid not in terminal._TERMINALS
+
+
+def test_teardown_runs_without_holding_the_registry_lock(monkeypatch):
+    """Process teardown can block for seconds; it must not pin `_LOCK`.
+
+    Otherwise a single hung shell stalls every attach and spawn for the whole
+    kill/wait budget.
+    """
+    sid = "toctou-teardown-lock"
+    _make_registered_term(monkeypatch, sid, unwatched_since=_expired_since())
+
+    observed = {}
+
+    def slow_killpg(*_a, **_k):
+        # `_LOCK` is an RLock, so probing it from the reaping thread itself
+        # would succeed even while held. Probe from a separate thread, and
+        # acquire/release there — an RLock may only be released by its owner.
+        def probe():
+            acquired = terminal._LOCK.acquire(timeout=0.5)
+            observed["free"] = acquired
+            if acquired:
+                terminal._LOCK.release()
+
+        holder = threading.Thread(target=probe)
+        holder.start()
+        holder.join(timeout=5)
+
+    monkeypatch.setattr(terminal.os, "killpg", slow_killpg)
+    terminal._reap_idle_terminals(time.time())
+
+    assert observed.get("free") is True, "_LOCK was held across process teardown"
+
+
+def test_a_header_write_failure_after_attach_still_unsubscribes(monkeypatch):
+    """Attaching before the headers must not be able to leak the subscriber.
+
+    `attach_terminal()` subscribes, and only then are the response headers
+    written. A client that dropped in that instant makes `end_headers()` raise
+    BrokenPipeError; if that escaped before the try/finally, the queue would
+    stay in `_subscribers` forever, pin `unwatched_since` at None, and make the
+    terminal permanently unreapable — the very leak this reaper exists to
+    prevent, self-inflicted.
+    """
+    import io
+    from types import SimpleNamespace
+
+    import api.routes as routes
+
+    sid = "sse-header-failure"
+    term = _make_registered_term(monkeypatch, sid, unwatched_since=_expired_since())
+
+    class _Handler:
+        headers = {}
+
+        def __init__(self):
+            self.wfile = io.BytesIO()
+            self.status = None
+
+        def send_response(self, status):
+            self.status = status
+
+        def send_header(self, _name, _value):
+            pass
+
+        def end_headers(self):
+            raise ConnectionResetError("client vanished")
+
+    monkeypatch.setattr(routes, "_embedded_terminal_gate_allows", lambda _handler: True)
+    monkeypatch.setattr(routes, "_sse_set_write_deadline", lambda _handler: None)
+
+    routes._handle_terminal_output(
+        _Handler(), SimpleNamespace(query=f"session_id={sid}")
+    )
+
+    assert term._subscribers == [], "the subscriber leaked on a header-write failure"
+    assert term.unwatched_since is not None, "the terminal was pinned as watched forever"
+    # The grace restarts from the detach, which is correct — but the terminal
+    # is reapable again once it elapses, instead of being pinned forever.
+    assert terminal._reap_idle_terminals(
+        term.unwatched_since + terminal._TERMINAL_IDLE_GRACE_SECONDS + 1
+    ) == 1