fix(agent): enforce marker-strip invariant with a single terminal sweep (#57491)

Follow-up to the per-site strips from the review gate. The two copy-site
strips are correct but positional — a copy site added after the assembly
loops would re-leak _db_persisted into the child-session flush. Add a single
terminal sweep (_strip_persistence_markers) run once on the fully-assembled
compressed list so the invariant 'no compacted message leaves compress()
carrying a persistence marker' is structural, not dependent on copy-site order.

- agent/context_compressor.py: _strip_persistence_markers() called before
  compress() returns; helper docstring notes the sweep is the authoritative guard
- tests/agent/test_context_compressor.py: structural regression — neuter the
  per-site helper to a leaking copy, assert the terminal sweep still strips
- tests/run_agent/test_compression_persistence.py: pin the fixture assumption
  behind the exact-equality row-count assertion

agent/context_compressor.py

@@ -94,12 +94,37 @@ def _fresh_compaction_message_copy(msg: Dict[str, Any]) -> Dict[str, Any]:
     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).
+
+    This strips at the copy site (clearest intent, and cheap), but the
+    authoritative guarantee is the single terminal sweep in ``compress()``
+    (``_strip_persistence_markers``): no message may leave ``compress()``
+    carrying ``_db_persisted`` regardless of how many intermediate copy sites
+    a future refactor adds.
     """
     fresh = msg.copy()
     fresh.pop(_DB_PERSISTED_MARKER, None)
     return fresh
 
 
+def _strip_persistence_markers(messages: List[Dict[str, Any]]) -> None:
+    """Enforce the compaction invariant: no assembled message carries a
+    session-store persistence marker.
+
+    ``compress()`` copies protected head/tail messages out of the live
+    cached-gateway transcript, which stamps ``_db_persisted`` on every message
+    over the life of the session.  If any copied dict keeps that marker, the
+    rotation flush to the child session skips it and the compacted transcript is
+    lost from ``state.db`` (#57491).  Stripping at each copy site is necessary
+    but *positional* — a copy site added after the assembly loops would re-leak.
+    This single terminal sweep makes the guarantee structural instead: run it
+    once on the fully-assembled list so the invariant holds no matter where the
+    copies happened.  Mutates in place (the dicts are compaction-local copies).
+    """
+    for msg in messages:
+        if isinstance(msg, dict):
+            msg.pop(_DB_PERSISTED_MARKER, None)
+
+
 # Appended to every standalone summary message (and to the merged-into-tail
 # prefix) so the model has an unambiguous "summary ends here" boundary.
 # Without it, weak models read the verbatim "## Active Task" quote as fresh
@@ -2984,4 +3009,10 @@ This compaction should PRIORITISE preserving all information related to the focu
             )
             logger.info("Compression #%d complete", self.compression_count)
 
+        # Enforced invariant (#57491): no compacted message may leave compress()
+        # carrying a session-store persistence marker. The per-site strips above
+        # are positional; this single terminal sweep makes it structural so a
+        # future copy site cannot re-leak the marker into the child-session flush.
+        _strip_persistence_markers(compressed)
+
         return compressed

tests/agent/test_context_compressor.py

@@ -352,6 +352,27 @@ class TestCompress:
         assert len(result) < len(msgs)
         assert all("_db_persisted" not in msg for msg in result)
 
+    def test_compress_terminal_sweep_strips_markers_even_if_a_copy_site_leaks(self, compressor):
+        """Regression for #57491, structural: even if a copy site fails to strip
+        the marker (simulating a future refactor that adds/reintroduces a leaky
+        copy), the single terminal sweep in compress() guarantees no compacted
+        message leaves carrying `_db_persisted`. Neuter the per-site helper to a
+        plain leaking copy and assert the invariant still holds."""
+        import agent.context_compressor as _cc
+
+        msgs = [
+            {"role": "user" if i % 2 == 0 else "assistant", "content": f"m{i}", "_db_persisted": True}
+            for i in range(10)
+        ]
+        # Make the per-site helper leak the marker (dict.copy keeps it).
+        with patch.object(_cc, "_fresh_compaction_message_copy", lambda m: m.copy()), \
+             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), (
+            "terminal sweep must strip _db_persisted even when a copy site leaks"
+        )
+
     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

@@ -207,6 +207,11 @@ class TestFlushAfterCompression:
             agent.compression_in_place = False
             agent._ensure_db_session()
 
+            # Plain marked messages only: the exact-equality assertion below
+            # relies on `compressed` containing no message that _flush filters
+            # for a reason INDEPENDENT of _db_persisted (ephemeral scaffolding,
+            # synthetic recovery turns). Keep this fixture free of such messages
+            # or the row count would legitimately differ from len(compressed).
             messages = [
                 {
                     "role": "user" if i % 2 == 0 else "assistant",