fix(gateway): reap only the background processes an abandoned turn created

An agent turn can spawn a long-running background subprocess (e.g.
`next build`) and later be abandoned via inactivity timeout, /stop,
/new, or a client disconnect. Before this fix the gateway interrupted
the agent loop but never touched the subprocess: it kept running
inside the gateway's cgroup, unbounded, until memory pressure starved
the event loop and made every platform/cron look hung (#76115).

The process registry already knew how to kill a process tree — the
missing piece was per-turn ownership: nothing distinguished a process
that predates the turn (must survive), a process the turn started and
finished successfully (must survive), and a process an abandoned turn
left running (must be reaped).

- tools/process_registry.py: snapshot_running_ids() captures a turn's
  starting baseline; kill_started_since() reaps only IDs created after
  it, scoped to one task_id.
- gateway/turn_context.py: TurnContext carries process_task_id +
  process_baseline so the timeout/interrupt paths can reach them.
- gateway/run.py: baseline is snapshotted right before the turn's
  executor task starts; the inactivity-timeout path and the explicit
  /stop|/new|disconnect interrupt path both reap via the same helper.
  A daemon-thread watchdog backs up the asyncio-based timeout poll,
  since a starved event loop is exactly the failure mode this bug
  causes. The turn's own worker clears its ownership markers the
  instant it finishes, closing a race where a /stop landing right
  after normal completion could reap a background process the turn
  deliberately left running.

Related but insufficient on their own: #37454 (cgroup ExecStopPost
reaper only fires on service restart) and #68915 (orphaned-pipe
grandchild detection, a registry bug not a turn-lifecycle gap).
Neither ties process cleanup to turn abandonment.

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:

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

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

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

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

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: