Files
hermes-webui/tests/test_terminal_idle_reaper.py
ai-ag2026 a31cdb7765 fix(terminal): unsubscribe when the SSE header write fails after attaching
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.
2026-07-25 03:17:30 +02:00

393 lines
14 KiB
Python

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