Release v0.51.327 — Release KQ (brick wave: #3829 + #3828 + #3822) (#3836)

Brick wave: stale-cache freshness (#3829), compression-tail preservation (#3828), interrupt-successor race (#3822, +bounded-guard hardening). Full suite 8271, Codex SAFE, Opus SHIP, CI 11/11. Co-authored-by: dso2ng <dso2ng@users.noreply.github.com> Co-authored-by: franksong2702 <franksong2702@users.noreply.github.com>

CHANGELOG.md

@@ -3,6 +3,13 @@
 
 ## [Unreleased]
 
+## [v0.51.327] — 2026-06-08 — Release KQ (brick wave: session-cache freshness + compression-tail + interrupt race)
+
+### Fixed
+- **`/api/session` no longer serves a stale cached conversation** that lags disk — recent assistant replies could disappear from a session until the server restarted because the in-process LRU held an older `Session` object after another object persisted new turns. `get_session()` now reloads from disk when the sidecar/index is strictly ahead of the cache. (#3829, @dso2ng)
+- **The latest assistant message no longer vanishes after compression-lineage stitching.** A compression continuation child was applying its own truncation watermark to its already-persisted sidecar tail during WebUI display merge, dropping the tail. (#3828, @dso2ng, follow-up to #3770/#3776)
+- **Interrupt-and-send no longer races a still-unwinding worker.** After Stop/interrupt clears the active stream id, a successor send could reuse the cached agent while the cancelled worker was still landing; `/api/chat/start` now blocks (409) on a same-session active run during the bounded unwind window. (#3822, @franksong2702, #3808)
+
 ## [v0.51.326] — 2026-06-08 — Release KP (mic STT probe + journal cleanup + schema guard + help hover)
 
 ### Fixed

api/models.py

@@ -2191,6 +2191,33 @@ def _cache_has_stale_unsaved_user_tail(cached, disk_session) -> bool:
     return _message_content_text(cached_tail) == _message_content_text(previous_disk_user)
 
 
+def _cached_session_lags_disk(cached) -> bool:
+    """Return True when a cached full session is older than its sidecar.
+
+    Active/reconnect paths can update the persisted sidecar through another
+    Session object while the LRU cache still holds an older object for the same
+    id. Serving the cache then makes recent assistant results disappear from
+    GET /api/session even though disk and _index.json are correct. Compare only
+    cheap metadata here; full reload happens only if disk is strictly ahead.
+    """
+    if cached is None:
+        return False
+    sid = getattr(cached, 'session_id', None)
+    if not sid:
+        return False
+    try:
+        disk_meta = Session.load_metadata_only(sid)
+    except Exception:
+        return False
+    if disk_meta is None:
+        return False
+    cached_count = len(getattr(cached, 'messages', None) or [])
+    disk_count = _parse_nonnegative_int(getattr(disk_meta, '_metadata_message_count', None))
+    if disk_count is None:
+        disk_count = _lookup_index_message_count(sid)
+    return bool(disk_count is not None and disk_count > cached_count)
+
+
 def get_session(sid, metadata_only=False):
     """Load a session, optionally with metadata only (skipping the messages array).
 
@@ -2220,6 +2247,18 @@ def get_session(sid, metadata_only=False):
                     SESSIONS.pop(sid, None)
             cached = None
     if cached is not None:
+        if not metadata_only and _cached_session_lags_disk(cached):
+            try:
+                disk_session = Session.load(sid)
+                with LOCK:
+                    SESSIONS[sid] = disk_session
+                    SESSIONS.move_to_end(sid)
+                cached = disk_session
+            except Exception:
+                logger.debug(
+                    "cached session disk-freshness check failed for session %s",
+                    sid, exc_info=True,
+                )
         if not metadata_only and _inactive_cache_tail_needs_disk_check(cached):
             try:
                 disk_session = Session.load(sid)

api/routes.py

@@ -2893,6 +2893,23 @@ def _message_window_for_display(messages, msg_limit=None, msg_before=None, expan
     return window, start_idx
 
 
+def _messages_start_with_visible_prefix(messages, prefix) -> bool:
+    """Return True when ``messages`` already replays ``prefix`` in display order."""
+    messages = list(messages or [])
+    prefix = list(prefix or [])
+    if not prefix:
+        return True
+    if len(messages) < len(prefix):
+        return False
+    try:
+        return all(
+            _session_message_visible_key(messages[idx]) == _session_message_visible_key(prefix_msg)
+            for idx, prefix_msg in enumerate(prefix)
+        )
+    except Exception:
+        return False
+
+
 def _webui_sidecar_lineage_messages_for_display(session, *, max_hops: int = 20) -> list:
     """Return WebUI sidecar messages stitched across compression snapshots.
 
@@ -2904,6 +2921,7 @@ def _webui_sidecar_lineage_messages_for_display(session, *, max_hops: int = 20)
     """
     segments = []
     current = session
+    session_messages = list(getattr(session, "messages", []) or [])
     seen = {str(getattr(session, "session_id", "") or "")}
     for _ in range(max(0, int(max_hops))):
         parent_id = str(getattr(current, "parent_session_id", "") or "").strip()
@@ -2912,6 +2930,11 @@ def _webui_sidecar_lineage_messages_for_display(session, *, max_hops: int = 20)
         parent = Session.load(parent_id)
         if not parent or not getattr(parent, "pre_compression_snapshot", False):
             break
+        if not segments and _messages_start_with_visible_prefix(
+            session_messages,
+            getattr(parent, "messages", []) or [],
+        ):
+            return session_messages
         segments.append(parent)
         seen.add(parent_id)
         current = parent
@@ -2929,7 +2952,7 @@ def _webui_sidecar_lineage_messages_for_display(session, *, max_hops: int = 20)
     return merge_session_messages_append_only(
         merged,
         getattr(session, "messages", []) or [],
-        truncation_watermark=getattr(session, "truncation_watermark", None),
+        truncation_watermark=None,
     )
 
 
@@ -3342,6 +3365,7 @@ from api.models import (
     merge_session_messages_append_only,
     _active_stream_ids,
     _session_message_merge_key,
+    _session_message_visible_key,
     _is_empty_partial_activity_message,
     _hide_from_default_sidebar,
     prune_session_from_index,
@@ -11439,6 +11463,52 @@ def _active_stream_blocks_chat_start(session, stream_id: str | None) -> bool:
     return False
 
 
+def _active_run_stream_for_session(session_id: str | None) -> str | None:
+    """Return a live worker stream for this session even if sidecar stream id is clear.
+
+    cancel_stream() intentionally clears ``session.active_stream_id`` before the
+    worker thread fully exits so Stop remains responsive. During that unwind
+    window ACTIVE_RUNS is the worker-lifecycle truth; a successor chat/start for
+    the same session must wait or it can reuse the cached agent while the old
+    interrupt is still landing (#3808).
+
+    Bounded: the post-cancel unwind is short (dominated by the worker finally's
+    ``_ckpt_thread.join(timeout=15)``), and ``unregister_active_run`` runs in that
+    finally, so a healthy worker leaves ACTIVE_RUNS within seconds. A detached /
+    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.)
+    """
+    sid = str(session_id or "").strip()
+    if not sid:
+        return None
+    ceiling = 180.0  # generous vs the 15s checkpoint-join unwind; finite to avoid permanent-409
+    now = time.time()
+    try:
+        from api import config as _live_config
+        with _live_config.ACTIVE_RUNS_LOCK:
+            for run_stream_id, raw in (_live_config.ACTIVE_RUNS or {}).items():
+                stream_id = str((raw or {}).get("stream_id") or run_stream_id or "").strip()
+                run_sid = str((raw or {}).get("session_id") or "").strip()
+                if run_sid != sid or not stream_id:
+                    continue
+                try:
+                    started_at = float((raw or {}).get("started_at") or 0)
+                except (TypeError, ValueError):
+                    started_at = 0.0
+                # Ignore a stale/wedged entry past the unwind ceiling so it can't
+                # block the session permanently.
+                if started_at and (now - started_at) > ceiling:
+                    continue
+                return stream_id
+    except Exception:
+        return None
+    return None
+
+
 def _start_chat_stream_for_session(
     s,
     *,
@@ -11492,6 +11562,14 @@ def _start_chat_stream_for_session(
                     }
                 needs_stale_cleanup = True
             else:
+                blocking_run_stream_id = _active_run_stream_for_session(s.session_id)
+                if blocking_run_stream_id:
+                    diag.stage("response_write") if diag else None
+                    return {
+                        "error": "session already has an active stream",
+                        "active_stream_id": blocking_run_stream_id,
+                        "_status": 409,
+                    }
                 needs_stale_cleanup = False
                 stream_id = uuid.uuid4().hex
                 diag.stage("save_pending_state") if diag else None

tests/test_session_lineage_full_transcript.py

@@ -274,3 +274,85 @@ def test_webui_fork_session_does_not_stitch_non_snapshot_parent(monkeypatch):
     assert [m["content"] for m in routes._webui_sidecar_lineage_messages_for_display(child)] == [
         "fork child only",
     ]
+
+
+def test_webui_lineage_display_keeps_child_tail_after_snapshot_watermark(monkeypatch):
+    """A child sidecar watermark must not delete the child's persisted continuation tail."""
+    parent = SimpleNamespace(
+        session_id="parent-watermark",
+        parent_session_id=None,
+        pre_compression_snapshot=True,
+        truncation_watermark=None,
+        messages=[
+            {"role": "user", "content": "parent user", "timestamp": 1000.0},
+            {"role": "assistant", "content": "parent assistant", "timestamp": 1001.0},
+        ],
+    )
+    child = SimpleNamespace(
+        session_id="child-watermark",
+        parent_session_id="parent-watermark",
+        pre_compression_snapshot=False,
+        truncation_watermark=1001.0,
+        messages=[
+            {"role": "user", "content": "child user", "timestamp": 1002.0},
+            {"role": "assistant", "content": "child assistant final", "timestamp": 1003.0},
+        ],
+    )
+    monkeypatch.setattr(routes.Session, "load", lambda sid: parent if sid == "parent-watermark" else None)
+
+    contents = [m["content"] for m in routes._webui_sidecar_lineage_messages_for_display(child)]
+
+    assert contents == [
+        "parent user",
+        "parent assistant",
+        "child user",
+        "child assistant final",
+    ]
+
+
+def test_webui_lineage_display_does_not_restitch_ancestor_when_child_replays_parent(monkeypatch):
+    """If the child sidecar already starts with its direct parent snapshot, use it as-is."""
+    grandparent = SimpleNamespace(
+        session_id="grandparent-replayed",
+        parent_session_id=None,
+        pre_compression_snapshot=True,
+        truncation_watermark=None,
+        messages=[
+            {"role": "user", "content": "grandparent user", "timestamp": 1000.0},
+            {"role": "assistant", "content": "grandparent assistant", "timestamp": 1001.0},
+        ],
+    )
+    parent = SimpleNamespace(
+        session_id="parent-replayed",
+        parent_session_id="grandparent-replayed",
+        pre_compression_snapshot=True,
+        truncation_watermark=1001.0,
+        messages=grandparent.messages + [
+            {"role": "user", "content": "parent user", "timestamp": 1002.0},
+            {"role": "assistant", "content": "parent assistant", "timestamp": 1003.0},
+        ],
+    )
+    child = SimpleNamespace(
+        session_id="child-replayed",
+        parent_session_id="parent-replayed",
+        pre_compression_snapshot=False,
+        truncation_watermark=1001.0,
+        messages=parent.messages + [
+            {"role": "user", "content": "followup user", "timestamp": 1004.0},
+            {"role": "assistant", "content": "latest final answer", "timestamp": 1005.0},
+        ],
+    )
+    by_id = {"parent-replayed": parent, "grandparent-replayed": grandparent}
+    monkeypatch.setattr(routes.Session, "load", lambda sid: by_id.get(sid))
+
+    contents = [m["content"] for m in routes._webui_sidecar_lineage_messages_for_display(child)]
+
+    assert contents == [
+        "grandparent user",
+        "grandparent assistant",
+        "parent user",
+        "parent assistant",
+        "followup user",
+        "latest final answer",
+    ]
+    assert contents[-1] == "latest final answer"

tests/test_stale_stream_cleanup.py

@@ -1,5 +1,6 @@
 import queue
 import threading
+import time
 from pathlib import Path
 
 import api.config as config
@@ -155,6 +156,200 @@ def test_chat_start_rechecks_active_stream_under_session_lock(monkeypatch, tmp_p
         routes.STREAMS.pop(existing_stream_id, None)
 
 
+def test_chat_start_blocks_same_session_active_run_after_cancel_clears_stream_id(monkeypatch, tmp_path):
+    """Regression for #3808: cancel clears active_stream_id before worker exit.
+
+    interrupt-and-send queues a successor message, then calls cancel_stream().
+    cancel_stream() intentionally clears session.active_stream_id so Stop remains
+    responsive, but the old worker remains in ACTIVE_RUNS until its finally block
+    unregisters it. chat/start must still block by session_id during that window.
+    """
+    config.STREAMS.clear()
+    config.ACTIVE_RUNS.clear()
+    config.SESSION_AGENT_LOCKS.clear()
+
+    class ChatStartSession:
+        session_id = "interrupt-send-session"
+
+        def __init__(self):
+            self.active_stream_id = None
+            self.pending_user_message = None
+            self.pending_attachments = []
+            self.pending_started_at = None
+            self.messages = []
+            self.title = "Interrupt Send"
+            self.worktree_path = None
+            self.workspace = None
+            self.model = None
+            self.model_provider = None
+
+        def save(self, *args, **kwargs):
+            return None
+
+    session = ChatStartSession()
+    old_stream_id = "old-cancelling-stream"
+    config.register_active_run(old_stream_id, session_id=session.session_id, phase="cancelling")
+
+    class NoopThread:
+        def __init__(self, *args, **kwargs):
+            self.args = args
+            self.kwargs = kwargs
+
+        def start(self):
+            return None
+
+    monkeypatch.setattr(routes.uuid, "uuid4", lambda: type("FakeUuid", (), {"hex": "new-stream"})())
+    monkeypatch.setattr(routes, "set_last_workspace", lambda workspace: None)
+    monkeypatch.setattr(routes, "create_stream_channel", lambda: queue.Queue())
+    monkeypatch.setattr(routes.threading, "Thread", NoopThread)
+
+    try:
+        response = routes._start_chat_stream_for_session(
+            session,
+            msg="successor prompt",
+            attachments=[],
+            workspace=str(tmp_path),
+            model="test-model",
+            model_provider=None,
+        )
+
+        assert response["_status"] == 409
+        assert response["active_stream_id"] == old_stream_id
+        assert session.active_stream_id is None
+        assert session.pending_user_message is None
+        assert "new-stream" not in routes.STREAMS
+    finally:
+        config.unregister_active_run(old_stream_id)
+
+
+def test_chat_start_allows_same_session_after_active_run_unregisters(monkeypatch, tmp_path):
+    """The #3808 guard must release once the old worker unregisters ACTIVE_RUNS."""
+    config.STREAMS.clear()
+    config.ACTIVE_RUNS.clear()
+    config.SESSION_AGENT_LOCKS.clear()
+
+    class ChatStartSession:
+        session_id = "interrupt-send-session-released"
+
+        def __init__(self):
+            self.active_stream_id = None
+            self.pending_user_message = None
+            self.pending_attachments = []
+            self.pending_started_at = None
+            self.messages = []
+            self.title = "Interrupt Send"
+            self.worktree_path = None
+            self.workspace = None
+            self.model = None
+            self.model_provider = None
+
+        def save(self, *args, **kwargs):
+            return None
+
+    session = ChatStartSession()
+
+    class NoopThread:
+        def __init__(self, *args, **kwargs):
+            self.args = args
+            self.kwargs = kwargs
+
+        def start(self):
+            return None
+
+    monkeypatch.setattr(routes.uuid, "uuid4", lambda: type("FakeUuid", (), {"hex": "new-stream"})())
+    monkeypatch.setattr(routes, "set_last_workspace", lambda workspace: None)
+    monkeypatch.setattr(routes, "create_stream_channel", lambda: queue.Queue())
+    monkeypatch.setattr(routes.threading, "Thread", NoopThread)
+
+    response = routes._start_chat_stream_for_session(
+        session,
+        msg="successor prompt",
+        attachments=[],
+        workspace=str(tmp_path),
+        model="test-model",
+        model_provider=None,
+    )
+
+    try:
+        assert "error" not in response
+        assert response["stream_id"] == "new-stream"
+        assert session.active_stream_id == "new-stream"
+        assert session.pending_user_message == "successor prompt"
+    finally:
+        routes.STREAMS.pop("new-stream", None)
+
+
+def test_chat_start_not_permanently_blocked_by_stale_active_run(monkeypatch, tmp_path):
+    """A wedged/detached ACTIVE_RUNS entry past the unwind ceiling must NOT 409 forever.
+
+    The #3808 successor guard waits on a same-session ACTIVE_RUNS entry during the
+    short post-cancel unwind. But unregister only runs in the worker finally, so a
+    worker stuck in a provider call (or leaked by SIGKILL without restart) would
+    block the session permanently. The guard ignores entries older than the 180s
+    ceiling so the user can recover. (Codex brick-gate hardening, #3822.)
+    """
+    config.STREAMS.clear()
+    config.ACTIVE_RUNS.clear()
+    config.SESSION_AGENT_LOCKS.clear()
+
+    class ChatStartSession:
+        session_id = "interrupt-send-session-stale"
+
+        def __init__(self):
+            self.active_stream_id = None
+            self.pending_user_message = None
+            self.pending_attachments = []
+            self.pending_started_at = None
+            self.messages = []
+            self.title = "Interrupt Send"
+            self.worktree_path = None
+            self.workspace = None
+            self.model = None
+            self.model_provider = None
+
+        def save(self, *args, **kwargs):
+            return None
+
+    session = ChatStartSession()
+    stale_stream_id = "wedged-old-stream"
+    config.register_active_run(stale_stream_id, session_id=session.session_id, phase="running")
+    # Age the entry well past the 180s unwind ceiling.
+    with config.ACTIVE_RUNS_LOCK:
+        config.ACTIVE_RUNS[stale_stream_id]["started_at"] = time.time() - 600
+
+    # The bounded guard should treat it as stale and NOT report it as blocking.
+    assert routes._active_run_stream_for_session(session.session_id) is None
+
+    class NoopThread:
+        def __init__(self, *args, **kwargs):
+            self.args = args
+            self.kwargs = kwargs
+
+        def start(self):
+            return None
+
+    monkeypatch.setattr(routes.uuid, "uuid4", lambda: type("FakeUuid", (), {"hex": "new-stream"})())
+    monkeypatch.setattr(routes, "set_last_workspace", lambda workspace: None)
+    monkeypatch.setattr(routes, "create_stream_channel", lambda: queue.Queue())
+    monkeypatch.setattr(routes.threading, "Thread", NoopThread)
+
+    try:
+        response = routes._start_chat_stream_for_session(
+            session,
+            msg="successor prompt",
+            attachments=[],
+            workspace=str(tmp_path),
+            model="test-model",
+            model_provider=None,
+        )
+        assert "error" not in response
+        assert response["stream_id"] == "new-stream"
+        assert session.active_stream_id == "new-stream"
+    finally:
+        config.unregister_active_run(stale_stream_id)
+        routes.STREAMS.pop("new-stream", None)
+
+
 def test_stale_stream_cleanup_does_not_clobber_concurrent_chat_start(monkeypatch):
     """Regression for #1533: stale cleanup must not erase a new stream id.
 

tests/test_webui_state_db_reconciliation.py

@@ -566,6 +566,49 @@ def test_api_session_reload_drops_stale_cached_user_tail_after_saved_assistant(m
     assert handler.response_json["session"]["message_count"] == 2
 
 
+def test_get_session_reloads_when_cached_session_lags_disk(monkeypatch, tmp_path):
+    import api.models as models
+
+    sid = "webui_reconcile_cache_lags_disk"
+    old_messages = [
+        {"role": "user", "content": "old user", "timestamp": 1000.0},
+        {"role": "assistant", "content": "old assistant", "timestamp": 1001.0},
+    ]
+    _install_test_session(monkeypatch, tmp_path, sid, old_messages)
+
+    cached = models.Session.load(sid)
+    assert cached is not None
+    cached.active_stream_id = "stream-cache-lags-disk"
+    cached.pending_user_message = "next prompt"
+    models.SESSIONS[sid] = cached
+
+    newer = models.Session(
+        session_id=sid,
+        title="Reconcile",
+        workspace=str(tmp_path),
+        model="test-model",
+        messages=old_messages + [
+            {"role": "user", "content": "new user", "timestamp": 1002.0},
+            {"role": "assistant", "content": "new final answer", "timestamp": 1003.0},
+        ],
+        created_at=1000.0,
+        updated_at=1003.0,
+        active_stream_id="stream-cache-lags-disk",
+        pending_user_message="next prompt",
+    )
+    newer.save(touch_updated_at=False)
+
+    loaded = models.get_session(sid)
+
+    assert [m["content"] for m in loaded.messages] == [
+        "old user",
+        "old assistant",
+        "new user",
+        "new final answer",
+    ]
+    assert models.SESSIONS[sid] is loaded
+
+
 def test_metadata_fast_path_uses_summary_without_full_merge_for_restamped_replays(monkeypatch, tmp_path):
     """Metadata-only /api/session must not full-read and merge transcripts.