fix(streams): only reap stale ACTIVE_RUNS entries when worker is gone from STREAMS (#4492 Codex gate)

Gate-found: reaping on started_at age alone could pop a genuinely-live worker
that's mid-teardown past the 180s ceiling (active_stream_id cleared during final
writeback, STREAMS entry still present) — ACTIVE_RUNS is worker-lifecycle truth
for health/background-wakeup/agent-cache consumers. Add a 'not in STREAMS'
guard (snapshot taken sequentially before ACTIVE_RUNS_LOCK, no nesting) and pop
by the real dict key. +1 regression test (toothed). Keeps the anti-permanent-409
age behavior intact.

Co-authored-by: ai-ag2026 <ai-ag2026@users.noreply.github.com>

api/routes.py

@@ -14139,6 +14139,14 @@ def _active_run_stream_for_session(session_id: str | None) -> str | None:
     now = time.time()
     try:
         from api import config as _live_config
+        # Snapshot the live-worker set BEFORE taking ACTIVE_RUNS_LOCK (sequential,
+        # not nested, so no lock-ordering/deadlock risk). A long turn whose
+        # active_stream_id was already cleared during final writeback can still be
+        # mid-teardown — its STREAMS entry present — past the age ceiling, so age
+        # alone must NOT pop its lifecycle row (health / background-wakeup /
+        # active-agent-cache consumers read ACTIVE_RUNS as worker-lifecycle truth).
+        with _live_config.STREAMS_LOCK:
+            live_stream_ids = set(_live_config.STREAMS.keys())
         stale_stream_ids = []
         with _live_config.ACTIVE_RUNS_LOCK:
             for run_stream_id, raw in list((_live_config.ACTIVE_RUNS or {}).items()):
@@ -14150,10 +14158,15 @@ def _active_run_stream_for_session(session_id: str | None) -> str | None:
                     started_at = float((raw or {}).get("started_at") or 0)
                 except (TypeError, ValueError):
                     started_at = 0.0
-                # Reconcile stale/wedged entries past the unwind ceiling so they
-                # cannot keep health/recovery polling in a half-alive state.
+                # Past the unwind ceiling: never block a successor on it (the
+                # anti-permanent-409 guarantee, #3822). Additionally reconcile the
+                # zombie out of ACTIVE_RUNS so health/recovery polling stops seeing a
+                # half-alive run — but ONLY when the worker is truly gone from
+                # STREAMS, so a still-live / still-tearing-down worker keeps its
+                # lifecycle row. Pop by the real dict key. (Codex gate, #4492)
                 if started_at and (now - started_at) > ceiling:
-                    stale_stream_ids.append(stream_id)
+                    if run_stream_id not in live_stream_ids and stream_id not in live_stream_ids:
+                        stale_stream_ids.append(run_stream_id)
                     continue
                 return stream_id
             for stale_stream_id in stale_stream_ids:

tests/test_stale_stream_cleanup.py

@@ -353,6 +353,34 @@ def test_chat_start_not_permanently_blocked_by_stale_active_run(monkeypatch, tmp
         routes.STREAMS.pop("new-stream", None)
 
 
+def test_live_worker_past_ceiling_is_not_reaped_from_active_runs():
+    """A still-live worker (present in STREAMS) past the age ceiling must NOT be
+    popped from ACTIVE_RUNS — only genuinely-gone workers are reconciled (#4492).
+
+    A long turn whose active_stream_id was cleared during final writeback can be
+    mid-teardown past the 180s ceiling while its STREAMS entry is still present;
+    reaping its lifecycle row then would lie to health / background-wakeup /
+    active-agent-cache consumers that read ACTIVE_RUNS as worker-lifecycle truth.
+    """
+    config.STREAMS.clear()
+    config.ACTIVE_RUNS.clear()
+    sid = "live-teardown-session"
+    live_stream_id = "still-alive-stream"
+    config.register_active_run(live_stream_id, session_id=sid, phase="running")
+    # Age the entry past the ceiling AND keep the worker present in STREAMS.
+    with config.ACTIVE_RUNS_LOCK:
+        config.ACTIVE_RUNS[live_stream_id]["started_at"] = time.time() - 600
+    config.STREAMS[live_stream_id] = object()
+    try:
+        # Not reported as blocking (past the ceiling) ...
+        assert routes._active_run_stream_for_session(sid) is None
+        # ... but the lifecycle row is preserved because the worker is still live.
+        assert live_stream_id in config.ACTIVE_RUNS
+    finally:
+        config.STREAMS.pop(live_stream_id, None)
+        config.unregister_active_run(live_stream_id)
+
+
 def test_stale_stream_cleanup_does_not_clobber_concurrent_chat_start(monkeypatch):
     """Regression for #1533: stale cleanup must not erase a new stream id.