fix(compression): roll the live transcript back when an in-place compaction commit fails (#99477)
`archive_and_compact()` is atomic: when it raises, every pre-compaction row is still `active = 1` and the compacted set was never inserted. The rotation branch already rolled the live transcript back to `messages_before_compression` in that case, but the in-place branch — the default (`compression_in_place` defaults to True) — did not, so `compress_context()` handed the caller the uncommitted compacted list. That list is marker-swept by `_strip_persistence_markers` (#57491) and the post-commit `stamp_db_persisted_markers` (#98450) never ran, so the next append-only flush treated the whole compacted transcript as new and INSERTed it on top of the rows it was supposed to replace. The active set then held the summary AND the turns it summarized: the next resume reloaded both, the token count went up, preflight fired again, and every failed attempt appended another copy of the protected head plus tail. The in-place rollback mirrors the rotation branch and is gated on `split_status != "in_place_committed"`, which is assigned on the statement immediately after the atomic commit returns, so a committed compaction can never be rolled back into a mismatch of the opposite sign. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EKrRS7LVgyHf2WQkEahSwu
This commit is contained in:
joaomarcos
@@ -5148,6 +5148,51 @@ def compress_context(
|
||||
"_proactive_prune_rearm_tokens"
|
||||
]
|
||||
)
|
||||
elif (
|
||||
in_place
|
||||
and split_status != "in_place_committed"
|
||||
and messages_before_compression is not None
|
||||
):
|
||||
# In-place sibling of the rotation rollback above (#99477).
|
||||
# archive_and_compact() is atomic, so a raise before it
|
||||
# returned means EVERY pre-compaction row is still
|
||||
# ``active = 1`` in state.db — nothing was archived and the
|
||||
# compacted set was never inserted. But ``compressed`` is
|
||||
# the marker-swept output of compress()
|
||||
# (_strip_persistence_markers, #57491) and the post-commit
|
||||
# ``stamp_db_persisted_markers`` never ran, so handing it
|
||||
# back makes the next append-only flush treat the whole
|
||||
# compacted transcript as new and INSERT it ON TOP of the
|
||||
# rows it was supposed to replace. The active set then holds
|
||||
# the summary AND the turns it summarized; the next resume
|
||||
# reloads both, the token count goes UP, preflight fires
|
||||
# again, and each failed attempt appends another copy of the
|
||||
# protected head + tail (#99477: ~15 real turns stored as
|
||||
# 3,814 rows, the first user message repeated 893 times).
|
||||
#
|
||||
# Gate on ``split_status`` rather than ``compacted_in_place``:
|
||||
# it is assigned on the statement immediately after the
|
||||
# atomic commit returns, so a committed compaction can never
|
||||
# be rolled back into a live/durable mismatch of the
|
||||
# opposite sign.
|
||||
#
|
||||
# The deepcopy carries each row's _DB_PERSISTED_MARKER from
|
||||
# the pre-compression snapshot, so the restored transcript is
|
||||
# correctly skipped by the flush, and replacing every dict
|
||||
# breaks _db_flush_scan_prefix identity (same reasoning as
|
||||
# the rotation branch — no explicit clear needed).
|
||||
messages[:] = copy.deepcopy(messages_before_compression)
|
||||
compressed = messages
|
||||
_compression_made_progress = False
|
||||
# Runway rolls back with the transcript, exactly as above:
|
||||
# compress() zeroed it in memory, and the durable clear only
|
||||
# rides the archive_and_compact that just failed.
|
||||
if "_proactive_prune_rearm_tokens" in _compressor_attempt_snapshot:
|
||||
agent.context_compressor._proactive_prune_rearm_tokens = (
|
||||
_compressor_attempt_snapshot[
|
||||
"_proactive_prune_rearm_tokens"
|
||||
]
|
||||
)
|
||||
split_status = (
|
||||
"aborted"
|
||||
if locals().get("old_session_id") is None and not in_place
|
||||
|
||||
@@ -0,0 +1,176 @@
|
||||
"""Regression test for #99477: a FAILED in-place compaction commit must roll
|
||||
the live transcript back to its pre-compression snapshot.
|
||||
|
||||
``compress()`` returns marker-swept copies (``_strip_persistence_markers``,
|
||||
#57491) and only the post-commit ``stamp_db_persisted_markers`` (#98450) makes
|
||||
them skippable by the append-only flush. ``archive_and_compact()`` is atomic,
|
||||
so when it raises — a concurrent queued-follow-up drain holding the write path,
|
||||
a lost compression lease, ``database is locked`` — NOTHING was archived and
|
||||
NOTHING was inserted: every pre-compaction row is still ``active = 1``.
|
||||
|
||||
Before this fix the rotation branch rolled the live list back but the in-place
|
||||
branch (the DEFAULT, ``compression_in_place`` defaults to True) did not, so the
|
||||
caller adopted the uncommitted, unstamped compacted list. The next
|
||||
``_persist_session`` walk then INSERTed the whole compacted transcript ON TOP of
|
||||
the rows it was supposed to replace: the active set ended up holding the summary
|
||||
AND the turns it summarized, the next resume reloaded both, the token count went
|
||||
UP, preflight fired again, and every failed attempt appended another copy of the
|
||||
protected head + tail (#99477: ~15 real turns stored as 3,814 rows, the first
|
||||
user message repeated 893 times).
|
||||
"""
|
||||
|
||||
import contextlib
|
||||
import os
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def _session_db(name):
|
||||
"""Real SessionDB on a temp file, closed before the directory is removed.
|
||||
|
||||
Windows holds the SQLite file open until the connection is closed, so
|
||||
letting TemporaryDirectory clean up first raises WinError 32 and masks the
|
||||
assertion result.
|
||||
"""
|
||||
from hermes_state import SessionDB
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
db = SessionDB(db_path=Path(tmp) / name)
|
||||
try:
|
||||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def _make_agent(session_db, session_id):
|
||||
with patch.dict(os.environ, {"OPENROUTER_API_KEY": "test-key"}):
|
||||
from run_agent import AIAgent
|
||||
|
||||
agent = AIAgent(
|
||||
api_key="test-key",
|
||||
base_url="https://openrouter.ai/api/v1",
|
||||
model="test/model",
|
||||
quiet_mode=True,
|
||||
session_db=session_db,
|
||||
session_id=session_id,
|
||||
skip_context_files=True,
|
||||
skip_memory=True,
|
||||
)
|
||||
agent.compression_in_place = True
|
||||
agent._session_db_created = True
|
||||
|
||||
def _fake_compress(messages, current_tokens=None, focus_topic=None, force=False):
|
||||
# Mirrors the real compress() contract: marker-swept fresh dicts.
|
||||
return [
|
||||
{"role": "user", "content": "[CONTEXT COMPACTION] summary of prior turns"},
|
||||
{"role": "assistant", "content": "kept reply 1"},
|
||||
{"role": "user", "content": "kept question"},
|
||||
{"role": "assistant", "content": "kept reply 2"},
|
||||
]
|
||||
|
||||
agent.context_compressor.compress = _fake_compress
|
||||
agent.context_compressor._last_compress_aborted = False
|
||||
agent.context_compressor._last_summary_error = None
|
||||
agent.context_compressor.compression_count = 1
|
||||
return agent
|
||||
|
||||
|
||||
def _seed(db, sid, n=8):
|
||||
db.create_session(sid, "gateway", model="test/model")
|
||||
for i in range(n):
|
||||
db.append_message(
|
||||
session_id=sid,
|
||||
role="user" if i % 2 == 0 else "assistant",
|
||||
content=f"seed msg {i}",
|
||||
)
|
||||
|
||||
|
||||
def _counts(db, sid):
|
||||
total = db._conn.execute(
|
||||
"SELECT COUNT(*) FROM messages WHERE session_id = ?", (sid,)
|
||||
).fetchone()[0]
|
||||
active = db._conn.execute(
|
||||
"SELECT COUNT(*) FROM messages WHERE session_id = ? AND active = 1", (sid,)
|
||||
).fetchone()[0]
|
||||
return total, active
|
||||
|
||||
|
||||
class TestInPlaceCommitFailureRollback:
|
||||
def test_failed_commit_does_not_reinsert_the_transcript(self):
|
||||
"""archive_and_compact raises → live list must return to the snapshot."""
|
||||
from hermes_state import SessionDB
|
||||
from agent.context_compressor import _DB_PERSISTED_MARKER
|
||||
from agent.conversation_compression import compress_context
|
||||
|
||||
with _session_db("rollback.db") as db:
|
||||
sid = "20260831_120000_rollback"
|
||||
_seed(db, sid, n=8)
|
||||
agent = _make_agent(db, sid)
|
||||
|
||||
messages = [
|
||||
{"role": "user" if i % 2 == 0 else "assistant", "content": f"m{i}"}
|
||||
for i in range(8)
|
||||
]
|
||||
agent._flush_messages_to_session_db(messages)
|
||||
assert _counts(db, sid) == (16, 16) # 8 seeded + 8 flushed
|
||||
|
||||
# The realistic trigger: a concurrent queued-follow-up drain holds
|
||||
# the write path, so the atomic commit raises instead of landing.
|
||||
# Raised from the DB method itself, so the compression path's own
|
||||
# error handling runs — nothing is stubbed out around it.
|
||||
def _locked(*_a, **_kw):
|
||||
raise RuntimeError("database is locked (concurrent drain)")
|
||||
|
||||
with patch.object(SessionDB, "archive_and_compact", _locked):
|
||||
compressed, _prompt = compress_context(
|
||||
agent, messages, approx_tokens=900_000, system_message="sys"
|
||||
)
|
||||
|
||||
# ── Precondition: the commit genuinely did not land. Without this
|
||||
# the assertions below would pass vacuously on a path that
|
||||
# actually compacted.
|
||||
assert _counts(db, sid) == (16, 16)
|
||||
|
||||
# ── The fix: the returned transcript is the pre-compression
|
||||
# snapshot, not the uncommitted compacted set.
|
||||
assert [m["content"] for m in compressed] == [
|
||||
f"m{i}" for i in range(8)
|
||||
], "failed in-place commit must not hand back the compacted set"
|
||||
assert all(
|
||||
m.get(_DB_PERSISTED_MARKER) for m in compressed
|
||||
), "restored rows must keep their persistence marker"
|
||||
|
||||
# ── The symptom: the post-compression persist walk must not
|
||||
# re-INSERT anything. Run it twice — multiple exit paths persist.
|
||||
agent._flush_messages_to_session_db(compressed)
|
||||
agent._flush_messages_to_session_db(compressed)
|
||||
assert _counts(db, sid) == (16, 16), (
|
||||
"a failed in-place compaction re-inserted the transcript"
|
||||
)
|
||||
|
||||
def test_successful_commit_still_compacts_in_place(self):
|
||||
"""The rollback must not fire when the commit landed (#98450 guard)."""
|
||||
from hermes_state import SessionDB
|
||||
from agent.context_compressor import _DB_PERSISTED_MARKER
|
||||
from agent.conversation_compression import compress_context
|
||||
|
||||
with _session_db("committed.db") as db:
|
||||
sid = "20260831_120001_committed"
|
||||
_seed(db, sid, n=8)
|
||||
agent = _make_agent(db, sid)
|
||||
agent._last_flushed_db_idx = 8
|
||||
|
||||
messages = [{"role": "user", "content": f"m{i}"} for i in range(8)]
|
||||
compressed, _prompt = compress_context(
|
||||
agent, messages, approx_tokens=900_000, system_message="sys"
|
||||
)
|
||||
|
||||
total, active = _counts(db, sid)
|
||||
assert active == len(compressed) == 4, "in-place commit must compact"
|
||||
assert total == 12 # 8 soft-archived + 4 active
|
||||
assert all(m.get(_DB_PERSISTED_MARKER) for m in compressed)
|
||||
|
||||
agent._flush_messages_to_session_db(compressed)
|
||||
assert _counts(db, sid) == (12, 4)
|
||||
Reference in New Issue
Block a user