fix(compaction): gate checkpoint replay/prune on current request eligibility

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 <noreply@anthropic.com>

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] = {

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)

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")

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"]