diff --git a/CHANGELOG.md b/CHANGELOG.md index 143250ec..162b6644 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/api/routes.py b/api/routes.py index 64c5c6d9..dd917f4b 100644 --- a/api/routes.py +++ b/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 diff --git a/api/streaming.py b/api/streaming.py index 7a2486bf..adddf30a 100644 --- a/api/streaming.py +++ b/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) diff --git a/tests/test_cancel_interrupt.py b/tests/test_cancel_interrupt.py index c2d35aac..375d0ef5 100644 --- a/tests/test_cancel_interrupt.py +++ b/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 diff --git a/tests/test_stale_stream_cleanup.py b/tests/test_stale_stream_cleanup.py index 8d5756a9..2fe67c65 100644 --- a/tests/test_stale_stream_cleanup.py +++ b/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.