fix(agent): protect batch-compaction markers from micro supersede/defrag

Phase 2 review findings on the salvage branch:

C1 (critical): batch and micro summary markers share
COMPRESSED_SUMMARY_METADATA_KEY, and compress() never reset micro state.
After micro absorbed exchanges 1..k, a batch compaction summarizing
1..m (m>k) could fire; the next micro pass's supersede then dropped the
batch marker (whose content the stale rolling summary does NOT contain)
and archive_and_compact immediately made the loss durable. Defrag had
the same hazard: it rewrote "the newest marker" even if that was a
batch marker. Empirically confirmed with a probe (batch marker content
destroyed in one pass).

Fix, three parts:
- Micro-created markers now carry MICRO_COMPACT_MARKER_KEY; supersede
  and defrag only ever touch micro-tagged markers. Rehydration in
  _resolve_compact_cursor tags the marker it absorbs (containment
  proof), which safely covers adopting a batch marker as the new
  rolling base after a reset.
- compress() success path resets micro rolling summary/cursor state so
  a stale summary can never claim cumulativeness over a batch marker.
- Regression tests for both directions plus the reset.

W4: _splice_micro_compact_result no longer strips _db_persisted stamps
from surviving messages. Micro archives in place under the SAME session
id (unlike batch's child-session rotation, #57491), so surviving stamps
are accurate; stripping them meant an archive_and_compact failure left
every previously-persisted message unstamped and the next append-only
flush re-inserted them all as duplicate active rows.

W5: finalize_turn micro gate now checks agent._persist_disabled —
persistence-isolated fork agents (background review) must not burn an
aux call per review turn, and must never archive_and_compact the
canonical session rows if their compressor ever gains a DB binding.

W1: _serialize_one_exchange now delegates to _serialize_for_summary
(was a ~70-line near-verbatim copy; one serializer, one place to fix).

S4: _find_one_exchange boundary guard rejects only assistant/tool
boundaries (the actual alternation hazard) instead of requiring user —
a stray mid-list system/injected message can no longer wedge the
cursor forever.

5 new regression tests; 38 micro/prune tests, 400 compression-suite
tests, 61 finalize/persist tests pass; ruff clean.

agent/context_compressor.py

@@ -137,6 +137,12 @@ LEGACY_SUMMARY_PREFIX = "[CONTEXT SUMMARY]:"
 # "is_compressed_summary" would reach the wire and trip exactly that.
 COMPRESSED_SUMMARY_METADATA_KEY = "_compressed_summary"
 COMPRESSED_SUMMARY_HAS_USER_TURN_KEY = "_compressed_summary_has_user_turn"
+# Distinguishes rolling micro-compaction markers from batch-compaction
+# markers (both carry COMPRESSED_SUMMARY_METADATA_KEY so resume/handoff
+# treat them alike). Supersede/defrag/rehydration must only ever touch
+# micro markers: a batch marker's content is NOT contained in the micro
+# rolling summary, so dropping or rewriting one destroys history.
+MICRO_COMPACT_MARKER_KEY = "_micro_compact_marker"
 _DB_PERSISTED_MARKER = "_db_persisted"
 
 _NO_USER_TASK_SENTINEL = "None. This session contains no user-authored turns."
@@ -5052,6 +5058,14 @@ This compaction should PRIORITISE preserving all information related to the focu
                 )
                 if recovered:
                     self._micro_compact_rolling_summary = recovered
