fix(agent): invalidate flush-scan cursor at the defrag marker-pop sibling site

The micro-compaction defrag pass (_defrag_rolling_summary) rewrites the
newest MICRO marker's content and pops _DB_PERSISTED_MARKER from the
LIVE dict in place — the same in-place pop class finalize_turn's fill
site was fixed for in #75170. Without invalidation the bounded
flush-scan cursor identity-skips the rewritten marker row and the
defragged rolling summary never reaches state.db (resume rehydrates a
stale summary).

The compressor holds no agent reference, so the pop site raises
_flush_scan_cursor_invalidated and the finalize_turn micro-compaction
block consumes it, setting agent._db_flush_scan_prefix = None.

The module-scope pop sites (context_compressor.py:175/224) operate on
fresh copies — identity-breaking by construction — and need no flag.

Follow-up to #75170 (fix-the-class sweep of _DB_PERSISTED_MARKER
in-place pops).

agent/context_compressor.py

@@ -2214,6 +2214,10 @@ class ContextCompressor(ContextEngine):
         self._micro_compact_consecutive_failures: int = 0
         self._micro_compact_last_failure_cursor: int = -1
         self._micro_compact_defrag_threshold_tokens: int = 2000
+        # Set by _defrag_rolling_summary when it pops _DB_PERSISTED_MARKER
+        # from a live dict in place; consumed by finalize_turn to invalidate
+        # the agent's bounded flush-scan cursor (sibling of the #75170 site).
+        self._flush_scan_cursor_invalidated: bool = False
         self._micro_compact_passes: int = 0
         self._micro_compact_tokens_saved_total: int = 0
         # Cadence: run a pass every Nth completed turn. Each pass rewrites
@@ -5376,6 +5380,15 @@ This compaction should PRIORITISE preserving all information related to the focu
                 # Content changed after a possible flush — clear the persisted
                 # stamp so the DB sync/flush rewrites the row.
                 entry.pop(_DB_PERSISTED_MARKER, None)
+                # Sibling of the finalize_turn pop site (#75170): this pop
+                # also strips the marker from a LIVE dict in place, so the
+                # bounded flush-scan cursor would identity-skip the rewritten
+                # marker and the defragged summary would never reach state.db.
+                # The compressor holds no agent reference, so raise a flag the
+                # finalizer consumes to invalidate agent._db_flush_scan_prefix.
+                # (The pop sites at module scope — fresh copies in
+                # strip-marker helpers — break identity and need no flag.)
+                self._flush_scan_cursor_invalidated = True
                 break
         logger.info(
             "Micro-compaction defrag: rolling summary re-summarized "

agent/turn_finalizer.py

@@ -384,6 +384,18 @@ def finalize_turn(
                 ):
                     _before = len(messages)
                     _compacted = _compressor._micro_compact(messages)
+                    # Micro-compaction defrag rewrites the newest MICRO
+                    # marker's content and pops _db_persisted from the live
+                    # dict in place — the sibling of the pop site above. The
+                    # compressor has no agent reference, so it raises a flag
+                    # for us to invalidate the bounded flush-scan cursor;
+                    # otherwise the rewritten marker row is identity-skipped
+                    # and the stale summary persists to state.db.
+                    if getattr(
+                        _compressor, "_flush_scan_cursor_invalidated", False
+                    ):
+                        _compressor._flush_scan_cursor_invalidated = False
+                        agent._db_flush_scan_prefix = None
                     if isinstance(_compacted, list) and _compacted:
                         messages[:] = _compacted
                     _after = len(messages)

tests/agent/test_micro_compaction.py

@@ -752,3 +752,64 @@ class TestMicroCompaction:
         assert not unstamped, (
             "splice must not strip _db_persisted from surviving messages"
         )
+
+
+class TestDefragFlushCursorInvalidation:
+    """Sibling of the finalize_turn pop site (#75170): defrag pops
+    _DB_PERSISTED_MARKER from the live marker dict in place, so the bounded
+    flush-scan cursor must be invalidated or the rewritten summary is
+    identity-skipped and never re-persisted."""
+
+    def _defrag_setup(self):
+        from agent.context_compressor import _DB_PERSISTED_MARKER
+
+        cc = _compressor(summary="FRESH DEFRAGGED SUMMARY")
+        messages = _conversation(exchanges=8)
+        messages = cc._micro_compact(list(messages))
+        # Simulate an incremental flush having stamped the marker row.
+        for m in messages:
+            if m.get(COMPRESSED_SUMMARY_METADATA_KEY):
+                m[_DB_PERSISTED_MARKER] = True
+        cc._micro_compact_rolling_summary = "x" * 40_000  # force defrag
+        return cc, messages
+
+    def test_defrag_marker_pop_raises_invalidation_flag(self):
+        from agent.context_compressor import _DB_PERSISTED_MARKER
+
+        cc, messages = self._defrag_setup()
+        assert cc._flush_scan_cursor_invalidated is False
+
+        result = cc._micro_compact(list(messages))
+
+        markers = _summary_markers(result)
+        assert len(markers) == 1
+        # The pop happened in place on the live dict...
+        assert not markers[0].get(_DB_PERSISTED_MARKER)
+        # ...so the compressor must flag the flush-scan cursor stale.
+        assert cc._flush_scan_cursor_invalidated is True
+
+    def test_no_defrag_no_flag(self):
+        cc = _compressor()
+        messages = _conversation(exchanges=8)
+        cc._micro_compact(list(messages))
+        assert cc._flush_scan_cursor_invalidated is False
+
+    def test_finalizer_consumes_flag_and_invalidates_agent_cursor(self):
+        """finalize_turn's micro-compaction block must translate the
+        compressor flag into agent._db_flush_scan_prefix = None (and reset
+        the flag) so the next flush re-examines the rewritten marker row."""
+        import inspect
+
+        from agent import turn_finalizer
+
+        src = inspect.getsource(turn_finalizer.finalize_turn)
+        micro_block = src.split("Post-turn micro-compaction", 1)[1]
+        micro_block = micro_block.split("agent._persist_session", 1)[0]
+        assert "_flush_scan_cursor_invalidated" in micro_block, (
+            "finalize_turn must consume the compressor's cursor-invalidation "
+            "flag raised by the defrag marker pop"
+        )
+        assert "agent._db_flush_scan_prefix = None" in micro_block, (
+            "finalize_turn must invalidate the bounded flush-scan cursor "
+            "when the defrag pop stripped a live marker's stamp"
+        )