diff --git a/api/routes.py b/api/routes.py index d97aa979..5627de77 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,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) diff --git a/api/terminal.py b/api/terminal.py index 2e6000c2..538509a2 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() @@ -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: diff --git a/tests/test_terminal_idle_reaper.py b/tests/test_terminal_idle_reaper.py new file mode 100644 index 00000000..62abdc05 --- /dev/null +++ b/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