ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

refactor(agent): tighten memory/compaction/prompt-cache helpers (pass 1, corpus parity)

Browse Source
This commit is contained in:
Teknium
2026-09-02 18:28:38 -07:00
parent 113f04616b
commit 4f20954c5f
10 changed files with 366 additions and 576 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/memory_manager.py
+38 -77
View File
@@ -4,14 +4,6 @@ Single integration point (run_agent.py) that fans out to registered providers.
The builtin provider is always allowed; only ONE external plugin provider may be
registered at a time — a second is rejected with a warning to prevent tool
schema bloat and conflicting memory backends.
Usage in run_agent.py:
self._memory_manager = MemoryManager()
self._memory_manager.add_provider(plugin_provider) # at most one external
prompt_parts.append(self._memory_manager.build_system_prompt())
context = self._memory_manager.prefetch_all(user_message) # pre-turn
self._memory_manager.sync_all(user_msg, assistant_response) # post-turn
self._memory_manager.queue_prefetch_all(user_msg)
"""
from __future__ import annotations
@@ -36,14 +28,12 @@ logger = logging.getLogger(__name__)
# historical best-effort contract (API v1).
_LEGACY_PRE_COMPRESS_API_VERSION = 1
# How long shutdown_all() waits for in-flight background sync/prefetch work to
# drain before abandoning it. Worker threads are daemon, so a wedged provider
# never blocks interpreter exit — it dies with the process past this window.
# How long shutdown_all() waits for in-flight background work to drain before
# abandoning it. Workers are daemon threads, so a wedged provider never blocks
# interpreter exit.
_SYNC_DRAIN_TIMEOUT_S = 5.0
_EXTERNAL_PREFETCH_TIMEOUT_S = 8.0
_VAR_KEYWORD = inspect.Parameter.VAR_KEYWORD
# ---------------------------------------------------------------------------
# Signature introspection (providers are duck-typed; call shapes vary)
@@ -58,7 +48,7 @@ def _signature_params(fn: Callable[..., Any]):
def _has_var_kwargs(params) -> bool:
return any(p.kind is _VAR_KEYWORD for p in params.values())
return any(p.kind is inspect.Parameter.VAR_KEYWORD for p in params.values())
def _accepts_require_checkpoint(fn: Callable[..., Any]) -> bool:
@@ -85,10 +75,9 @@ def _accepts_require_checkpoint(fn: Callable[..., Any]) -> bool:
def _ctx_bound(fn: Callable[[], Any]) -> Callable[[], Any]:
"""Bind ``fn`` to the CALLER's contextvars for execution on another thread.
Profile isolation in multi-profile processes (gateway multiplexer, dashboard,
cron) is a ContextVar-scoped HERMES_HOME override; worker threads start with
empty contexts, so an unbound provider resolving config paths or secrets
from a worker would silently land on the default profile.
Profile isolation is a ContextVar-scoped HERMES_HOME override; worker threads
start with empty contexts, so an unbound provider resolving config paths or
secrets from a worker would silently land on the default profile.
"""
return partial(contextvars.copy_context().run, fn)
@@ -100,12 +89,10 @@ def _ctx_bound(fn: Callable[[], Any]) -> Callable[[], Any]:
def normalize_tool_schema(schema: Any) -> Optional[Dict[str, Any]]:
"""Return a bare function-tool dict with a resolvable top-level ``name``, else None.
Providers should return ``{"name", "description", "parameters"}`` which callers
wrap as ``{"type": "function", "function": schema}``. Some return the already
wrapped OpenAI form; wrapping that twice yields a ``function`` with no ``name``
and strict providers (e.g. DeepSeek) reject the ENTIRE request (HTTP 400),
disabling every tool. Both shapes are normalized here so callers can skip
nameless entries with a warning instead of poisoning the request.
Providers should return ``{"name", "description", "parameters"}``; some return
the already wrapped OpenAI form. Wrapping that twice yields a ``function`` with
no ``name`` and strict providers (e.g. DeepSeek) reject the ENTIRE request,
so both shapes are normalized and nameless entries can be skipped.
"""
if not isinstance(schema, dict):
return None
@@ -248,9 +235,8 @@ class StreamingContextScrubber:
The one-shot ``sanitize_context`` regex needs both tags in one string, so a
span opened in one delta and closed in a later one would leak to the UI.
This state machine holds back partial-tag tails between ``feed()`` calls and
drops everything inside a span (including the system-note line). Create a
fresh scrubber (or ``reset()``) per top-level response; call ``flush()`` at
end of stream.
drops everything inside a span. Create a fresh scrubber (or ``reset()``) per
top-level response; call ``flush()`` at end of stream.
"""
_OPEN_TAG = "<memory-context>"
@@ -279,7 +265,7 @@ class StreamingContextScrubber:
# Hold back a potential partial close tag; drop the rest.
held = self._max_partial_suffix(buf, self._CLOSE_TAG)
self._buf = buf[-held:] if held else ""
return "".join(out)
break
buf = buf[idx + len(self._CLOSE_TAG):]
self._in_span = False
else:
@@ -292,7 +278,7 @@ class StreamingContextScrubber:
self._append_visible(out, buf[:-held] if held else buf)
if held:
self._buf = buf[-held:]
return "".join(out)
break
if idx > 0:
self._append_visible(out, buf[:idx])
buf = buf[idx + len(self._OPEN_TAG):]
@@ -307,12 +293,9 @@ class StreamingContextScrubber:
memory context is worse than a truncated answer. Otherwise the held tail
was not a real tag and is emitted verbatim.
"""
if self._in_span:
self._buf = ""
self._in_span = False
return ""
tail = self._buf
tail = "" if self._in_span else self._buf
self._buf = ""
self._in_span = False
return tail
@staticmethod
@@ -333,21 +316,16 @@ class StreamingContextScrubber:
idx = buf_lower.find(self._OPEN_TAG, search_start)
if idx == -1:
return -1
if self._is_block_boundary(buf, idx) and self._has_block_opener_suffix(buf, idx):
after_idx = idx + len(self._OPEN_TAG)
if self._is_block_boundary(buf, idx) and after_idx < len(buf) and buf[after_idx] in "\r\n":
return idx
search_start = idx + 1
def _max_pending_open_suffix(self, buf: str) -> int:
"""Hold a complete boundary tag at the buffer end until the following char confirms it."""
if not buf.lower().endswith(self._OPEN_TAG):
return 0
if not self._is_block_boundary(buf, len(buf) - len(self._OPEN_TAG)):
return 0
return len(self._OPEN_TAG)
def _has_block_opener_suffix(self, buf: str, idx: int) -> bool:
after_idx = idx + len(self._OPEN_TAG)
return after_idx < len(buf) and buf[after_idx] in "\r\n"
if buf.lower().endswith(self._OPEN_TAG) and self._is_block_boundary(buf, len(buf) - len(self._OPEN_TAG)):
return len(self._OPEN_TAG)
return 0
def _is_block_boundary(self, buf: str, idx: int) -> bool:
if idx == 0:
@@ -477,18 +455,15 @@ class MemoryManager:
# Core tool names are reserved: built-ins always win at agent init, so a
# shadowing provider tool would linger in ``_tool_to_provider`` and
# hijack dispatch. Reject it at the door, like the TTS/browser/search
# provider registries do.
# hijack dispatch. Reject it at the door.
from toolsets import _HERMES_CORE_TOOLS
_core_tool_names = set(_HERMES_CORE_TOOLS)
for raw_schema in provider.get_tool_schemas():
schema = normalize_tool_schema(raw_schema)
if schema is None:
continue
tool_name = schema["name"]
if tool_name in _core_tool_names:
if tool_name in _HERMES_CORE_TOOLS:
logger.warning(
"Memory provider '%s' tool '%s' shadows a reserved core "
"tool name; registration ignored. Core tools always win — "
@@ -540,9 +515,7 @@ class MemoryManager:
A /skill or /bundle turn expands into a model-facing message embedding the
whole skill body; feeding that to providers pollutes stores/embeddings with
prompt scaffolding. Recover just the user's instruction once, for the whole
fan-out. Non-skill text passes through; a bare invocation (no instruction)
yields None since there is nothing worth remembering.
prompt scaffolding. A bare invocation (no instruction) yields None.
"""
return extract_user_instruction_from_skill_message(text)
@@ -638,13 +611,9 @@ class MemoryManager:
def queue_prefetch_all(self, query: str, *, session_id: str = "") -> None:
"""Queue background prefetch on all providers for the next turn (see ``sync_all``)."""
providers = list(self._providers)
if not providers:
return
clean_query = self._strip_skill_scaffolding(query)
clean_query = self._strip_skill_scaffolding(query) if providers else None
if not clean_query:
return
self._submit_background(
lambda: self._each_provider(
"queue_prefetch failed (non-fatal)",
@@ -674,15 +643,11 @@ class MemoryManager:
Never inline: a provider's ``sync_turn`` may block on a network/daemon
call for minutes, which kept ``run_conversation`` open after the user saw
the response, so every interface showed the agent "running" and follow-up
messages triggered interrupts. The single worker also serializes writes so
turn N lands before turn N+1 without provider-side ordering logic.
the response. The single worker also serializes writes so turn N lands
before turn N+1 without provider-side ordering logic.
"""
providers = list(self._providers)
if not providers:
return
clean_user_content = self._strip_skill_scaffolding(user_content)
clean_user_content = self._strip_skill_scaffolding(user_content) if providers else None
if not clean_user_content:
return
@@ -797,7 +762,6 @@ class MemoryManager:
"""
from toolsets import _HERMES_CORE_TOOLS
_core_tool_names = set(_HERMES_CORE_TOOLS)
schemas: List[Dict[str, Any]] = []
seen = set()
@@ -812,7 +776,7 @@ class MemoryManager:
)
continue
name = schema["name"]
if name not in _core_tool_names and name not in seen:
if name not in _HERMES_CORE_TOOLS and name not in seen:
schemas.append(schema)
seen.add(name)
@@ -872,11 +836,10 @@ class MemoryManager:
"""Queue old-session extraction + provider rebinding as ONE serialized task.
``on_session_end`` (LLM-bound extraction, seconds) must run strictly
BEFORE ``on_session_switch`` rebinds provider-internal session state;
an ad-hoc thread raced the inline switch and misattributed transcripts
to the new session. One task on the single FIFO worker gives both an
immediate return and ordering against every other provider write. If
the executor is unavailable, ``_submit_background`` runs it inline.
BEFORE ``on_session_switch`` rebinds provider-internal session state; an
ad-hoc thread raced the inline switch and misattributed transcripts. One
task on the single FIFO worker gives both an immediate return and
ordering against every other provider write.
"""
if not self._providers:
return
@@ -910,10 +873,9 @@ class MemoryManager:
) -> None:
"""Notify providers that ``AIAgent.session_id`` rotated without teardown.
Fires on ``/resume``, ``/branch``, ``/reset``, ``/new`` and compression;
providers refresh cached per-session state so later writes land in the
right record. ``rewound=True`` (``/undo``) means the id is unchanged but
the transcript was truncated.
Fires on ``/resume``, ``/branch``, ``/reset``, ``/new`` and compression.
``rewound=True`` (``/undo``) means the id is unchanged but the transcript
was truncated.
"""
if not new_session_id:
return
@@ -1074,8 +1036,7 @@ class MemoryManager:
) -> None:
"""Mirror a built-in memory tool call to external providers.
Single entry point the agent loop calls after the ``memory`` tool runs:
gates on a committed write, expands single-op and batched ``operations``
Gates on a committed write, expands single-op and batched ``operations``
shapes, keeps only mutating actions, and forwards ``old_text`` plus the
per-op provenance from ``build_metadata`` (the loop knows session/task/
tool-call identity the manager does not).
agent/memory_provider.py
+16 -21
View File
@@ -1,11 +1,9 @@
"""Abstract base class for pluggable memory providers.
Memory providers give the agent persistent recall across sessions. Plugins
ship in ``plugins/memory/<name>/`` and are activated via ``memory.provider``;
MemoryManager allows only ONE external provider at a time (tool-schema bloat,
conflicting backends). Lifecycle is driven by MemoryManager: initialize ->
system_prompt_block / prefetch / sync_turn per turn -> tool dispatch ->
shutdown, plus the optional ``on_*`` hooks below.
Plugins ship in ``plugins/memory/<name>/`` and are activated via ``memory.provider``;
MemoryManager allows only ONE external provider at a time. Lifecycle is driven by
MemoryManager: initialize -> system_prompt_block / prefetch / sync_turn per turn ->
tool dispatch -> shutdown, plus the optional ``on_*`` hooks below.
"""
from __future__ import annotations
@@ -96,8 +94,7 @@ class MemoryProvider(ABC):
def unavailable_reason(self) -> str:
"""Short user-facing hint for the "provider unavailable" warning (e.g.
which package to install) — ``initialize()`` never runs when unavailable,
so this is the only place such a diagnostic can surface."""
which package to install); ``initialize()`` never runs when unavailable."""
return ""
def system_prompt_block(self) -> str:
@@ -167,13 +164,11 @@ class MemoryProvider(ABC):
"""session_id reassigned mid-process (/resume, /branch, /reset, /new,
gateway equivalents, context compression) without a provider teardown.
Update or reset any per-session state cached in ``initialize()`` so
later writes land in the right record. ``parent_session_id`` carries
lineage for /branch, compression and /resume ("" when none). ``reset``
is True only for a genuinely new conversation (/reset, /new) — flush
per-session buffers; False when the logical conversation continues
under a new id. ``rewound``: same id but the transcript was truncated,
so invalidate per-turn document state.
Update or reset per-session state cached in ``initialize()`` so later
writes land in the right record. ``parent_session_id`` carries lineage
("" when none). ``reset`` is True only for a genuinely new conversation
(/reset, /new) — flush per-session buffers. ``rewound``: same id but the
transcript was truncated, so invalidate per-turn document state.
"""
def on_pre_compress(self, messages: List[Dict[str, Any]]) -> str:
@@ -191,9 +186,9 @@ class MemoryProvider(ABC):
Each field: ``key``, ``description``, optional ``secret`` (goes to .env),
``required``, ``default``, ``choices``, ``type`` (text | integer |
number | boolean), ``minimum`` / ``maximum`` / ``step`` (numeric,
Dashboard rendering), ``url`` (where to get the credential), ``env_var``
(explicit secret env var; default auto-generated).
number | boolean), ``minimum`` / ``maximum`` / ``step`` (numeric),
``url`` (where to get the credential), ``env_var`` (explicit secret env
var; default auto-generated).
"""
return []
@@ -216,7 +211,7 @@ class MemoryProvider(ABC):
def backup_paths(self) -> List[str]:
"""Absolute paths of provider state OUTSIDE HERMES_HOME (e.g. ``~/.honcho``)
so ``hermes backup`` can capture them under ``_external/`` and
``hermes import`` restore them; paths outside the home dir are skipped.
MUST work without ``initialize()`` or network — resolve from config/env."""
so ``hermes backup``/``hermes import`` can capture and restore them; paths
outside the home dir are skipped. MUST work without ``initialize()`` or
network — resolve from config/env."""
return []
agent/message_content.py
+3 -10
View File
@@ -9,9 +9,7 @@ _TEXT_KEYS = ("text", "content", "input_text", "output_text", "summary_text")
def _field(value: Any, key: str) -> Any:
if isinstance(value, Mapping):
return value.get(key)
return getattr(value, key, None)
return value.get(key) if isinstance(value, Mapping) else getattr(value, key, None)
def _text_from_part(part: Any) -> str:
@@ -19,11 +17,8 @@ def _text_from_part(part: Any) -> str:
return ""
if isinstance(part, str):
return part
part_type = str(_field(part, "type") or "").strip().lower()
if part_type in _NON_TEXT_PART_TYPES:
if str(_field(part, "type") or "").strip().lower() in _NON_TEXT_PART_TYPES:
return ""
for key in _TEXT_KEYS:
text = _field(part, key)
if isinstance(text, str):
@@ -38,9 +33,7 @@ def flatten_message_text(content: Any, *, sep: str = "\n") -> str:
if isinstance(content, str):
return content
if isinstance(content, list):
chunks = [_text_from_part(part) for part in content]
return sep.join(chunk for chunk in chunks if chunk)
return sep.join(chunk for chunk in map(_text_from_part, content) if chunk)
text = _text_from_part(content)
if text:
return text
agent/message_metadata.py
+3 -5
View File
@@ -20,9 +20,8 @@ def stamp_message_timestamp(
) -> _Message:
"""Attach a creation timestamp without replacing source-provided time.
Gateway adapters can supply the platform event time. All other callers use
the local wall clock at the point the message enters the live transcript.
Returning the same mapping keeps the helper convenient at append sites.
Gateway adapters can supply the platform event time; all other callers use
the local wall clock. Returns the same mapping for use at append sites.
"""
if message.get("timestamp") is None:
message["timestamp"] = wall_time() if timestamp is None else timestamp
@@ -36,6 +35,5 @@ def append_message(
timestamp: Optional[float] = None,
) -> _Message:
"""Stamp and append one live transcript message."""
stamp_message_timestamp(message, timestamp=timestamp)
messages.append(message)
messages.append(stamp_message_timestamp(message, timestamp=timestamp))
return message
agent/message_sanitization.py
+94 -143
View File
@@ -1,9 +1,9 @@
"""Message and tool-payload sanitization helpers.
Pure functions (extracted from ``run_agent.py``) that walk OpenAI-format message
lists and structured payloads, repairing or stripping characters that would
crash ``json.dumps`` in the OpenAI SDK or be rejected upstream. Stateless except
for documented in-place mutation; ``run_agent`` re-exports them for old imports.
Pure functions that walk OpenAI-format message lists and structured payloads,
repairing or stripping characters that would crash ``json.dumps`` in the OpenAI
SDK or be rejected upstream. Stateless except for documented in-place mutation;
``run_agent`` re-exports them for old imports.
"""
from __future__ import annotations
@@ -150,22 +150,16 @@ def _escape_invalid_chars_in_json_strings(raw: str) -> str:
n = len(raw)
while i < n:
ch = raw[i]
if in_string:
if ch == "\\" and i + 1 < n:
out.append(ch)
out.append(raw[i + 1])
i += 2
continue
if ch == '"':
in_string = False
out.append(ch)
elif ord(ch) < 0x20:
out.append(f"\\u{ord(ch):04x}")
else:
out.append(ch)
if in_string and ch == "\\" and i + 1 < n:
out.append(raw[i:i + 2])
i += 2
continue
if ch == '"':
in_string = not in_string
out.append(ch)
elif in_string and ord(ch) < 0x20:
out.append(f"\\u{ord(ch):04x}")
else:
if ch == '"':
in_string = True
out.append(ch)
i += 1
return "".join(out)
@@ -177,6 +171,14 @@ def _escape_invalid_chars_in_json_strings(raw: str) -> str:
_FULL_ARGS_LOG_BOUND = 100_000
def _loads_ok(text: str) -> bool:
try:
json.loads(text)
return True
except json.JSONDecodeError:
return False
def _repair_tool_call_arguments(raw_args: str, tool_name: str = "?") -> str:
"""Repair malformed tool_call argument JSON (truncation, trailing commas,
Python ``None``, literal control chars); returns ``"{}"`` if unrepairable so
@@ -209,47 +211,34 @@ def _repair_tool_call_arguments(raw_args: str, tool_name: str = "?") -> str:
# Passes 1-3: strip trailing commas, close unclosed structures, then trim
# excess closers (bounded).
fixed = re.sub(r',\s*([}\]])', r'\1', raw_stripped)
open_curly = fixed.count('{') - fixed.count('}')
open_bracket = fixed.count('[') - fixed.count(']')
if open_curly > 0:
fixed += '}' * open_curly
if open_bracket > 0:
fixed += ']' * open_bracket
fixed += '}' * max(0, fixed.count('{') - fixed.count('}'))
fixed += ']' * max(0, fixed.count('[') - fixed.count(']'))
for _ in range(50):
try:
json.loads(fixed)
if _loads_ok(fixed):
break
if (fixed.endswith('}') and fixed.count('}') > fixed.count('{')) or (
fixed.endswith(']') and fixed.count(']') > fixed.count('[')
):
fixed = fixed[:-1]
else:
break
except json.JSONDecodeError:
if (fixed.endswith('}') and fixed.count('}') > fixed.count('{')) or (
fixed.endswith(']') and fixed.count(']') > fixed.count('[')
):
fixed = fixed[:-1]
else:
break
try:
json.loads(fixed)
if _loads_ok(fixed):
logger.warning(
"Repaired malformed tool_call arguments for %s: %s → %s",
tool_name, raw_stripped[:80], fixed[:80],
)
return fixed
except json.JSONDecodeError:
pass
# Pass 4: escape control chars inside strings (strict=False alone fails
# when other malformations are present too), then retry.
try:
escaped = _escape_invalid_chars_in_json_strings(fixed)
if escaped != fixed:
json.loads(escaped)
logger.warning(
"Repaired control-char-laced tool_call arguments for %s: %s → %s",
tool_name, raw_stripped[:80], escaped[:80],
)
return escaped
except (json.JSONDecodeError, TypeError, ValueError):
pass
escaped = _escape_invalid_chars_in_json_strings(fixed)
if escaped != fixed and _loads_ok(escaped):
logger.warning(
"Repaired control-char-laced tool_call arguments for %s: %s → %s",
tool_name, raw_stripped[:80], escaped[:80],
)
return escaped
logger.warning(
"Unrepairable tool_call arguments for %s — "
@@ -265,9 +254,7 @@ def close_interrupted_tool_sequence(messages: list, final_response: Any = None)
A transcript ending on a raw ``tool`` message makes the next user message
land as ``tool → user`` — a role-alternation violation strict providers
(Gemini, Claude) answer by hallucinating a continuation and dropping prior
context. ``finalize_turn`` covers the happy interrupt path; the retry/backoff
early-returns in ``conversation_loop`` need this shared helper. Mutates in
place; returns True if a closing turn was appended.
context. Mutates in place; returns True if a closing turn was appended.
"""
if not messages:
return False
@@ -287,12 +274,11 @@ def close_interrupted_tool_sequence(messages: list, final_response: Any = None)
def serialized_messages_bytes(messages: list) -> int:
"""Exact serialized byte size of the ``messages`` payload (HTTP 413 recovery).
A 413 is a BYTE-size error, but the token estimator deliberately prices an
image at a flat per-image cost, so it cannot score recovery from an
image-dominated 413 (compaction frees megabytes while the estimate barely
moves → "no progress"). This measures what the provider actually rejected,
identically before and after each pass. Non-serializable values fall back to
``str()`` so a malformed message can never crash recovery.
A 413 is a BYTE-size error, but the token estimator prices an image at a
flat per-image cost, so it cannot score recovery from an image-dominated
413. This measures what the provider actually rejected, identically before
and after each pass. Non-serializable values fall back to ``str()`` so a
malformed message can never crash recovery.
"""
if not isinstance(messages, list) or not messages:
return 0
@@ -306,18 +292,18 @@ def serialized_messages_bytes(messages: list) -> int:
return sum(len(str(m)) for m in messages)
_IMAGE_PART_TYPES = {"image_url", "image", "input_image"}
def _strip_images_from_messages(messages: list) -> bool:
"""Remove image content parts from all messages in-place (server rejected images).
Preserves alternation invariants: ``tool`` messages and assistant messages
carrying ``tool_calls`` whose content was entirely images are replaced with a
placeholder, NOT deleted (deleting orphans the paired ``tool_call_id`` →
HTTP 400); other now-empty messages (synthetic image-only attachment turns)
are dropped. Any rewritten message also loses its ``api_content`` sidecar —
it carries the exact bytes previously sent, i.e. the images being removed,
and would be substituted back on the wire next turn.
Returns True if any image parts were removed.
HTTP 400); other now-empty messages are dropped. Any rewritten message also
loses its ``api_content`` sidecar — it carries the exact bytes previously
sent, i.e. the images being removed. Returns True if any image parts were removed.
"""
from agent.turn_context import drop_stale_api_content
@@ -329,13 +315,12 @@ def _strip_images_from_messages(messages: list) -> bool:
content = msg.get("content")
if not isinstance(content, list):
continue
new_parts = []
for part in content:
if isinstance(part, dict) and part.get("type") in {"image_url", "image", "input_image"}:
found = True
else:
new_parts.append(part)
new_parts = [
part for part in content
if not (isinstance(part, dict) and part.get("type") in _IMAGE_PART_TYPES)
]
if len(new_parts) < len(content):
found = True
if new_parts:
msg["content"] = new_parts
elif msg.get("role") == "tool" or msg.get("tool_calls"):
@@ -368,8 +353,7 @@ _IMAGE_REJECTION_PHRASES = (
"model does not support image",
# DashScope-style gateways reject non-text blocks with this generic body.
"unexpected item type in content",
# ChatGPT-account Codex backend rejects data:image URLs in input_image
# ("Invalid 'input[N].content[K].image_url'. Expected a valid URL ...");
# ChatGPT-account Codex backend rejects data:image URLs in input_image;
# keyed on the field-path apostrophe so other URL errors don't false-trip.
"image_url'. expected",
# ChatGPT-account Codex wording for corrupt/unsupported native image payloads.
@@ -380,8 +364,7 @@ _IMAGE_REJECTION_PHRASES = (
# OpenRouter HTTP 404 when no upstream endpoint accepts image input (passes
# the 4xx gate; without this the gateway queue wedges behind the stuck turn).
"no endpoints found that support image input",
# Kimi/Moonshot et al. reject truncated/corrupt image bytes baked into
# immutable history ("prepare image failed ... failed to decode image").
# Kimi/Moonshot et al. reject truncated/corrupt image bytes baked into history.
"failed to decode image",
)
@@ -426,8 +409,8 @@ __all__ = [
# coalescing, and duplicate-id repair.
#
# NOT consolidated on purpose: agent/transports/codex_event_projector's
# _deterministic_call_id maps codex app-server ITEM ids (`codex_<type>_<item_id>`),
# not chat tool-call content; merging would change ids and invalidate caches.
# _deterministic_call_id maps codex app-server ITEM ids, not chat tool-call
# content; merging would change ids and invalidate caches.
#
# HARD INVARIANT: everything here stays deterministic (never uuid4) and
# byte-identical for existing inputs — these ids feed prompt-cache prefixes.
@@ -455,17 +438,12 @@ def _expand_tool_id_variants(values: tuple[Any, ...]) -> frozenset[str]:
"""
variants: set[str] = set()
for raw in values:
if not isinstance(raw, str):
continue
value = raw.strip()
value = raw.strip() if isinstance(raw, str) else ""
if not value:
continue
variants.add(value)
if "|" in value:
for part in value.split("|"):
part = part.strip()
if part:
variants.add(part)
variants.update(p for p in (part.strip() for part in value.split("|")) if p)
return frozenset(variants)
@@ -489,9 +467,7 @@ def coalesce_tool_call_id(tc: Any) -> str:
Returns ``""`` when neither is set.
"""
for raw in (_tc_field(tc, "call_id"), _tc_field(tc, "id")):
if not isinstance(raw, str):
continue
value = raw.strip()
value = raw.strip() if isinstance(raw, str) else ""
if value:
return value.split("|", 1)[0].strip() or value
return ""
@@ -501,20 +477,18 @@ def uniquify_tool_call_ids(tool_calls: list) -> list:
"""Ensure every tool call in one assistant turn has a distinct id.
Some providers reuse one id across calls in a batch; the pre-API sanitizer
then keeps only the first call/result pair per id (the later result silently
vanishes) and strict providers reject duplicates outright. First occurrence
keeps its id; later collisions get a deterministic ``<id>_d<n>`` suffix
(never uuid4 — cache-prefix stability). Mutates entries in place (SDK models
/ SimpleNamespace / dicts) and returns the same list. Blank ids are left for
the deterministic fallback in ``build_assistant_message``.
then keeps only the first call/result pair per id and strict providers
reject duplicates outright. First occurrence keeps its id; later collisions
get a deterministic ``<id>_d<n>`` suffix (never uuid4 — cache-prefix
stability). Mutates entries in place (SDK models / SimpleNamespace / dicts)
and returns the same list. Blank ids are left for the deterministic fallback
in ``build_assistant_message``.
"""
seen: set = set()
for tc in tool_calls or []:
# Same coalescing rule as coalesce_tool_call_id, tolerant of non-string ids.
raw = _tc_field(tc, "call_id") or _tc_field(tc, "id") or ""
raw = raw.strip() if isinstance(raw, str) else ""
if not raw:
continue
# Composite Responses ids ("call_x|fc_y") collide on the call half —
# that's the pairing key providers enforce per turn.
cid = raw.split("|", 1)[0]
@@ -524,10 +498,9 @@ def uniquify_tool_call_ids(tool_calls: list) -> list:
seen.add(cid)
continue
n = 2
new_id = f"{cid}_d{n}"
while new_id in seen:
while f"{cid}_d{n}" in seen:
n += 1
new_id = f"{cid}_d{n}"
new_id = f"{cid}_d{n}"
seen.add(new_id)
def _renamed(value):
@@ -539,10 +512,7 @@ def uniquify_tool_call_ids(tool_calls: list) -> list:
try:
if isinstance(tc, dict):
if tc.get("id"):
tc["id"] = _renamed(tc["id"])
else:
tc["id"] = new_id
tc["id"] = _renamed(tc["id"]) if tc.get("id") else new_id
if tc.get("call_id"):
tc["call_id"] = new_id
else:
@@ -605,13 +575,13 @@ def matches_reasoning_echo_family(
from utils import base_url_host_matches
_, raw_providers, lowered_providers, model_subs, hosts = _REASONING_ECHO_RULE_BY_FAMILY[family]
provider_lower = (provider or "").lower()
model_lower = (model or "").lower()
if provider in raw_providers or provider_lower in lowered_providers:
return True
if any(sub in model_lower for sub in model_subs):
return True
return any(base_url_host_matches(base_url, host) for host in hosts)
return (
provider in raw_providers
or (provider or "").lower() in lowered_providers
or any(sub in model_lower for sub in model_subs)
or any(base_url_host_matches(base_url, host) for host in hosts)
)
def reasoning_echo_family(provider: Any, model: Any, base_url: Any) -> "str | None":
@@ -639,10 +609,8 @@ def stale_thinking_reaches_wire(
can look over-threshold to preflight yet fully tail-protected to the walk —
an infinite ineffective compaction loop.
* ``codex_responses``: the Responses input builder never reads the text keys
(continuity rides the encrypted ``codex_reasoning_items`` sidecar, already
charged by both estimators) → False.
* echo-back families: ``apply_reasoning_content_policy`` replays stored
``reasoning_content`` verbatim on every assistant turn → True.
(continuity rides the encrypted ``codex_reasoning_items`` sidecar) → False.
* echo-back families: stored ``reasoning_content`` is replayed verbatim → True.
* everything else: stripped or one-space-padded at send time → False.
"""
if (api_mode or "") == "codex_responses":
@@ -664,50 +632,33 @@ def apply_reasoning_content_policy(
# 1. Explicit reasoning_content set. Require-side: preserve verbatim,
# upgrading legacy "" placeholders to " " (DeepSeek V4 400s on ""). Strict
# side: strip entirely — a reasoning primary pads history with " ", then a
# fallback to Mistral/Cerebras/Groq replays the pad and 422s. This covers
# the rebuild path; reapply_reasoning_echo covers already-built api_messages.
# fallback to Mistral/Cerebras/Groq replays the pad and 422s.
existing = source_msg.get("reasoning_content")
if isinstance(existing, str):
if not needs_thinking_pad:
api_msg.pop("reasoning_content", None)
elif existing == "":
api_msg["reasoning_content"] = " "
else:
api_msg["reasoning_content"] = existing
api_msg["reasoning_content"] = existing or " "
return
# 2. Cross-provider poisoned history: tool_calls + 'reasoning' but no
# 'reasoning_content' key means the reasoning text came from ANOTHER
# provider (DeepSeek's own build pins reasoning_content for tool-call
# turns). Pad with " " to satisfy the API without leaking foreign CoT.
normalized_reasoning = source_msg.get("reasoning")
if (
needs_thinking_pad
and source_msg.get("tool_calls")
and isinstance(normalized_reasoning, str)
and normalized_reasoning
):
api_msg["reasoning_content"] = " "
return
# 3. Healthy session: promote internal 'reasoning' → 'reasoning_content'
# (must precede the unconditional pad so real reasoning isn't overwritten),
# but only for echo-back providers — strict ones reject the field.
if isinstance(normalized_reasoning, str) and normalized_reasoning:
if needs_thinking_pad:
has_reasoning = isinstance(normalized_reasoning, str) and bool(normalized_reasoning)
if needs_thinking_pad:
# 2. Cross-provider poisoned history: tool_calls + 'reasoning' but no
# 'reasoning_content' key means the reasoning text came from ANOTHER
# provider (DeepSeek's own build pins reasoning_content for tool-call
# turns). Pad with " " to satisfy the API without leaking foreign CoT.
# 3. Healthy session: promote internal 'reasoning' → 'reasoning_content'.
# 4. No reasoning at all: every assistant turn still needs the field;
# " " (not "") because DeepSeek V4 rejects empty string.
if has_reasoning and not source_msg.get("tool_calls"):
api_msg["reasoning_content"] = normalized_reasoning
else:
api_msg.pop("reasoning_content", None)
api_msg["reasoning_content"] = " "
return
# 4. Require-side with no reasoning at all: every assistant turn needs the
# field; " " (not "") because DeepSeek V4 rejects empty string.
if needs_thinking_pad:
api_msg["reasoning_content"] = " "
return
# 5. reasoning_content present but not a string (e.g. None after
# compaction) — never pass null to the API.
# 5. Strict side: never carry the field (incl. a non-string value such as
# None after compaction — never pass null to the API).
api_msg.pop("reasoning_content", None)
agent/micro_compaction.py
+66 -103
View File
@@ -20,6 +20,16 @@ from agent.model_metadata import estimate_messages_tokens_rough, estimate_tokens
logger = logging.getLogger("agent.context_compressor")
def _is_micro_marker(entry: Any) -> bool:
"""True for a summary marker provably absorbed into the rolling summary (micro, not batch)."""
from agent.context_compressor import COMPRESSED_SUMMARY_METADATA_KEY, MICRO_COMPACT_MARKER_KEY
return (
isinstance(entry, dict)
and bool(entry.get(COMPRESSED_SUMMARY_METADATA_KEY))
and bool(entry.get(MICRO_COMPACT_MARKER_KEY))
)
class MicroCompactionMixin:
"""Rolling micro-compaction; host must be a ``ContextCompressor``."""
@@ -34,12 +44,13 @@ class MicroCompactionMixin:
Uses the in-memory cursor when valid; otherwise scans for the last summary marker.
"""
from agent.context_compressor import MICRO_COMPACT_MARKER_KEY
if self._micro_compact_cursor > head_end and self._micro_compact_cursor < tail_start:
if head_end < self._micro_compact_cursor < tail_start:
return self._micro_compact_cursor
last_summary_idx = -1
for idx in range(head_end, tail_start):
if self._is_context_summary_message(messages[idx]):
last_summary_idx = idx
cursor = head_end
if last_summary_idx >= head_end:
cursor = last_summary_idx + 1
# Resumed session: rehydrate the rolling summary from the surviving marker so the next
@@ -57,8 +68,6 @@ class MicroCompactionMixin:
"Micro-compaction: recovered rolling summary from "
"transcript (%d chars)", len(recovered),
)
else:
cursor = head_end
self._micro_compact_cursor = cursor
return cursor
@@ -73,40 +82,29 @@ class MicroCompactionMixin:
Returns ``(exchange_start, exchange_end)`` or ``None``. Spans assistant+tool rows up to the
next user message; user turns are never absorbed (alternation safety, verbatim user text).
"""
limit = min(tail_start, len(messages))
idx = start
n = len(messages)
if idx >= n or idx >= tail_start:
return None
# Skip user messages and (assistant-role) summary markers to reach a real assistant message;
# otherwise a rehydrated cursor could absorb the marker itself.
while idx < tail_start and idx < n:
while idx < limit:
msg = messages[idx]
if msg.get("role") == "assistant" and not self._is_context_summary_message(msg):
break
idx += 1
if idx >= tail_start or idx >= n:
if idx >= limit:
return None
exchange_start = idx
idx += 1
while idx < tail_start and idx < n:
while idx < limit:
msg = messages[idx]
role = msg.get("role")
if role not in ("assistant", "tool"):
break
if self._is_context_summary_message(msg):
if msg.get("role") not in ("assistant", "tool") or self._is_context_summary_message(msg):
break
idx += 1
if idx <= exchange_start:
return None
# Boundary must close the turn: a mid-turn stop at tail_start would put the assistant marker
# beside assistant/tool rows. Any other role is a safe splice (avoids wedging the cursor).
if idx >= n:
if idx >= len(messages):
return None
boundary = messages[idx]
if not isinstance(boundary, dict) or boundary.get("role") in ("assistant", "tool"):
@@ -128,10 +126,7 @@ class MicroCompactionMixin:
exchange_text: str,
) -> List[Dict[str, str]]:
"""Build the prompt messages for a single-exchange micro-summary."""
if existing_summary.strip():
summary_block = existing_summary
else:
summary_block = "(No previous summary yet.)"
summary_block = existing_summary if existing_summary.strip() else "(No previous summary yet.)"
user_prompt = (
"You are a summarization agent creating a compact record of an "
@@ -166,14 +161,11 @@ class MicroCompactionMixin:
from agent.auxiliary_client import aux_interrupt_protection, call_llm
from agent.context_compressor import _response_finish_reason
messages = self._build_micro_summary_prompt(
self._micro_compact_rolling_summary,
exchange_text,
)
call_kwargs = {
"task": "compression",
"messages": messages,
"messages": self._build_micro_summary_prompt(
self._micro_compact_rolling_summary, exchange_text,
),
"max_tokens": min(1500, self.max_summary_tokens or 1500),
"temperature": 0.1,
}
@@ -195,8 +187,7 @@ class MicroCompactionMixin:
logger.info("micro-summarization call failed: %s", exc)
return None
# A length stop means a partial merge; leave the exchange unabsorbed so a later pass retries
# (pi#7048).
# A length stop means a partial merge; leave the exchange unabsorbed so a later pass retries.
if _response_finish_reason(response) == "length":
logger.warning(
"micro-summarization output hit the token cap "
@@ -217,8 +208,7 @@ class MicroCompactionMixin:
return None
from agent.agent_runtime_helpers import strip_think_blocks
stripped = strip_think_blocks(None, content).strip()
return stripped if stripped else None
return strip_think_blocks(None, content).strip() or None
def _needs_defrag(self) -> bool:
"""Return True when the rolling summary is large enough to defrag."""
@@ -233,11 +223,7 @@ class MicroCompactionMixin:
Transcript-shape-neutral (no splice, no cursor move). Returns True when it rewrote.
"""
from agent.context_compressor import (
_DB_PERSISTED_MARKER,
COMPRESSED_SUMMARY_METADATA_KEY,
MICRO_COMPACT_MARKER_KEY,
)
from agent.context_compressor import _DB_PERSISTED_MARKER
old_summary = self._micro_compact_rolling_summary
if not old_summary.strip():
return False
@@ -250,18 +236,13 @@ class MicroCompactionMixin:
self._micro_compact_rolling_summary = fresh_summary
# Rewrite only the newest MICRO marker (resume rehydrates from it); a batch marker holds
# history we lack.
for idx in range(len(messages) - 1, -1, -1):
entry = messages[idx]
if (
isinstance(entry, dict)
and entry.get(COMPRESSED_SUMMARY_METADATA_KEY)
and entry.get(MICRO_COMPACT_MARKER_KEY)
):
for entry in reversed(messages):
if _is_micro_marker(entry):
entry["content"] = self._render_micro_marker_content(fresh_summary)
# Content changed: clear the persisted stamp so the DB sync rewrites the row.
entry.pop(_DB_PERSISTED_MARKER, None)
# In-place pop on a live dict would be identity-skipped by the bounded flush scan;
# flag the finalizer (#75170).
# flag the finalizer.
self._flush_scan_cursor_invalidated = True
break
logger.info(
@@ -270,6 +251,10 @@ class MicroCompactionMixin:
)
return True
def _reset_micro_failure_tracking(self) -> None:
self._micro_compact_consecutive_failures = 0
self._micro_compact_last_failure_cursor = -1
def _micro_compact(
self,
messages: List[Dict[str, Any]],
@@ -299,7 +284,6 @@ class MicroCompactionMixin:
head_size = self._protect_head_size(messages)
compress_start = self._align_boundary_forward(messages, head_size)
compress_end = self._find_tail_cut_by_tokens(messages, compress_start)
if compress_start >= compress_end:
return messages
@@ -310,16 +294,21 @@ class MicroCompactionMixin:
exchange = self._find_one_exchange(messages, cursor, compress_end)
if exchange is None:
return messages
exchange_start, exchange_end = exchange
# Telemetry baseline; taken only once an exchange exists so no-op turns don't pay.
_started_at = time.monotonic()
_tokens_before = estimate_messages_tokens_rough(messages)
_messages_before = n_messages
def _elapsed_ms() -> int:
return int((time.monotonic() - _started_at) * 1000)
def _telemetry(outcome: str, result: List[Dict[str, Any]], **extra: Any) -> None:
self._emit_micro_compaction_telemetry(
outcome=outcome,
messages_before=n_messages,
messages_after=len(result),
tokens_before=_tokens_before,
duration_ms=int((time.monotonic() - _started_at) * 1000),
**extra,
)
# Defrag rewrites summary text/marker in place (no splice, no cursor move) instead of
# absorbing this turn.
@@ -327,15 +316,10 @@ class MicroCompactionMixin:
defragged = self._defrag_rolling_summary(messages)
if defragged:
self._sync_micro_compact_to_db(messages)
self._micro_compact_consecutive_failures = 0
self._micro_compact_last_failure_cursor = -1
self._emit_micro_compaction_telemetry(
outcome="defrag" if defragged else "defrag_failed",
messages_before=_messages_before,
messages_after=len(messages),
tokens_before=_tokens_before,
self._reset_micro_failure_tracking()
_telemetry(
"defrag" if defragged else "defrag_failed", messages,
tokens_after=estimate_messages_tokens_rough(messages),
duration_ms=_elapsed_ms(),
)
return messages
@@ -353,6 +337,7 @@ class MicroCompactionMixin:
self._micro_compact_consecutive_failures = 1
self._micro_compact_last_failure_cursor = exchange_start
_outcome = "summarize_failed"
if self._micro_compact_consecutive_failures >= _MICRO_COMPACT_MAX_CONSECUTIVE_FAILURES:
logger.info(
"Micro-compaction: skipping exchange at cursor %d "
@@ -361,40 +346,25 @@ class MicroCompactionMixin:
)
# Skip the stuck exchange; it stays in the transcript for batch compression/defrag.
self._micro_compact_cursor = exchange_end
self._micro_compact_consecutive_failures = 0
self._micro_compact_last_failure_cursor = -1
self._reset_micro_failure_tracking()
_outcome = "exchange_skipped"
else:
_outcome = "summarize_failed"
self._emit_micro_compaction_telemetry(
outcome=_outcome,
messages_before=_messages_before,
messages_after=len(messages),
tokens_before=_tokens_before,
tokens_after=_tokens_before,
exchange_tokens=_exchange_tokens,
duration_ms=_elapsed_ms(),
_telemetry(
_outcome, messages, tokens_after=_tokens_before, exchange_tokens=_exchange_tokens,
)
return messages
self._micro_compact_rolling_summary = updated_summary
self._micro_compact_cursor = exchange_end
self._micro_compact_consecutive_failures = 0
self._micro_compact_last_failure_cursor = -1
self._reset_micro_failure_tracking()
result = self._splice_micro_compact_result(
messages, exchange_start, exchange_end, supersede=_cumulative,
)
self._micro_compact_cursor = self._cursor_after_splice(result, exchange_start + 1)
self._sync_micro_compact_to_db(result)
self._emit_micro_compaction_telemetry(
outcome="absorbed",
messages_before=_messages_before,
messages_after=len(result),
tokens_before=_tokens_before,
tokens_after=estimate_messages_tokens_rough(result),
exchange_tokens=_exchange_tokens,
duration_ms=_elapsed_ms(),
_telemetry(
"absorbed", result,
tokens_after=estimate_messages_tokens_rough(result), exchange_tokens=_exchange_tokens,
)
return result
@@ -457,7 +427,7 @@ class MicroCompactionMixin:
delta = tokens_after - tokens_before
self._micro_compact_tokens_saved_total -= delta
self._micro_compact_passes += 1
# Cached reads only: the lazy properties can fire a synchronous /models probe (#32221).
# Cached reads only: the lazy properties can fire a synchronous /models probe.
threshold = self._threshold_tokens
context_limit = self._resolved_context_length
occupancy = None
@@ -510,13 +480,13 @@ class MicroCompactionMixin:
return
try:
# Every row except the marker is a carried-forward original: archive pre-splice
# originals rewind-style (#86366).
# originals rewind-style.
session_db.archive_and_compact(
session_id,
compacted_messages,
tail_count=max(0, len(compacted_messages) - 1),
)
# Shared post-commit stamp site with batch commit and proactive prune (#98450).
# Shared post-commit stamp site with batch commit and proactive prune.
stamp_db_persisted_markers(compacted_messages)
except Exception:
logger.info(
@@ -550,8 +520,7 @@ class MicroCompactionMixin:
COMPRESSED_SUMMARY_METADATA_KEY: True,
# Micro marker: eligible for supersede/defrag; batch markers never carry this key.
MICRO_COMPACT_MARKER_KEY: True,
# Micro markers absorb only assistant/tool content; user turns stay in the transcript
# (#64650).
# Micro markers absorb only assistant/tool content; user turns stay in the transcript.
COMPRESSED_SUMMARY_HAS_USER_TURN_KEY: False,
}
@@ -560,12 +529,7 @@ class MicroCompactionMixin:
# Cumulative summary: keep only the newest marker. Drop an older one only if supersede AND
# it has MICRO_COMPACT_MARKER_KEY (provably absorbed); a batch marker holds MORE history.
if supersede:
marker_idxs = [
i for i, m in enumerate(result)
if isinstance(m, dict)
and m.get(COMPRESSED_SUMMARY_METADATA_KEY)
and m.get(MICRO_COMPACT_MARKER_KEY)
]
marker_idxs = [i for i, m in enumerate(result) if _is_micro_marker(m)]
if len(marker_idxs) > 1:
superseded = set(marker_idxs[:-1])
result = [m for i, m in enumerate(result) if i not in superseded]
@@ -602,19 +566,18 @@ class MicroCompactionMixin:
from agent.context_compressor import COMPRESSED_SUMMARY_METADATA_KEY
from agent.turn_context import drop_stale_api_content
def _plain_user(m: Any) -> bool:
return (
isinstance(m, dict)
and m.get("role") == "user"
and not m.get(COMPRESSED_SUMMARY_METADATA_KEY)
and isinstance(m.get("content"), str)
)
merged: List[Dict[str, Any]] = []
for msg in result:
prev = merged[-1] if merged else None
if (
isinstance(msg, dict)
and isinstance(prev, dict)
and msg.get("role") == "user"
and prev.get("role") == "user"
and not msg.get(COMPRESSED_SUMMARY_METADATA_KEY)
and not prev.get(COMPRESSED_SUMMARY_METADATA_KEY)
and isinstance(prev.get("content"), str)
and isinstance(msg.get("content"), str)
):
if _plain_user(msg) and _plain_user(prev):
prev_content = prev["content"]
new_content = msg["content"]
prev["content"] = (
agent/native_compaction.py
+47 -72
View File
@@ -1,24 +1,12 @@
"""Native OpenAI Responses server-side compaction — gpt-5.6 on direct OpenAI routes only.
Including ``context_management=[{"type": "compaction", "compact_threshold": N}]``
in a ``/v1/responses`` request makes the server summarize older context into an
opaque ``compaction`` item (``encrypted_content``, sealed to the issuing
endpoint) once the input crosses N tokens; replaying that item stands in for
the pruned history. Docs: https://developers.openai.com/api/docs/guides/compaction
Support is deliberately narrow (live-verified):
* gpt-5.6 family only — gpt-5.1/5.2 fail server-side (HTTP 500 blocking, a
permanent stall streaming) with no structured "unsupported" rejection, so an
explicit model-family check is the only safe gate.
* Direct OpenAI routes only (api.openai.com or the ChatGPT Codex backend) —
other Responses surfaces would 400 on the field and cannot mint/decrypt the blob.
Hermes' local compressor stays armed as fallback owner: the native threshold is
clamped below the local trigger so the server compacts first, and captured
compaction items ride the existing ``codex_reasoning_items`` sidecar (persistence,
replay, cross-issuer stamping, kill switch). This module stays free of
transport/adapter imports so transport, adapter, and loop share the gate
without cycles; ``context_compressor`` and ``message_content`` sit below it.
``context_management=[{"type": "compaction", "compact_threshold": N}]`` makes the server
summarize older context into an opaque ``compaction`` item (sealed to the issuing
endpoint) once the input crosses N tokens. Support is deliberately narrow (live-verified):
gpt-5.6 family only (5.1/5.2 fail server-side with no structured rejection) on direct
OpenAI routes (api.openai.com or the ChatGPT Codex backend). Hermes' local compressor stays
armed as fallback: the native threshold is clamped below the local trigger, and captured
compaction items ride the existing ``codex_reasoning_items`` sidecar. No transport imports.
"""
from __future__ import annotations
@@ -81,6 +69,17 @@ def is_direct_openai_route(
return hostname == "api.openai.com"
def _positive_int(value: Any, *, reject: tuple = (bool,)) -> Optional[int]:
"""``int(value)`` when it is a positive integer-like (never a bool), else None."""
if value is None or isinstance(value, reject):
return None
try:
parsed = int(value)
except (TypeError, ValueError):
return None
return parsed if parsed > 0 else None
def resolve_compact_threshold(
configured_threshold: Any,
local_trigger_tokens: Any = None,
@@ -92,15 +91,7 @@ def resolve_compact_threshold(
explicit positive integer is absolute unless it must be clamped so native
compaction fires first. Booleans are never thresholds.
"""
local = None
try:
if local_trigger_tokens is not None and not isinstance(local_trigger_tokens, bool):
local = int(local_trigger_tokens)
except (TypeError, ValueError):
local = None
if local is not None and local <= 0:
local = None
local = _positive_int(local_trigger_tokens)
upper = None
if local is not None:
if local > LOCAL_TRIGGER_SAFETY_MARGIN:
@@ -108,15 +99,8 @@ def resolve_compact_threshold(
else:
upper = max(1_024, int(local * 0.8))
try:
configured = (
None
if isinstance(configured_threshold, (bool, float))
else int(configured_threshold)
)
except (TypeError, ValueError):
configured = None
if configured is None or configured <= 0:
configured = _positive_int(configured_threshold, reject=(bool, float))
if configured is None:
return upper if upper is not None else DEFAULT_COMPACT_THRESHOLD
if upper is None:
return configured
@@ -215,15 +199,16 @@ def _extract_item_text(item: Any) -> Optional[str]:
parts = []
for part in content:
if isinstance(part, str):
if part.strip():
parts.append(part.strip())
candidates = (part,)
elif isinstance(part, dict):
part_text = part.get("text") or part.get("input_text") or part.get("output_text")
if isinstance(part_text, str) and part_text.strip():
parts.append(part_text.strip())
part_meta = part.get("metadata")
if isinstance(part_meta, dict) and isinstance(part_meta.get("text"), str) and part_meta["text"].strip():
parts.append(part_meta["text"].strip())
candidates = (
part.get("text") or part.get("input_text") or part.get("output_text"),
part_meta.get("text") if isinstance(part_meta, dict) else None,
)
else:
continue
parts.extend(c.strip() for c in candidates if isinstance(c, str) and c.strip())
text = " ".join(parts)
return text if text.strip() else None
@@ -241,15 +226,13 @@ def _has_retainable_image_content(item: Any) -> bool:
content = item.get("content")
if not isinstance(content, list):
return False
for part in content:
if not isinstance(part, dict):
continue
if str(part.get("type") or "").strip().lower() != "input_image":
continue
image_url = part.get("image_url")
if isinstance(image_url, str) and image_url.strip():
return True
return False
return any(
isinstance(part, dict)
and str(part.get("type") or "").strip().lower() == "input_image"
and isinstance(part.get("image_url"), str)
and part["image_url"].strip()
for part in content
)
# Canonical provenance check (metadata marker, then canonical prefix classifier).
@@ -259,6 +242,10 @@ def _has_retainable_image_content(item: Any) -> bool:
_is_summary_item = is_compaction_summary_message
def _is_compaction_item(item: Any) -> bool:
return isinstance(item, dict) and item.get("type") == "compaction"
def prune_pre_checkpoint_items(
items: List[Dict[str, Any]],
retained_user_token_budget: int = RETAINED_USER_MESSAGE_TOKEN_BUDGET,
@@ -297,17 +284,13 @@ def prune_pre_checkpoint_items(
last_cp = None
for i, item in enumerate(items):
if isinstance(item, dict) and item.get("type") == "compaction":
if _is_compaction_item(item):
last_cp = i
if last_cp is None:
return items
first_cp = last_cp
while (
first_cp > 0
and isinstance(items[first_cp - 1], dict)
and items[first_cp - 1].get("type") == "compaction"
):
while first_cp > 0 and _is_compaction_item(items[first_cp - 1]):
first_cp -= 1
pre = items[:first_cp]
@@ -362,17 +345,14 @@ def prune_pre_checkpoint_items(
continue
text = _extract_item_text(item)
has_retainable_image = is_user and _has_retainable_image_content(item)
if text is None and not has_retainable_image:
continue
if text is None:
if not (is_user and _has_retainable_image_content(item)):
continue
text = ""
if is_summary:
_retain_summary(text, item)
elif is_user:
if user_remaining <= 0:
continue
elif user_remaining > 0:
cost = _approx_tokens(text)
if cost <= user_remaining:
retained_reversed.append(item)
@@ -428,10 +408,7 @@ def has_compaction_checkpoint(items: Any) -> bool:
reasoning, and exists in exactly one place: anything that rewrites or
discards the sidecar must ask this first or lose the compacted history.
"""
return any(
isinstance(item, dict) and item.get("type") == "compaction"
for item in (items if isinstance(items, list) else ())
)
return isinstance(items, list) and any(_is_compaction_item(item) for item in items)
def merge_interim_reasoning_items(
@@ -445,9 +422,7 @@ def merge_interim_reasoning_items(
win, but prior checkpoints are prepended unless the newer payload has its own.
"""
kept_checkpoints = [
item
for item in (prior_items if isinstance(prior_items, list) else [])
if isinstance(item, dict) and item.get("type") == "compaction"
item for item in (prior_items if isinstance(prior_items, list) else []) if _is_compaction_item(item)
]
new_list = list(new_items) if isinstance(new_items, list) else []
if has_compaction_checkpoint(new_list) or not kept_checkpoints:
agent/prompt_cache_boundary.py
+9 -17
View File
@@ -1,19 +1,12 @@
"""Builder-declared stable prefixes for Anthropic prompt caching.
Skill/webhook/cron builders concatenate a large static scaffold with a small
volatile invocation tail into one user-message string. Only the builder knows
where the tail begins, so it registers the stable prefix here and the cache
planner places a breakpoint at that boundary instead of caching the whole
message. Re-parsing marker strings out of the message at request time is
deliberately avoided: markers can legitimately appear inside skill bodies or
event payloads, and any delimiter heuristic then shrinks the cached prefix or
silently absorbs volatile bytes into it.
Process-local by design: a webhook/cron fire is built and sent by the same
process, and any miss (restart, eviction, historic message) falls back to the
whole-message policy. The split only applies while the message is one of the
plan's marked endpoints; once it rotates out it ships as one block again
(one-time re-ingest in long interactive sessions, never for webhook/cron).
Skill/webhook/cron builders concatenate a large static scaffold with a small volatile
tail into one user-message string. Only the builder knows where the tail begins, so it
registers the stable prefix here and the cache planner places a breakpoint at that
boundary. Re-parsing marker strings at request time is deliberately avoided: markers
can legitimately appear inside skill bodies, and any delimiter heuristic then shrinks
the cached prefix or absorbs volatile bytes. Process-local by design: any miss falls
back to the whole-message policy.
"""
import threading
@@ -22,9 +15,8 @@ from typing import Optional
# A couple dozen active scaffolds is generous for one gateway process.
_MAX_ENTRIES = 32
# Entries hold whole expanded skill bodies, so also bound total retained chars
# (1-4x bytes). The newest entry is always kept so one oversized scaffold still
# gets a boundary instead of silently disabling the split.
# Entries hold whole expanded skill bodies, so also bound total retained chars.
# The newest entry is always kept so one oversized scaffold still gets a boundary.
_MAX_CHARS = 4 * 1024 * 1024
_lock = threading.Lock()
agent/prompt_cache_scope.py
+16 -33
View File
@@ -1,30 +1,21 @@
"""Rotation-stable logical cache scope for prompt_cache_key derivation.
Legacy compression rotation (``compression.in_place: false``) mints a new
physical ``session_id`` mid-conversation, which moved the conversation into a
fresh cache bucket each time. ``resolve_prompt_cache_scope()`` instead maps
the physical id to the ROOT of its compression lineage via
``SessionDB.get_compression_lineage()`` — NOT ``get_conversation_root`` /
``_conversation_root_id`` (the Portal-attribution walk), which follows
``parent_session_id`` blindly and would collapse /branch children and delegate
trees into one id. The two resolvers are intentionally different.
Legacy compression rotation mints a new physical ``session_id`` mid-conversation,
which moved the conversation into a fresh cache bucket each time.
``resolve_prompt_cache_scope()`` maps the physical id to the ROOT of its compression
lineage via ``SessionDB.get_compression_lineage()`` — NOT ``get_conversation_root``
(the Portal-attribution walk), which follows ``parent_session_id`` blindly and would
collapse /branch children and delegate trees into one id.
Scope boundaries: rotation children walk back to the original segment; ``/new``
starts a fresh scope; ``/branch`` children, delegate subagents, and tool-tagged
children are explicit fork children with their own isolated scope; cron fires
keep their physical id (the per-fire timestamp is stripped later).
Hosts that mint one physical id per RESPONSE (Studio group chat, ``/v1/responses``
with client-managed history) carry no lineage, so the walk returns the physical
id and the scope moves every reply. Hermes must not infer the conversation from
id SYNTAX (that collides client-supplied ids); the host declares it via
``gateway_session_key`` (``X-Hermes-Session-Key`` / ``build_session_key``),
consumed by ``declared_conversation_scope()``, which wins over the lineage walk.
The declared key is hashed to ``gwk_<sha256[:24]>`` because it embeds
platform/chat/user identifiers and leaves the process as a provider routing key.
Resolution is memoized per (agent, session_id, db-present): the lineage walk
runs once per transcript segment, never per API call.
keep their physical id. Hosts that mint one physical id per RESPONSE (Studio group
chat, ``/v1/responses`` with client-managed history) carry no lineage, so they
declare the conversation via ``gateway_session_key`` (``X-Hermes-Session-Key``),
consumed by ``declared_conversation_scope()``, which wins over the lineage walk and
is hashed to ``gwk_<sha256[:24]>`` because it embeds platform/chat/user identifiers.
Resolution is memoized per (agent, session_id, db-present).
"""
import hashlib
@@ -38,10 +29,7 @@ _DECLARED_SCOPE_PREFIX = "gwk_"
def _lineage_root(session_id: str, session_db: Any) -> Optional[str]:
"""Compression-lineage root of *session_id*, or None.
Tolerates non-list results from test doubles / partially built agents.
"""
"""Compression-lineage root of *session_id*, or None (tolerates test-double results)."""
if session_db is None:
return None
try:
@@ -66,8 +54,7 @@ def _agent_source(
lands, use the SAME resolver persistence uses
(``run_agent._session_source_for_agent``), not ``agent.platform``: they
diverge under ``HERMES_SESSION_SOURCE``, and the declared scope is memoized
immediately, so both sides of a ``/new`` would otherwise miss the boundary
recorded under the override and hash the same scope.
immediately, so both sides of a ``/new`` would otherwise hash the same scope.
"""
if row_source is None and session_id and session_db is not None:
try:
@@ -105,9 +92,7 @@ def _conversation_generation(session_key: str, source: str, session_db: Any) ->
if not callable(reader):
return ""
generation = reader(session_key, source)
if generation is None:
return ""
return str(int(generation))
return "" if generation is None else str(int(generation))
def declared_conversation_scope(agent: Any) -> Optional[str]:
@@ -173,9 +158,7 @@ def resolve_prompt_cache_scope(agent: Any) -> str:
memo = getattr(agent, _MEMO_ATTR, None)
if isinstance(memo, tuple) and len(memo) == 2 and memo[0] == key:
return memo[1]
root = declared_conversation_scope(agent) or (
_lineage_root(sid, db) if db is not None else None
)
root = declared_conversation_scope(agent) or _lineage_root(sid, db)
scope = root or sid
# Memoize on success, with no DB, or when the agent never persists a row
# (background-review forks hold a DB handle but set _persist_disabled).
agent/prompt_caching.py
+74 -95
View File
@@ -39,6 +39,13 @@ def envelope_tool_part_cache_markers_supported(
return not _is_litellm_route((provider or "").strip().lower(), base_url or "")
def _text_part(text: str, cache_marker: dict | None = None) -> dict:
part: dict = {"type": "text", "text": text}
if cache_marker is not None:
part["cache_control"] = cache_marker
return part
def _apply_cache_marker(
msg: dict,
cache_marker: dict,
@@ -68,30 +75,22 @@ def _apply_cache_marker(
return
if isinstance(content, str):
if role == "user":
stable_prefix = find_stable_prefix(content)
if stable_prefix is not None:
suffix = content[len(stable_prefix):]
if suffix.strip():
# Builder-declared boundary: the scaffold carries the
# breakpoint and the volatile tail rides unmarked, so a
# changed ticket ID/timestamp no longer invalidates the
# skill body. Request-local only — the stored message
# stays a plain string.
msg["content"] = [
{"type": "text", "text": stable_prefix, "cache_control": cache_marker},
{"type": "text", "text": suffix},
]
return
msg["content"] = [
{"type": "text", "text": content, "cache_control": cache_marker}
]
stable_prefix = find_stable_prefix(content) if role == "user" else None
if stable_prefix is not None and content[len(stable_prefix):].strip():
# Builder-declared boundary: the scaffold carries the breakpoint and
# the volatile tail rides unmarked, so a changed ticket ID/timestamp
# no longer invalidates the skill body. Request-local only — the
# stored message stays a plain string.
msg["content"] = [
_text_part(stable_prefix, cache_marker),
_text_part(content[len(stable_prefix):]),
]
else:
msg["content"] = [_text_part(content, cache_marker)]
return
if isinstance(content, list) and content:
last = content[-1]
if isinstance(last, dict):
last["cache_control"] = cache_marker
if isinstance(content, list) and content and isinstance(content[-1], dict):
content[-1]["cache_control"] = cache_marker
def _can_carry_marker(
@@ -111,13 +110,11 @@ def _can_carry_marker(
if msg.get("role") == "tool" and not tool_part_markers:
return False
content = msg.get("content")
if content is None or content == "":
return False
if isinstance(content, list):
# Mirrors _apply_cache_marker (marks only the LAST part): a list whose
# last element isn't a dict cannot receive a marker.
return bool(content) and isinstance(content[-1], dict)
return isinstance(content, str)
return isinstance(content, str) and content != ""
def _build_marker(ttl: str) -> Dict[str, str]:
@@ -141,11 +138,10 @@ ALIBABA_FAMILY_PROVIDERS = frozenset({
})
# 1h-tier ALLOW-list: only routes wire-measured to retain a 1h marker (delayed
# read past 5 minutes with no intervening call — an intervening read renews the
# window and masks expiry). Other opencode routes stay clamped because they are
# UNMEASURED, not known-bad. Note opencode-go labels every write
# `ephemeral_5m_input_tokens` regardless of requested ttl; that label is not
# evidence of the retention window.
# read past 5 minutes with no intervening call). Other opencode routes stay
# clamped because they are UNMEASURED, not known-bad. opencode-go labels every
# write `ephemeral_5m_input_tokens` regardless of requested ttl; that label is
# not evidence of the retention window.
MEASURED_1H_PROVIDERS = frozenset({
"opencode-go",
})
@@ -181,20 +177,19 @@ def effective_cache_ttl(
"""Clamp a requested cache TTL to what the destination route supports.
Qwen/Alibaba routes document a five-minute window and drop the ``1h``
tier, so a configured ``1h`` regresses to ``5m`` there instead of creating
a false 1h-cache expectation — except on ``MEASURED_1H_PROVIDERS``, which
keep ``1h`` minus any ``NO_1H_TIER_MODELS`` model. The measured-route check
runs BEFORE the generic Qwen clamp, which would otherwise swallow every
Qwen model on it. ``None`` resolves to ``5m``.
tier, so a configured ``1h`` regresses to ``5m`` there — except on
``MEASURED_1H_PROVIDERS``, which keep ``1h`` minus any ``NO_1H_TIER_MODELS``
model. The measured-route check runs BEFORE the generic Qwen clamp, which
would otherwise swallow every Qwen model on it. ``None`` resolves to ``5m``.
"""
if ttl != "1h":
return ttl or "5m"
if (provider or "").lower() in MEASURED_1H_PROVIDERS:
# Checked BEFORE the generic Qwen clamp (which would swallow every Qwen
# model on this route); the per-model denial stays nested so an
# opencode-go observation cannot reclamp the same model on another route.
provider_lower = (provider or "").lower()
if provider_lower in MEASURED_1H_PROVIDERS:
# The per-model denial stays nested so an opencode-go observation
# cannot reclamp the same model on another route.
return "5m" if _flat_model(model) in NO_1H_TIER_MODELS else "1h"
if is_qwen_model(model) or (provider or "").lower() in ALIBABA_FAMILY_PROVIDERS:
if is_qwen_model(model) or provider_lower in ALIBABA_FAMILY_PROVIDERS:
return "5m"
return "1h"
@@ -229,23 +224,23 @@ def _apply_system_cache_markers(
):
suffix = content[len(static_system_prefix):]
if suffix.strip():
suffix_part: dict = {"type": "text", "text": suffix}
if mark_suffix:
suffix_part["cache_control"] = cache_marker
message["content"] = [
{"type": "text", "text": static_system_prefix, "cache_control": cache_marker},
suffix_part,
_text_part(static_system_prefix, cache_marker),
_text_part(suffix, cache_marker if mark_suffix else None),
]
return 2 if mark_suffix else 1
_apply_cache_marker(message, cache_marker, native_anthropic=native_anthropic)
return 1
if not fallback_to_whole:
elif not fallback_to_whole:
return 0
_apply_cache_marker(message, cache_marker, native_anthropic=native_anthropic)
return 1
def _has_part_marker(content: Any) -> bool:
return isinstance(content, list) and any(
isinstance(part, dict) and "cache_control" in part for part in content
)
def strip_anthropic_cache_control(
api_messages: List[Dict[str, Any]],
) -> List[Dict[str, Any]]:
@@ -283,7 +278,7 @@ def strip_anthropic_cache_control(
and "cache_control" in content[0]
and "cache_control" not in content[1]
)
if any(isinstance(part, dict) and "cache_control" in part for part in content):
if _has_part_marker(content):
content = [
{k: v for k, v in part.items() if k != "cache_control"}
if isinstance(part, dict) and "cache_control" in part
@@ -318,18 +313,15 @@ def strip_anthropic_tool_cache_control(tools: List[Dict[str, Any]] | None) -> Li
def _count_cache_markers(messages: List[Dict[str, Any]], tools: List[Dict[str, Any]]) -> int:
"""Count the wire-visible cache markers in a request-local plan."""
count = sum(
1
for message in messages
if isinstance(message, dict) and "cache_control" in message
)
count += sum(
1
for message in messages
if isinstance(message, dict) and isinstance(message.get("content"), list)
for part in message["content"]
if isinstance(part, dict) and "cache_control" in part
)
count = 0
for message in messages:
if not isinstance(message, dict):
continue
count += "cache_control" in message
if isinstance(message.get("content"), list):
count += sum(
1 for part in message["content"] if isinstance(part, dict) and "cache_control" in part
)
return count + sum(
1 for tool in tools if isinstance(tool, dict) and "cache_control" in tool
)
@@ -339,6 +331,16 @@ def _completed_transaction_endpoint_indexes(
messages: List[Dict[str, Any]], *, native_anthropic: bool,
) -> List[int]:
"""Select legal ends of completed tool runs and ordinary turns."""
def _tool_run_end(start: int) -> int:
end = start
while end < len(messages):
result = messages[end]
if not isinstance(result, dict) or result.get("role") != "tool":
break
end += 1
return end
endpoints: List[int] = []
index = 0
while index < len(messages):
@@ -346,37 +348,21 @@ def _completed_transaction_endpoint_indexes(
if not isinstance(message, dict) or message.get("role") == "system":
index += 1
continue
role = message.get("role")
if message.get("role") == "assistant" and message.get("tool_calls"):
result_start = index + 1
result_end = result_start
while result_end < len(messages):
result = messages[result_end]
if not isinstance(result, dict) or result.get("role") != "tool":
break
result_end += 1
if result_end > result_start:
endpoint = result_end - 1
if _can_carry_marker(messages[endpoint], native_anthropic):
endpoints.append(endpoint)
if role == "assistant" and message.get("tool_calls"):
result_end = _tool_run_end(index + 1)
if result_end > index + 1 and _can_carry_marker(messages[result_end - 1], native_anthropic):
endpoints.append(result_end - 1)
index = result_end
continue
if message.get("role") == "tool":
while index < len(messages):
result = messages[index]
if not isinstance(result, dict) or result.get("role") != "tool":
break
index += 1
if role == "tool":
index = _tool_run_end(index)
continue
if message.get("role") == "user" and index + 1 < len(messages):
index += 1
continue
if (
message.get("role") == "assistant"
and message.get("content") in (None, "")
if (role == "user" and index + 1 < len(messages)) or (
role == "assistant" and message.get("content") in (None, "")
):
index += 1
continue
@@ -470,14 +456,7 @@ def apply_anthropic_cache_control(
marker = _build_marker(cache_ttl)
for i, msg in enumerate(messages):
if not isinstance(msg, dict):
continue
content = msg.get("content")
has_marker = "cache_control" in msg or (
isinstance(content, list)
and any(isinstance(part, dict) and "cache_control" in part for part in content)
)
if has_marker:
if isinstance(msg, dict) and ("cache_control" in msg or _has_part_marker(msg.get("content"))):
messages[i] = strip_anthropic_cache_control([dict(msg)])[0]
breakpoints_used = 0