From cdceca42e107f0e51ab9ff50e1cc881ad76b577e Mon Sep 17 00:00:00 2001 From: Austin Pickett Date: Thu, 17 Sep 2026 16:22:05 -0400 Subject: [PATCH] test(gateway): cover watchdog fallback for empty agent holder and raising snapshot The reachable gap is agent_holder=[None] during turn setup (run_turn_runner fills the holder later); a real AIAgent always has _last_activity_ts, so seconds_since_activity=None is not a state production produces. Replace the None-returning fake with a snapshot-raises fail-safe case and add the empty-holder test. Both are red against the pre-fallback watchdog loop. --- .../test_abandoned_turn_process_cleanup.py | 84 +++++++++++++------ 1 file changed, 57 insertions(+), 27 deletions(-) diff --git a/tests/gateway/test_abandoned_turn_process_cleanup.py b/tests/gateway/test_abandoned_turn_process_cleanup.py index 3f37453090..c0a6201486 100644 --- a/tests/gateway/test_abandoned_turn_process_cleanup.py +++ b/tests/gateway/test_abandoned_turn_process_cleanup.py @@ -22,12 +22,19 @@ class _IdleAgent: self.interrupts.append(reason) -class _UnavailableActivityAgent: +class _RaisingActivityAgent: + """Agent whose activity snapshot read raises (fail-safe case). + + A real AIAgent always carries ``_last_activity_ts``; the snapshot is only + missing when the diagnostic read itself fails. The watchdog must still bound + the turn instead of skipping every poll. + """ + def __init__(self): self.interrupts = [] def get_activity_summary(self): - return {"seconds_since_activity": None} + raise RuntimeError("activity snapshot unavailable") def interrupt(self, reason): self.interrupts.append(reason) @@ -37,24 +44,12 @@ def _state(): return threading.Event(), threading.Event(), threading.Lock() -def test_thread_watchdog_times_out_when_agent_activity_is_unavailable(monkeypatch): - agent = _UnavailableActivityAgent() - worker_done, timeout_fired, cleanup_lock = _state() - calls = [] - monkeypatch.setattr( - process_registry, - "kill_started_since", - lambda task_id, baseline, *, source: calls.append( - (task_id, baseline, source) - ) - or 0, - ) - +def _run_watchdog(agent_holder, task_id, *, worker_done, timeout_fired, cleanup_lock): watchdog = threading.Thread( target=_watch_gateway_turn_inactivity, kwargs={ - "agent_holder": [agent], - "task_id": "session-without-agent", + "agent_holder": agent_holder, + "task_id": task_id, "process_baseline": frozenset(), "timeout": 0.03, "worker_done": worker_done, @@ -64,21 +59,56 @@ def test_thread_watchdog_times_out_when_agent_activity_is_unavailable(monkeypatc }, ) watchdog.start() - watchdog.join(timeout=1) + watchdog.join(timeout=2) if watchdog.is_alive(): worker_done.set() - watchdog.join(timeout=1) - + watchdog.join(timeout=2) assert not watchdog.is_alive() + return watchdog + + +def test_thread_watchdog_times_out_before_agent_exists(monkeypatch): + """``agent_holder`` is still ``[None]`` while the turn is being set up + (``run_turn_runner`` fills it later). A turn wedged in that window has no + activity snapshot, so the watchdog must fall back to elapsed wall-clock time + and still reap the turn rather than skip every poll forever.""" + worker_done, timeout_fired, cleanup_lock = _state() + calls = [] + monkeypatch.setattr( + process_registry, + "kill_started_since", + lambda task_id, baseline, *, source: calls.append((task_id, baseline, source)) or 0, + ) + + _run_watchdog( + [None], "session-before-agent", + worker_done=worker_done, timeout_fired=timeout_fired, cleanup_lock=cleanup_lock, + ) + + assert timeout_fired.is_set() + assert calls == [("session-before-agent", frozenset(), "gateway_turn_timeout")] + + +def test_thread_watchdog_times_out_when_activity_snapshot_raises(monkeypatch): + """Fail-safe: a snapshot read that raises must not disable the watchdog; the + wall-clock fallback bounds the turn.""" + agent = _RaisingActivityAgent() + worker_done, timeout_fired, cleanup_lock = _state() + calls = [] + monkeypatch.setattr( + process_registry, + "kill_started_since", + lambda task_id, baseline, *, source: calls.append((task_id, baseline, source)) or 0, + ) + + _run_watchdog( + [agent], "session-snapshot-raises", + worker_done=worker_done, timeout_fired=timeout_fired, cleanup_lock=cleanup_lock, + ) + assert timeout_fired.is_set() assert agent.interrupts == ["Execution timed out (inactivity)"] - assert calls == [ - ( - "session-without-agent", - frozenset(), - "gateway_turn_timeout", - ) - ] + assert calls == [("session-snapshot-raises", frozenset(), "gateway_turn_timeout")] def test_thread_watchdog_reaps_only_processes_created_by_timed_out_turn(monkeypatch):