diff --git a/agent/context_compressor.py b/agent/context_compressor.py index 841564a4b2..bcb9e01cef 100644 --- a/agent/context_compressor.py +++ b/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 " diff --git a/agent/turn_finalizer.py b/agent/turn_finalizer.py index 992adcdef1..d4d6a23e78 100644 --- a/agent/turn_finalizer.py +++ b/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) diff --git a/tests/agent/test_micro_compaction.py b/tests/agent/test_micro_compaction.py index c3528d10f9..0292bf51a2 100644 --- a/tests/agent/test_micro_compaction.py +++ b/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" + )