From b96f0f6d6f061a74b4a1aefc358d13ee5f2a6370 Mon Sep 17 00:00:00 2001 From: nesquena-hermes Date: Mon, 8 Jun 2026 11:06:59 -0700 Subject: [PATCH] =?UTF-8?q?Release=20v0.51.327=20=E2=80=94=20Release=20KQ?= =?UTF-8?q?=20(brick=20wave:=20#3829=20+=20#3828=20+=20#3822)=20(#3836)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Co-authored-by: franksong2702 --- CHANGELOG.md | 7 + api/models.py | 39 ++++ api/routes.py | 80 ++++++- tests/test_session_lineage_full_transcript.py | 82 ++++++++ tests/test_stale_stream_cleanup.py | 195 ++++++++++++++++++ tests/test_webui_state_db_reconciliation.py | 43 ++++ 6 files changed, 445 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8ff26a54..98030f94 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/api/models.py b/api/models.py index 9fd54e2a..df1965ac 100644 --- a/api/models.py +++ b/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) diff --git a/api/routes.py b/api/routes.py index eb54c0a2..2f2d02c0 100644 --- a/api/routes.py +++ b/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 diff --git a/tests/test_session_lineage_full_transcript.py b/tests/test_session_lineage_full_transcript.py index 11f7550c..e0812888 100644 --- a/tests/test_session_lineage_full_transcript.py +++ b/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" diff --git a/tests/test_stale_stream_cleanup.py b/tests/test_stale_stream_cleanup.py index de0f37e4..18f6fd82 100644 --- a/tests/test_stale_stream_cleanup.py +++ b/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. diff --git a/tests/test_webui_state_db_reconciliation.py b/tests/test_webui_state_db_reconciliation.py index 28da6bab..ef0baa07 100644 --- a/tests/test_webui_state_db_reconciliation.py +++ b/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.