fix(agent): keep native compaction checkpoints out of the thinking-only drop

A type="compaction" item is the server-side stand-in for history that has
already been pruned, and it rides the same codex_reasoning_items sidecar as
per-turn reasoning. e00965a7e taught compaction pruning to filter that
sidecar instead of popping it so checkpoints survive on every retained
message.

The thinking-only sanitizer reaches the same sidecar from the other
direction and asks a coarser question: does any item have type ==
"reasoning"? A commentary turn carrying both a reasoning item and a
checkpoint answers yes, so the whole message is dropped from the wire copy
and the only copy of the checkpoint goes with it. The request then carries
neither the compacted history nor the checkpoint standing in for it.

Extract has_compaction_checkpoint() into agent/native_compaction.py — the
module that owns the concept, and where merge_interim_reasoning_items()
already spelled the same predicate inline — and consult it before the
thinking-only verdict. A reasoning-only carrier is still dropped.
This commit is contained in:
Drexuxux
2026-08-09 03:23:07 +03:00
committed by Teknium
parent e029e300ca
commit 6c2d4efd02
3 changed files with 74 additions and 5 deletions
+17 -5
View File
@@ -303,6 +303,22 @@ def is_native_compaction_rejection(error: Any, status_code: Any = None) -> bool:
return any(marker in text for marker in rejection_markers)
def has_compaction_checkpoint(items: Any) -> bool:
"""Does this ``codex_reasoning_items`` sidecar carry a compaction checkpoint?
A ``type: "compaction"`` item is the server-side stand-in for history that
has already been pruned — cumulative context, not per-turn reasoning. It
rides the same sidecar as ordinary reasoning items, so anything that
rewrites or discards that sidecar (or the message carrying it) has to ask
this question first: the checkpoint exists in exactly one place, and the
request that loses it loses the compacted history with it.
"""
return any(
isinstance(item, dict) and item.get("type") == "compaction"
for item in (items if isinstance(items, list) else ())
)
def merge_interim_reasoning_items(
prior_items: Any,
new_items: Any,
@@ -324,10 +340,6 @@ def merge_interim_reasoning_items(
if isinstance(item, dict) and item.get("type") == "compaction"
]
new_list = list(new_items) if isinstance(new_items, list) else []
new_has_checkpoint = any(
isinstance(item, dict) and item.get("type") == "compaction"
for item in new_list
)
if new_has_checkpoint or not kept_checkpoints:
if has_compaction_checkpoint(new_list) or not kept_checkpoints:
return new_list
return kept_checkpoints + new_list
+10
View File
@@ -4668,6 +4668,16 @@ class AIAgent:
# empty-turn handling instead of being dropped here.
codex_items = msg.get("codex_reasoning_items")
if drop_codex_reasoning_items and isinstance(codex_items, list):
# A native compaction checkpoint rides this same sidecar and is
# the server-side stand-in for already-pruned history. Dropping
# the turn takes the checkpoint with it — the request then carries
# neither the compacted history nor the checkpoint that replaces
# it. Compaction pruning filters items for this reason rather than
# popping the key; a carrier is never thinking-only.
from agent.native_compaction import has_compaction_checkpoint
if has_compaction_checkpoint(codex_items):
return False
return any(
isinstance(item, dict) and item.get("type") == "reasoning"
for item in codex_items
@@ -178,3 +178,50 @@ class TestDropThinkingOnlyAndMergeUsers:
assert out[0]["role"] == "system"
assert out[1]["role"] == "user"
assert out[1]["content"] == "u1\n\nu2"
# ---------------------------------------------------------------------------
# Native compaction checkpoints ride the same sidecar
# ---------------------------------------------------------------------------
class TestCompactionCheckpointCarrier:
"""``type: "compaction"`` items are cumulative context, not per-turn
reasoning: they stand in for history the server already pruned, and they
exist in exactly one place. Compaction pruning filters the sidecar rather
than popping it so they survive on every retained message; dropping the
carrier message defeats that from the other direction.
"""
CHECKPOINT = {"type": "compaction", "encrypted_content": "ckpt"}
REASONING = {"type": "reasoning", "encrypted_content": "per-turn"}
def _carrier(self, items):
return {"role": "assistant", "content": "", "codex_reasoning_items": list(items)}
def test_reasoning_only_carrier_is_still_thinking_only(self):
"""The existing contract is unchanged for ordinary reasoning turns."""
assert AIAgent._is_thinking_only_assistant(self._carrier([self.REASONING]))
def test_checkpoint_alongside_reasoning_is_not_thinking_only(self):
msg = self._carrier([self.REASONING, self.CHECKPOINT])
assert not AIAgent._is_thinking_only_assistant(msg)
def test_checkpoint_order_does_not_matter(self):
msg = self._carrier([self.CHECKPOINT, self.REASONING])
assert not AIAgent._is_thinking_only_assistant(msg)
def test_checkpoint_survives_the_sanitizer_pass(self):
msgs = [
{"role": "user", "content": "u1"},
self._carrier([self.REASONING, self.CHECKPOINT]),
{"role": "user", "content": "u2"},
]
out = AIAgent._drop_thinking_only_and_merge_users(msgs)
surviving = [
item
for m in out
for item in (m.get("codex_reasoning_items") or [])
if item.get("type") == "compaction"
]
assert surviving == [self.CHECKPOINT]