David/HERMES-GIELIE
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(compression): guard state replay after compacted anchors

Browse source
This commit is contained in:
Frank Song 2026-06-22 17:55:01 +08:00 committed by nesquena-hermes
parent 1e47baf0f0
commit 0567b056c0
2 changed files with 266 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

108
api/models.py
View file

@ -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,

158
tests/test_issue1217_transcript_compaction.py
View file

@ -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"},