RomanNum3ral
/
hermes-agent
mirror of https://github.com/NousResearch/hermes-agent.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix(compression): prune pre-checkpoint history on native compaction replay 

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.
This commit is contained in:
Teknium 2026-08-13 01:41:07 -07:00
parent e4aeb65599
commit e4b3b91b62
3 changed files with 254 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
agent/codex_responses_adapter.py
View File

@ -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)
# ---------------------------------------------------------------------------

122
agent/native_compaction.py
View File

@ -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.

120
tests/run_agent/test_native_compaction.py
View File

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