fix(terminal): close the reaper's selection→reconnect→close TOCTOU

`_terminals_to_reap()` snapshots victims under `_LOCK` and then releases it,
but `subscribe()` clears `unwatched_since` and appends to `_subscribers` under
the terminal's own `_sub_lock`, which the selection pass never held. A viewer
reconnecting in that gap left object identity — all `close_terminal(expected=…)`
ever checked — still true, so the reaper killed a terminal that now had a live
viewer.

- `_claim_reap_victim()` re-establishes the *whole* selection predicate while
  holding `_LOCK` and the terminal's `_sub_lock`, and removes the registry
  entry in that same critical section: same object, no subscribers, and
  `unwatched_since` still set and still expired. A dead process is still
  reaped unconditionally — it cannot come back, and a viewer watching a corpse
  must not pin the entry forever.
- `attach_terminal()` does the registry lookup and the subscribe atomically
  under `_LOCK`, replacing `get_terminal()` + `term.subscribe()` in the SSE
  route. It takes the same two locks in the same order as the claim, so
  exactly one side wins: the viewer attaches (and the reap is refused), or the
  entry is already gone (and the route answers "terminal not running" before
  committing a 200, instead of streaming from a corpse). Registration is the
  authority — a registered-but-`closed` terminal still attaches so the viewer
  receives `terminal_closed`.
- Teardown moved to `_teardown_terminal()` and runs after both locks are
  released; `killpg`+`wait` can block ~2.5s and must not pin the registry.
- Dropped the duplicated `self.last_activity = time.time()` the rebase left in
  `put_output()`.

Tests: a barrier regression that pauses the reaper after selection, attaches a
viewer through the production path, resumes, and asserts the terminal stays
registered/live and still receives output; the inverse where the reap claims
first and the attach reports not-running; refreshed-grace and replaced-entry
claims; dead-with-viewer; and a probe that `_LOCK` is free during teardown.
Three of them fail against an identity-only claim.

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,7 +17854,24 @@ 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
+
+    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
     try:
         while True:
             try:

api/terminal.py

@@ -132,7 +132,6 @@ class TerminalSession:
                 self.unwatched_since = time.time()
 
     def put_output(self, event: str, payload: dict) -> None:
-        self.last_activity = time.time()
         self.last_activity = time.time()
         with self._sub_lock:
             item = (self._next_output_seq, event, payload)
@@ -455,13 +454,55 @@ def _terminals_to_reap(now: float) -> list[tuple[str, TerminalSession]]:
     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``. Returns the count
-    closed. ``expected=term`` guards against closing a restart replacement."""
+    """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):
-        if close_terminal(sid, expected=term):
-            reaped += 1
+        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
 
 
@@ -601,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")
@@ -644,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:
@@ -671,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

@@ -9,6 +9,8 @@ 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
 
@@ -154,3 +156,187 @@ def test_reaper_thread_starts_idempotently(monkeypatch):
     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"