hermes-webui/tests/test_issue6623_cancel_owner_race.py

"""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"
    )