+                    # Rehydration is containment proof: this marker's text now
+                    # lives inside the rolling summary, so it becomes
+                    # supersede/defrag-eligible. This also covers a BATCH
+                    # marker adopted as the rolling base after a batch
+                    # compaction reset — safe precisely because we just
+                    # absorbed its content. Markers whose content we did NOT
+                    # absorb never get the key and are never dropped.
+                    messages[last_summary_idx][MICRO_COMPACT_MARKER_KEY] = True
                     logger.info(
                         "Micro-compaction: recovered rolling summary from "
                         "transcript (%d chars)", len(recovered),
@@ -5132,17 +5146,22 @@ This compaction should PRIORITISE preserving all information related to the focu
         if idx <= exchange_start:
             return None
 
-        # Splice-boundary guard: the message right after the exchange must be
-        # a user turn (or a summary marker, which stands in for one
-        # structurally).  If the walk stopped because it ran into *tail_start*
-        # mid-turn, splicing here would leave the assistant-role marker
-        # adjacent to the turn's remaining assistant/tool messages — invalid
-        # alternation.  Skip this pass; the tail recedes as the conversation
-        # grows and the turn becomes absorbable later.
+        # Splice-boundary guard: the message right after the exchange must
+        # close the turn. If the walk stopped because it ran into
+        # *tail_start* mid-turn (boundary is assistant or tool — including
+        # an assistant-role summary marker), splicing here would leave the
+        # assistant-role marker adjacent to the turn's remaining
+        # assistant/tool messages — invalid alternation. Skip this pass; the
+        # tail recedes as the conversation grows and the turn becomes
+        # absorbable later. Any other boundary role (user, or a stray
+        # system/injected message) is a safe splice point — the marker is
+        # assistant-role, so no same-role adjacency is possible — and
+        # accepting them keeps one odd message from wedging the cursor
+        # forever.
         if idx >= n:
             return None
         boundary = messages[idx]
-        if not isinstance(boundary, dict) or boundary.get("role") != "user":
+        if not isinstance(boundary, dict) or boundary.get("role") in ("assistant", "tool"):
             return None
         return (exchange_start, idx)
 
@@ -5154,72 +5173,11 @@ This compaction should PRIORITISE preserving all information related to the focu
     ) -> str:
         """Serialize a single exchange for the micro-summarizer.
 
-        Uses the same content-max truncation and redaction as the batch
-        ``_serialize_for_summary`` method, but scoped to one exchange.
+        Delegates to the batch path's ``_serialize_for_summary`` (same
+        truncation, redaction, think-block stripping, and media labeling),
+        scoped to one exchange — one serializer, one place to fix.
         """
