diff --git a/gateway/run.py b/gateway/run.py index 70950a11c5..3092fd31c5 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -40,6 +40,7 @@ import sys import signal import threading import time +import traceback from collections import OrderedDict from contextvars import copy_context from pathlib import Path @@ -3002,6 +3003,69 @@ def _reap_gateway_turn_processes( return killed +_TURN_STACK_DUMP_FRAME_MARKERS = ( + "run_conversation", + "run_sync", + "_run_sync_with_timeout_lifecycle", + "finalize_turn", + "end_turn", + "run_in_session", +) + + +def _dump_wedged_turn_stacks(task_id: str) -> None: + """Log the stack of every thread that looks like turn work, at reap time. + + When the inactivity reaper fires, the model loop is usually long done and + the worker thread is wedged somewhere in post-turn finalization — but the + reaper's hard interrupt frees it, so the blocked frame is gone before + anyone can attach a profiler. A live incident (Aug 2026, WhatsApp session + on a Relay-corrupted scope stack) wedged EVERY turn for exactly the + 1800s timeout between "Turn ended" and run_sync returning, and the wedge + point was unrecoverable post-mortem. Dumping the stacks here, BEFORE the + interrupt, names the frame. + + Best-effort and bounded: pure in-process frame walking (no signals, no + external tools), only threads whose stack mentions a turn-machinery + marker are logged, output capped per thread. Must never raise into the + reaper. + """ + try: + frames = sys._current_frames() + names = {t.ident: t.name for t in threading.enumerate()} + dumped = 0 + for ident, frame in frames.items(): + if ident == threading.get_ident(): + continue # the reaper itself + stack = traceback.format_stack(frame) + joined = "".join(stack) + if not any(marker in joined for marker in _TURN_STACK_DUMP_FRAME_MARKERS): + continue + dumped += 1 + if dumped > 8: + logger.error( + "Wedged-turn stack dump for task %s truncated: more than " + "8 candidate threads", + task_id, + ) + break + logger.error( + "Wedged-turn stack dump (task=%s thread=%s ident=%s):\n%s", + task_id, + names.get(ident, "?"), + ident, + "".join(stack[-25:]), + ) + if dumped == 0: + logger.error( + "Wedged-turn stack dump for task %s: no thread with " + "turn-machinery frames found (worker may have already exited)", + task_id, + ) + except Exception: + logger.debug("Wedged-turn stack dump failed", exc_info=True) + + def _abandon_timed_out_gateway_turn( *, agent_holder, @@ -3018,6 +3082,11 @@ def _abandon_timed_out_gateway_turn( return False timeout_fired.set() + # Capture the wedged worker's stack BEFORE interrupting it — the + # interrupt frees the blocked frame, destroying the only evidence of + # where the turn was stuck (see _dump_wedged_turn_stacks). + _dump_wedged_turn_stacks(task_id) + agent = agent_holder[0] if agent_holder else None if agent is not None: try: diff --git a/tests/gateway/test_abandoned_turn_process_cleanup.py b/tests/gateway/test_abandoned_turn_process_cleanup.py index c63a930b6a..da0fcdaa0c 100644 --- a/tests/gateway/test_abandoned_turn_process_cleanup.py +++ b/tests/gateway/test_abandoned_turn_process_cleanup.py @@ -232,3 +232,98 @@ def test_timeout_abandon_propagates_is_still_current_to_the_reap(monkeypatch): # reap was skipped because a newer turn already claimed the session. assert agent.interrupts == ["Execution timed out (inactivity)"] assert calls == [] + + +# --------------------------------------------------------------------------- +# Wedged-turn stack dump at reap time (Aug 2026 zombie-turn incident): +# the reaper's interrupt frees the blocked frame, so the dump must run +# BEFORE the interrupt and must capture the actual wedged stack. +# --------------------------------------------------------------------------- + + +def _run_wedged_worker(release: threading.Event, entered: threading.Event): + """Worker blocked inside a frame named like turn machinery.""" + + def run_sync(): # marker frame the dump filter matches on + entered.set() + release.wait(timeout=30.0) + + run_sync() + + +def test_reaper_dumps_wedged_worker_stack_before_interrupt(monkeypatch, caplog): + import logging + + from gateway.run import _dump_wedged_turn_stacks + + release = threading.Event() + entered = threading.Event() + worker = threading.Thread( + target=_run_wedged_worker, + args=(release, entered), + name="wedged-test-worker", + daemon=True, + ) + worker.start() + try: + assert entered.wait(timeout=5.0) + with caplog.at_level(logging.ERROR, logger="gateway.run"): + _dump_wedged_turn_stacks("task-wedge-test") + dumps = [ + r for r in caplog.records if "Wedged-turn stack dump" in r.getMessage() + ] + assert dumps, "no stack dump was logged" + joined = "\n".join(r.getMessage() for r in dumps) + assert "wedged-test-worker" in joined + assert "run_sync" in joined + assert "release.wait" in joined # the actual blocked line is named + finally: + release.set() + worker.join(timeout=5.0) + + +def test_abandon_timed_out_turn_dumps_stacks_before_interrupt(monkeypatch): + """The dump hook runs inside the reaper, before the agent interrupt.""" + import gateway.run as gateway_run + + order = [] + monkeypatch.setattr( + gateway_run, + "_dump_wedged_turn_stacks", + lambda task_id: order.append(("dump", task_id)), + ) + monkeypatch.setattr( + gateway_run, + "_reap_gateway_turn_processes", + lambda *a, **k: order.append(("reap",)), + ) + + class _Agent: + def interrupt(self, reason): + order.append(("interrupt", reason)) + + worker_done, timeout_fired, cleanup_lock = _state() + assert _abandon_timed_out_gateway_turn( + agent_holder=[_Agent()], + task_id="t-dump-order", + process_baseline=frozenset(), + worker_done=worker_done, + timeout_fired=timeout_fired, + cleanup_lock=cleanup_lock, + ) + assert order[0] == ("dump", "t-dump-order") + assert ("interrupt", order[1][1]) == order[1] + assert order[-1] == ("reap",) + + +def test_dump_wedged_turn_stacks_never_raises(monkeypatch): + import gateway.run as gateway_run + + monkeypatch.setattr( + gateway_run.sys, + "_current_frames", + lambda: (_ for _ in ()).throw(RuntimeError("boom")), + ) + from gateway.run import _dump_wedged_turn_stacks + + _dump_wedged_turn_stacks("t-no-raise") # must not raise