From e4b3b91b6266c5c873e080aa5b6a343558648b5f Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Thu, 13 Aug 2026 01:41:07 -0700 Subject: [PATCH] fix(compression): prune pre-checkpoint history on native compaction replay MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Live verification (gpt-5.6 @ api.openai.com) proved the Responses server renders NOTHING placed before a replayed compaction checkpoint: a fact stated in a pre-checkpoint input item is invisible to the model, while the same item after the checkpoint recalls perfectly. Hermes was replaying the full pre-checkpoint transcript anyway — dead upload weight, and worse, every plaintext user ask from before the boundary silently vanished from the model's view, surviving only inside the opaque server summary. That is the goal-drift failure mode reported against native compaction sessions. Codex CLI never hits this because it rebuilds history client-side after compaction, retaining user messages verbatim under a token budget. This change is the wire-level equivalent: when a replayed checkpoint is present, _chat_messages_to_responses_input restructures the input as [newest checkpoint run] + [retained pre-checkpoint user messages, newest-first within a 64K-token budget] + [post-checkpoint tail] Histories without a checkpoint are returned unchanged, so non-native sessions see a byte-identical wire. --- agent/codex_responses_adapter.py | 13 ++- agent/native_compaction.py | 122 ++++++++++++++++++++++ tests/run_agent/test_native_compaction.py | 120 +++++++++++++++++++++ 3 files changed, 254 insertions(+), 1 deletion(-) diff --git a/agent/codex_responses_adapter.py b/agent/codex_responses_adapter.py index 074fa942f0faa..2b431cff720b9 100644 --- a/agent/codex_responses_adapter.py +++ b/agent/codex_responses_adapter.py @@ -690,7 +690,18 @@ def _chat_messages_to_responses_input( "output": output_value, }) - return items + # Native server-side compaction: when a replayed checkpoint is present, + # restructure the wire around it. The server renders nothing placed + # before a compaction item (live-verified Aug 2026), so pre-checkpoint + # history is dead upload weight and — worse — the user's plaintext asks + # from before the boundary silently vanish from the model's view. Keep + # the newest checkpoint first, retain pre-checkpoint USER messages + # verbatim within a token budget (Codex CLI parity), and leave the + # post-checkpoint tail untouched. Self-gating: histories without a + # checkpoint (every non-native session) return unchanged. + from agent.native_compaction import prune_pre_checkpoint_items + + return prune_pre_checkpoint_items(items) # --------------------------------------------------------------------------- diff --git a/agent/native_compaction.py b/agent/native_compaction.py index 10395c462b321..caceda8aeda6e 100644 --- a/agent/native_compaction.py +++ b/agent/native_compaction.py @@ -144,6 +144,128 @@ def native_compaction_context_management( return [{"type": "compaction", "compact_threshold": threshold}] +# Retention budget for plaintext user messages carried across a native +# compaction boundary (mirrors Codex CLI's RETAINED_MESSAGE_TOKEN_BUDGET). +# Live verification (Aug 2026, gpt-5.6 @ api.openai.com): the server renders +# NOTHING placed before a replayed compaction checkpoint — a fact stated in a +# pre-checkpoint input item is invisible to the model ("NONE" recall), while +# the same item placed after the checkpoint recalls perfectly. Without +# retention, every plaintext user ask from before the compaction survives +# only as whatever the opaque server summary kept — the goal-drift failure +# mode. Codex CLI solves this by rebuilding history with user messages +# retained verbatim; ``prune_pre_checkpoint_items`` is our wire-level +# equivalent. +RETAINED_USER_MESSAGE_TOKEN_BUDGET = 64_000 + + +def _approx_tokens(text: str) -> int: + """Cheap chars//4 token estimate — same shape Codex uses for retention.""" + return max(1, len(text) // 4) + + +def _user_item_text(item: Dict[str, Any]) -> Optional[str]: + """Extract the retained-budget text of a user-role input item. + + Returns None when the item carries no measurable text (empty message). + Multimodal list content is measured by its ``input_text`` parts; images + count as zero, matching Codex's retention accounting. + """ + content = item.get("content") + if isinstance(content, str): + return content if content.strip() else None + if isinstance(content, list): + text = "".join( + part.get("text", "") + for part in content + if isinstance(part, dict) and part.get("type") == "input_text" + ) + return text if text.strip() or content else None + return None + + +def prune_pre_checkpoint_items( + items: List[Dict[str, Any]], + retained_user_token_budget: int = RETAINED_USER_MESSAGE_TOKEN_BUDGET, +) -> List[Dict[str, Any]]: + """Restructure Responses input around the newest compaction checkpoint. + + The server drops every input item that precedes a replayed ``compaction`` + item (live-verified Aug 2026), so sending pre-checkpoint history is dead + weight AND silently erases the user's plaintext asks. When a checkpoint + is present, rebuild the wire as:: + + [checkpoint run] + [retained user messages (newest-first budget)] + [post] + + - The NEWEST contiguous run of checkpoints wins (the server can emit + more than one compaction item in a single response — live-observed + Aug 2026 — and they arrive adjacent; a run from a newer response + cumulatively carries prior windows, so older runs are dropped). + - Retained user messages are the user-role items from before the + checkpoint, kept verbatim newest-first within + ``retained_user_token_budget``; the boundary message is head-truncated + when it only partially fits (string content only). + - Everything after the checkpoint is untouched, so function_call / + function_call_output pairing is preserved (a checkpoint is captured on + an assistant response, and that response's own calls and their outputs + are all emitted after its reasoning items). + - No checkpoint in ``items`` → returned unchanged (self-gating: non-native + routes and kill-switched sessions never see a restructured wire). + + Deterministic for a given history, so the request prefix stays stable + across turns and server-side prompt caching keeps working. + """ + last_cp = None + for i, item in enumerate(items): + if isinstance(item, dict) and item.get("type") == "compaction": + last_cp = i + if last_cp is None: + return items + + # Extend backwards over the contiguous run ending at last_cp. + first_cp = last_cp + while ( + first_cp > 0 + and isinstance(items[first_cp - 1], dict) + and items[first_cp - 1].get("type") == "compaction" + ): + first_cp -= 1 + + pre = items[:first_cp] + checkpoint_run = items[first_cp : last_cp + 1] + post = items[last_cp + 1 :] + + retained_reversed: List[Dict[str, Any]] = [] + remaining = max(0, int(retained_user_token_budget)) + for item in reversed(pre): + if not isinstance(item, dict) or item.get("role") != "user": + continue + # Skip typed items (function_call_output etc. never carry role=user, + # but stay defensive about future shapes). + if "type" in item and item.get("type") != "message": + continue + if remaining <= 0: + break + text = _user_item_text(item) + if text is None: + continue + cost = _approx_tokens(text) + if cost <= remaining: + retained_reversed.append(item) + remaining -= cost + elif isinstance(item.get("content"), str): + # Head-truncate the boundary message: goals are usually stated + # up front, so the head is the valuable end. + truncated = dict(item) + truncated["content"] = item["content"][: remaining * 4] + if truncated["content"].strip(): + retained_reversed.append(truncated) + remaining = 0 + # Multimodal boundary message that doesn't fit whole: skip rather + # than rewrite parts. + + return checkpoint_run + list(reversed(retained_reversed)) + post + + def is_native_compaction_rejection(error: Any) -> bool: """True when a provider error names the context_management field. diff --git a/tests/run_agent/test_native_compaction.py b/tests/run_agent/test_native_compaction.py index 55956758be959..24aaee5aaaca0 100644 --- a/tests/run_agent/test_native_compaction.py +++ b/tests/run_agent/test_native_compaction.py @@ -405,3 +405,123 @@ class TestAgentInitConfig: agent.codex_responses_native_compaction = True kwargs = agent._build_api_kwargs([{"role": "user", "content": "hi"}]) assert "context_management" not in kwargs + + +class TestPrunePreCheckpointItems: + """Wire restructure around a replayed checkpoint (live-verified Aug 2026: + the server renders nothing placed before a compaction input item).""" + + def _items(self): + return [ + {"role": "user", "content": "goal: build the runner"}, + {"role": "assistant", "content": "ack"}, + {"role": "user", "content": "constraint: no sleep-polling"}, + {"type": "function_call", "call_id": "c1", "name": "t", "arguments": "{}"}, + {"type": "function_call_output", "call_id": "c1", "output": "out"}, + {"type": "compaction", "encrypted_content": "blobA"}, + {"role": "assistant", "content": "post-cp answer"}, + {"role": "user", "content": "next ask"}, + ] + + def test_no_checkpoint_is_identity(self): + from agent.native_compaction import prune_pre_checkpoint_items + + items = [i for i in self._items() if i.get("type") != "compaction"] + assert prune_pre_checkpoint_items(list(items)) == items + + def test_checkpoint_leads_and_pre_history_dropped(self): + from agent.native_compaction import prune_pre_checkpoint_items + + out = prune_pre_checkpoint_items(self._items()) + assert out[0] == {"type": "compaction", "encrypted_content": "blobA"} + # Pre-checkpoint assistant + tool traffic is gone (server never saw it + # anyway); post-checkpoint tail is intact and ordered. + assert {"role": "assistant", "content": "ack"} not in out + assert all(i.get("call_id") != "c1" for i in out if isinstance(i, dict)) + assert out[-2:] == self._items()[-2:] + + def test_pre_checkpoint_user_messages_retained_in_order(self): + from agent.native_compaction import prune_pre_checkpoint_items + + out = prune_pre_checkpoint_items(self._items()) + users = [i["content"] for i in out if i.get("role") == "user"] + assert users == [ + "goal: build the runner", + "constraint: no sleep-polling", + "next ask", + ] + # Retained users sit between the checkpoint and the post tail. + assert out.index({"role": "user", "content": "goal: build the runner"}) == 1 + + def test_newest_checkpoint_run_wins(self): + from agent.native_compaction import prune_pre_checkpoint_items + + items = [ + {"type": "compaction", "encrypted_content": "old"}, + {"role": "user", "content": "mid ask"}, + {"type": "compaction", "encrypted_content": "newA"}, + {"type": "compaction", "encrypted_content": "newB"}, + {"role": "user", "content": "tail ask"}, + ] + out = prune_pre_checkpoint_items(items) + blobs = [i["encrypted_content"] for i in out if i.get("type") == "compaction"] + assert blobs == ["newA", "newB"] + assert [i["content"] for i in out if i.get("role") == "user"] == [ + "mid ask", + "tail ask", + ] + + def test_retention_budget_newest_first_with_truncation(self): + from agent.native_compaction import prune_pre_checkpoint_items + + old = {"role": "user", "content": "x" * 4000} # ~1000 tokens + newer = {"role": "user", "content": "y" * 2000} # ~500 tokens + items = [old, newer, {"type": "compaction", "encrypted_content": "b"}] + out = prune_pre_checkpoint_items(items, retained_user_token_budget=600) + users = [i["content"] for i in out if i.get("role") == "user"] + # Newest kept whole; boundary (older) head-truncated to remaining budget. + assert users[-1] == "y" * 2000 + assert users[0] == "x" * 400 # (600-500)*4 chars + assert out[0]["type"] == "compaction" + + def test_zero_budget_keeps_only_post_tail(self): + from agent.native_compaction import prune_pre_checkpoint_items + + out = prune_pre_checkpoint_items(self._items(), retained_user_token_budget=0) + users = [i["content"] for i in out if i.get("role") == "user"] + assert users == ["next ask"] + + def test_adapter_applies_prune_end_to_end(self): + from agent.codex_responses_adapter import _chat_messages_to_responses_input + + msgs = [ + {"role": "user", "content": "the goal"}, + { + "role": "assistant", + "content": "ok", + "codex_reasoning_items": [ + {"type": "compaction", "encrypted_content": "blob"} + ], + }, + {"role": "user", "content": "follow-up"}, + ] + items = _chat_messages_to_responses_input(msgs) + assert items[0] == {"type": "compaction", "encrypted_content": "blob"} + users = [i["content"] for i in items if i.get("role") == "user"] + assert users == ["the goal", "follow-up"] + # The checkpoint's own turn content is emitted AFTER the checkpoint + # in wire order (sidecar items lead the assistant branch), so it + # survives in the post tail — call pairing for that turn is intact. + assert {"role": "assistant", "content": "ok"} in items + assert items.index({"role": "assistant", "content": "ok"}) > 0 + + def test_adapter_without_checkpoint_unchanged_shape(self): + from agent.codex_responses_adapter import _chat_messages_to_responses_input + + msgs = [ + {"role": "user", "content": "hi"}, + {"role": "assistant", "content": "hello"}, + {"role": "user", "content": "again"}, + ] + items = _chat_messages_to_responses_input(msgs) + assert [i.get("role") for i in items] == ["user", "assistant", "user"]