-        from agent.agent_runtime_helpers import strip_think_blocks
-
-        parts = []
-        for msg in messages[start:end]:
-            role = msg.get("role", "unknown")
-            content = msg.get("content")
-            if isinstance(content, list):
-                text_parts: list[str] = []
-                for part in content:
-                    if isinstance(part, dict):
-                        ptype = part.get("type")
-                        if ptype == "text":
-                            text_parts.append(part.get("text", ""))
-                        elif ptype in {"image", "image_url", "input_image"}:
-                            text_parts.append(_image_part_label(part))
-                        else:
-                            text_parts.append(f"[{ptype or 'attachment'}]")
-                    elif isinstance(part, str):
-                        text_parts.append(part)
-                content = "\n".join(text_parts)
-            content = _redact_compaction_text(content or "")
-            content = _MEDIA_DIRECTIVE_RE.sub("[media attachment]", content)
-            if role == "assistant" and content:
-                content = strip_think_blocks(None, content)
-
-            if role == "tool":
-                tool_id = msg.get("tool_call_id", "")
-                if len(content) > self._CONTENT_MAX:
-                    content = content[:self._CONTENT_HEAD] + "\n...[truncated]...\n" + content[-self._CONTENT_TAIL:]
-                parts.append(f"[TOOL RESULT {tool_id}]: {content}")
-                continue
-
-            if role == "assistant":
-                if len(content) > self._CONTENT_MAX:
-                    content = content[:self._CONTENT_HEAD] + "\n...[truncated]...\n" + content[-self._CONTENT_TAIL:]
-                tool_calls = msg.get("tool_calls", [])
-                if tool_calls:
-                    tc_parts = []
-                    for tc in tool_calls:
-                        if isinstance(tc, dict):
-                            fn = tc.get("function", {})
-                            name = fn.get("name", "?")
-                            args = _redact_compaction_text(fn.get("arguments", ""))
-                            if len(args) > self._TOOL_ARGS_MAX:
-                                args = args[:self._TOOL_ARGS_HEAD] + "..."
-                            tc_parts.append(f"  {name}({args})")
-                        else:
-                            fn = getattr(tc, "function", None)
-                            name = getattr(fn, "name", "?") if fn else "?"
-                            tc_parts.append(f"  {name}(...)")
-                    content += "\n[Tool calls:\n" + "\n".join(tc_parts) + "\n]"
-                parts.append(f"[ASSISTANT]: {content}")
-                continue
-
-            if role == "user":
-                if len(content) > self._CONTENT_MAX:
-                    content = content[:self._CONTENT_HEAD] + "\n...[truncated]...\n" + content[-self._CONTENT_TAIL:]
-                parts.append(f"[USER]: {content}")
-                continue
-
-            parts.append(f"[{role.upper()}]: {content}")
-
-        return "\n---\n".join(parts)
+        return self._serialize_for_summary(messages[start:end])
 
     def _build_micro_summary_prompt(
         self,
@@ -5349,11 +5307,17 @@ This compaction should PRIORITISE preserving all information related to the focu
             self._micro_compact_rolling_summary = old_summary
             return False
         self._micro_compact_rolling_summary = fresh_summary
-        # Rewrite the newest marker's content in place so the transcript and
-        # the in-memory summary stay in step (resume rehydrates from it).
+        # Rewrite the newest MICRO marker's content in place so the transcript
+        # and the in-memory summary stay in step (resume rehydrates from it).
+        # Scoped to micro-tagged markers: rewriting a batch-compaction marker
+        # would overwrite history the rolling summary does not contain.
         for idx in range(len(messages) - 1, -1, -1):
             entry = messages[idx]
-            if isinstance(entry, dict) and entry.get(COMPRESSED_SUMMARY_METADATA_KEY):
+            if (
+                isinstance(entry, dict)
+                and entry.get(COMPRESSED_SUMMARY_METADATA_KEY)
+                and entry.get(MICRO_COMPACT_MARKER_KEY)
+            ):
                 entry["content"] = self._render_micro_marker_content(fresh_summary)
                 # Content changed after a possible flush — clear the persisted
                 # stamp so the DB sync/flush rewrites the row.
@@ -5700,6 +5664,10 @@ This compaction should PRIORITISE preserving all information related to the focu
             "role": "assistant",
             "content": self._render_micro_marker_content(summary_text),
             COMPRESSED_SUMMARY_METADATA_KEY: True,
+            # Micro-created marker: eligible for supersede/defrag rewrites.
+            # Batch markers never carry this key and are never touched —
+            # their content is not contained in the rolling summary.
+            MICRO_COMPACT_MARKER_KEY: True,
             # Honest provenance (#64650): this marker absorbs only
             # assistant/tool content — user turns are never micro-compacted,
             # so they remain in the transcript and _transcript_has_real_user_turn
@@ -5715,22 +5683,39 @@ This compaction should PRIORITISE preserving all information related to the focu
         # its own prefix/heading/end-marker scaffolding — so the transcript
         # grows with every turn instead of shrinking, which defeats the point.
         # Keep only the newest marker.
-        # Only drop earlier markers when this one demonstrably contains them:
-        # the rolling summary must have been non-empty going into this pass.
-        # A pass that started from nothing (a resume that could not rehydrate)
-        # produces a marker covering one exchange, and dropping the previous
-        # marker would throw away the entire compacted history.
+        # Two containment gates before dropping an earlier marker:
+        # 1. supersede (the rolling summary was non-empty going into this
+        #    pass) — a pass that started from nothing (a resume that could
+        #    not rehydrate) covers one exchange, and dropping the previous
+        #    marker would throw away the entire compacted history.
+        # 2. MICRO_COMPACT_MARKER_KEY on the candidate — only markers whose
+        #    text is provably inside the rolling summary (created by our own
+        #    splice, or rehydrated into the summary by
+        #    _resolve_compact_cursor) carry it. A batch-compaction marker
+        #    that landed after our last pass holds MORE history than the
+        #    stale rolling summary; dropping it would destroy that history.
         if supersede:
             marker_idxs = [
                 i for i, m in enumerate(result)
-                if isinstance(m, dict) and m.get(COMPRESSED_SUMMARY_METADATA_KEY)
+                if isinstance(m, dict)
+                and m.get(COMPRESSED_SUMMARY_METADATA_KEY)
+                and m.get(MICRO_COMPACT_MARKER_KEY)
             ]
             if len(marker_idxs) > 1:
                 superseded = set(marker_idxs[:-1])
                 result = [m for i, m in enumerate(result) if i not in superseded]
                 result = self._merge_adjacent_user_turns(result)
 
-        _strip_persistence_markers(result)
+        # NOTE: deliberately NO _strip_persistence_markers here. The batch
+        # path strips because compress() copies head/tail into a rotated
+        # child session (#57491); micro-compaction archives in place under
+        # the SAME session id, and the surviving dicts' _db_persisted stamps
+        # are accurate. Stripping them meant an archive_and_compact failure
+        # left every previously-persisted message unstamped, and the next
+        # append-only flush re-inserted them as duplicate active rows on top
+        # of the still-active originals. _sync_micro_compact_to_db re-stamps
+        # everything after a SUCCESSFUL archive; on failure the old stamps
+        # keep the flush idempotent (only the new marker row is appended).
         return result
 
     @staticmethod
@@ -6468,6 +6453,19 @@ This compaction should PRIORITISE preserving all information related to the focu
         _strip_persistence_markers(compressed)
         self._last_compression_made_progress = True
 
+        # Batch compaction invalidates micro-compaction state: the batch
+        # marker now holds MORE history than the in-memory rolling summary
+        # (it summarized everything in the window, including exchanges micro
+        # never absorbed). Keeping the stale summary would let the next micro
+        # pass supersede-drop or defrag-rewrite content it does not contain.
+        # Reset instead; the next micro pass rehydrates from the batch marker
+        # via _resolve_compact_cursor, which re-tags it as micro-eligible
+        # only after absorbing its content into the rolling summary.
+        self._micro_compact_rolling_summary = ""
+        self._micro_compact_cursor = 0
+        self._micro_compact_consecutive_failures = 0
+        self._micro_compact_last_failure_cursor = -1
+
         return compressed
 
 

agent/turn_finalizer.py

@@ -368,6 +368,13 @@ def finalize_turn(
                     and getattr(_compressor, '_micro_compact_enabled', False) is True
                     and callable(getattr(_compressor, '_micro_compact', None))
                     and final_response
+                    # Persistence-isolated agents (background review fork)
+                    # must not micro-compact: the pass burns a real aux-LLM
+                    # call on a throwaway replay transcript, and if the
+                    # compressor ever holds a session_db binding it would
+                    # archive_and_compact the CANONICAL session rows — the
+                    # exact write class _persist_disabled exists to stop.
+                    and not getattr(agent, "_persist_disabled", False)
                 ):
                     _before = len(messages)
                     _compacted = _compressor._micro_compact(messages)

tests/agent/test_micro_compaction.py

@@ -631,3 +631,124 @@ class TestMicroCompaction:
         marker = _summary_markers(result)[0]
 
         assert marker[COMPRESSED_SUMMARY_HAS_USER_TURN_KEY] is False
+
+    def test_supersede_never_drops_a_batch_compaction_marker(self):
+        """A batch marker holds history the rolling summary does NOT contain.
+
+        Sequence: micro absorbs some exchanges (rolling summary = those k
+        exchanges only), then batch compaction fires and its marker
+        summarizes MORE (exchanges 1..m). The next micro pass's supersede
+        must not treat the batch marker as redundant — dropping it destroys
+        everything batch summarized beyond exchange k. Only micro-tagged
+        markers (whose text is provably inside the rolling summary) may be
+        superseded.
+        """
+        cc = _compressor(summary="MICRO SUMMARY (exchanges 1..k only)")
+        msgs = _conversation(exchanges=8)
+        msgs = cc._micro_compact(msgs)
+        assert cc._micro_compact_rolling_summary
+
+        # Simulate a batch-compaction marker replacing the middle (batch
+        # markers carry the shared metadata key but NOT the micro tag).
+        batch_marker = {
+            "role": "user",
+            "content": "[batch summary] CRITICAL HISTORY: exchanges 1..m",
+            COMPRESSED_SUMMARY_METADATA_KEY: True,
+        }
+        micro_idx = next(
+            i for i, m in enumerate(msgs)
+            if m.get(COMPRESSED_SUMMARY_METADATA_KEY)
+        )
+        msgs = msgs[:micro_idx] + [batch_marker] + msgs[micro_idx + 3:]
+
+        out = cc._micro_compact(msgs)
+
+        assert any(
+            "CRITICAL HISTORY" in str(m.get("content")) for m in out
+        ), "batch-compaction summary destroyed by micro supersede"
+
+    def test_defrag_never_rewrites_a_batch_compaction_marker(self):
+        """Defrag rewrites only micro-tagged markers, never batch markers."""
+        cc = _compressor(summary="DEFRAGGED")
+        msgs = [{"role": "system", "content": "sys"}]
+        msgs.append({
+            "role": "user",
+            "content": "[batch summary] CRITICAL HISTORY: exchanges 1..m",
+            COMPRESSED_SUMMARY_METADATA_KEY: True,
+        })
+        for i in range(6):
+            msgs.append({"role": "user", "content": f"q{i}"})
+            msgs.append({"role": "assistant", "content": f"a{i} " + "z" * 400})
+
+        cc._micro_compact_rolling_summary = "x" * 40_000  # force defrag
+        result = cc._micro_compact(list(msgs))
+
+        batch = [m for m in result if "CRITICAL HISTORY" in str(m.get("content"))]
+        assert batch, "batch marker content overwritten by defrag"
+
+    def test_batch_compress_resets_micro_state(self):
+        """compress() success path invalidates the stale rolling summary.
+
+        Without the reset, the in-memory micro summary (exchanges 1..k)
+        outlives a batch compaction whose marker covers 1..m — the next
+        micro pass would then treat its stale summary as cumulative.
+        """
+        cc = _compressor()
+        msgs = _conversation(exchanges=8)
+        msgs = cc._micro_compact(msgs)
+        assert cc._micro_compact_rolling_summary
+        assert cc._micro_compact_cursor > 0
+
+        cc.compress(msgs, force=True)
+
+        assert cc._micro_compact_rolling_summary == ""
+        assert cc._micro_compact_cursor == 0
+
+    def test_persist_disabled_agent_never_micro_compacts(self):
+        """finalize_turn must skip micro-compaction on isolated fork agents.
+
+        The background-review fork sets _persist_disabled=True; running a
+        pass there burns an aux-LLM call on a throwaway replay transcript
+        and, if the compressor ever holds a DB binding, would
+        archive_and_compact the CANONICAL session rows.
+        """
+        import inspect
+
+        from agent import turn_finalizer
+
+        src = inspect.getsource(turn_finalizer.finalize_turn)
+        micro_block = src.split("Post-turn micro-compaction", 1)[1]
+        # Scope to the micro block only: stop at the persist call that follows.
+        micro_block = micro_block.split("agent._persist_session", 1)[0]
+        assert "_persist_disabled" in micro_block, (
+            "micro-compaction gate must check agent._persist_disabled"
+        )
+
+    def test_splice_preserves_db_persisted_stamps(self):
+        """Surviving messages keep their _db_persisted stamps through a splice.
+
+        Micro-compaction archives in place under the SAME session id, so the
+        stamps on untouched messages stay accurate. Stripping them (as the
+        batch path does for its child-session rotation) meant an
+        archive_and_compact failure left every previously-persisted message
+        unstamped and the next append-only flush re-inserted them all as
+        duplicate active rows.
+        """
+        from agent.context_compressor import _DB_PERSISTED_MARKER
+
+        cc = _compressor()
+        messages = _conversation(exchanges=8)
+        for m in messages:
+            m[_DB_PERSISTED_MARKER] = True
+
+        # No DB bound -> _sync_micro_compact_to_db no-ops (the failure shape).
+        result = cc._micro_compact(messages)
+
+        unstamped = [
+            m for m in result
+            if not m.get(_DB_PERSISTED_MARKER)
+            and not m.get(COMPRESSED_SUMMARY_METADATA_KEY)
+        ]
+        assert not unstamped, (
+            "splice must not strip _db_persisted from surviving messages"
+        )