fix(streaming): reject obsolete cancel finalizer after successor completes (#6636)

Prevents a delayed cancelled streaming worker from writing back stale output over a completed successor's result, and clears the Gateway writeback-owner entry on teardown/pre-start-cancel to avoid a slow SESSION_WRITEBACK_OWNERS leak.

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

api/config.py

@@ -8879,6 +8879,52 @@ def unregister_stream_owner(stream_id: str) -> None:
         STREAM_SESSION_OWNERS.pop(stream_id, None)
 
 
+# ── Per-session writeback-ownership registry (#6623 re-gate) ────────────────
+# Maps session_id -> stream_id of the turn that currently owns the session's
+# writeback. Written whenever a turn is admitted (route layer, next to
+# session.active_stream_id), REPLACED when a successor turn is admitted, and
+# NEVER cleared by cancel_stream() — cancel eagerly pops STREAMS/ACTIVE_RUNS
+# and clears ``active_stream_id``, so a delayed finalizer from an old worker
+# cannot tell "the session advanced to a successor" apart from "cancel simply
+# cleared the field" by looking at its own (possibly LRU-evicted, detached)
+# snapshot. This record survives cancel cleanup: the owning worker's own
+# finally clears the entry, and only while it still owns it.
+SESSION_WRITEBACK_OWNERS: dict = {}
+SESSION_WRITEBACK_OWNERS_LOCK = threading.Lock()
+
+
+def register_session_writeback_owner(session_id: str, stream_id: str) -> None:
+    """Record the stream that currently owns a session's writeback."""
+    session_id = str(session_id or "").strip()
+    stream_id = str(stream_id or "").strip()
+    if not session_id or not stream_id:
+        return
+    with SESSION_WRITEBACK_OWNERS_LOCK:
+        SESSION_WRITEBACK_OWNERS[session_id] = stream_id
+
+
+def session_writeback_owner(session_id: str) -> str | None:
+    """Return the stream that currently owns the session's writeback, if any."""
+    session_id = str(session_id or "").strip()
+    if not session_id:
+        return None
+    with SESSION_WRITEBACK_OWNERS_LOCK:
+        owner = SESSION_WRITEBACK_OWNERS.get(session_id)
+    owner = str(owner or "").strip()
+    return owner or None
+
+
+def clear_session_writeback_owner_if_owned(session_id: str, stream_id: str) -> None:
+    """Forget the writeback-ownership entry only while ``stream_id`` still owns it."""
+    session_id = str(session_id or "").strip()
+    stream_id = str(stream_id or "").strip()
+    if not session_id or not stream_id:
+        return
+    with SESSION_WRITEBACK_OWNERS_LOCK:
+        if SESSION_WRITEBACK_OWNERS.get(session_id) == stream_id:
+            SESSION_WRITEBACK_OWNERS.pop(session_id, None)
+
+
 # ── Gateway capability cache ─────────────────────────────────────────────────
 # Probes /v1/capabilities once per base_url/api-key pair and caches the result
 # for 60 s so guarded-turn routing decisions do not add latency on every chat

api/gateway_chat.py

@@ -25,6 +25,7 @@ from api.config import (
     STREAM_REASONING_TEXT,
     _get_session_agent_lock,
     _parse_provider_qualified_model_id,
+    clear_session_writeback_owner_if_owned,
     coerce_reasoning_effort_for_model,
     gateway_approval_unavailable_reason,
     gateway_supports_approval,
@@ -848,6 +849,10 @@ def _run_gateway_chat_streaming(
         # Cancelled before the worker started; release the owner entry the route
         # layer registered so STREAM_SESSION_OWNERS does not leak (no teardown finally runs).
         unregister_stream_owner(stream_id)
+        # Also release the writeback-owner entry the route layer registered, so
+        # SESSION_WRITEBACK_OWNERS does not leak on this pre-start cancellation
+        # path (the teardown finally below never runs when we early-return here).
+        clear_session_writeback_owner_if_owned(session_id, stream_id)
         return
     register_active_run(
         stream_id,
@@ -1394,3 +1399,8 @@ def _run_gateway_chat_streaming(
         _clear_gateway_run_starting(stream_id)
         unregister_stream_owner(stream_id)
         unregister_active_run(stream_id)
+        # Release the writeback-owner entry the route layer registered for this
+        # Gateway run so SESSION_WRITEBACK_OWNERS does not grow unbounded across
+        # the process lifetime (compare-and-clear: only clears if still owned by
+        # this stream, mirroring the local streaming teardown).
+        clear_session_writeback_owner_if_owned(session_id, stream_id)

api/routes.py

@@ -2841,6 +2841,7 @@ from api.config import (
     ACTIVE_RUNS,
     ACTIVE_RUNS_LOCK,
     register_stream_owner,
+    register_session_writeback_owner,
     stream_owner_session_id,
     unregister_stream_owner,
     CHAT_LOCK,
@@ -2905,6 +2906,37 @@ def _kanban_unknown_endpoint(handler, parsed, method: str) -> bool:
     ) or True
 
 
+# A cancelled worker that stays in ACTIVE_RUNS longer than this is treated as
+# stuck (e.g. blocked in C-level provider I/O and never reaching its finally).
+# Once the cancel has been outstanding past this grace window, the run row can
+# no longer protect the session's active_stream_id/pending_* from stale
+# cleanup: _clear_stale_stream_state() clears them, and every delayed cancel
+# finalizer (api/streaming.py _finalize_cancelled_turn) is generation-guarded
+# under the session lock — it no-ops unless the session still points at the
+# cancelled stream — so clearing early cannot clobber a newer turn (#6623).
+_STALE_CANCELLED_RUN_GRACE_SECONDS = 60.0
+
+
+def _cancelled_run_is_stale(run_entry) -> bool:
+    """Return True when an ACTIVE_RUNS row belongs to a cancel that has been
+    outstanding longer than the stale grace window.
+
+    ``cancelled_at`` is stamped by cancel_stream() when it flips the run to
+    phase="cancelling". ``started_at`` is accepted as a fallback anchor so runs
+    cancelled before the stamp was introduced are still reclaimed eventually.
+    """
+    if not isinstance(run_entry, dict) or run_entry.get("phase") != "cancelling":
+        return False
+    anchor = run_entry.get("cancelled_at") or run_entry.get("started_at")
+    if not anchor:
+        return False
+    try:
+        age = time.time() - float(anchor)
+    except (TypeError, ValueError):
+        return False
+    return age >= _STALE_CANCELLED_RUN_GRACE_SECONDS
+
+
 def _clear_stale_stream_state(session) -> bool:
     """Clear persisted streaming flags when the in-memory stream no longer exists.
 
@@ -2932,13 +2964,33 @@ def _clear_stale_stream_state(session) -> bool:
     except Exception:
         worker_alive = False
     if worker_alive:
-        logger.debug(
-            "_clear_stale_stream_state: stream %s for session %s missing SSE channel "
-            "but worker bookkeeping is still active; deferring stale cleanup",
+        # #6623: a worker stuck in C-level I/O may never reach its finally to
+        # unregister the run, so ACTIVE_RUNS could hold the row forever and
+        # block stale cleanup indefinitely. A *cancelled* run (cancel_stream()
+        # stamped phase="cancelling" + cancelled_at) that has not unwound past
+        # the grace window is treated as stale — clear the session anyway. The
+        # _stream_writeback_is_current() guard rejects any eventual writeback
+        # from the stuck worker, so this cannot clobber a newer turn.
+        try:
+            with _live_config.ACTIVE_RUNS_LOCK:
+                run_entry = dict((_live_config.ACTIVE_RUNS or {}).get(stream_id) or {})
+        except Exception:
+            run_entry = {}
+        if not _cancelled_run_is_stale(run_entry):
+            logger.debug(
+                "_clear_stale_stream_state: stream %s for session %s missing SSE channel "
+                "but worker bookkeeping is still active; deferring stale cleanup",
+                stream_id,
+                getattr(session, "session_id", "?"),
+            )
+            return False
+        logger.info(
+            "_clear_stale_stream_state: stream %s for session %s missing SSE channel and "
+            "cancelled run is stale (cancelled_at=%s); clearing stale stream state (#6623)",
             stream_id,
             getattr(session, "session_id", "?"),
+            run_entry.get("cancelled_at"),
         )
-        return False
     grace_seconds = 30.0
     try:
         from api.models import _REPAIR_STALE_PENDING_GRACE_SECONDS
@@ -20991,6 +21043,7 @@ def _handle_btw(handler, body):
     ephemeral.save()
     stream_id = uuid.uuid4().hex
     ephemeral.active_stream_id = stream_id
+    register_session_writeback_owner(ephemeral.session_id, stream_id)
     ephemeral.save()
     stream = create_stream_channel()
     register_stream_owner(stream_id, ephemeral.session_id)
@@ -21041,6 +21094,7 @@ def _handle_background(handler, body):
     bg.save()
     stream_id = uuid.uuid4().hex
     bg.active_stream_id = stream_id
+    register_session_writeback_owner(bg.session_id, stream_id)
     bg.save()
     stream = create_stream_channel()
     register_stream_owner(stream_id, bg.session_id)
@@ -21180,6 +21234,7 @@ def _prepare_chat_start_session_for_stream(
     s.model = model
     s.model_provider = model_provider
     s.active_stream_id = stream_id
+    register_session_writeback_owner(s.session_id, stream_id)
     s.post_compression_context_tokens_estimate = None
     s.pending_user_message = msg
     s.pending_attachments = attachments
@@ -21261,9 +21316,14 @@ def _active_run_stream_for_session(session_id: str | None) -> str | None:
     wedged worker that never reaches its finally (e.g. stuck in a provider call,
     or leaked by SIGKILL without restart) must NOT 409 the session forever — so an
     entry older than the unwind ceiling (180s) is treated as stale and ignored
-    here. A legitimately long-running turn keeps ``active_stream_id`` SET
-    and is handled by ``_active_stream_blocks_chat_start`` above; this guard only
-    covers the cleared-stream-id unwind window. (Codex brick-gate hardening, #3822.)
+    here. For a phase="cancelling" row the ceiling is anchored on the cancel time
+    (``cancelled_at``), never on the original run start: cancel_stream() removes
+    STREAMS itself, so absence from STREAMS is not worker-death proof, and a
+    long-running turn that was just cancelled must not be reaped (and a successor
+    admitted) while the old worker is still alive (#6623). A legitimately
+    long-running turn keeps ``active_stream_id`` SET and is handled by
+    ``_active_stream_blocks_chat_start`` above; this guard only covers the
+    cleared-stream-id unwind window. (Codex brick-gate hardening, #3822.)
     """
     sid = str(session_id or "").strip()
     if not sid:
@@ -21291,13 +21351,35 @@ 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
+                # #6623 re-gate: cancel_stream() removes STREAMS itself, so a
+                # missing SSE channel is NOT proof that a cancelled worker is
+                # dead. For a phase="cancelling" row the unwind ceiling must be
+                # anchored on the CANCEL time (cancelled_at, started_at as a
+                # legacy fallback), never on the original run start: a turn that
+                # ran for minutes and was just cancelled would otherwise be
+                # reaped the instant its started_at crosses the ceiling and a
+                # successor admitted while the old worker is still alive. A
+                # recently cancelled run therefore keeps blocking a successor
+                # (this function returns its stream id) until the worker either
+                # unwinds (its finally unregisters the row within seconds) or
+                # the cancel itself has been outstanding past the ceiling.
+                _run_phase = str((raw or {}).get("phase") or "").strip()
+                if _run_phase == "cancelling":
+                    try:
+                        _age_anchor = float(
+                            (raw or {}).get("cancelled_at") or started_at or 0
+                        )
+                    except (TypeError, ValueError):
+                        _age_anchor = started_at
+                else:
+                    _age_anchor = started_at
                 # 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 _age_anchor and (now - _age_anchor) > 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

api/streaming.py

@@ -37,6 +37,9 @@ from api.config import (
     _set_thread_env, _clear_thread_env,
     register_active_run, update_active_run, unregister_active_run,
     unregister_stream_owner,
+    stream_owner_session_id,
+    session_writeback_owner,
+    clear_session_writeback_owner_if_owned,
     SESSION_AGENT_LOCKS, SESSION_AGENT_LOCKS_LOCK,
     resolve_model_provider,
     resolve_custom_provider_connection,
@@ -1968,8 +1971,149 @@ def _cleanup_ephemeral_cancelled_turn(session) -> None:
         logger.debug("Failed to clean up ephemeral cancelled session", exc_info=True)
 
 
-def _finalize_cancelled_turn(session, *, ephemeral: bool = False, message: str = 'Task cancelled.') -> None:
-    """Finalize a cancelled turn for persistent or ephemeral sessions."""
+def _resolve_current_session_for_write(session):
+    """Resolve the CURRENT session object for a delayed-cancel write.
+
+    The worker thread holds a snapshot ``Session`` object captured when the
+    turn started. ``cancel_stream()`` clears ``active_stream_id`` on the
+    object it resolves and saves it, after which the session is eligible for
+    LRU eviction; a successor turn may then be admitted through a DISTINCT
+    object lazily reloaded by ``get_session()``. Writing to the worker's
+    snapshot after that point serializes stale state over the successor
+    (#6623 re-gate), so delayed-cancel writes MUST target the object that
+    ``get_session()`` currently resolves under the canonical per-session
+    lock.
+
+    Returns ``None`` (fail closed) when the current object cannot be
+    resolved — missing/deleted session, transient error, or no session id.
+    Callers MUST no-op on ``None`` and must NEVER fall back to the
+    worker-held snapshot: that object may be a detached LRU-evicted
+    generation whose save would overwrite the current session state.
+    """
+    sid = getattr(session, "session_id", None)
+    if not sid:
+        return None
+    try:
+        return get_session(sid)
+    except Exception:
+        logger.debug(
+            "Failed to resolve current session %s for delayed-cancel write; "
+            "failing closed (no write)",
+            sid,
+            exc_info=True,
+        )
+        return None
+
+
+def _merge_process_wakeup_pause_into_current_session(
+    session,
+    *,
+    classification: str,
+    model=None,
+    provider=None,
+) -> dict | None:
+    """Record a process-wakeup provider-unavailable pause on the CURRENT session.
+
+    The worker-held ``session`` may be a detached snapshot (LRU-evicted and
+    replaced by a distinct object through which a successor was admitted and
+    saved). The pause is session-wide suppression metadata that must survive
+    regardless of stream ownership, so it is merged into the current object —
+    resolved under the canonical per-session lock — and saved there. Saving
+    the worker's snapshot instead would serialize stale state over the
+    successor even when the generation-guarded finalizer later no-ops
+    (#6623 re-gate). If the current object cannot be resolved the pause is
+    skipped (fail closed) — it is never written through the worker-held
+    snapshot.
+    """
+    current = _resolve_current_session_for_write(session)
+    if current is None:
+        logger.info(
+            "Skipping process-wakeup pause for unresolvable session %s "
+            "(fail closed — no detached-snapshot save)",
+            getattr(session, "session_id", None),
+        )
+        return None
+    recorded = record_process_wakeup_provider_unavailable_pause(
+        current,
+        classification=classification,
+        model=model,
+        provider=provider,
+    )
+    if recorded:
+        try:
+            current.save(touch_updated_at=False)
+        except Exception:
+            logger.debug("Failed to save process-wakeup pause", exc_info=True)
+    return recorded
+
+
+def _finalize_cancelled_turn(
+    session,
+    *,
+    ephemeral: bool = False,
+    message: str = 'Task cancelled.',
+    stream_id: str | None = None,
+) -> None:
+    """Finalize a cancelled turn for persistent or ephemeral sessions.
+
+    Generation-guarded (#6623 re-gate): when ``stream_id`` is provided, the
+    finalizer only acts while the session has not advanced past that stream.
+    The worker-held ``session`` object is only a snapshot: cancel_stream()
+    clears ``active_stream_id`` eagerly, and the session may have been
+    LRU-evicted and lazily reloaded as a DISTINCT object through which a
+    successor turn was admitted and saved. Finalization authority is therefore
+    bound to the per-session writeback-ownership record
+    (``SESSION_WRITEBACK_OWNERS``), which survives cancel cleanup, AND to the
+    CURRENT session object resolved by canonical session id under the
+    per-session agent lock (callers hold it). A successor owns the session now
+    and a delayed finalizer from the old worker MUST no-op against it instead
+    of clearing pending fields, materializing the pending user turn, appending
+    markers, unlinking the session, or saving. ``active_stream_id is None`` on
+    a worker-held snapshot is NOT proof that no successor exists — cancel
+    cleanup clears that field eagerly. And a MISSING ownership record is NOT
+    proof that the session is idle: the successor's own teardown clears the
+    record when it completes, so ``owner is None`` must fail closed too.
+    """
+    if stream_id:
+        session_id = getattr(session, "session_id", None)
+        current = _resolve_current_session_for_write(session) if session_id else None
+        if current is None:
+            logger.info(
+                "Skipping stale cancelled-turn finalize for session %s stream %s; "
+                "current session cannot be resolved (fail closed)",
+                session_id,
+                stream_id,
+            )
+            return
+        # Immutable stream/generation authority: the writeback-ownership record
+        # survives cancel cleanup and is replaced only by a successor admission,
+        # so authority must EQUAL this worker's exact stream_id. ``None`` —
+        # successor completed and its teardown cleared the entry, record never
+        # registered, or already reaped/deleted — must FAIL CLOSED: it is not
+        # proof that no successor exists, and the old worker would otherwise
+        # append/save its obsolete cancellation over the completed successor.
+        owner = session_writeback_owner(session_id) if session_id else None
+        if owner != stream_id:
+            logger.info(
+                "Skipping stale cancelled-turn finalize for session %s stream %s; "
+                "writeback owner=%r (missing, replaced, or unresolvable — fail closed)",
+                session_id,
+                stream_id,
+                owner,
+            )
+            return
+        _current = getattr(current, "active_stream_id", None)
+        if _current is not None and _current != stream_id:
+            logger.info(
+                "Skipping stale cancelled-turn finalize for session %s stream %s; "
+                "active_stream_id=%s (successor owns the writeback)",
+                session_id,
+                stream_id,
+                _current,
+            )
+            return
+        # Finalize against the CURRENT object — never the worker's snapshot.
+        session = current
     if ephemeral:
         _cleanup_ephemeral_cancelled_turn(session)
         return
@@ -7633,6 +7777,13 @@ def _run_agent_streaming(
         # already registered the stream owner, so release it here to avoid
         # leaking a STREAM_SESSION_OWNERS entry that the teardown finally never sees.
         unregister_stream_owner(stream_id)
+        try:
+            clear_session_writeback_owner_if_owned(session_id, stream_id)
+        except Exception:
+            logger.debug(
+                "Failed to clear session writeback owner for stream %s", stream_id,
+                exc_info=True,
+            )
         return
     register_active_run(
         stream_id,
@@ -8125,7 +8276,7 @@ def _run_agent_streaming(
         # Check for pre-flight cancel (user cancelled before agent even started)
         if cancel_event.is_set():
             with _agent_lock:
-                _finalize_cancelled_turn(s, ephemeral=ephemeral, message='Task cancelled before start.')
+                _finalize_cancelled_turn(s, ephemeral=ephemeral, message='Task cancelled before start.', stream_id=stream_id)
             put('cancel', _cancel_event_payload('Cancelled before start'))
             return
 
@@ -9407,7 +9558,7 @@ def _run_agent_streaming(
                     except Exception:
                         logger.debug("Failed to interrupt agent before start")
                     with _agent_lock:
-                        _finalize_cancelled_turn(s, ephemeral=ephemeral, message='Task cancelled before start.')
+                        _finalize_cancelled_turn(s, ephemeral=ephemeral, message='Task cancelled before start.', stream_id=stream_id)
                     put('cancel', _cancel_event_payload('Cancelled by user'))
                     return
 
@@ -9618,10 +9769,11 @@ def _run_agent_streaming(
                 if _ckpt_thread is not None:
                     _ckpt_thread.join(timeout=15)
                 if ephemeral:
-                    _cleanup_ephemeral_cancelled_turn(s)
+                    with _agent_lock:
+                        _finalize_cancelled_turn(s, ephemeral=True, stream_id=stream_id)
                 else:
                     with _agent_lock:
-                        _finalize_cancelled_turn(s, ephemeral=False)
+                        _finalize_cancelled_turn(s, ephemeral=False, stream_id=stream_id)
                         try:
                             append_turn_journal_event_for_stream(
                                 s.session_id,
@@ -9663,7 +9815,7 @@ def _run_agent_streaming(
                 _ckpt_thread.join(timeout=15)
             if cancel_event.is_set():
                 with _agent_lock:
-                    _finalize_cancelled_turn(s, ephemeral=False)
+                    _finalize_cancelled_turn(s, ephemeral=False, stream_id=stream_id)
                     try:
                         append_turn_journal_event_for_stream(
                             s.session_id,
@@ -9724,7 +9876,7 @@ def _run_agent_streaming(
                         if isinstance(result, dict):
                             result = {**result, 'messages': _result_messages}
                     if cancel_event.is_set():
-                        _finalize_cancelled_turn(s, ephemeral=False)
+                        _finalize_cancelled_turn(s, ephemeral=False, stream_id=stream_id)
                         try:
                             append_turn_journal_event_for_stream(
                                 s.session_id,
@@ -9976,7 +10128,7 @@ def _run_agent_streaming(
                 # _token_sent tracks whether on_token() was called (any streamed text)
                 if _terminal_failure or (not _assistant_added and not _token_sent):
                     if cancel_event.is_set():
-                        _finalize_cancelled_turn(s, ephemeral=ephemeral)
+                        _finalize_cancelled_turn(s, ephemeral=ephemeral, stream_id=stream_id)
                         if not ephemeral:
                             try:
                                 append_turn_journal_event_for_stream(
@@ -10637,7 +10789,7 @@ def _run_agent_streaming(
                         except Exception:
                             logger.debug("Failed to append assistant_started turn journal event", exc_info=True)
                 if cancel_event.is_set():
-                    _finalize_cancelled_turn(s, ephemeral=False)
+                    _finalize_cancelled_turn(s, ephemeral=False, stream_id=stream_id)
                     try:
                         append_turn_journal_event_for_stream(
                             s.session_id,
@@ -10655,7 +10807,7 @@ def _run_agent_streaming(
                 with _stream_writeback_stage(_writeback_timings, "session_save"):
                     s.save()
                 if cancel_event.is_set():
-                    _finalize_cancelled_turn(s, ephemeral=False)
+                    _finalize_cancelled_turn(s, ephemeral=False, stream_id=stream_id)
                     try:
                         append_turn_journal_event_for_stream(
                             s.session_id,
@@ -10759,7 +10911,7 @@ def _run_agent_streaming(
             _lock_ctx = _agent_lock if _agent_lock is not None else contextlib.nullcontext()
             with _lock_ctx:
                 if cancel_event.is_set():
-                    _finalize_cancelled_turn(s, ephemeral=False)
+                    _finalize_cancelled_turn(s, ephemeral=False, stream_id=stream_id)
                     try:
                         append_turn_journal_event_for_stream(
                             s.session_id,
@@ -10791,7 +10943,7 @@ def _run_agent_streaming(
                             s.save(touch_updated_at=False)
                         except Exception:
                             logger.debug("Failed to persist restored process wakeup pause", exc_info=True)
-                        _finalize_cancelled_turn(s, ephemeral=False)
+                        _finalize_cancelled_turn(s, ephemeral=False, stream_id=stream_id)
                         try:
                             append_turn_journal_event_for_stream(
                                 s.session_id,
@@ -10814,7 +10966,7 @@ def _run_agent_streaming(
                             s.save(touch_updated_at=False)
                         except Exception:
                             logger.debug("Failed to persist restored process wakeup pause", exc_info=True)
-                        _finalize_cancelled_turn(s, ephemeral=False)
+                        _finalize_cancelled_turn(s, ephemeral=False, stream_id=stream_id)
                         try:
                             append_turn_journal_event_for_stream(
                                 s.session_id,
@@ -11177,13 +11329,22 @@ def _run_agent_streaming(
                         and _turn_pending_source == 'process_wakeup'
                         and _exc_is_credential_pool_empty
                     ):
-                        record_process_wakeup_provider_unavailable_pause(
+                        # Merge the pause into the CURRENT session object under
+                        # the canonical lock. The worker-held ``s`` may be a
+                        # detached snapshot (LRU-evicted + replaced by a
+                        # distinct object through which a successor was
+                        # admitted); saving it would serialize stale state over
+                        # the successor even though the generation-guarded
+                        # finalizer below later no-ops. The pause is
+                        # session-wide suppression metadata that must survive
+                        # regardless of stream ownership (#6623 re-gate).
+                        _wakeup_pause_recorded = _merge_process_wakeup_pause_into_current_session(
                             s,
                             classification=_classification['type'],
                             model=_turn_route_model,
                             provider=_turn_route_provider,
                         )
-                    _finalize_cancelled_turn(s, ephemeral=ephemeral)
+                    _finalize_cancelled_turn(s, ephemeral=ephemeral, stream_id=stream_id)
                     if not ephemeral:
                         try:
                             append_turn_journal_event_for_stream(
@@ -11352,21 +11513,17 @@ def _run_agent_streaming(
             with _lock_ctx:
                 if not ephemeral and not _stream_writeback_is_current(s, stream_id):
                     if _turn_pending_source == 'process_wakeup':
-                        _pause = record_process_wakeup_provider_unavailable_pause(
+                        # #6623 re-gate: merge the pause into the CURRENT
+                        # session object under the canonical lock — never save
+                        # the worker's detached snapshot. The helper fails
+                        # closed (no write) when the current object cannot be
+                        # resolved.
+                        _merge_process_wakeup_pause_into_current_session(
                             s,
                             classification=_exc_type,
                             model=_turn_route_model,
                             provider=_turn_route_provider,
                         )
-                        if _pause is not None:
-                            try:
-                                s.save(touch_updated_at=False)
-                            except Exception:
-                                logger.debug(
-                                    "Failed to persist stale-stream process_wakeup pause for session %s",
-                                    getattr(s, 'session_id', session_id),
-                                    exc_info=True,
-                                )
                     logger.info(
                         "Skipping stale stream error writeback for session %s stream %s; active_stream_id=%s",
                         getattr(s, 'session_id', session_id),
@@ -11512,6 +11669,16 @@ def _run_agent_streaming(
             # Clean up the stream-owner registry so stale stream_id→session_id
             # mappings do not accumulate over thousands of completed streams (#6351).
             unregister_stream_owner(stream_id)
+            # Release the session's writeback-ownership entry only while this
+            # stream still owns it (#6623 re-gate): a successor admitted after
+            # cancel must keep its registry claim.
+            try:
+                clear_session_writeback_owner_if_owned(session_id, stream_id)
+            except Exception:
+                logger.debug(
+                    "Failed to clear session writeback owner for stream %s", stream_id,
+                    exc_info=True,
+                )
             # NOTE: do NOT discard PENDING_GOAL_CONTINUATION here. The marker
             # is set by goal_continue (line ~3328) inside the SAME function
             # call and consumed atomically by `_start_chat_stream_for_session`
@@ -11721,6 +11888,7 @@ def cancel_stream(stream_id: str) -> bool:
     _snap_tool_calls = None
     _snap_flag = None
     _snap_agent = None
+    _snap_owner_session_id = None
     _cancel_session_payload = None
 
     with streams_lock:
@@ -11732,6 +11900,16 @@ def cancel_stream(stream_id: str) -> bool:
         # cancel must snapshot them too or it loses the already-streamed text.
         _snap_flag = cancel_flags.get(stream_id)
         _snap_agent = agent_instances.get(stream_id)
+        # Capture the stream owner WHILE the stream still exists (#6623). The
+        # just-starting worker takes its `q is None -> unregister_stream_owner`
+        # early path (api/streaming.py `_run_agent_streaming` and
+        # api/gateway_chat.py `_run_gateway_chat_streaming`) the instant
+        # STREAMS[stream_id] is popped below, so reading the owner AFTER the
+        # pop can race that teardown and yield None — leaving the session's
+        # active_stream_id/pending_* stuck while cancel still returns True.
+        # Reading it here, under streams_lock and before any pop, gives cancel
+        # a stable owner to resolve session cleanup against.
+        _snap_owner_session_id = stream_owner_session_id(stream_id)
         _snap_partial_text = partial_texts.get(stream_id, '')
         if not _snap_partial_text:
             _live_partials = getattr(_live_config, 'STREAM_PARTIAL_TEXT', partial_texts)
@@ -11770,8 +11948,10 @@ def cancel_stream(stream_id: str) -> bool:
 
     # 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")
+    # health polling sees during that detached window. Stamp cancelled_at so
+    # _clear_stale_stream_state() can eventually reclaim the session if the
+    # worker is stuck in C-level I/O and never reaches its finally (#6623).
+    update_active_run(stream_id, phase="cancelling", cancelled_at=time.time())
 
     # 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).
@@ -11847,6 +12027,15 @@ def cancel_stream(stream_id: str) -> bool:
     _cancel_session_id = getattr(agent, 'session_id', None) if agent else None
     if not _cancel_session_id and active_run_session_id:
         _cancel_session_id = active_run_session_id
+    # Third fallback: stream owner registry — populated before the worker
+    # thread starts, so it's always available even for early cancels that
+    # race ahead of AGENT_INSTANCES and ACTIVE_RUNS (#6623). The owner is
+    # read UNDER streams_lock above (while STREAMS[stream_id] still exists),
+    # NOT here after the eager pop: the just-starting worker unregisters the
+    # owner the instant the stream map entry disappears, so a post-pop lookup
+    # would race that teardown and return None.
+    if not _cancel_session_id and _snap_owner_session_id:
+        _cancel_session_id = _snap_owner_session_id
     # Use the snapshots captured under streams_lock above (the worker's finally
     # may have popped the live buffers by now via agent.interrupt()). For the
     # ACTIVE_RUNS-only path (stream absent) the snapshots are None → fall back to

tests/test_issue6623_cancel_owner_race.py

@@ -0,0 +1,970 @@
+"""Regression tests for #6623: early-Stop cancel race + stale cancelled worker.
+
+Covers the two CHANGES_REQUESTED items on PR #6636:
+
+1. ``cancel_stream()`` must capture the stream owner registry entry WHILE the
+   stream still exists (under ``STREAMS_LOCK``, before the eager
+   ``STREAMS.pop``). The just-starting worker takes its
+   ``q is None -> unregister_stream_owner`` early path the instant the stream
+   map entry disappears, so a post-pop owner lookup races that teardown and
+   returns None — leaving ``session.active_stream_id`` / pending_* stuck while
+   the HTTP cancel path still reports success.
+
+2. A worker stuck in C-level I/O may never reach its ``finally`` to unregister
+   the run, so ``ACTIVE_RUNS`` can hold the row forever and
+   ``_clear_stale_stream_state()`` defers stale cleanup indefinitely. The fix
+   stamps ``cancelled_at`` on the run when cancel_stream() flips it to
+   phase="cancelling", and ``_clear_stale_stream_state()`` reclaims the session
+   once the cancel has been outstanding past the grace window.
+
+3. (RE-GATE 20:36) A delayed cancel finalizer must not be authorized by a
+   MISSING writeback-ownership record: the successor's own teardown clears
+   ``SESSION_WRITEBACK_OWNERS[session_id]`` when it completes, so ``owner is
+   None`` with ``active_stream_id is None`` again is exactly the ambiguous
+   state that lets an obsolete finalizer serialize over the completed
+   successor. ``None`` — and any unresolvable current session — must fail
+   closed. The stale-stream process-wakeup branch must likewise merge the
+   pause into the canonical current session, never save the worker's detached
+   snapshot.
+"""
+import queue
+import threading
+import time
+
+import pytest
+
+import api.config as config
+import api.models as models
+import api.streaming as streaming
+from api.models import Session
+from unittest.mock import Mock, patch
+
+
+@pytest.fixture(autouse=True)
+def _isolate_sessions(tmp_path, monkeypatch):
+    session_dir = tmp_path / "sessions"
+    session_dir.mkdir()
+    index_file = session_dir / "_index.json"
+    monkeypatch.setattr(models, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(models, "SESSION_INDEX_FILE", index_file)
+    monkeypatch.setattr(streaming, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(config, "SESSION_INDEX_FILE", index_file, raising=False)
+    models.SESSIONS.clear()
+    config.STREAMS.clear()
+    config.CANCEL_FLAGS.clear()
+    config.AGENT_INSTANCES.clear()
+    config.ACTIVE_RUNS.clear()
+    config.STREAM_SESSION_OWNERS.clear()
+    config.SESSION_WRITEBACK_OWNERS.clear()
+    config.SESSION_AGENT_LOCKS.clear()
+    yield
+    models.SESSIONS.clear()
+    config.STREAMS.clear()
+    config.CANCEL_FLAGS.clear()
+    config.AGENT_INSTANCES.clear()
+    config.ACTIVE_RUNS.clear()
+    config.STREAM_SESSION_OWNERS.clear()
+    config.SESSION_WRITEBACK_OWNERS.clear()
+    config.SESSION_AGENT_LOCKS.clear()
+
+
+class _PoppingStreams(dict):
+    """STREAMS stand-in that retires the stream owner the instant the stream
+    entry is popped — exactly what the just-starting worker does in its
+    ``q is None -> unregister_stream_owner`` early path once cancel_stream()
+    eagerly pops ``STREAMS[stream_id]``."""
+
+    def pop(self, key, *default):
+        value = dict.pop(self, key, *default)
+        config.unregister_stream_owner(key)
+        return value
+
+
+def test_issue6623_owner_must_be_captured_before_stream_pop(tmp_path, monkeypatch):
+    """Deterministic repro of the nesquena-hermes interleaving:
+
+    1. Stop sees the stream while AGENT_INSTANCES and ACTIVE_RUNS are empty.
+    2. Stop removes STREAMS[stream_id] (via the _PoppingStreams wrapper, which
+       synchronously retires the owner exactly like the worker's early path).
+    3. Stop then resolves the owner — the post-pop lookup would return None,
+       skipping session cleanup while cancel still returns True.
+
+    The fix must capture the owner under the lock BEFORE the pop, so session
+    cleanup still runs (get_session called exactly once) and the persisted
+    active_stream_id / pending fields are cleared.
+    """
+    sid = "sess_owner_race"
+    stream_id = "stream_owner_race"
+
+    s = Session(session_id=sid, title="Owner race", messages=[])
+    s.active_stream_id = stream_id
+    s.pending_user_message = "hello"
+    s.pending_attachments = []
+    s.pending_started_at = 1234567890.0
+    s.save = Mock()
+    models.SESSIONS[sid] = s
+
+    # Real owner registration, exactly like the route layer before worker start.
+    config.register_stream_owner(stream_id, sid)
+
+    # STREAMS present; NO AGENT_INSTANCES, NO ACTIVE_RUNS (early-Stop race).
+    wrapper = _PoppingStreams()
+    wrapper[stream_id] = queue.Queue()
+    monkeypatch.setattr(config, "STREAMS", wrapper)
+    monkeypatch.setattr(streaming, "STREAMS", wrapper)
+    config.CANCEL_FLAGS[stream_id] = threading.Event()
+
+    with patch("api.streaming.get_session", return_value=s) as m_get_session:
+        result = streaming.cancel_stream(stream_id)
+
+    assert result is True
+    assert m_get_session.call_count == 1, (
+        f"Expected 'get_session' to be called once. Called {m_get_session.call_count} times. "
+        "The stream owner must be captured BEFORE the STREAMS pop."
+    )
+    assert s.active_stream_id is None
+    assert s.pending_user_message is None
+    assert s.pending_attachments == []
+    assert s.pending_started_at is None
+    s.save.assert_called_once()
+    # No owner leak: the simulated worker teardown retired the registry entry,
+    # and cancel captured the owner before that teardown could hide it.
+    assert config.STREAM_SESSION_OWNERS.get(stream_id) is None
+
+
+def test_issue6623_newer_stream_stale_writeback_still_rejected(tmp_path, monkeypatch):
+    """Control: with the owner captured before the pop, a session whose
+    active_stream_id has rotated to a NEWER stream must still be left alone —
+    _stream_writeback_is_current() keeps rejecting the stale writeback, no
+    cancel marker is appended, and no save happens."""
+    sid = "sess_owner_rotated"
+    stream_id = "old-stream-owner-race"
+
+    s = Session(
+        session_id=sid,
+        title="Rotated stream",
+        messages=[{"role": "user", "content": "newer prompt"}],
+    )
+    s.active_stream_id = "newer-stream"
+    s.pending_user_message = "newer prompt"
+    s.pending_started_at = 456.0
+    s.save = Mock()
+    models.SESSIONS[sid] = s
+
+    config.register_stream_owner(stream_id, sid)
+
+    wrapper = _PoppingStreams()
+    wrapper[stream_id] = queue.Queue()
+    monkeypatch.setattr(config, "STREAMS", wrapper)
+    monkeypatch.setattr(streaming, "STREAMS", wrapper)
+    config.CANCEL_FLAGS[stream_id] = threading.Event()
+
+    with patch("api.streaming.get_session", return_value=s) as m_get_session:
+        result = streaming.cancel_stream(stream_id)
+
+    assert result is True
+    assert m_get_session.call_count == 1
+    assert s.active_stream_id == "newer-stream"
+    assert s.pending_user_message == "newer prompt"
+    s.save.assert_not_called()
+    assert all(
+        str(m.get("content", "")) != "*Task cancelled.*" for m in s.messages
+    ), "stale cancel writeback must still be rejected for a newer stream"
+    assert config.STREAM_SESSION_OWNERS.get(stream_id) is None
+
+
+def test_issue6623_stale_cancelled_run_cleared_after_grace(tmp_path, monkeypatch):
+    """A cancelled run whose cancel has been outstanding past the grace window
+    is treated as stale: _clear_stale_stream_state() clears the session even
+    though the worker row is still in ACTIVE_RUNS (worker stuck in C-level I/O
+    never reached its finally)."""
+    import api.routes as routes
+
+    sid = "stale_cancelled_sid"
+    stream_id = "stale-cancelled-stream"
+    s = Session(session_id=sid, title="Stale cancelled", messages=[])
+    s.active_stream_id = stream_id
+    s.save()
+    models.SESSIONS[sid] = s
+
+    config.register_active_run(
+        stream_id,
+        session_id=sid,
+        phase="cancelling",
+        cancelled_at=time.time() - 120.0,
+    )
+
+    assert routes._clear_stale_stream_state(s) is True
+    assert s.active_stream_id is None
+
+
+def test_issue6623_stale_cancelled_run_without_cancelled_at_reclaimed_via_started_at(
+    tmp_path, monkeypatch
+):
+    """Legacy run cancelled before the cancelled_at stamp existed: the
+    started_at anchor still reclaims the session once the run is old enough."""
+    import api.routes as routes
+
+    sid = "legacy_stale_cancelled_sid"
+    stream_id = "legacy-stale-cancelled-stream"
+    s = Session(session_id=sid, title="Legacy stale cancelled", messages=[])
+    s.active_stream_id = stream_id
+    s.save()
+    models.SESSIONS[sid] = s
+
+    config.register_active_run(stream_id, session_id=sid, phase="cancelling")
+    with config.ACTIVE_RUNS_LOCK:
+        config.ACTIVE_RUNS[stream_id]["started_at"] = time.time() - 120.0
+
+    assert routes._clear_stale_stream_state(s) is True
+    assert s.active_stream_id is None
+
+
+def test_issue6623_fresh_cancelled_run_still_deferred(tmp_path, monkeypatch):
+    """Control: a recently cancelled run (inside the grace window) must still
+    defer stale cleanup — the worker may legitimately be unwinding."""
+    import api.routes as routes
+
+    sid = "fresh_cancelled_sid"
+    stream_id = "fresh-cancelled-stream"
+    s = Session(session_id=sid, title="Fresh cancelled", messages=[])
+    s.active_stream_id = stream_id
+    s.save()
+    models.SESSIONS[sid] = s
+
+    config.register_active_run(
+        stream_id,
+        session_id=sid,
+        phase="cancelling",
+        cancelled_at=time.time() - 5.0,
+    )
+
+    assert routes._clear_stale_stream_state(s) is False
+    assert s.active_stream_id == stream_id
+
+
+def test_issue6623_delayed_cancel_finalizer_gated_by_stream_ownership():
+    """RE-GATE unit control (20:36): ``_finalize_cancelled_turn(..., stream_id=...)``
+    must no-op entirely (no clearing, no marker append, no save) unless the
+    immutable writeback-ownership record still names the old worker's exact
+    stream_id. ``None`` — successor completed and its teardown cleared the
+    entry, or the record was never registered — must FAIL CLOSED, and so must
+    an unresolvable current session (deleted/unloadable)."""
+    old_stream = "finalizer-old-stream"
+
+    # Successor owns the session -> the delayed finalizer must no-op.
+    s = Session(
+        session_id="finalizer-gate-successor",
+        messages=[{"role": "user", "content": "newer prompt"}],
+    )
+    s.active_stream_id = "newer-stream"
+    s.pending_user_message = "newer prompt"
+    s.pending_started_at = 456.0
+    s.save = Mock()
+    models.SESSIONS[s.session_id] = s
+    config.register_session_writeback_owner(s.session_id, "newer-stream")
+    streaming._finalize_cancelled_turn(s, ephemeral=False, stream_id=old_stream)
+    assert s.active_stream_id == "newer-stream"
+    assert s.pending_user_message == "newer prompt"
+    s.save.assert_not_called()
+    assert streaming._session_has_cancel_marker(s) is False
+
+    # Ownership record missing (successor completed and its teardown cleared
+    # it) and active_stream_id is None again -> ``None`` must NOT authorize:
+    # the finalizer no-ops instead of appending/saving the obsolete
+    # cancellation over the completed successor.
+    s2 = Session(session_id="finalizer-gate-nostream", messages=[])
+    s2.active_stream_id = None
+    s2.save = Mock()
+    models.SESSIONS[s2.session_id] = s2
+    streaming._finalize_cancelled_turn(s2, ephemeral=False, stream_id=old_stream)
+    s2.save.assert_not_called()
+    assert s2.active_stream_id is None
+    assert streaming._session_has_cancel_marker(s2) is False
+
+    # Current session cannot be resolved (not cached, no file on disk) ->
+    # fail closed: no terminal write from a detached worker.
+    s4 = Session(session_id="finalizer-gate-unresolvable", messages=[])
+    s4.active_stream_id = None
+    s4.save = Mock()
+    streaming._finalize_cancelled_turn(s4, ephemeral=False, stream_id=old_stream)
+    s4.save.assert_not_called()
+    assert streaming._session_has_cancel_marker(s4) is False
+
+    # Session still points at the cancelled stream AND the worker still owns
+    # the writeback record -> the finalizer runs.
+    s3 = Session(session_id="finalizer-gate-owns", messages=[])
+    s3.active_stream_id = old_stream
+    s3.save = Mock()
+    models.SESSIONS[s3.session_id] = s3
+    config.register_session_writeback_owner(s3.session_id, old_stream)
+    streaming._finalize_cancelled_turn(s3, ephemeral=False, stream_id=old_stream)
+    s3.save.assert_called_once()
+    assert s3.active_stream_id is None
+    assert streaming._session_has_cancel_marker(s3) is True
+
+
+def test_issue6623_recent_cancel_not_reaped_despite_old_started_at(tmp_path, monkeypatch):
+    """RE-GATE control for the cancellation-age anchor: a just-cancelled turn
+    whose ORIGINAL started_at is far past the 180s unwind ceiling must NOT be
+    reaped (and a successor admitted) — cancel_stream() removes STREAMS
+    itself, so absence from STREAMS is not proof the worker is dead. The
+    reaping anchor for a phase="cancelling" row is the cancel time."""
+    import api.routes as routes
+
+    sid = "recent_cancel_old_start"
+    old_stream = "recent-cancel-old-start-stream"
+    config.register_active_run(
+        old_stream,
+        session_id=sid,
+        started_at=time.time() - 400.0,
+        phase="cancelling",
+        cancelled_at=time.time() - 5.0,
+    )
+
+    assert routes._active_run_stream_for_session(sid) == old_stream
+    assert old_stream in config.ACTIVE_RUNS
+
+
+def test_issue6623_stale_recovery_successor_survives_delayed_cancel_finalizer(
+    tmp_path, monkeypatch
+):
+    """RE-GATE production-composed regression (#6623): a long-running worker
+    that is cancelled, reaped by stale recovery, and superseded must not
+    clobber the successor when it finally returns from the provider boundary
+    and unwinds through the delayed cancel finalizer.
+
+    Sequence, all through real production code paths:
+
+    1. Old turn owns the session with an ACTIVE_RUNS row whose original
+       started_at is long past the 180s unwind ceiling.
+    2. cancel_stream() pops STREAMS, stamps phase="cancelling" + cancelled_at,
+       clears the session, writes the cancel marker, and saves.
+    3. The cancel stays outstanding past the 180s ceiling (worker stuck in
+       provider I/O) -> _active_run_stream_for_session() reaps the row and a
+       successor turn is admitted (active_stream_id/pending_* rotated, newer
+       user message appended, persisted).
+    4. The old worker is released from the provider boundary and runs the
+       delayed cancel finalizer under the session lock — exactly the call
+       sites at api/streaming.py:9686/9747.
+    5. The successor's active_stream_id, pending fields, messages, and saved
+       state must all survive.
+    """
+    import api.routes as routes
+
+    sid = "sess_stale_successor"
+    old_stream = "old-stale-cancelled-stream"
+    newer_stream = "newer-successor-stream"
+
+    s = Session(
+        session_id=sid,
+        title="Stale successor",
+        messages=[
+            {"role": "user", "content": "old prompt"},
+            {"role": "assistant", "content": "partial old answer"},
+        ],
+    )
+    s.active_stream_id = old_stream
+    s.pending_user_message = "old prompt"
+    s.pending_attachments = []
+    s.pending_started_at = 1000.0
+    s.pending_user_source = "webui"
+    s.save()
+    models.SESSIONS[sid] = s
+
+    config.register_stream_owner(old_stream, sid)
+    config.STREAMS[old_stream] = queue.Queue()
+    config.CANCEL_FLAGS[old_stream] = threading.Event()
+    config.register_active_run(
+        old_stream,
+        session_id=sid,
+        started_at=time.time() - 400.0,  # original run far past the 180s ceiling
+        phase="running",
+    )
+
+    # 2) Cancel through the real production path.
+    with patch("api.streaming.get_session", return_value=s):
+        assert streaming.cancel_stream(old_stream) is True
+    assert s.active_stream_id is None
+    assert streaming._session_has_cancel_marker(s)
+
+    # 3) Stale recovery: the cancel has been outstanding past the unwind
+    # ceiling (stuck worker never reached its finally) -> the run row is
+    # reaped and the successor may be admitted.
+    with config.ACTIVE_RUNS_LOCK:
+        config.ACTIVE_RUNS[old_stream]["cancelled_at"] = time.time() - 200.0
+    assert routes._active_run_stream_for_session(sid) is None
+    assert old_stream not in config.ACTIVE_RUNS
+
+    # Admit the successor the way the route layer does: under the session
+    # lock, rotate active_stream_id/pending_*, append the newer user turn,
+    # and persist.
+    _lock = streaming._get_session_agent_lock(sid)
+    with _lock:
+        s.active_stream_id = newer_stream
+        s.pending_user_message = "newer prompt"
+        s.pending_attachments = []
+        s.pending_started_at = 2000.0
+        s.pending_user_source = "webui"
+        s.messages.append(
+            {"role": "user", "content": "newer prompt", "timestamp": 2000}
+        )
+        s.save()
+    _messages_before_release = len(s.messages)
+
+    # 4) Release the old worker from the provider boundary; it unwinds through
+    # the delayed cancel finalizer under the session lock.
+    _errors = []
+    _release = threading.Event()
+
+    def _old_worker_return():
+        _release.wait()  # the provider boundary the worker was blocked in
+        try:
+            with _lock:
+                streaming._finalize_cancelled_turn(
+                    s, ephemeral=False, stream_id=old_stream
+                )
+        except Exception as exc:  # pragma: no cover - failure surface
+            _errors.append(exc)
+
+    _worker = threading.Thread(target=_old_worker_return, daemon=True)
+    _worker.start()
+    _release.set()
+    _worker.join(timeout=10)
+    assert not _worker.is_alive(), "old worker thread did not unwind"
+    assert not _errors
+
+    # 5) The successor's state must survive untouched.
+    assert s.active_stream_id == newer_stream
+    assert s.pending_user_message == "newer prompt"
+    assert s.pending_attachments == []
+    assert s.pending_started_at == 2000.0
+    assert s.pending_user_source == "webui"
+    assert len(s.messages) == _messages_before_release, (
+        "delayed cancel finalizer must not append anything over the successor turn"
+    )
+    assert any(
+        str(m.get("content", "")) == "newer prompt" and m.get("role") == "user"
+        for m in s.messages
+    )
+    # The persisted state survives on disk too.
+    disk = models.Session.load(sid)
+    assert disk.active_stream_id == newer_stream
+    assert disk.pending_user_message == "newer prompt"
+    assert disk.pending_started_at == 2000.0
+    assert disk.pending_user_source == "webui"
+    assert any(
+        str(m.get("content", "")) == "newer prompt" and m.get("role") == "user"
+        for m in disk.messages
+    )
+
+
+def test_issue6623_writeback_owner_released_only_while_still_owned():
+    """RE-GATE unit: the writeback-ownership record is replaced by a successor
+    admission, and a worker's teardown clears it ONLY while it still owns it —
+    the old worker's finally must never erase the successor's claim."""
+    config.register_session_writeback_owner("sess_owner_release", "stream-a")
+    # A successor replaced the claim; the old worker's finally must not clear it.
+    config.register_session_writeback_owner("sess_owner_release", "stream-b")
+    config.clear_session_writeback_owner_if_owned("sess_owner_release", "stream-a")
+    assert config.session_writeback_owner("sess_owner_release") == "stream-b"
+    # The current owner's finally clears it.
+    config.clear_session_writeback_owner_if_owned("sess_owner_release", "stream-b")
+    assert config.session_writeback_owner("sess_owner_release") is None
+
+
+def test_issue6623_replaced_session_successor_survives_delayed_cancel_finalizer(
+    tmp_path, monkeypatch
+):
+    """RE-GATE regression (#6623): the delayed cancel finalizer must resolve
+    the CURRENT session object and bind authority to the writeback-ownership
+    record — a deterministic schedule with TWO distinct Session instances:
+
+    1. Old turn owns the session; cancel_stream() clears + saves it.
+    2. REAL LRU eviction (production ``_evict_sessions_over_cap``) drops the
+       old object; ``get_session()`` lazily reloads a DISTINCT object.
+    3. A successor turn is admitted and persisted through the replacement.
+    4. The old worker unwinds through the real cancel-finalizer path holding
+       its stale snapshot.
+    5. Cache and disk must retain the successor's fields/messages; the old
+       generation must not make a terminal write.
+
+    The pre-fix guard accepted ``active_stream_id is None`` on the worker-held
+    snapshot (cancel_stream clears it eagerly), so it proceeded to serialize
+    the stale snapshot over the successor — this test fails without the fix.
+    """
+    sid = "sess_replaced_successor"
+    old_stream = "old-replaced-stream"
+    newer_stream = "newer-replaced-stream"
+
+    s_old = Session(
+        session_id=sid,
+        title="Replaced successor",
+        messages=[
+            {"role": "user", "content": "old prompt"},
+            {"role": "assistant", "content": "partial old answer"},
+        ],
+    )
+    s_old.active_stream_id = old_stream
+    s_old.pending_user_message = "old prompt"
+    s_old.pending_attachments = []
+    s_old.pending_started_at = 1000.0
+    s_old.pending_user_source = "webui"
+    s_old.save()
+    models.SESSIONS[sid] = s_old
+
+    config.register_stream_owner(old_stream, sid)
+    config.register_session_writeback_owner(sid, old_stream)
+    config.STREAMS[old_stream] = queue.Queue()
+    config.CANCEL_FLAGS[old_stream] = threading.Event()
+    config.register_active_run(old_stream, session_id=sid, phase="running")
+
+    # 2) Cancel through the real production path.
+    assert streaming.cancel_stream(old_stream) is True
+    assert s_old.active_stream_id is None
+    assert streaming._session_has_cancel_marker(s_old)
+    _old_messages_after_cancel = len(s_old.messages)
+
+    # 3) REAL LRU eviction: drop the (now idle, persisted) old object via the
+    # production eviction pass, then lazily reload a DISTINCT object.
+    monkeypatch.setattr(models, "SESSIONS_MAX", 2)
+    for i in range(4):
+        _filler = Session(
+            session_id=f"sess_replaced_filler_{i}", title="filler", messages=[]
+        )
+        _filler.save()
+        models.SESSIONS[_filler.session_id] = _filler
+    with config.LOCK:
+        _evicted = models._evict_sessions_over_cap(0)
+    assert sid not in models.SESSIONS, (
+        f"old generation should be LRU-evicted (evicted={_evicted})"
+    )
+
+    s_new = models.get_session(sid)
+    assert s_new is not s_old, "lazy reload must yield a DISTINCT Session instance"
+
+    # 4) Admit the successor the way the route layer does: under the session
+    # lock, rotate active_stream_id/pending_*, register the writeback owner,
+    # append the newer user turn, and persist.
+    _lock = streaming._get_session_agent_lock(sid)
+    with _lock:
+        s_new.active_stream_id = newer_stream
+        s_new.pending_user_message = "newer prompt"
+        s_new.pending_attachments = []
+        s_new.pending_started_at = 2000.0
+        s_new.pending_user_source = "webui"
+        s_new.messages.append(
+            {"role": "user", "content": "newer prompt", "timestamp": 2000}
+        )
+        config.register_session_writeback_owner(sid, newer_stream)
+        s_new.save()
+    _messages_before_release = len(s_new.messages)
+
+    # 5) Release the old worker: it unwinds through the REAL cancel-finalizer
+    # path holding its STALE snapshot (s_old) under the session lock.
+    with _lock:
+        streaming._finalize_cancelled_turn(s_old, ephemeral=False, stream_id=old_stream)
+
+    # 6) The old generation made no terminal write: its in-memory snapshot
+    # must be untouched (no delayed marker appended, no stale save).
+    assert len(s_old.messages) == _old_messages_after_cancel, (
+        "stale snapshot must not receive a delayed terminal write"
+    )
+    # The successor's cache object survives untouched.
+    assert s_new.active_stream_id == newer_stream
+    assert s_new.pending_user_message == "newer prompt"
+    assert s_new.pending_attachments == []
+    assert s_new.pending_started_at == 2000.0
+    assert s_new.pending_user_source == "webui"
+    assert len(s_new.messages) == _messages_before_release, (
+        "delayed cancel finalizer must not append anything over the successor turn"
+    )
+    # The writeback-ownership record still names the successor.
+    assert config.session_writeback_owner(sid) == newer_stream
+    # The persisted state survives on disk too.
+    disk = models.Session.load(sid)
+    assert disk.active_stream_id == newer_stream
+    assert disk.pending_user_message == "newer prompt"
+    assert disk.pending_started_at == 2000.0
+    assert disk.pending_user_source == "webui"
+    assert any(
+        str(m.get("content", "")) == "newer prompt" and m.get("role") == "user"
+        for m in disk.messages
+    )
+
+
+def test_issue6623_replaced_session_completed_successor_still_protected_by_ownership_record(
+    tmp_path, monkeypatch
+):
+    """RE-GATE control (#6623): even after the successor's own turn completed
+    (so the CURRENT object's ``active_stream_id`` is None again — exactly the
+    ambiguous state the pre-fix guard accepted), the writeback-ownership record
+    still proves the session advanced past the old stream. The delayed
+    finalizer must no-op: the stale snapshot's save() would otherwise
+    overwrite the completed successor transcript on disk."""
+    sid = "sess_completed_successor"
+    old_stream = "old-completed-stream"
+    newer_stream = "newer-completed-stream"
+
+    s_old = Session(
+        session_id=sid,
+        title="Completed successor",
+        messages=[{"role": "user", "content": "old prompt"}],
+    )
+    s_old.active_stream_id = old_stream
+    s_old.pending_user_message = "old prompt"
+    s_old.pending_attachments = []
+    s_old.pending_started_at = 1000.0
+    s_old.pending_user_source = "webui"
+    s_old.save()
+    models.SESSIONS[sid] = s_old
+    config.register_session_writeback_owner(sid, old_stream)
+
+    config.register_stream_owner(old_stream, sid)
+    config.STREAMS[old_stream] = queue.Queue()
+    config.CANCEL_FLAGS[old_stream] = threading.Event()
+    config.register_active_run(old_stream, session_id=sid, phase="running")
+    assert streaming.cancel_stream(old_stream) is True
+
+    monkeypatch.setattr(models, "SESSIONS_MAX", 2)
+    for i in range(4):
+        _filler = Session(
+            session_id=f"sess_completed_filler_{i}", title="filler", messages=[]
+        )
+        _filler.save()
+        models.SESSIONS[_filler.session_id] = _filler
+    with config.LOCK:
+        models._evict_sessions_over_cap(0)
+    s_new = models.get_session(sid)
+    assert s_new is not s_old
+
+    with streaming._get_session_agent_lock(sid):
+        s_new.active_stream_id = newer_stream
+        s_new.pending_user_message = "newer prompt"
+        s_new.pending_attachments = []
+        s_new.pending_started_at = 2000.0
+        s_new.pending_user_source = "webui"
+        s_new.messages.append(
+            {"role": "user", "content": "newer prompt", "timestamp": 2000}
+        )
+        config.register_session_writeback_owner(sid, newer_stream)
+        s_new.save()
+        # The successor turn then completes normally: active_stream_id and
+        # pending fields are cleared, the transcript persists. The ownership
+        # record is NOT touched by normal completion.
+        s_new.active_stream_id = None
+        s_new.pending_user_message = None
+        s_new.pending_attachments = []
+        s_new.pending_started_at = None
+        s_new.pending_user_source = None
+        s_new.save()
+    _messages_after_successor_completion = len(s_new.messages)
+
+    with streaming._get_session_agent_lock(sid):
+        streaming._finalize_cancelled_turn(s_old, ephemeral=False, stream_id=old_stream)
+
+    assert config.session_writeback_owner(sid) == newer_stream
+    disk = models.Session.load(sid)
+    assert disk.active_stream_id is None
+    assert len(disk.messages) == _messages_after_successor_completion, (
+        "delayed finalizer must not mutate the completed successor transcript"
+    )
+    assert any(
+        str(m.get("content", "")) == "newer prompt" and m.get("role") == "user"
+        for m in disk.messages
+    )
+
+
+def test_issue6623_replaced_session_process_wakeup_pause_merges_into_current(
+    tmp_path, monkeypatch
+):
+    """RE-GATE regression (#6623): the credential-pool process-wakeup
+    exception branch must merge the pause into the CURRENT session object
+    under the canonical lock — never save a detached worker snapshot.
+
+    Same replacement schedule as the finalizer regression: cancel the old
+    generation, force REAL LRU eviction, lazily reload a distinct object,
+    admit and persist a successor through it, then run the pause branch with
+    the old worker's stale snapshot. The pause must land on the current object
+    + disk while the successor state survives; the stale snapshot must not be
+    the write target."""
+    sid = "sess_replaced_pause"
+    old_stream = "old-pause-stream"
+    newer_stream = "newer-pause-stream"
+
+    s_old = Session(
+        session_id=sid,
+        title="Replaced pause",
+        messages=[{"role": "user", "content": "old prompt"}],
+    )
+    s_old.active_stream_id = old_stream
+    s_old.pending_user_message = "old prompt"
+    s_old.pending_attachments = []
+    s_old.pending_started_at = 1000.0
+    s_old.pending_user_source = "process_wakeup"
+    s_old.save()
+    models.SESSIONS[sid] = s_old
+    config.register_session_writeback_owner(sid, old_stream)
+
+    config.register_stream_owner(old_stream, sid)
+    config.STREAMS[old_stream] = queue.Queue()
+    config.CANCEL_FLAGS[old_stream] = threading.Event()
+    config.register_active_run(old_stream, session_id=sid, phase="running")
+    assert streaming.cancel_stream(old_stream) is True
+
+    monkeypatch.setattr(models, "SESSIONS_MAX", 2)
+    for i in range(4):
+        _filler = Session(
+            session_id=f"sess_pause_filler_{i}", title="filler", messages=[]
+        )
+        _filler.save()
+        models.SESSIONS[_filler.session_id] = _filler
+    with config.LOCK:
+        models._evict_sessions_over_cap(0)
+    assert sid not in models.SESSIONS
+    s_new = models.get_session(sid)
+    assert s_new is not s_old
+
+    with streaming._get_session_agent_lock(sid):
+        s_new.active_stream_id = newer_stream
+        s_new.pending_user_message = "newer prompt"
+        s_new.pending_attachments = []
+        s_new.pending_started_at = 2000.0
+        s_new.pending_user_source = "webui"
+        s_new.messages.append(
+            {"role": "user", "content": "newer prompt", "timestamp": 2000}
+        )
+        config.register_session_writeback_owner(sid, newer_stream)
+        s_new.save()
+
+    # The old worker unwinds through the credential-pool process-wakeup
+    # branch holding its stale snapshot, under the session lock.
+    with streaming._get_session_agent_lock(sid):
+        _recorded = streaming._merge_process_wakeup_pause_into_current_session(
+            s_old,
+            classification="credential_pool_empty",
+            model="test-model",
+            provider="test-provider",
+        )
+    assert _recorded is not None
+    # The stale snapshot itself was NOT the write target: the default empty
+    # pause dict must not carry the recorded pause.
+    assert not (s_old.process_wakeup_pause or {}).get("paused")
+    # The pause merged into the CURRENT object and persisted.
+    assert s_new.process_wakeup_pause is not None
+    assert s_new.process_wakeup_pause.get("paused") is True
+    assert s_new.process_wakeup_pause.get("classification") == "credential_pool_empty"
+    assert s_new.process_wakeup_pause.get("model") == "test-model"
+    disk = models.Session.load(sid)
+    assert disk.process_wakeup_pause is not None
+    assert disk.process_wakeup_pause.get("classification") == "credential_pool_empty"
+    # Successor state survives on disk.
+    assert disk.active_stream_id == newer_stream
+    assert disk.pending_user_message == "newer prompt"
+    assert any(
+        str(m.get("content", "")) == "newer prompt" and m.get("role") == "user"
+        for m in disk.messages
+    )
+
+
+def test_issue6623_completed_successor_teardown_cleared_owner_blocks_old_finalizer(
+    tmp_path, monkeypatch
+):
+    """RE-GATE regression (#6623, 20:36): a delayed old finalizer must not be
+    authorized by a MISSING writeback-ownership record.
+
+    The successor is admitted, persists its result, then COMPLETES — and its
+    teardown clears the ownership entry (``clear_session_writeback_owner_if_owned``
+    in the final teardown of ``_run_agent_streaming``). When the old cancelled
+    worker finally resumes, ``owner is None`` AND the current object's
+    ``active_stream_id`` is None again — the exact ambiguous state the pre-fix
+    guard accepted (``owner is not None`` / ``active_stream_id is not None``
+    both false). ``None`` must fail closed: the old worker must not
+    append/save its obsolete cancellation over the completed successor.
+
+    This is the sequence the earlier completed-successor control did NOT
+    exercise (it kept the ownership record alive); production completion
+    clears it.
+    """
+    sid = "sess_teardown_cleared_owner"
+    old_stream = "old-teardown-stream"
+    newer_stream = "newer-teardown-stream"
+
+    s_old = Session(
+        session_id=sid,
+        title="Teardown-cleared owner",
+        messages=[{"role": "user", "content": "old prompt"}],
+    )
+    s_old.active_stream_id = old_stream
+    s_old.pending_user_message = "old prompt"
+    s_old.pending_attachments = []
+    s_old.pending_started_at = 1000.0
+    s_old.pending_user_source = "webui"
+    s_old.save()
+    models.SESSIONS[sid] = s_old
+    config.register_session_writeback_owner(sid, old_stream)
+
+    config.register_stream_owner(old_stream, sid)
+    config.STREAMS[old_stream] = queue.Queue()
+    config.CANCEL_FLAGS[old_stream] = threading.Event()
+    config.register_active_run(old_stream, session_id=sid, phase="running")
+
+    # 2) Cancel the old turn through the real production path.
+    assert streaming.cancel_stream(old_stream) is True
+    assert s_old.active_stream_id is None
+    assert streaming._session_has_cancel_marker(s_old)
+    _old_messages_after_cancel = len(s_old.messages)
+
+    # 3) REAL LRU eviction + lazy reload of a DISTINCT current object.
+    monkeypatch.setattr(models, "SESSIONS_MAX", 2)
+    for i in range(4):
+        _filler = Session(
+            session_id=f"sess_teardown_filler_{i}", title="filler", messages=[]
+        )
+        _filler.save()
+        models.SESSIONS[_filler.session_id] = _filler
+    with config.LOCK:
+        models._evict_sessions_over_cap(0)
+    s_new = models.get_session(sid)
+    assert s_new is not s_old, "lazy reload must yield a DISTINCT Session instance"
+
+    # 4) Admit the successor, let it COMPLETE, then run its teardown: the
+    # ownership entry is released (compare-and-clear) exactly like the final
+    # teardown in _run_agent_streaming.
+    with streaming._get_session_agent_lock(sid):
+        s_new.active_stream_id = newer_stream
+        s_new.pending_user_message = "newer prompt"
+        s_new.pending_attachments = []
+        s_new.pending_started_at = 2000.0
+        s_new.pending_user_source = "webui"
+        s_new.messages.append(
+            {"role": "user", "content": "newer prompt", "timestamp": 2000}
+        )
+        config.register_session_writeback_owner(sid, newer_stream)
+        s_new.save()
+        # Successor completes: active_stream_id + pending fields cleared.
+        s_new.active_stream_id = None
+        s_new.pending_user_message = None
+        s_new.pending_attachments = []
+        s_new.pending_started_at = None
+        s_new.pending_user_source = None
+        s_new.save()
+        # Successor teardown: release ownership while still owning it.
+        config.clear_session_writeback_owner_if_owned(sid, newer_stream)
+    assert config.session_writeback_owner(sid) is None
+    _messages_after_successor = len(s_new.messages)
+    # The old turn's OWN cancel marker is legitimately on disk (cancel_stream
+    # persisted it); the successor transcript may carry it. What must NOT
+    # happen is the delayed finalizer appending a NEW one.
+    _cancel_markers_before = sum(
+        1 for m in s_new.messages
+        if str(m.get("content", "")).startswith("**Task cancelled:**")
+    )
+
+    # 5) The old worker resumes and unwinds through the delayed cancel
+    # finalizer holding its stale snapshot, under the session lock.
+    with streaming._get_session_agent_lock(sid):
+        streaming._finalize_cancelled_turn(s_old, ephemeral=False, stream_id=old_stream)
+
+    # 6) Fail closed: the old generation made no terminal write.
+    assert len(s_old.messages) == _old_messages_after_cancel, (
+        "stale snapshot must not receive a delayed terminal write"
+    )
+    # The successor's cache object and disk state survive untouched.
+    assert s_new.active_stream_id is None
+    assert s_new.pending_user_message is None
+    assert len(s_new.messages) == _messages_after_successor, (
+        "delayed cancel finalizer must not append over the completed successor"
+    )
+    assert sum(
+        1 for m in s_new.messages
+        if str(m.get("content", "")).startswith("**Task cancelled:**")
+    ) == _cancel_markers_before, (
+        "delayed cancel finalizer must not append a new cancel marker"
+    )
+    disk = models.Session.load(sid)
+    assert disk.active_stream_id is None
+    assert len(disk.messages) == _messages_after_successor
+    assert any(
+        str(m.get("content", "")) == "newer prompt" and m.get("role") == "user"
+        for m in disk.messages
+    )
+    assert sum(
+        1 for m in disk.messages
+        if str(m.get("content", "")).startswith("**Task cancelled:**")
+    ) == _cancel_markers_before, (
+        "disk must not gain a new cancel marker from the stale finalizer"
+    )
+
+
+def test_issue6623_unresolvable_current_session_fails_closed_for_pause_merge(
+    tmp_path, monkeypatch
+):
+    """RE-GATE regression (#6623, 20:36): the process-wakeup pause merge must
+    fail closed (no write, no exception) when the canonical current session
+    cannot be resolved — the worker-held snapshot is never the write target.
+    """
+    sid = "sess_unresolvable_pause"
+    s_old = Session(session_id=sid, messages=[])
+    s_old.save = Mock()
+    # NOT cached, no file on disk -> get_session() resolves nothing.
+    recorded = streaming._merge_process_wakeup_pause_into_current_session(
+        s_old,
+        classification="credential_pool_empty",
+        model="test-model",
+        provider="test-provider",
+    )
+    assert recorded is None
+    s_old.save.assert_not_called()
+    assert not (s_old.process_wakeup_pause or {}).get("paused")
+
+
+def test_issue6636_gateway_prestart_cancel_clears_writeback_owner(tmp_path, monkeypatch):
+    """RE-GATE (maintainer fix): the Gateway worker's pre-start cancellation path
+    (``q is None`` early return in ``_run_gateway_chat_streaming``) must release
+    the SESSION_WRITEBACK_OWNERS entry the route layer registered, or every
+    cancelled-before-start Gateway run leaks a permanent owner entry (unbounded
+    process-lifetime growth). Revert-sensitive: fails if the
+    clear_session_writeback_owner_if_owned call on that path is removed.
+    """
+    import api.gateway_chat as gateway_chat
+
+    session_id = "sess_gw_prestart"
+    stream_id = "stream-gw-prestart"
+    # The route layer registered the writeback owner before dispatching the worker.
+    config.register_session_writeback_owner(session_id, stream_id)
+    assert config.session_writeback_owner(session_id) == stream_id
+    # Simulate cancel-before-start: the stream map has no queue for this id, so the
+    # worker takes its `q is None` early-return teardown path.
+    config.STREAMS.pop(stream_id, None)
+
+    gateway_chat._run_gateway_chat_streaming(
+        session_id,
+        [],
+        "test-model",
+        None,
+        stream_id,
+        None,
+    )
+
+    assert config.session_writeback_owner(session_id) is None, (
+        "pre-start Gateway cancellation must clear the writeback owner it registered"
+    )
+
+
+def test_issue6636_gateway_clear_only_affects_owned_stream(tmp_path, monkeypatch):
+    """The Gateway teardown clear must be compare-and-clear: if a successor has
+    already taken writeback ownership by the time the old worker tears down, the
+    old worker's clear must NOT evict the successor's entry.
+    """
+    session_id = "sess_gw_successor"
+    old_stream = "stream-gw-old"
+    new_stream = "stream-gw-new"
+    config.register_session_writeback_owner(session_id, old_stream)
+    # Successor takes ownership before the old worker's teardown runs.
+    config.register_session_writeback_owner(session_id, new_stream)
+    # Old worker's teardown compare-and-clear must be a no-op (it no longer owns it).
+    config.clear_session_writeback_owner_if_owned(session_id, old_stream)
+    assert config.session_writeback_owner(session_id) == new_stream, (
+        "old worker teardown must not evict the successor's writeback ownership"
+    )