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]