From 11c5aae104cb95b5141744dcb277448ef8b24dce Mon Sep 17 00:00:00 2001 From: joaomarcos Date: Fri, 14 Aug 2026 04:43:58 -0300 Subject: [PATCH] fix(compaction): gate checkpoint replay/prune on current request eligibility MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A captured native-compaction checkpoint lives in the persisted codex_reasoning_items sidecar, but the wire restructure that follows it (prune_pre_checkpoint_items) ran unconditionally: the native gate only decided whether context_management went into the request, and no signal from it ever reached _chat_messages_to_responses_input. So a single checkpoint kept deleting every pre-checkpoint item from all later requests — after a mid-session swap out of the gpt-5.6 family, after compression.enabled: false, after the rejection kill switch, and after a session resume that reloads the sidecar from state.db. The model receiving the opaque blob was no longer the one able to decode it, and nothing was logged. Thread a single native_compaction_eligible boolean, derived from the same value that gates the context_management field, into the converter. When ineligible: do not replay type: "compaction" items and do not prune. Safe because native compaction never truncates Hermes' local history, so the fallback still carries the full conversation. All Responses call sites are covered: build_kwargs and convert_messages derive the flag via _native_compaction_active, the auxiliary/compression client is explicitly ineligible, and the converter defaults to False (pre-feature wire) so future call sites are safe by construction. Fixes #85914 Co-Authored-By: Claude Opus 5 --- agent/auxiliary_client.py | 9 +- agent/codex_responses_adapter.py | 42 ++++++- agent/transports/codex.py | 27 +++- tests/run_agent/test_native_compaction.py | 142 +++++++++++++++++++++- 4 files changed, 215 insertions(+), 5 deletions(-) diff --git a/agent/auxiliary_client.py b/agent/auxiliary_client.py index f5e827b7c5..e6b26493db 100644 --- a/agent/auxiliary_client.py +++ b/agent/auxiliary_client.py @@ -1500,8 +1500,15 @@ class _CodexCompletionsAdapter: # build_kwargs, so they need the same guard applied independently. _host_for_input = str(getattr(self._client, "base_url", "") or "") _is_github_for_input = base_url_host_matches(_host_for_input, "githubcopilot.com") + # Auxiliary calls never send ``context_management`` (native + # compaction is a main-turn feature), so they must never replay a + # compaction checkpoint from the replayed history nor let one + # restructure this request — the summarizer/aggregator model is + # usually not even the one that minted the blob. input_items = _chat_messages_to_responses_input( - replay_messages, is_github_responses=_is_github_for_input, + replay_messages, + is_github_responses=_is_github_for_input, + native_compaction_eligible=False, ) resp_kwargs: Dict[str, Any] = { diff --git a/agent/codex_responses_adapter.py b/agent/codex_responses_adapter.py index 2b431cff72..ac1129d4fc 100644 --- a/agent/codex_responses_adapter.py +++ b/agent/codex_responses_adapter.py @@ -414,6 +414,7 @@ def _chat_messages_to_responses_input( is_github_responses: bool = False, replay_encrypted_reasoning: bool = True, current_issuer_kind: Optional[str] = None, + native_compaction_eligible: bool = False, ) -> List[Dict[str, Any]]: """Convert internal chat-style messages to Responses input items. @@ -458,6 +459,24 @@ def _chat_messages_to_responses_input( ``replay_encrypted_reasoning=False`` is the session-wide kill switch (drops ALL replay); ``current_issuer_kind`` is the per-item filter that runs only when replay is still enabled. + + ``native_compaction_eligible`` mirrors, for THIS request, the decision + made by ``native_compaction.native_compaction_context_management`` — it + is True only when that gate returned a payload, i.e. when the request + actually carries ``context_management``. It controls two things that + must never outlive the gate: replaying ``type: "compaction"`` checkpoint + items, and restructuring the wire around them + (``prune_pre_checkpoint_items``). Checkpoints are persisted in the + ``codex_reasoning_items`` sidecar and survive a mid-session model swap, + a ``compression.enabled: false`` flip, the rejection kill switch and a + resumed session; without this flag a single captured checkpoint would + keep deleting every pre-checkpoint item from every later request, on a + model that cannot decrypt the blob (#85914). Default False = pre-feature + wire, which is also correct for every caller that never sends + ``context_management`` (auxiliary/compression client, ad-hoc + ``convert_messages``). Dropping the checkpoint costs nothing: Hermes' + local history is never truncated by native compaction, so the full + conversation is still on the wire. """ items: List[Dict[str, Any]] = [] seen_item_ids: set = set() @@ -499,6 +518,20 @@ def _chat_messages_to_responses_input( item_id = ri.get("id") if item_id and item_id in seen_item_ids: continue + # Native-compaction gate: a checkpoint is only + # meaningful to the endpoint/model that minted it + # AND only while this request still asks for + # server-side compaction. Once the gate closes + # (model swapped out of the gpt-5.6 family, + # compression disabled, rejection kill switch), + # the persisted checkpoint must not be replayed — + # replaying it is what makes the wire restructure + # below erase pre-checkpoint history forever. + if ( + ri.get("type") == "compaction" + and not native_compaction_eligible + ): + continue # Cross-issuer guard: drop reasoning blocks that # were minted by a different Responses endpoint. # The current endpoint cannot decrypt foreign @@ -697,8 +730,13 @@ def _chat_messages_to_responses_input( # 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. + # post-checkpoint tail untouched. Gated on the CURRENT request's native + # eligibility, not merely on the presence of a checkpoint: a persisted + # checkpoint outlives the gate, and pruning for a request that carries no + # ``context_management`` deletes history the server never compacted. + if not native_compaction_eligible: + return items + from agent.native_compaction import prune_pre_checkpoint_items return prune_pre_checkpoint_items(items) diff --git a/agent/transports/codex.py b/agent/transports/codex.py index fac925fdba..919e6e6fdd 100644 --- a/agent/transports/codex.py +++ b/agent/transports/codex.py @@ -263,6 +263,21 @@ def _is_post_tool_replay(messages: Optional[List[Dict[str, Any]]]) -> bool: return False +def _native_compaction_active(context_management: Any) -> bool: + """Is THIS request natively compacted? + + True only when the caller's eligibility gate + (``native_compaction.native_compaction_context_management``) produced a + non-empty payload. Every native-compaction side effect on the wire — + sending ``context_management``, replaying a ``type: "compaction"`` + checkpoint, restructuring the input around it — hangs off this one + predicate, so a checkpoint that outlives the gate (model swapped out of + the gpt-5.6 family, compression disabled, rejection kill switch, resumed + session) cannot keep reshaping requests on its own. + """ + return isinstance(context_management, list) and bool(context_management) + + class ResponsesApiTransport(ProviderTransport): """Transport for api_mode='codex_responses'. @@ -303,6 +318,9 @@ class ResponsesApiTransport(ProviderTransport): kwargs.get("replay_encrypted_reasoning", True) ), current_issuer_kind=issuer, + native_compaction_eligible=_native_compaction_active( + kwargs.get("context_management") + ), ) def convert_tools(self, tools: List[Dict[str, Any]]) -> Any: @@ -377,6 +395,12 @@ class ResponsesApiTransport(ProviderTransport): # agent.native_compaction.native_compaction_context_management(); # None means the field is never added to the request. context_management = params.get("context_management") + # Single source of truth for "this request is natively compacted": + # the same value decides whether the field goes out AND whether the + # converter may replay/prune around a compaction checkpoint. Keeping + # them derived from one expression is what stops a persisted + # checkpoint from restructuring the wire after the gate closes. + native_compaction_active = _native_compaction_active(context_management) # Resolve the issuing endpoint for this call. Stashed on the # transport so normalize_response can stamp it onto reasoning @@ -471,6 +495,7 @@ class ResponsesApiTransport(ProviderTransport): is_github_responses=is_github_responses, replay_encrypted_reasoning=replay_encrypted_reasoning, current_issuer_kind=issuer_kind, + native_compaction_eligible=native_compaction_active, ), "store": False, } @@ -478,7 +503,7 @@ class ResponsesApiTransport(ProviderTransport): kwargs["tools"] = response_tools kwargs["tool_choice"] = "auto" kwargs["parallel_tool_calls"] = True - if isinstance(context_management, list) and context_management: + if native_compaction_active: kwargs["context_management"] = context_management session_id = params.get("session_id") diff --git a/tests/run_agent/test_native_compaction.py b/tests/run_agent/test_native_compaction.py index e62595e3b0..6518b44fbf 100644 --- a/tests/run_agent/test_native_compaction.py +++ b/tests/run_agent/test_native_compaction.py @@ -349,6 +349,7 @@ class TestResponseCapture: {"role": "user", "content": "next"}, ], current_issuer_kind="codex_backend", + native_compaction_eligible=True, ) replayed = [item for item in items if item.get("type") == "compaction"] assert len(replayed) == 1 @@ -375,6 +376,7 @@ class TestResponseCapture: {"role": "user", "content": "next"}, ], current_issuer_kind="xai_responses", + native_compaction_eligible=True, ) assert all(item.get("type") != "compaction" for item in items) @@ -549,7 +551,7 @@ class TestPrunePreCheckpointItems: }, {"role": "user", "content": "follow-up"}, ] - items = _chat_messages_to_responses_input(msgs) + items = _chat_messages_to_responses_input(msgs, native_compaction_eligible=True) 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"] @@ -569,3 +571,141 @@ class TestPrunePreCheckpointItems: ] items = _chat_messages_to_responses_input(msgs) assert [i.get("role") for i in items] == ["user", "assistant", "user"] + + +class TestCheckpointGatedOnCurrentEligibility: + """A captured checkpoint must not outlive the native gate. + + The checkpoint is persisted in the ``codex_reasoning_items`` sidecar, so + it survives a mid-session model swap, ``compression.enabled: false``, the + rejection kill switch and a resumed session. Every one of those closes the + gate; if the wire kept being restructured around the stale checkpoint, + pre-checkpoint history would be deleted from requests that were never + natively compacted — on a model that cannot even decrypt the blob. + """ + + def _history(self): + return [ + {"role": "user", "content": "goal: ship the migration"}, + {"role": "assistant", "content": "on it"}, + {"role": "user", "content": "detail A"}, + { + "role": "assistant", + "content": "checkpointed turn", + "codex_reasoning_items": [ + { + "type": "compaction", + "encrypted_content": "blob", + "_issuer_kind": "codex_backend", + } + ], + }, + {"role": "user", "content": "next ask"}, + ] + + def test_ineligible_request_keeps_pre_feature_wire(self): + from agent.codex_responses_adapter import _chat_messages_to_responses_input + + history = self._history() + items = _chat_messages_to_responses_input( + history, + current_issuer_kind="codex_backend", + native_compaction_eligible=False, + ) + pre_feature = _chat_messages_to_responses_input( + [ + {k: v for k, v in msg.items() if k != "codex_reasoning_items"} + for msg in history + ], + ) + assert items == pre_feature + # Specifically: no checkpoint on the wire, no deleted history. + assert all(i.get("type") != "compaction" for i in items) + assert {"role": "assistant", "content": "on it"} in items + + def test_eligible_request_still_restructures(self): + from agent.codex_responses_adapter import _chat_messages_to_responses_input + + items = _chat_messages_to_responses_input( + self._history(), + current_issuer_kind="codex_backend", + native_compaction_eligible=True, + ) + assert items[0]["type"] == "compaction" + assert {"role": "assistant", "content": "on it"} not in items + + def test_converter_defaults_to_ineligible(self): + from agent.codex_responses_adapter import _chat_messages_to_responses_input + + items = _chat_messages_to_responses_input(self._history()) + assert all(i.get("type") != "compaction" for i in items) + assert {"role": "assistant", "content": "on it"} in items + + def test_build_kwargs_without_field_does_not_prune(self): + """Model swapped out of the gpt-5.6 family / kill switch fired: + the gate returns None, so the wire must be the pre-feature one.""" + from agent.transports.codex import ResponsesApiTransport + + kwargs = ResponsesApiTransport().build_kwargs( + model="gpt-5.2", + messages=self._history(), + context_management=None, + ) + assert "context_management" not in kwargs + assert all(i.get("type") != "compaction" for i in kwargs["input"]) + assert {"role": "assistant", "content": "on it"} in kwargs["input"] + + def test_build_kwargs_with_field_prunes(self): + from agent.transports.codex import ResponsesApiTransport + + kwargs = ResponsesApiTransport().build_kwargs( + model="gpt-5.6", + messages=self._history(), + is_codex_backend=True, + context_management=[{"type": "compaction", "compact_threshold": 4000}], + ) + assert kwargs["input"][0]["type"] == "compaction" + assert {"role": "assistant", "content": "on it"} not in kwargs["input"] + + def test_convert_messages_defaults_to_ineligible(self): + from agent.transports.codex import ResponsesApiTransport + + items = ResponsesApiTransport().convert_messages( + self._history(), is_codex_backend=True + ) + assert all(i.get("type") != "compaction" for i in items) + assert {"role": "assistant", "content": "on it"} in items + + def test_auxiliary_responses_adapter_never_prunes(self, monkeypatch): + """Auxiliary calls (compression, flush_memories, MoA) replay real + session history but never send ``context_management`` — so a + checkpoint in that history must not restructure their request.""" + import agent.codex_responses_adapter as adapter + from agent.auxiliary_client import _CodexCompletionsAdapter + + seen = {} + real = adapter._chat_messages_to_responses_input + + def _spy(messages, **kw): + seen.update(kw) + return real(messages, **kw) + + monkeypatch.setattr(adapter, "_chat_messages_to_responses_input", _spy) + + class _Responses: + def create(self, **kwargs): + seen["input"] = kwargs.get("input") + raise RuntimeError("stop before network") + + class _Client: + base_url = "https://chatgpt.com/backend-api/codex" + responses = _Responses() + + with pytest.raises(RuntimeError, match="stop before network"): + _CodexCompletionsAdapter(_Client(), "gpt-5.2").create( + messages=self._history() + ) + + assert seen.get("native_compaction_eligible") is False + assert all(i.get("type") != "compaction" for i in seen["input"]) + assert {"role": "assistant", "content": "on it"} in seen["input"]