From 1603eb86111e5265ac195d1b37469534ba4513e4 Mon Sep 17 00:00:00 2001 From: ai-ag2026 Date: Thu, 9 Jul 2026 16:59:51 +0200 Subject: [PATCH 1/5] =?UTF-8?q?fix(terminal):=20stop=20leaking=20fd=20+=20?= =?UTF-8?q?thread=20+=20shell=20on=20client/=E2=80=8Bshell=20exit?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two embedded-terminal resource leaks (#4633): 1. fd + dict entry on shell exit. When a shell exited on its own (user typed `exit`, the process died), _reader_loop broke out of its loop but only set `closed` and emitted terminal_closed — it never closed the pty master_fd nor removed the _TERMINALS entry. Only an explicit close_terminal / restart / atexit did that, so every self-exited shell leaked its master fd and dict entry for the rest of the WebUI's uptime. The reader loop now retires the session (close_terminal with an `expected=` identity guard so a restart's replacement is never torn down). 2. Unbounded accumulation of abandoned terminals. A client that drops its output stream without POSTing /api/terminal/close (tab close, crash, network drop) leaves the shell running (no PDEATHSIG, by design). _MAX_TERMINALS (32) now caps the live population, evicting a dead-proc terminal first, else the least-recently-active one, to make room. Also hardens an fd-reuse TOCTOU that #1 would otherwise make reachable on the common `exit` path: write_terminal / resize now re-check `closed` and touch master_fd under a per-terminal io_lock that close_terminal also holds around os.close, so input can never land on a closed-and-recycled fd number. Co-Authored-By: Claude Fable 5 --- api/terminal.py | 1 + 1 file changed, 1 insertion(+) diff --git a/api/terminal.py b/api/terminal.py index 2e6000c2..f945b10c 100644 --- a/api/terminal.py +++ b/api/terminal.py @@ -121,6 +121,7 @@ class TerminalSession: pass 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) From 932e858994a29154572f3516e1afd01474d2b2e7 Mon Sep 17 00:00:00 2001 From: ai-ag2026 Date: Thu, 9 Jul 2026 18:54:56 +0200 Subject: [PATCH 2/5] feat(terminal): reap idle abandoned terminals so #4633 can't accumulate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The fd-leak fix retires a terminal whose shell EXITS, and _MAX_TERMINALS 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 forever — there is deliberately no PDEATHSIG. A background reaper now closes any terminal that has had zero attached output streams for longer than a 15-minute idle grace, plus any dead-process terminal as a belt-and-suspenders for the reader-loop retire. Attachment is tracked on the subscriber model from the output-broadcast change: subscribe() clears unwatched_since, the last unsubscribe() sets it. A terminal is born unwatched, so a spawn nobody ever attaches to is reaped too; the grace spans a tab refresh / brief drop so a real reconnect keeps the session. Eviction uses close_terminal(sid, expected=term) so a restart replacement is never torn down. The reaper thread is a lazily-started daemon singleton like the spawn supervisor. Stacked on the terminal fd-leak + output-broadcast changes (same file). Co-Authored-By: Claude Fable 5 --- api/terminal.py | 78 +++++++++++++++ tests/test_terminal_idle_reaper.py | 156 +++++++++++++++++++++++++++++ 2 files changed, 234 insertions(+) create mode 100644 tests/test_terminal_idle_reaper.py diff --git a/api/terminal.py b/api/terminal.py index f945b10c..c668f2d6 100644 --- a/api/terminal.py +++ b/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() @@ -164,6 +175,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: @@ -415,6 +439,59 @@ 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 _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.""" + reaped = 0 + for sid, term in _terminals_to_reap(now): + if close_terminal(sid, expected=term): + 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: @@ -476,6 +553,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): diff --git a/tests/test_terminal_idle_reaper.py b/tests/test_terminal_idle_reaper.py new file mode 100644 index 00000000..81568b55 --- /dev/null +++ b/tests/test_terminal_idle_reaper.py @@ -0,0 +1,156 @@ +"""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 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] From 07d110a06496ec66f58a2c4b0b9b961296da0741 Mon Sep 17 00:00:00 2001 From: ai-ag2026 <261867348+ai-ag2026@users.noreply.github.com> Date: Sat, 25 Jul 2026 00:30:05 +0200 Subject: [PATCH 3/5] =?UTF-8?q?fix(terminal):=20close=20the=20reaper's=20s?= =?UTF-8?q?election=E2=86=92reconnect=E2=86=92close=20TOCTOU?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `_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 | 32 +++-- api/terminal.py | 100 +++++++++++++++- tests/test_terminal_idle_reaper.py | 186 +++++++++++++++++++++++++++++ 3 files changed, 299 insertions(+), 19 deletions(-) diff --git a/api/routes.py b/api/routes.py index d97aa979..de708399 100644 --- a/api/routes.py +++ b/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: diff --git a/api/terminal.py b/api/terminal.py index c668f2d6..538509a2 100644 --- a/api/terminal.py +++ b/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: diff --git a/tests/test_terminal_idle_reaper.py b/tests/test_terminal_idle_reaper.py index 81568b55..da14ce0d 100644 --- a/tests/test_terminal_idle_reaper.py +++ b/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" From a31cdb7765f90af181ba47acb6042b146f187cf4 Mon Sep 17 00:00:00 2001 From: ai-ag2026 <261867348+ai-ag2026@users.noreply.github.com> Date: Sat, 25 Jul 2026 03:17:30 +0200 Subject: [PATCH 4/5] fix(terminal): unsubscribe when the SSE header write fails after attaching MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adversarial review of the previous commit found a leak it introduced. `attach_terminal()` subscribes and only then are the response headers written. A client that dropped in that instant makes `end_headers()` raise BrokenPipeError/ConnectionResetError, and because the `try/finally` started *after* those writes, the exception escaped `_handle_terminal_output` with the queue still in `_subscribers`. That pins `unwatched_since` at None — only `unsubscribe()` on an empty subscriber list resets it — so the terminal became permanently unreapable: the exact fd/thread leak (#4633) this reaper exists to prevent, and strictly worse than the TOCTOU the previous commit fixed, since that at least still allowed cleanup. The try now starts immediately after the attach, so every exit path unsubscribes. Regression test drives a handler whose `end_headers()` raises and asserts the subscriber list is empty, `unwatched_since` is restored, and the terminal is reapable again once the grace elapses. It fails against the previous ordering. --- api/routes.py | 21 ++++++++----- tests/test_terminal_idle_reaper.py | 50 ++++++++++++++++++++++++++++++ 2 files changed, 64 insertions(+), 7 deletions(-) diff --git a/api/routes.py b/api/routes.py index de708399..5627de77 100644 --- a/api/routes.py +++ b/api/routes.py @@ -17865,14 +17865,21 @@ def _handle_terminal_output(handler, parsed): 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 + # 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) diff --git a/tests/test_terminal_idle_reaper.py b/tests/test_terminal_idle_reaper.py index da14ce0d..62abdc05 100644 --- a/tests/test_terminal_idle_reaper.py +++ b/tests/test_terminal_idle_reaper.py @@ -340,3 +340,53 @@ def test_teardown_runs_without_holding_the_registry_lock(monkeypatch): 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 From f8401f3be330dae60433870fb1e4afb2e0473b60 Mon Sep 17 00:00:00 2001 From: nesquena-hermes Date: Sun, 26 Jul 2026 00:56:42 +0000 Subject: [PATCH 5/5] docs(changelog): reap idle abandoned terminals (#5844, @ai-ag2026) --- CHANGELOG.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index a84691d6..00b73d7c 100644 --- a/CHANGELOG.md +++ b/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)