diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index 791c3ddb649b..dd7b75796b5c 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -1236,8 +1236,29 @@ def compress_context( # Flush any un-persisted current-turn messages to the OLD # session before ending it, so they survive in the preserved # parent transcript (#47202). (In-place skips this — see above.) + # + # Pass the already-durable prefix as conversation_history so + # the flush skips it by identity (#68196). Preflight + # compression runs BEFORE the normal turn flush has stamped + # the cold-resumed history dicts with _DB_PERSISTED_MARKER, so + # without a boundary _flush_messages_to_session_db treats every + # restored row as new and re-appends the whole transcript to + # the parent. turn_context anchors _persist_user_message_idx at + # the current-turn user message before preflight runs, so + # messages[:idx] is exactly the persisted prefix; only the + # current turn's new messages get written. + current_idx = getattr(agent, "_persist_user_message_idx", None) + persisted_history = ( + messages[:current_idx] + if isinstance(current_idx, int) + and 0 <= current_idx <= len(messages) + else None + ) try: - agent._flush_messages_to_session_db(messages) + agent._flush_messages_to_session_db( + messages, + conversation_history=persisted_history, + ) except Exception: pass # best-effort — don't block compression on a flush error # Propagate title to the new session with auto-numbering diff --git a/tests/agent/test_rotation_flush_persisted_boundary_68196.py b/tests/agent/test_rotation_flush_persisted_boundary_68196.py new file mode 100644 index 000000000000..c082be359149 --- /dev/null +++ b/tests/agent/test_rotation_flush_persisted_boundary_68196.py @@ -0,0 +1,118 @@ +"""Regression (#68196): rotating preflight compression must not re-append the +already-persisted transcript to the parent session on cold resume. + +On the first turn after a cold Desktop resume, the stored rows are handed to +``run_conversation()`` as ``conversation_history`` and live in the message list +as plain dicts that have NOT yet been stamped with ``_DB_PERSISTED_MARKER`` — +the normal turn flush (which stamps them) runs *after* preflight compression. + +The legacy rotation branch in ``agent/conversation_compression.py`` flushes the +current turn to the OLD session before ending it (#47202). It used to call +``agent._flush_messages_to_session_db(messages)`` with no history boundary, so +``_flush_messages_to_session_db`` saw an empty ``history_ids`` set and treated +every restored row as new — durably appending the whole transcript to the +parent a second time. The fix passes ``messages[:_persist_user_message_idx]`` +(the already-durable prefix ``turn_context`` anchors before preflight runs) as +``conversation_history`` so only the current turn's new messages are written. + +Without the fix the parent grows to 5 rows (the two originals + a duplicate of +both + the new turn). With it the parent holds exactly the two originals plus +the single new turn. +""" + +from __future__ import annotations + +import os +import time +from pathlib import Path +from unittest.mock import MagicMock, patch + +from hermes_state import SessionDB + + +def _build_agent_with_db(db: SessionDB, session_id: str): + """Build an AIAgent wired to ``db`` and pinned to ``session_id``. + + Mirrors the helper in ``test_compression_concurrent_fork.py``: stub the + compressor so it returns deterministic output without an LLM call, and pin + ``compression_in_place=False`` so the legacy rotation path is exercised + regardless of the global default. + """ + 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=db, + session_id=session_id, + skip_context_files=True, + skip_memory=True, + ) + + compressor = MagicMock() + + def _compress(*_a, **_kw): + time.sleep(0.01) + return [ + {"role": "user", "content": "[CONTEXT COMPACTION] summary"}, + {"role": "user", "content": "tail"}, + ] + + compressor.compress.side_effect = _compress + compressor.compression_count = 1 + compressor.last_prompt_tokens = 0 + compressor.last_completion_tokens = 0 + compressor._last_summary_error = None + compressor._last_compress_aborted = False + compressor._last_aux_model_failure_model = None + compressor._last_aux_model_failure_error = None + agent.context_compressor = compressor + agent.compression_in_place = False + return agent + + +def _contents(rows): + return [r.get("content") for r in rows] + + +def test_rotation_flush_does_not_duplicate_persisted_prefix(tmp_path: Path) -> None: + """Cold-resume + rotating preflight compression keeps the parent transcript + at (persisted prefix + one new turn) — no second copy of the durable rows.""" + db = SessionDB(db_path=tmp_path / "state.db") + + parent_sid = "COLD_RESUME_PARENT" + db.create_session(parent_sid, source="desktop") + + # Two durable rows already in the parent. + db.append_message(parent_sid, "user", "persisted question") + db.append_message(parent_sid, "assistant", "persisted answer") + + # Cold resume: the stored rows come back as plain dicts, unstamped, and the + # live turn appends one new user message on top. + loaded = db.get_messages_as_conversation(parent_sid) + assert _contents(loaded) == ["persisted question", "persisted answer"] + messages = [*loaded, {"role": "user", "content": "new turn"}] + + agent = _build_agent_with_db(db, parent_sid) + # turn_context anchors this at the current-turn user message before preflight + # compression runs; emulate that anchor. + agent._persist_user_message_idx = len(messages) - 1 + + agent._compress_context(messages, "sys", approx_tokens=120_000) + + # The flush at the rotation boundary lands on the OLD (parent) session, + # which is then ended. Read it back verbatim (include_inactive to be robust + # to the end_session bookkeeping). + parent_rows = db.get_messages_as_conversation(parent_sid, include_inactive=True) + contents = _contents(parent_rows) + + assert contents.count("persisted question") == 1, ( + "Rotation flush re-appended the already-persisted prefix to the parent " + f"(#68196). Parent transcript is {contents!r}; expected the two durable " + "rows plus only the new turn." + ) + assert contents.count("persisted answer") == 1 + assert contents == ["persisted question", "persisted answer", "new turn"]