From 53559aaf86b84dadae83cd9bb605ca476f9a0606 Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Fri, 31 Jul 2026 17:37:44 +0530 Subject: [PATCH] fix(agent): protect batch-compaction markers from micro supersede/defrag MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 164 +++++++++++++-------------- agent/turn_finalizer.py | 7 ++ tests/agent/test_micro_compaction.py | 121 ++++++++++++++++++++ 3 files changed, 209 insertions(+), 83 deletions(-) diff --git a/agent/context_compressor.py b/agent/context_compressor.py index 7ff6f92bf59..c4a58c10ac8 100644 --- a/agent/context_compressor.py +++ b/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 diff --git a/agent/turn_finalizer.py b/agent/turn_finalizer.py index 5078eed9d3e..86f3f509926 100644 --- a/agent/turn_finalizer.py +++ b/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) diff --git a/tests/agent/test_micro_compaction.py b/tests/agent/test_micro_compaction.py index 543360be4a1..c3528d10f97 100644 --- a/tests/agent/test_micro_compaction.py +++ b/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" + )