fix(agent): pass persisted-prefix boundary when rotation flushes on cold resume (#68196)

The legacy rotation branch in agent/conversation_compression.py flushes the
current turn to the OLD session before ending it (#47202) via
_flush_messages_to_session_db(messages) with no conversation_history boundary.

On the first turn after a cold Desktop resume, the restored transcript rows
live in the message list as plain dicts that have not yet been stamped with
_DB_PERSISTED_MARKER — the normal turn flush that stamps them runs after
preflight compression. With no boundary, _flush_messages_to_session_db builds
an empty history_ids set and treats every restored row as new, durably
re-appending the whole transcript to the parent session. Repeated
restart/resume + threshold compression keeps growing the parent transcript.

Pass messages[:_persist_user_message_idx] (the already-durable prefix that
turn_context anchors before preflight runs, guarded for int/bounds) as
conversation_history so the flush skips the persisted rows by identity and
writes only the current turn's new messages.

Adds a regression test that pre-populates SQLite, cold-loads the transcript,
appends one current user row, and forces rotating compression: it fails before
this change (parent grows to 5 rows) and passes after (parent holds the two
originals plus the single new turn).

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

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"]