Merge pull request #4502 from nesquena/stage-4492

Release v0.51.519 — reconcile stale active-run registry entries (#4492)

CHANGELOG.md

@@ -3,6 +3,12 @@
 
 ## [Unreleased]
 
+## [v0.51.519] — 2026-06-19 — Release SD (reconcile stale active-run registry entries)
+
+### Fixed
+
+- **A genuinely-dead streaming run no longer lingers in the active-run registry advertising a half-alive state (#4492).** When a run passed the bounded unwind ceiling, `_active_run_stream_for_session` already declined to block a successor turn on it, but the zombie entry stayed in `ACTIVE_RUNS`, so health and recovery polling kept seeing it. It now reconciles those entries out of the registry — but only when the worker is truly gone from the live `STREAMS` map, so a long turn that is merely mid-teardown keeps its lifecycle row. `cancel_stream` also stamps the run `phase="cancelling"` immediately so the registry reflects the cancel during the detached teardown window. Thanks @ai-ag2026.
+
 ## [v0.51.518] — 2026-06-19 — Release SC (read-only memory writes report 403, not 500)
 
 ### Fixed

api/routes.py

@@ -14139,8 +14139,17 @@ 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 (_live_config.ACTIVE_RUNS or {}).items():
+            for run_stream_id, raw in list((_live_config.ACTIVE_RUNS or {}).items()):
                 stream_id = str((raw or {}).get("stream_id") or run_stream_id or "").strip()
                 run_sid = str((raw or {}).get("session_id") or "").strip()
                 if run_sid != sid or not stream_id:
@@ -14149,11 +14158,19 @@ 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
-                # Ignore a stale/wedged entry past the unwind ceiling so it can't
-                # block the session permanently.
+                # 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:
+                    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:
+                (_live_config.ACTIVE_RUNS or {}).pop(stale_stream_id, None)
     except Exception:
         return None
     return None

api/streaming.py

@@ -8903,6 +8903,11 @@ def cancel_stream(stream_id: str) -> bool:
         if active_run_entry and not active_run_session_id:
             active_run_session_id = str(active_run_entry.get("session_id") or "").strip() or None
 
+    # Mark the worker lifecycle registry immediately. The SSE maps may be popped
+    # below while the worker is still unwinding; ACTIVE_RUNS is what recovery /
+    # health polling sees during that detached window.
+    update_active_run(stream_id, phase="cancelling")
+
     # Set WebUI layer cancel flag. Prefer the snapshot captured under the lock;
     # fall back to a fresh lookup for the ACTIVE_RUNS-only path (stream absent).
     flag = _snap_flag if _snap_flag is not None else cancel_flags.get(stream_id)

tests/test_cancel_interrupt.py

@@ -127,6 +127,7 @@ class TestCancelInterrupt:
             result = cancel_stream(stream_id)
 
         assert result is True
+        assert ACTIVE_RUNS[stream_id]["phase"] == "cancelling"
         mock_agent.interrupt.assert_called_once_with("Cancelled by user")
         assert mock_session.active_stream_id is None
         assert mock_session.pending_user_message is None

tests/test_stale_stream_cleanup.py

@@ -319,6 +319,9 @@ def test_chat_start_not_permanently_blocked_by_stale_active_run(monkeypatch, tmp
 
     # The bounded guard should treat it as stale and NOT report it as blocking.
     assert routes._active_run_stream_for_session(session.session_id) is None
+    # It must also reconcile the zombie registry entry immediately so health /
+    # recovery polling does not keep advertising a half-alive run forever.
+    assert stale_stream_id not in config.ACTIVE_RUNS
 
     class NoopThread:
         def __init__(self, *args, **kwargs):
@@ -350,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.