From 0567b056c09932209f185713ddfdf6d53e4ccb15 Mon Sep 17 00:00:00 2001 From: Frank Song Date: Mon, 22 Jun 2026 17:55:01 +0800 Subject: [PATCH] fix(compression): guard state replay after compacted anchors --- api/models.py | 108 ++++++++++++ tests/test_issue1217_transcript_compaction.py | 158 ++++++++++++++++++ 2 files changed, 266 insertions(+) diff --git a/api/models.py b/api/models.py index 28f36f01..a41b3f50 100644 --- a/api/models.py +++ b/api/models.py @@ -14,6 +14,7 @@ from contextlib import closing from pathlib import Path import api.config as _cfg +from api.compression_anchor import is_context_compression_marker from api.config import ( SESSION_DIR, SESSION_INDEX_FILE, SESSIONS, SESSIONS_MAX, LOCK, STREAMS, STREAMS_LOCK, DEFAULT_WORKSPACE, DEFAULT_MODEL, PROJECTS_FILE, HOME, @@ -4898,6 +4899,93 @@ def state_db_delta_after_context(sidecar_context: list, state_messages: list) -> return state_messages[best_len:] +def _normalized_compression_anchor_text(value) -> str: + return " ".join(str(value or "").split()).strip()[:160] + + +def _compression_anchor_timestamp_as_float(value) -> float | None: + if value in (None, ""): + return None + try: + return float(value) + except (TypeError, ValueError): + pass + try: + return datetime.datetime.fromisoformat(str(value).strip().replace("Z", "+00:00")).timestamp() + except Exception: + return None + + +def _context_messages_include_compression_marker(messages: list) -> bool: + for message in messages or []: + if not is_context_compression_marker(message): + continue + text = _message_content_text(message).lower().lstrip() + # Only prompt compaction summaries require fail-closed state.db replay. + # Other compression-adjacent summaries, such as Session Arc Summary, + # keep the existing prefix-delta behavior so fresh follow-ups survive. + if text.startswith("[context compaction") or text.startswith("context compaction"): + return True + return False + + +def _state_db_anchor_index(state_messages: list, anchor_key) -> int | None: + if not isinstance(anchor_key, dict): + return None + + anchor_role = str(anchor_key.get("role") or "").strip().lower() + anchor_text = _normalized_compression_anchor_text(anchor_key.get("text")) + anchor_attachments = anchor_key.get("attachments") + anchor_ts = _compression_anchor_timestamp_as_float(anchor_key.get("ts")) + + if not anchor_role: + return None + + # Do not attempt text-only fallback when timestamp is unavailable. Text-based + # fallback can match stale legacy rows if the anchor timestamp was lost, + # which can re-introduce old state.db rows after compaction. + if anchor_ts is None: + return None + + if anchor_attachments in (None, ""): + expected_attachments = 0 + else: + try: + expected_attachments = int(anchor_attachments) + except (TypeError, ValueError): + expected_attachments = 0 + + exact_timestamp_matches = [] + for idx, message in enumerate(state_messages or []): + if _message_role(message) != anchor_role: + continue + + attachments = message.get("attachments") if isinstance(message, dict) else None + attach_count = len(attachments) if isinstance(attachments, list) else 0 + if attach_count != expected_attachments: + continue + + # Attachment-only or timestamp-only anchors have no stable text payload. + # In that shape the timestamp + role + attachment count is the boundary; + # apply text comparison only when the anchor actually captured text. + message_text = _normalized_compression_anchor_text(_message_content_text(message)) + if anchor_text and message_text != anchor_text: + continue + + message_ts = _compression_anchor_timestamp_as_float( + message.get("timestamp") if isinstance(message, dict) else None + ) + if message_ts is None: + continue + if abs(message_ts - anchor_ts) <= 1e-6: + exact_timestamp_matches.append(idx) + continue + + if exact_timestamp_matches: + return exact_timestamp_matches[-1] + return None + + def _insert_state_message_chronologically(messages: list, msg: dict) -> bool: """Insert a state.db-only row before newer sidecar rows when safe. @@ -5199,15 +5287,35 @@ def reconciled_state_db_messages_for_session( if session is None: return [] local_messages = [] + using_context_messages = False if prefer_context: context_messages = getattr(session, 'context_messages', None) if isinstance(context_messages, list) and context_messages: local_messages = context_messages + using_context_messages = True if not local_messages: local_messages = getattr(session, 'messages', None) or [] if state_messages is None: state_messages = get_state_db_session_messages(getattr(session, 'session_id', None)) if prefer_context and local_messages: + if using_context_messages: + compressed_context = _context_messages_include_compression_marker(local_messages) + anchor_key = getattr(session, "compression_anchor_message_key", None) + if compressed_context: + if not anchor_key: + logger.debug( + "Compressed context for session %s has no compression anchor; using context_messages only", + getattr(session, "session_id", None), + ) + return list(local_messages) + anchor_index = _state_db_anchor_index(state_messages, anchor_key) + if anchor_index is None: + logger.debug( + "Compressed context for session %s has an unverifiable compression anchor; using context_messages only", + getattr(session, "session_id", None), + ) + return list(local_messages) + state_messages = list(state_messages or [])[anchor_index + 1 :] state_messages = state_db_delta_after_context(local_messages, state_messages) return merge_session_messages_append_only( local_messages, diff --git a/tests/test_issue1217_transcript_compaction.py b/tests/test_issue1217_transcript_compaction.py index eced9261..15e65705 100644 --- a/tests/test_issue1217_transcript_compaction.py +++ b/tests/test_issue1217_transcript_compaction.py @@ -430,6 +430,164 @@ def test_prefer_context_reconcile_keeps_state_delta_when_no_mirrored_prefix(): assert reconciled == sidecar_context + state_messages +def test_prefer_context_reconcile_uses_compression_anchor_boundary_for_compacted_context(): + sidecar_context = [ + {"role": "assistant", "content": "[CONTEXT COMPACTION — REFERENCE ONLY] compacted"}, + {"role": "user", "content": "fresh kept question"}, + {"role": "assistant", "content": "fresh kept answer"}, + ] + state_messages = [ + {"role": "user", "content": "old question", "timestamp": 1.0}, + {"role": "assistant", "content": "old answer", "timestamp": 2.0}, + {"role": "user", "content": "fresh kept question", "timestamp": 3.0}, + {"role": "assistant", "content": "fresh kept answer", "timestamp": 4.0}, + {"role": "user", "content": "new after compress", "timestamp": 5.0}, + ] + session = SimpleNamespace( + session_id="anchor-boundary", + context_messages=sidecar_context, + messages=[], + compression_anchor_message_key={ + "role": "assistant", + "text": "fresh kept answer", + "ts": 4.0, + "attachments": 0, + }, + ) + + reconciled = reconciled_state_db_messages_for_session( + session, prefer_context=True, state_messages=state_messages + ) + + assert reconciled == sidecar_context + [ + {"role": "user", "content": "new after compress", "timestamp": 5.0}, + ] + + +def test_prefer_context_reconcile_fails_closed_when_compression_anchor_missing(): + sidecar_context = [ + {"role": "assistant", "content": "[CONTEXT COMPACTION — REFERENCE ONLY] compacted"}, + {"role": "user", "content": "fresh kept question"}, + {"role": "assistant", "content": "fresh kept answer"}, + ] + state_messages = [ + {"role": "user", "content": "old question", "timestamp": 1.0}, + {"role": "assistant", "content": "old answer", "timestamp": 2.0}, + {"role": "user", "content": "fresh kept question", "timestamp": 3.0}, + {"role": "assistant", "content": "fresh kept answer", "timestamp": 4.0}, + {"role": "user", "content": "new after compress", "timestamp": 5.0}, + ] + session = SimpleNamespace( + session_id="anchor-missing", + context_messages=sidecar_context, + messages=[], + compression_anchor_message_key={ + "role": "assistant", + "text": "missing boundary", + "ts": 999.0, + "attachments": 0, + }, + ) + + reconciled = reconciled_state_db_messages_for_session( + session, prefer_context=True, state_messages=state_messages + ) + + assert reconciled == sidecar_context + + +def test_prefer_context_reconcile_fails_closed_when_compressed_context_has_no_anchor_key(): + sidecar_context = [ + {"role": "assistant", "content": "[CONTEXT COMPACTION — REFERENCE ONLY] compacted"}, + {"role": "user", "content": "new question"}, + ] + state_messages = [ + {"role": "user", "content": "old question", "timestamp": 1.0}, + {"role": "assistant", "content": "old answer", "timestamp": 2.0}, + {"role": "user", "content": "new question", "timestamp": 3.0}, + ] + session = SimpleNamespace( + session_id="compressed-anchor-none", + context_messages=sidecar_context, + messages=[], + compression_anchor_message_key=None, + ) + + reconciled = reconciled_state_db_messages_for_session( + session, prefer_context=True, state_messages=state_messages + ) + + assert reconciled == sidecar_context + + +def test_prefer_context_reconcile_accepts_iso_compression_anchor_timestamp(): + sidecar_context = [ + {"role": "assistant", "content": "[CONTEXT COMPACTION — REFERENCE ONLY] compacted"}, + {"role": "user", "content": "fresh kept question"}, + {"role": "assistant", "content": "fresh kept answer"}, + ] + state_messages = [ + {"role": "user", "content": "old question", "timestamp": "2026-06-22T09:00:00Z"}, + {"role": "assistant", "content": "old answer", "timestamp": "2026-06-22T09:01:00Z"}, + {"role": "user", "content": "fresh kept question", "timestamp": "2026-06-22T09:02:00Z"}, + {"role": "assistant", "content": "fresh kept answer", "timestamp": "2026-06-22T09:03:00Z"}, + {"role": "user", "content": "new after compress", "timestamp": "2026-06-22T09:04:00Z"}, + ] + session = SimpleNamespace( + session_id="iso-anchor-boundary", + context_messages=sidecar_context, + messages=[], + compression_anchor_message_key={ + "role": "assistant", + "text": "fresh kept answer", + "ts": "2026-06-22T09:03:00Z", + "attachments": 0, + }, + ) + + reconciled = reconciled_state_db_messages_for_session( + session, prefer_context=True, state_messages=state_messages + ) + + assert reconciled == sidecar_context + [ + {"role": "user", "content": "new after compress", "timestamp": "2026-06-22T09:04:00Z"}, + ] + + +def test_prefer_context_reconcile_fails_closed_when_compression_anchor_ts_is_missing(): + sidecar_context = [ + {"role": "assistant", "content": "[CONTEXT COMPACTION — REFERENCE ONLY] compacted"}, + {"role": "user", "content": "latest kept question"}, + {"role": "assistant", "content": "latest kept answer"}, + ] + state_messages = [ + {"role": "user", "content": "original long prompt"}, + {"role": "assistant", "content": "old answer", "timestamp": 1.0}, + {"role": "user", "content": "old tool-heavy follow-up", "timestamp": 2.0}, + {"role": "assistant", "content": "old tool-heavy answer", "timestamp": 3.0}, + {"role": "user", "content": "latest kept question", "timestamp": 4.0}, + {"role": "assistant", "content": "latest kept answer", "timestamp": 5.0}, + {"role": "user", "content": "new after compression", "timestamp": 6.0}, + ] + session = SimpleNamespace( + session_id="anchor-ts-missing", + context_messages=sidecar_context, + messages=[], + compression_anchor_message_key={ + "role": "user", + "text": "original long prompt", + "ts": None, + "attachments": 0, + }, + ) + + reconciled = reconciled_state_db_messages_for_session( + session, prefer_context=True, state_messages=state_messages + ) + + assert reconciled == sidecar_context + + def test_prefer_context_reconcile_strips_small_mirrored_context_prefix(): sidecar_context = [ {"role": "user", "content": "[Session Arc Summary] compacted"},