diff --git a/gateway/run.py b/gateway/run.py index 5ddaba02f309b..ffbe288706d24 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -2690,6 +2690,116 @@ _INTERRUPT_REASON_SSE_DISCONNECT = "SSE client disconnected" _INTERRUPT_REASON_GATEWAY_SHUTDOWN = "Gateway shutting down" _INTERRUPT_REASON_GATEWAY_RESTART = "Gateway restarting" + +def _reap_gateway_turn_processes( + task_id: str, + process_baseline, + *, + source: str, +) -> int: + """Reap only background processes created by one abandoned turn.""" + from tools.process_registry import process_registry + + try: + killed = process_registry.kill_started_since( + task_id, + process_baseline, + source=source, + ) + except Exception: + # Runs on a detached daemon thread (interrupt and timeout call + # sites both fire-and-forget it) — an uncaught exception here + # would only surface via threading.excepthook, bypassing the + # app's logger. Swallow and log through the normal channel instead. + logger.warning( + "Failed to reap background processes for turn %s (%s)", + task_id, + source, + exc_info=True, + ) + return 0 + if killed: + logger.warning( + "Reaped %d background process(es) created by abandoned turn %s (%s)", + killed, + task_id, + source, + ) + return killed + + +def _abandon_timed_out_gateway_turn( + *, + agent_holder, + task_id: str, + process_baseline, + worker_done: threading.Event, + timeout_fired: threading.Event, + cleanup_lock: threading.Lock, +) -> bool: + """Interrupt one timed-out turn and reap only processes it created.""" + with cleanup_lock: + if worker_done.is_set() or timeout_fired.is_set(): + return False + timeout_fired.set() + + agent = agent_holder[0] if agent_holder else None + if agent is not None and hasattr(agent, "interrupt"): + try: + agent.interrupt(_INTERRUPT_REASON_TIMEOUT) + except Exception: + logger.debug("Timed-out agent interrupt failed", exc_info=True) + + try: + _reap_gateway_turn_processes( + task_id, + process_baseline, + source="gateway_turn_timeout", + ) + except Exception: + logger.warning( + "Failed to reap background processes for timed-out turn %s", + task_id, + exc_info=True, + ) + return True + + +def _watch_gateway_turn_inactivity( + *, + agent_holder, + task_id: str, + process_baseline, + timeout: float, + worker_done: threading.Event, + timeout_fired: threading.Event, + cleanup_lock: threading.Lock, + poll_interval: float = 5.0, +) -> None: + """Thread watchdog that remains runnable when gateway asyncio is starved.""" + while not worker_done.wait(max(0.01, poll_interval)): + agent = agent_holder[0] if agent_holder else None + if agent is None or not hasattr(agent, "get_activity_summary"): + continue + try: + idle_seconds = float( + agent.get_activity_summary().get("seconds_since_activity", 0.0) + ) + except Exception: + continue + if idle_seconds < timeout: + continue + _abandon_timed_out_gateway_turn( + agent_holder=agent_holder, + task_id=task_id, + process_baseline=process_baseline, + worker_done=worker_done, + timeout_fired=timeout_fired, + cleanup_lock=cleanup_lock, + ) + return + + _CONTROL_INTERRUPT_MESSAGES = frozenset( { _INTERRUPT_REASON_STOP.lower(), @@ -4755,6 +4865,11 @@ class TurnRunner: agent.thinking_progress = ctx._thinking_enabled # Store agent reference for interrupt support ctx.agent_holder[0] = agent + # Publish turn ownership for explicit /stop, /new, disconnect, and + # shutdown interrupts. Older session processes are outside this + # baseline and remain alive. + agent._gateway_turn_process_task_id = ctx.process_task_id + agent._gateway_turn_process_baseline = ctx.process_baseline # Capture the full tool definitions for transcript logging ctx.tools_holder[0] = agent.tools if hasattr(agent, 'tools') else None @@ -22132,6 +22247,20 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew running_agent = _iac_state.turn.agent if _iac_state else None if running_agent and running_agent is not _AGENT_PENDING_SENTINEL: running_agent.interrupt(interrupt_reason) + _process_task_id = getattr( + running_agent, "_gateway_turn_process_task_id", "" + ) + _process_baseline = getattr( + running_agent, "_gateway_turn_process_baseline", None + ) + if _process_task_id and _process_baseline is not None: + threading.Thread( + target=_reap_gateway_turn_processes, + args=(_process_task_id, _process_baseline), + kwargs={"source": "gateway_turn_interrupt"}, + name=f"gateway-turn-reaper-{_process_task_id[:12]}", + daemon=True, + ).start() self._invalidate_session_run_generation(session_key, reason=invalidation_reason) adapter = self._adapter_for_source(source) interrupt_session_activity = getattr( @@ -24095,8 +24224,62 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew _agent_warning_raw = _float_env("HERMES_AGENT_TIMEOUT_WARNING", 900) _agent_warning = _agent_warning_raw if _agent_warning_raw > 0 else None _warning_fired = False + + # A background=true process intentionally survives a successful + # turn, so capture existing IDs and reap only children created by + # THIS turn if it times out. The daemon watchdog is independent of + # asyncio: cgroup memory reclaim may starve the event loop that runs + # the normal timeout poll, but it need not also postpone cleanup + # until the loop recovers (#76115). + from tools.process_registry import process_registry + + _turn_task_id = session_id or "" + _turn_process_baseline = process_registry.snapshot_running_ids(_turn_task_id) + turn_ctx.process_task_id = _turn_task_id + turn_ctx.process_baseline = _turn_process_baseline + _turn_worker_done = threading.Event() + _turn_timeout_fired = threading.Event() + _turn_cleanup_lock = threading.Lock() + + def _run_sync_with_timeout_lifecycle(): + try: + return run_sync() + finally: + _turn_worker_done.set() + # `.turn.agent` on the session state is only reset to + # _AGENT_PENDING_SENTINEL when the *next* turn is + # claimed (see _session_state(...).turn.agent = ... at + # claim time), so a stale reference to this exact agent + # instance stays reachable from + # _interrupt_and_clear_session() until then. Clearing + # the ownership markers here — the instant this turn's + # own worker finishes — closes that window: an + # explicit /stop landing on the already-finished turn + # no longer reaps background work the turn deliberately + # left running (#76115). + _finished_agent = agent_holder[0] if agent_holder else None + if _finished_agent is not None: + _finished_agent._gateway_turn_process_task_id = "" + _finished_agent._gateway_turn_process_baseline = frozenset() + + if _agent_timeout is not None: + threading.Thread( + target=_watch_gateway_turn_inactivity, + kwargs={ + "agent_holder": agent_holder, + "task_id": _turn_task_id, + "process_baseline": _turn_process_baseline, + "timeout": _agent_timeout, + "worker_done": _turn_worker_done, + "timeout_fired": _turn_timeout_fired, + "cleanup_lock": _turn_cleanup_lock, + "poll_interval": 5.0, + }, + name=f"gateway-turn-watchdog-{_turn_task_id[:12]}", + daemon=True, + ).start() _executor_task = asyncio.ensure_future( - self._run_in_executor_with_context(run_sync) + self._run_in_executor_with_context(_run_sync_with_timeout_lifecycle) ) _inactivity_timeout = False @@ -24158,6 +24341,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew done, _ = await asyncio.wait( {_executor_task}, timeout=_POLL_INTERVAL ) + if _turn_timeout_fired.is_set(): + _inactivity_timeout = True + break if done: response = _executor_task.result() break @@ -24191,6 +24377,19 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew logger.debug("Inactivity warning send error: %s", _warn_err) if _idle_secs >= _agent_timeout: _inactivity_timeout = True + threading.Thread( + target=_abandon_timed_out_gateway_turn, + kwargs={ + "agent_holder": agent_holder, + "task_id": _turn_task_id, + "process_baseline": _turn_process_baseline, + "worker_done": _turn_worker_done, + "timeout_fired": _turn_timeout_fired, + "cleanup_lock": _turn_cleanup_lock, + }, + name=f"gateway-turn-reaper-{_turn_task_id[:12]}", + daemon=True, + ).start() break # Backup interrupt check (same as unlimited path). if not _interrupt_detected.is_set() and session_key: diff --git a/gateway/turn_context.py b/gateway/turn_context.py index 03a6d404c168e..ac727e2b4f3a8 100644 --- a/gateway/turn_context.py +++ b/gateway/turn_context.py @@ -84,6 +84,8 @@ class TurnContext: session_id: Optional[str] = None session_key: Optional[str] = None run_generation: Optional[int] = None + process_task_id: str = "" + process_baseline: frozenset[str] = field(default_factory=frozenset) _interrupt_depth: int = 0 event_message_id: Optional[str] = None moa_config: Optional[dict] = None diff --git a/tests/gateway/test_abandoned_turn_process_cleanup.py b/tests/gateway/test_abandoned_turn_process_cleanup.py new file mode 100644 index 0000000000000..6d4a029e86758 --- /dev/null +++ b/tests/gateway/test_abandoned_turn_process_cleanup.py @@ -0,0 +1,114 @@ +"""Regression coverage for abandoned gateway-turn subprocess cleanup (#76115).""" + +import threading + +from gateway.run import ( + _abandon_timed_out_gateway_turn, + _watch_gateway_turn_inactivity, +) +from tools.process_registry import process_registry + + +class _IdleAgent: + def __init__(self, idle_seconds=60.0): + self.idle_seconds = idle_seconds + self.interrupts = [] + + def get_activity_summary(self): + return {"seconds_since_activity": self.idle_seconds} + + def interrupt(self, reason): + self.interrupts.append(reason) + + +def _state(): + return threading.Event(), threading.Event(), threading.Lock() + + +def test_thread_watchdog_reaps_only_processes_created_by_timed_out_turn(monkeypatch): + agent = _IdleAgent() + 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 1, + ) + + watchdog = threading.Thread( + target=_watch_gateway_turn_inactivity, + kwargs={ + "agent_holder": [agent], + "task_id": "session-a", + "process_baseline": frozenset({"proc_existing"}), + "timeout": 30.0, + "worker_done": worker_done, + "timeout_fired": timeout_fired, + "cleanup_lock": cleanup_lock, + "poll_interval": 0.01, + }, + ) + watchdog.start() + watchdog.join(timeout=1) + + assert not watchdog.is_alive() + assert timeout_fired.is_set() + assert agent.interrupts == ["Execution timed out (inactivity)"] + assert calls == [ + ( + "session-a", + frozenset({"proc_existing"}), + "gateway_turn_timeout", + ) + ] + + +def test_completed_worker_wins_race_and_preserves_background_process(monkeypatch): + agent = _IdleAgent() + worker_done, timeout_fired, cleanup_lock = _state() + worker_done.set() + monkeypatch.setattr( + process_registry, + "kill_started_since", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("completed turn must not reap background work") + ), + ) + + assert not _abandon_timed_out_gateway_turn( + agent_holder=[agent], + task_id="session-a", + process_baseline=frozenset(), + worker_done=worker_done, + timeout_fired=timeout_fired, + cleanup_lock=cleanup_lock, + ) + assert not timeout_fired.is_set() + assert agent.interrupts == [] + + +def test_timeout_cleanup_is_idempotent(monkeypatch): + agent = _IdleAgent() + worker_done, timeout_fired, cleanup_lock = _state() + calls = [] + monkeypatch.setattr( + process_registry, + "kill_started_since", + lambda *_args, **_kwargs: calls.append(True) or 0, + ) + kwargs = { + "agent_holder": [agent], + "task_id": "session-a", + "process_baseline": frozenset(), + "worker_done": worker_done, + "timeout_fired": timeout_fired, + "cleanup_lock": cleanup_lock, + } + + assert _abandon_timed_out_gateway_turn(**kwargs) + assert not _abandon_timed_out_gateway_turn(**kwargs) + assert len(calls) == 1 + assert len(agent.interrupts) == 1 diff --git a/tests/gateway/test_tool_response_drop_recovery.py b/tests/gateway/test_tool_response_drop_recovery.py index 7f17c4f3fddb8..1a133021b9f02 100644 --- a/tests/gateway/test_tool_response_drop_recovery.py +++ b/tests/gateway/test_tool_response_drop_recovery.py @@ -256,7 +256,9 @@ class TestPostStopInterruptSwallow: assert "send it again" in response @pytest.mark.asyncio - async def test_interrupt_and_clear_session_evicts_cached_agent(self): + async def test_interrupt_and_clear_session_evicts_cached_agent( + self, monkeypatch + ): """The control-interrupt path must evict the session's cached agent so its ``_interrupt_requested`` flag cannot leak into the next turn.""" import threading @@ -271,6 +273,8 @@ class TestPostStopInterruptSwallow: self.interrupt_reasons.append(reason) agent = _RecordingAgent() + agent._gateway_turn_process_task_id = "session-123" + agent._gateway_turn_process_baseline = frozenset({"proc_existing"}) session_key = "agent:main:telegram:dm:12345" source = SessionSource( platform=Platform.TELEGRAM, chat_id="12345", chat_type="dm" @@ -291,6 +295,16 @@ class TestPostStopInterruptSwallow: runner._release_running_agent_state = ( lambda key, **kw: released.append(key) ) + reaped = [] + reaped_event = threading.Event() + from tools.process_registry import process_registry + + def _record_reap(task_id, baseline, *, source): + reaped.append((task_id, baseline, source)) + reaped_event.set() + return 1 + + monkeypatch.setattr(process_registry, "kill_started_since", _record_reap) await runner._interrupt_and_clear_session( session_key, @@ -301,6 +315,14 @@ class TestPostStopInterruptSwallow: assert agent.interrupt_reasons == [_INTERRUPT_REASON_STOP] assert released == [session_key] + assert reaped_event.wait(1) + assert reaped == [ + ( + "session-123", + frozenset({"proc_existing"}), + "gateway_turn_interrupt", + ) + ] assert session_key not in runner._agent_cache, ( "Cached agent with a set interrupt flag must be evicted on /stop " "so the flag cannot kill the session's next message (#44212)" diff --git a/tests/tools/test_process_registry.py b/tests/tools/test_process_registry.py index fe04da6beed22..dd69764aa3f72 100644 --- a/tests/tools/test_process_registry.py +++ b/tests/tools/test_process_registry.py @@ -55,6 +55,43 @@ def _spawn_python_sleep(seconds: float) -> subprocess.Popen: ) +def test_kill_started_since_preserves_preexisting_and_foreign_processes(registry): + old = _make_session(sid="proc_old", task_id="session-a") + finished = _make_session( + sid="proc_finished", task_id="session-a", exited=True, exit_code=0 + ) + registry._running[old.id] = old + registry._finished[finished.id] = finished + baseline = registry.snapshot_running_ids("session-a") + + new = _make_session(sid="proc_new", task_id="session-a") + foreign = _make_session(sid="proc_foreign", task_id="session-b") + registry._running[new.id] = new + registry._running[foreign.id] = foreign + + calls = [] + + def fake_kill(session_id, **kwargs): + calls.append((session_id, kwargs)) + return {"status": "killed"} + + registry.kill_process = fake_kill + + assert baseline == frozenset({"proc_old"}) + assert registry.kill_started_since( + "session-a", baseline, source="gateway_turn_timeout" + ) == 1 + assert calls == [ + ( + "proc_new", + { + "source": "gateway_turn_timeout", + "consume_output": True, + }, + ) + ] + + def _wait_until(predicate, timeout: float = 5.0, interval: float = 0.05) -> bool: """Poll a predicate until it returns truthy or the timeout elapses.""" deadline = time.monotonic() + timeout diff --git a/tools/process_registry.py b/tools/process_registry.py index 1cfc7ee7d0cc7..e683a341f6902 100644 --- a/tools/process_registry.py +++ b/tools/process_registry.py @@ -1929,6 +1929,50 @@ class ProcessRegistry: with self._lock: return any(not s.exited for s in self._running.values()) + def snapshot_running_ids(self, task_id: str) -> frozenset[str]: + """Capture running process IDs owned by ``task_id``. + + Gateway turns use this as a boundary marker: if a turn times out, only + processes absent from its starting snapshot belong to the abandoned + turn. Older session processes must survive because background tasks + intentionally span successful turns. + """ + with self._lock: + return frozenset( + s.id + for s in self._running.values() + if s.task_id == task_id and not s.exited + ) + + def kill_started_since( + self, + task_id: str, + baseline_ids, + *, + source: str, + ) -> int: + """Kill processes created for ``task_id`` after a prior snapshot.""" + baseline = frozenset(baseline_ids or ()) + with self._lock: + targets = [ + s + for s in self._running.values() + if s.task_id == task_id and s.id not in baseline and not s.exited + ] + + killed = 0 + for session in targets: + result = self.kill_process( + session.id, + source=source, + # Abandoned-turn output must not enqueue a synthetic follow-up + # that revives work the timeout deliberately stopped. + consume_output=True, + ) + if result.get("status") in {"killed", "already_exited"}: + killed += 1 + return killed + def kill_all(self, task_id: str = None) -> int: """Kill all running processes, optionally filtered by task_id. Returns count killed.""" with self._lock: