feat(gateway): dump wedged worker stacks when the turn reaper fires
When the inactivity reaper interrupts a timed-out turn, the interrupt frees the blocked frame — destroying the only evidence of where the turn was wedged. The Aug 2026 zombie-turn incident (WhatsApp session, Relay-corrupted scope stack) wedged every turn for exactly the 1800s timeout somewhere between 'Turn ended' and run_sync returning, and the wedge point was unprovable post-mortem. The reaper now logs the stack of every thread with turn-machinery frames BEFORE interrupting, so the next occurrence names the exact blocked line. Best-effort, bounded (8 threads, 25 frames), pure in-process, never raises into the reaper.
This commit is contained in:
Teknium
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user