fix(agent): strip _db_persisted when assembling rotation compression transcript (#57491)

Shallow messages[i].copy() during context compression propagated the
_db_persisted marker from cached gateway incremental flushes into the
post-rotation compressed list. _flush_messages_to_session_db then skipped
every row when writing to the new child session, so gateway restarts
lost the compacted transcript (severe amnesia).

Strip the marker in _fresh_compaction_message_copy() and add regression
tests for rotation flush + compressor assembly.

Fixes #57491

agent/context_compressor.py

@@ -84,6 +84,21 @@ LEGACY_SUMMARY_PREFIX = "[CONTEXT SUMMARY]:"
 # poisoning every subsequent request in the session — a bare key like
 # "is_compressed_summary" would reach the wire and trip exactly that.
 COMPRESSED_SUMMARY_METADATA_KEY = "_compressed_summary"
+_DB_PERSISTED_MARKER = "_db_persisted"
+
+
+def _fresh_compaction_message_copy(msg: Dict[str, Any]) -> Dict[str, Any]:
+    """Copy a message for compaction assembly without persistence markers.
+
+    Live cached-gateway transcripts stamp ``_db_persisted`` during incremental
+    flushes.  Shallow ``.copy()`` propagates that marker into the post-rotation
+    compressed list, so ``_flush_messages_to_session_db`` skips every row when
+    writing to the new child session (#57491).
+    """
+    fresh = msg.copy()
+    fresh.pop(_DB_PERSISTED_MARKER, None)
+    return fresh
+
 
 # Appended to every standalone summary message (and to the merged-into-tail
 # prefix) so the model has an unambiguous "summary ends here" boundary.
@@ -2834,7 +2849,7 @@ This compaction should PRIORITISE preserving all information related to the focu
         # Phase 4: Assemble compressed message list
         compressed = []
         for i in range(compress_start):
-            msg = messages[i].copy()
+            msg = _fresh_compaction_message_copy(messages[i])
             if i == 0 and msg.get("role") == "system":
                 existing = msg.get("content")
                 _compression_note = "[Note: Some earlier conversation turns have been compacted into a handoff summary to preserve context space. The current session state may still reflect earlier work, so build on that summary and state rather than re-doing work. Your persistent memory (MEMORY.md, USER.md) remains fully authoritative regardless of compaction.]"
@@ -2907,7 +2922,7 @@ This compaction should PRIORITISE preserving all information related to the focu
             })
 
         for i in range(compress_end, n_messages):
-            msg = messages[i].copy()
+            msg = _fresh_compaction_message_copy(messages[i])
             if _merge_summary_into_tail and i == compress_end:
                 # Merge the summary into the first tail message, but place
                 # the END MARKER at the very end so the model sees an

tests/agent/test_context_compressor.py

@@ -341,6 +341,17 @@ class TestCompress:
         # original content is present in either case.
         assert msgs[-2]["content"] in result[-2]["content"]
 
+    def test_compress_strips_db_persisted_from_assembled_messages(self, compressor):
+        """Regression for #57491: shallow copies must not carry flush markers."""
+        msgs = [
+            {"role": "user" if i % 2 == 0 else "assistant", "content": f"m{i}", "_db_persisted": True}
+            for i in range(10)
+        ]
+        with patch("agent.context_compressor.call_llm", side_effect=RuntimeError("no provider")):
+            result = compressor.compress(msgs)
+        assert len(result) < len(msgs)
+        assert all("_db_persisted" not in msg for msg in result)
+
     def test_protect_first_n_decays_after_first_compression(self):
         """Regression for #11996: protect_first_n must protect early turns on
         the FIRST compaction but DECAY afterwards, so the same early user

tests/run_agent/test_compression_persistence.py

@@ -191,6 +191,48 @@ class TestFlushAfterCompression:
                 "final answer",
             ]
 
+    def test_rotation_child_session_flushes_full_compressed_transcript_with_markers(self):
+        """Regression for #57491: live cached-agent markers must not block child flush."""
+        from agent.conversation_compression import compress_context
+        from hermes_state import SessionDB
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            db_path = Path(tmpdir) / "test.db"
+            db = SessionDB(db_path=db_path)
+            parent_sid = "20260701_152840_parent"
+            db.create_session(parent_sid, "gateway", model="test/model")
+
+            agent = self._make_agent(db)
+            agent.session_id = parent_sid
+            agent.compression_in_place = False
+            agent._ensure_db_session()
+
+            messages = [
+                {
+                    "role": "user" if i % 2 == 0 else "assistant",
+                    "content": f"message {i}",
+                    "_db_persisted": True,
+                }
+                for i in range(12)
+            ]
+
+            with patch("agent.context_compressor.call_llm", side_effect=RuntimeError("no provider")):
+                compressed, _ = compress_context(
+                    agent, messages, approx_tokens=100_000, system_message="sys"
+                )
+
+            assert agent.session_id != parent_sid
+            child_sid = agent.session_id
+
+            agent._flush_messages_to_session_db(compressed, None)
+
+            child_rows = db.get_messages(child_sid)
+            assert len(child_rows) == len(compressed), (
+                f"Expected {len(compressed)} rows in child session, got {len(child_rows)}. "
+                f"_db_persisted marker propagation bug (#57491)."
+            )
+            db.close()
+
 
 # ---------------------------------------------------------------------------
 # Part 2: Gateway-side — history_offset after session split