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

Merge branch 'simp/r3-06-p2-MOA' into simp/integration3

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:42:47 -07:00
parent 4dcf8ce7cf e216c32e60
commit 7b4977bfa5
4 changed files with 326 additions and 567 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/memory_manager.py
+141 -242
View File
@@ -46,28 +46,21 @@ def _has_var_kwargs(params) -> bool:
def _accepts_require_checkpoint(fn: Callable[..., Any]) -> bool:
"""True if ``fn`` can receive the ``require_checkpoint`` keyword.
"""True if ``fn`` can receive the ``require_checkpoint`` keyword (unreadable signatures -> False).
v2 providers written against the docs example use the bare ``on_pre_compress(self,
messages)`` shape; passing the keyword would raise TypeError, which the host would
re-raise as a checkpoint failure even though the durable write succeeded.
Unreadable signatures conservatively report False.
Bare-shape v2 providers (``on_pre_compress(self, messages)``) would raise TypeError on the
keyword, which the host would re-raise as a checkpoint failure despite a successful write.
"""
params = _signature_params(fn)
if params is None:
return False
param = params.get("require_checkpoint")
return _has_var_kwargs(params) or (
param is not None and param.kind in (inspect.Parameter.KEYWORD_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD)
)
kind = getattr(params.get("require_checkpoint"), "kind", None)
return _has_var_kwargs(params) or kind in (inspect.Parameter.KEYWORD_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD)
def _ctx_bound(fn: Callable[[], Any]) -> Callable[[], Any]:
"""Bind ``fn`` to the CALLER's contextvars for execution on another thread.
Profile isolation is a ContextVar-scoped HERMES_HOME override; an unbound provider
on a worker thread would silently resolve paths/secrets against the default profile.
"""
"""Bind ``fn`` to the CALLER's contextvars for another thread: profile isolation is a
ContextVar-scoped HERMES_HOME override, and an unbound worker would silently use the default profile."""
return partial(contextvars.copy_context().run, fn)
@@ -88,10 +81,8 @@ def normalize_tool_schema(schema: Any) -> Optional[Dict[str, Any]]:
return schema if name and isinstance(name, str) else None
def memory_provider_tools_enabled(
enabled_toolsets: Optional[List[str]], disabled_toolsets: Optional[List[str]] = None, *,
memory_tool_present: bool = False,
) -> bool:
def memory_provider_tools_enabled(enabled_toolsets: Optional[List[str]], disabled_toolsets: Optional[List[str]] = None,
*, memory_tool_present: bool = False) -> bool:
"""Return whether external memory-provider tools should be exposed."""
if disabled_toolsets and "memory" in disabled_toolsets:
return False
@@ -121,11 +112,9 @@ def memory_provider_tools_exposed(agent: Any) -> bool:
never advertises tools absent from the tool surface.
"""
tools = getattr(agent, "tools", None)
return memory_provider_tools_enabled(
getattr(agent, "enabled_toolsets", None),
getattr(agent, "disabled_toolsets", None),
memory_tool_present=isinstance(tools, (list, tuple)) and any(_tool_name(t) == "memory" for t in tools),
)
present = isinstance(tools, (list, tuple)) and any(_tool_name(t) == "memory" for t in tools)
enabled, disabled = getattr(agent, "enabled_toolsets", None), getattr(agent, "disabled_toolsets", None)
return memory_provider_tools_enabled(enabled, disabled, memory_tool_present=present)
def inject_memory_provider_tools(agent: Any) -> int:
@@ -138,7 +127,8 @@ def inject_memory_provider_tools(agent: Any) -> int:
if not memory_provider_tools_exposed(agent):
# Say so once: a silent 0 leaves the provider looking "half on" with no clue which
# config key (platform_toolsets / disabled_toolsets) gated it.
_providers = [p for p in (getattr(memory_manager, "providers", None) or []) if getattr(p, "name", "") != "builtin"]
_providers = [p for p in getattr(memory_manager, "providers", None) or []
if getattr(p, "name", "") != "builtin"]
if _providers:
logger.info(
"Memory provider(s) %s configured but the 'memory' toolset is "
@@ -192,10 +182,9 @@ def sanitize_context(text: str) -> str:
class StreamingContextScrubber:
"""Stateful scrubber for streaming text whose memory-context spans may straddle deltas.
``sanitize_context`` needs both tags in one string, so a span split across deltas
would leak to the UI. This holds back partial-tag tails between ``feed()`` calls and
drops everything inside a span. One scrubber (or ``reset()``) per top-level
response; call ``flush()`` at end of stream.
``sanitize_context`` needs both tags in one string, so a split span would leak to the UI;
this holds back partial-tag tails between ``feed()`` calls and drops span interiors.
One scrubber (or ``reset()``) per top-level response; call ``flush()`` at end of stream.
"""
_OPEN_TAG = "<memory-context>"
@@ -218,13 +207,16 @@ class StreamingContextScrubber:
out: list[str] = []
while buf:
if self._in_span:
idx = buf.lower().find(self._CLOSE_TAG)
held = self._max_partial_suffix(buf, self._CLOSE_TAG) # potential partial close tag
tag = self._CLOSE_TAG
idx = buf.lower().find(tag)
held = self._max_partial_suffix(buf, tag) # potential partial close tag
else:
idx = self._find_boundary_open_tag(buf)
held = self._max_pending_open_suffix(buf) or self._max_partial_suffix(buf, self._OPEN_TAG)
tag = self._OPEN_TAG
idx = self._find_boundary_open_tag(buf)
# A complete boundary tag at the buffer end is held until the next char confirms it.
n = len(tag)
pending = n if buf.lower().endswith(tag) and self._ends_at_block_boundary(buf[:-n]) else 0
held = pending or self._max_partial_suffix(buf, tag)
if idx == -1:
# Hold back the possible partial tag; inside a span the rest is dropped.
if not self._in_span:
@@ -248,41 +240,26 @@ class StreamingContextScrubber:
@staticmethod
def _max_partial_suffix(buf: str, tag: str) -> int:
"""Length of the longest buf-suffix that is a (case-insensitive) prefix of ``tag``, else 0."""
tag_lower = tag.lower()
buf_lower = buf.lower()
for i in range(min(len(buf_lower), len(tag_lower) - 1), 0, -1):
if tag_lower.startswith(buf_lower[-i:]):
return i
return 0
tag_lower, buf_lower = tag.lower(), buf.lower()
span = range(min(len(buf_lower), len(tag_lower) - 1), 0, -1)
return next((i for i in span if tag_lower.startswith(buf_lower[-i:])), 0)
def _find_boundary_open_tag(self, buf: str) -> int:
"""Find an opening fence only when it starts a block-like span (own line, newline after)."""
buf_lower = buf.lower()
search_start = 0
while True:
idx = buf_lower.find(self._OPEN_TAG, search_start)
if idx == -1:
return -1
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":
buf_lower, tag_len = buf.lower(), len(self._OPEN_TAG)
idx = buf_lower.find(self._OPEN_TAG)
while idx != -1:
after_idx = idx + tag_len
if self._ends_at_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 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
idx = buf_lower.find(self._OPEN_TAG, idx + 1)
return -1
def _ends_at_block_boundary(self, text: str) -> bool:
"""Whether emitting ``text`` leaves the stream at a line start (blank tail after the last newline)."""
last_newline = text.rfind("\n")
if last_newline == -1:
return self._at_block_boundary and text.strip() == ""
return text[last_newline + 1:].strip() == ""
def _is_block_boundary(self, buf: str, idx: int) -> bool:
return self._at_block_boundary if idx == 0 else self._ends_at_block_boundary(buf[:idx])
"""Whether emitting ``text`` leaves the stream at a line start (blank tail after the last newline;
no newline at all -> only whitespace and already at a boundary)."""
head, sep, tail = text.rpartition("\n")
return tail.strip() == "" and (bool(sep) or self._at_block_boundary)
def _append_visible(self, out: list[str], text: str) -> None:
if text:
@@ -307,11 +284,6 @@ def build_memory_context_block(raw_context: str) -> str:
)
def _nonblank(text: Any) -> Any:
"""Return ``text`` when it has non-whitespace content, else None."""
return text if text and text.strip() else None
class MemoryManager:
"""Builtin provider (always first) plus at most one external provider.
@@ -323,11 +295,11 @@ class MemoryManager:
self._providers: List[MemoryProvider] = []
self._tool_to_provider: Dict[str, MemoryProvider] = {}
self._has_external: bool = False
self._external_prefetch_timeout = (
_EXTERNAL_PREFETCH_TIMEOUT_S if external_prefetch_timeout is None else float(external_prefetch_timeout)
)
if self._external_prefetch_timeout <= 0:
timeout = external_prefetch_timeout
timeout = _EXTERNAL_PREFETCH_TIMEOUT_S if timeout is None else float(timeout)
if timeout <= 0:
raise ValueError("external_prefetch_timeout must be positive")
self._external_prefetch_timeout = timeout
self._external_prefetch_threads: Dict[str, threading.Thread] = {}
self._external_prefetch_lock = threading.Lock()
# Single-worker background executor for end-of-turn sync/prefetch, created lazily so
@@ -342,15 +314,10 @@ class MemoryManager:
"status": "not_started", "abandoned_writes": 0, "abandoned_prefetches": 0, "active_tasks": 0,
}
def _each_provider(
self, label: str, call: Callable[[MemoryProvider], Any], *,
level: int = logging.DEBUG, providers: Optional[List[MemoryProvider]] = None, exc_info: bool = False,
) -> List[Any]:
"""Call ``call(provider)`` for each provider, logging and swallowing failures.
``label`` completes the log line ``Memory provider '<name>' <label>: <exc>``.
Returns the successful results in provider order.
"""
def _each_provider(self, label: str, call: Callable[[MemoryProvider], Any], *, level: int = logging.DEBUG,
providers: Optional[List[MemoryProvider]] = None, exc_info: bool = False) -> List[Any]:
"""Call ``call(provider)`` per provider, logging+swallowing failures; returns successes in order.
``label`` completes the log line ``Memory provider '<name>' <label>: <exc>``."""
results: List[Any] = []
for provider in self._providers if providers is None else providers:
try:
@@ -409,10 +376,9 @@ class MemoryManager:
def build_system_prompt(self) -> str:
"""Join every provider's non-empty ``system_prompt_block()`` with blank lines."""
blocks = self._each_provider(
"system_prompt_block() failed", lambda p: _nonblank(p.system_prompt_block()), level=logging.WARNING,
)
return "\n\n".join(b for b in blocks if b)
blocks = self._each_provider("system_prompt_block() failed", lambda p: p.system_prompt_block(),
level=logging.WARNING)
return "\n\n".join(b for b in blocks if b and b.strip())
# A /skill or /bundle turn embeds the whole skill body in the model-facing message;
# providers get just the user's instruction (None for a bare invocation).
@@ -424,17 +390,13 @@ class MemoryManager:
if not clean_query:
return ""
parts = self._each_provider(
"prefetch failed (non-fatal)",
lambda p: _nonblank(self._prefetch_provider(p, clean_query, session_id=session_id)),
"prefetch failed (non-fatal)", lambda p: self._prefetch_provider(p, clean_query, session_id=session_id),
)
return "\n\n".join(p for p in parts if p)
return "\n\n".join(p for p in parts if p and p.strip())
def _prefetch_provider(self, provider: MemoryProvider, query: str, *, session_id: str = "") -> str:
"""Run one provider's prefetch; external providers are bounded by a timeout.
A stuck external call keeps running on its daemon thread and the provider is
skipped on later turns until it returns.
"""
"""Run one provider's prefetch; external providers are bounded by a timeout. A stuck external
call keeps running on its daemon thread and the provider is skipped on later turns until it returns."""
if provider.name == "builtin":
return provider.prefetch(query, session_id=session_id)
@@ -449,11 +411,9 @@ class MemoryManager:
thread = threading.Thread(target=_ctx_bound(_run), daemon=True, name=f"memory-prefetch-{provider.name}")
with self._external_prefetch_lock:
existing = self._external_prefetch_threads.get(provider.name)
if existing is not None:
if existing.is_alive():
logger.debug("Memory provider '%s' prefetch is still running; skipping this turn", provider.name)
return ""
self._external_prefetch_threads.pop(provider.name, None)
if existing is not None and existing.is_alive():
logger.debug("Memory provider '%s' prefetch is still running; skipping this turn", provider.name)
return ""
self._external_prefetch_threads[provider.name] = thread
thread.start()
@@ -473,21 +433,15 @@ class MemoryManager:
return result_box.get("value", "")
def describe_recall(self) -> str:
"""Deterministic recall indicator line (e.g. ``"🧠 Provider — recalled 3 memories"``).
Call right after :meth:`prefetch_all` so the user SEES memory was used regardless
of whether the model mentions it; ``""`` when no provider injected memory.
"""
"""Deterministic recall indicator line (e.g. ``"🧠 Provider — recalled 3 memories"``); ``""`` if none.
Call right after :meth:`prefetch_all` so the user SEES memory was used even if the model is silent."""
segments: List[str] = []
for status in self._each_provider("recall_status failed (non-fatal)", lambda p: p.recall_status()):
if status is None:
continue
if status.count == 1:
detail = "recalled 1 memory"
elif status.count > 1:
detail = f"recalled {status.count} memories"
else: # content injected but no discrete count (reflect)
detail = "recalled relevant memory"
# count <= 0: content injected but no discrete count (reflect)
detail = ("recalled 1 memory" if status.count == 1 else f"recalled {status.count} memories"
if status.count > 1 else "recalled relevant memory")
segments.append(f"{status.glyph} {status.provider_label} — {detail}")
return " ".join(segments)
@@ -497,14 +451,10 @@ class MemoryManager:
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)",
lambda p: p.queue_prefetch(clean_query, session_id=session_id),
providers=providers,
),
kind="prefetch",
)
self._submit_background(lambda: self._each_provider(
"queue_prefetch failed (non-fatal)", lambda p: p.queue_prefetch(clean_query, session_id=session_id),
providers=providers,
), kind="prefetch")
@staticmethod
def _provider_sync_accepts_messages(provider: MemoryProvider) -> bool:
@@ -512,15 +462,12 @@ class MemoryManager:
params = _signature_params(provider.sync_turn)
return params is None or _has_var_kwargs(params) or "messages" in params
def sync_all(
self, user_content: str, assistant_content: str, *,
session_id: str = "", messages: Optional[List[Dict[str, Any]]] = None,
) -> None:
def sync_all(self, user_content: str, assistant_content: str, *, session_id: str = "",
messages: Optional[List[Dict[str, Any]]] = None) -> None:
"""Sync a completed turn to all providers on the background worker.
Never inline: a provider's ``sync_turn`` may block for minutes, which kept
``run_conversation`` open after the user saw the response. The single worker
also serializes writes so turn N lands before turn N+1.
Never inline: a provider's ``sync_turn`` may block for minutes, which kept ``run_conversation``
open after the user saw the response. The single worker also serializes writes (turn N before N+1).
"""
providers = list(self._providers)
clean_user_content = self._strip_skill_scaffolding(user_content) if providers else None
@@ -538,13 +485,21 @@ class MemoryManager:
)
def _submit_background(self, fn, *, kind: str = "write") -> None:
"""Queue ``fn`` on the serialized worker and track its durability class.
Runs under the caller's contextvars (``_ctx_bound``). If the executor is
unavailable outside shutdown, run inline — the historical fail-safe.
"""
"""Queue ``fn`` on the serialized worker (created lazily; None once shutting down) and track its
durability class. Runs under the caller's contextvars (``_ctx_bound``). If the executor is
unavailable outside shutdown, run inline — the historical fail-safe."""
fn = _ctx_bound(fn)
executor = self._get_sync_executor()
executor = None if self._shutting_down else self._sync_executor
if executor is None and not self._shutting_down:
with self._sync_executor_lock:
if self._sync_executor is None and not self._shutting_down:
try:
# Daemon workers: a wedged provider must never block interpreter exit.
from tools.daemon_pool import DaemonThreadPoolExecutor
self._sync_executor = DaemonThreadPoolExecutor(max_workers=1, thread_name_prefix="mem-sync")
except Exception as e: # pragma: no cover - resource exhaustion
logger.warning("Failed to create memory sync executor: %s", e)
executor = self._sync_executor
future = None
try:
# Submit+track atomically with the shutdown snapshot. The callback is attached
@@ -562,53 +517,31 @@ class MemoryManager:
return
if future is not None:
future.add_done_callback(self._forget_background_future)
else:
try:
fn()
except Exception as e: # pragma: no cover - fn guards internally
logger.debug("Inline memory background task failed: %s", e)
return
try:
fn()
except Exception as e: # pragma: no cover - fn guards internally
logger.debug("Inline memory background task failed: %s", e)
def _forget_background_future(self, future: Future) -> None:
with self._sync_executor_lock:
self._background_futures.pop(future, None)
def _get_sync_executor(self) -> Optional[ThreadPoolExecutor]:
"""Lazily create the single-worker background executor (None once shutting down)."""
if self._shutting_down:
return None
if self._sync_executor is not None:
return self._sync_executor
with self._sync_executor_lock:
if self._sync_executor is None and not self._shutting_down:
try:
# Daemon workers: a wedged provider must never block interpreter exit.
from tools.daemon_pool import DaemonThreadPoolExecutor
self._sync_executor = DaemonThreadPoolExecutor(max_workers=1, thread_name_prefix="mem-sync")
except Exception as e: # pragma: no cover - resource exhaustion
logger.warning("Failed to create memory sync executor: %s", e)
return self._sync_executor
def flush_pending(self, timeout: Optional[float] = None) -> bool:
"""Block until queued sync/prefetch work has drained (False on timeout).
With a single worker, a sentinel task completing proves every earlier task ran.
"""
With a single worker, a sentinel task completing proves every earlier task ran."""
executor = self._sync_executor
if executor is None:
return True
try:
executor.submit(lambda: None).result(timeout=timeout)
except RuntimeError:
return True # executor already shut down — nothing pending
except Exception:
return False
except Exception as e:
return isinstance(e, RuntimeError) # executor already shut down — nothing pending
return True
def get_all_tool_schemas(self) -> List[Dict[str, Any]]:
"""Collect deduplicated tool schemas from all providers.
Reserved core tool names are skipped: :meth:`add_provider` refuses to route them.
"""
"""Collect deduplicated tool schemas from all providers; reserved core tool names are
skipped because :meth:`add_provider` refuses to route them."""
from toolsets import _HERMES_CORE_TOOLS
schemas: List[Dict[str, Any]] = []
@@ -630,7 +563,7 @@ class MemoryManager:
return schemas
def get_all_tool_names(self) -> set:
return set(self._tool_to_provider.keys())
return set(self._tool_to_provider)
def has_tool(self, tool_name: str) -> bool:
return tool_name in self._tool_to_provider
@@ -650,19 +583,15 @@ class MemoryManager:
self._each_provider("on_turn_start failed", lambda p: p.on_turn_start(turn_number, message, **kwargs))
def on_session_end(self, messages: List[Dict[str, Any]]) -> None:
self._each_provider(
"on_session_end failed", lambda p: p.on_session_end(messages), level=logging.WARNING, exc_info=True,
)
self._each_provider("on_session_end failed", lambda p: p.on_session_end(messages), level=logging.WARNING,
exc_info=True)
def commit_session_boundary_async(
self, messages: List[Dict[str, Any]], *,
new_session_id: str, parent_session_id: str = "", reason: str = "new_session",
) -> None:
def commit_session_boundary_async(self, messages: List[Dict[str, Any]], *, new_session_id: str,
parent_session_id: str = "", reason: str = "new_session") -> None:
"""Queue old-session extraction + provider rebinding as ONE serialized task.
``on_session_end`` (LLM-bound, seconds) must run strictly BEFORE ``on_session_switch``
rebinds provider session state; an ad-hoc thread raced the inline switch and
misattributed transcripts. One FIFO task gives immediate return plus ordering.
``on_session_end`` (LLM-bound, seconds) must run strictly BEFORE ``on_session_switch`` rebinds
provider state; an ad-hoc thread raced the inline switch and misattributed transcripts.
"""
if not self._providers:
return
@@ -680,10 +609,8 @@ class MemoryManager:
self._submit_background(_run)
def on_session_switch(
self, new_session_id: str, *,
parent_session_id: str = "", reset: bool = False, rewound: bool = False, **kwargs,
) -> None:
def on_session_switch(self, new_session_id: str, *, parent_session_id: str = "", reset: bool = False,
rewound: bool = False, **kwargs) -> None:
"""Notify providers that ``AIAgent.session_id`` rotated without teardown
(``/resume``, ``/branch``, ``/reset``, ``/new``, compression). ``rewound=True``
(``/undo``): same id, truncated transcript."""
@@ -704,57 +631,45 @@ class MemoryManager:
except (TypeError, ValueError):
return None
@classmethod
def _is_checkpoint_provider(cls, provider: MemoryProvider, api_version: int) -> bool:
version = cls._checkpoint_api_version(provider)
return (_LEGACY_PRE_COMPRESS_API_VERSION if version is None else version) >= api_version
def supports_pre_compress_checkpoint(self, api_version: int = PRE_COMPRESS_CHECKPOINT_API_VERSION) -> bool:
"""Return whether an active provider guarantees checkpoint API support."""
return any(
(version := self._checkpoint_api_version(p)) is not None and version >= api_version
for p in self._providers
)
versions = (self._checkpoint_api_version(p) for p in self._providers)
return any(v is not None and v >= api_version for v in versions)
def on_pre_compress(
self, messages: List[Dict[str, Any]], *,
evidence_messages: Optional[List[Dict[str, Any]]] = None,
require_checkpoint: bool = False,
checkpoint_api_version: int = PRE_COMPRESS_CHECKPOINT_API_VERSION,
) -> str:
def on_pre_compress(self, messages: List[Dict[str, Any]], *,
evidence_messages: Optional[List[Dict[str, Any]]] = None, require_checkpoint: bool = False,
checkpoint_api_version: int = PRE_COMPRESS_CHECKPOINT_API_VERSION) -> str:
"""Notify providers before compression; return their combined summary-prompt text.
``messages`` is the raw transcript (the v1 contract). ``evidence_messages`` is the
host-normalized list handed only to checkpoint (v2+) providers. With
``require_checkpoint``, at least one checkpoint provider must succeed — its
exception propagates so the caller keeps the uncompressed transcript.
``messages`` is the raw v1 transcript; ``evidence_messages`` is the host-normalized list handed
only to checkpoint (v2+) providers. With ``require_checkpoint`` at least one checkpoint provider
must succeed — its exception propagates so the caller keeps the uncompressed transcript.
"""
parts = []
checkpoint_succeeded = False
for provider in self._providers:
is_checkpoint_provider = self._is_checkpoint_provider(provider, checkpoint_api_version)
provider_messages = messages
version = self._checkpoint_api_version(provider)
if version is None:
version = _LEGACY_PRE_COMPRESS_API_VERSION
is_checkpoint_provider = version >= checkpoint_api_version
use_evidence = is_checkpoint_provider and evidence_messages is not None
provider_messages = evidence_messages if use_evidence else messages
kwargs: Dict[str, Any] = {}
if is_checkpoint_provider:
if evidence_messages is not None:
provider_messages = evidence_messages
# v1 providers and bare-shape v2 providers never see the signal.
if _accepts_require_checkpoint(provider.on_pre_compress):
kwargs["require_checkpoint"] = require_checkpoint
# v1 providers and bare-shape v2 providers never see the signal.
if is_checkpoint_provider and _accepts_require_checkpoint(provider.on_pre_compress):
kwargs["require_checkpoint"] = require_checkpoint
try:
result = provider.on_pre_compress(provider_messages, **kwargs)
if result and result.strip():
parts.append(result)
checkpoint_succeeded = checkpoint_succeeded or is_checkpoint_provider
except Exception as e:
logger.debug("Memory provider '%s' on_pre_compress failed: %s", provider.name, e)
if require_checkpoint and is_checkpoint_provider:
raise
continue
checkpoint_succeeded = checkpoint_succeeded or is_checkpoint_provider
if require_checkpoint and not checkpoint_succeeded:
raise RuntimeError(
"No active memory provider completed pre-compress checkpoint "
f"API v{checkpoint_api_version}"
f"No active memory provider completed pre-compress checkpoint API v{checkpoint_api_version}"
)
return "\n\n".join(parts)
@@ -767,9 +682,8 @@ class MemoryManager:
accepted = sum(p.kind is not inspect.Parameter.VAR_POSITIONAL for p in params.values())
return "positional" if accepted >= 4 else "legacy"
def on_memory_write(
self, action: str, target: str, content: str, metadata: Optional[Dict[str, Any]] = None,
) -> None:
def on_memory_write(self, action: str, target: str, content: str,
metadata: Optional[Dict[str, Any]] = None) -> None:
"""Notify external providers when the built-in memory tool writes (skips builtin, the source)."""
def _notify(provider: MemoryProvider) -> None:
@@ -781,9 +695,8 @@ class MemoryManager:
else:
provider.on_memory_write(action, target, content, metadata=dict(metadata or {}))
self._each_provider(
"on_memory_write failed", _notify, providers=[p for p in self._providers if p.name != "builtin"],
)
external = [p for p in self._providers if p.name != "builtin"]
self._each_provider("on_memory_write failed", _notify, providers=external)
# Actions mirrored to external providers; non-mutating results (errors, staged) are
# filtered by ``notify_memory_tool_write`` first.
@@ -791,11 +704,8 @@ class MemoryManager:
@staticmethod
def _memory_tool_result_succeeded(result: Any) -> bool:
"""True only when the built-in memory tool actually committed a write.
Fails closed (non-JSON, non-dict, missing ``success``, staged for approval) so
providers never mirror a write that did not land.
"""
"""True only when the built-in memory tool actually committed a write. Fails closed (non-JSON,
non-dict, missing ``success``, staged for approval) so providers never mirror a write that did not land."""
if isinstance(result, str):
try:
result = json.loads(result)
@@ -803,23 +713,19 @@ class MemoryManager:
return False
return isinstance(result, dict) and result.get("success") is True and result.get("staged") is not True
def notify_memory_tool_write(
self, tool_result: Any, tool_args: Dict[str, Any], *,
build_metadata: Optional[Callable[[], Dict[str, Any]]] = None,
) -> None:
def notify_memory_tool_write(self, tool_result: Any, tool_args: Dict[str, Any], *,
build_metadata: Optional[Callable[[], Dict[str, Any]]] = None) -> None:
"""Mirror a built-in memory tool call to external providers.
Gates on a committed write, expands single-op and batched ``operations`` shapes,
keeps only mutating actions, and forwards ``old_text`` plus per-op provenance from
``build_metadata`` (the loop knows session/task/tool-call identity; we do not).
Gates on a committed write, expands single-op and batched ``operations`` shapes, keeps only
mutating actions, and forwards ``old_text`` plus provenance from ``build_metadata`` (the loop
knows session/task/tool-call identity; we do not).
"""
if not self._memory_tool_result_succeeded(tool_result):
return
target = str(tool_args.get("target") or "memory")
operations = tool_args.get("operations")
if not (isinstance(operations, list) and operations):
operations = [tool_args]
for op in operations:
for op in operations if isinstance(operations, list) and operations else [tool_args]:
action = str(op.get("action") or "") if isinstance(op, dict) else ""
if action not in self._MIRRORED_MEMORY_ACTIONS:
continue
@@ -841,9 +747,8 @@ class MemoryManager:
def shutdown_all(self) -> None:
"""Drain the background executor (bounded), then shut providers down in reverse order."""
self._drain_sync_executor()
self._each_provider(
"shutdown failed", lambda p: p.shutdown(), level=logging.WARNING, providers=list(reversed(self._providers)),
)
self._each_provider("shutdown failed", lambda p: p.shutdown(), level=logging.WARNING,
providers=self._providers[::-1])
@property
def shutdown_drain_state(self) -> Dict[str, Any]:
@@ -860,8 +765,7 @@ class MemoryManager:
tracked = dict(self._background_futures)
self._shutdown_drain_state = {
"status": "draining" if executor is not None else "drained",
"abandoned_writes": 0,
"abandoned_prefetches": 0,
"abandoned_writes": 0, "abandoned_prefetches": 0,
"active_tasks": sum(not future.done() for future in tracked),
}
if executor is None:
@@ -871,21 +775,17 @@ class MemoryManager:
# tracked futures lets the worker run every queued task in order up to the deadline.
executor.shutdown(wait=False, cancel_futures=False)
_, pending = wait(tuple(tracked), timeout=_SYNC_DRAIN_TIMEOUT_S)
if not pending:
with self._sync_executor_lock:
self._shutdown_drain_state.update(status="drained", active_tasks=0)
return
cancelled = [tracked[future] for future in pending if future.cancel()]
active_tasks = len(pending) - len(cancelled)
abandoned_prefetches = cancelled.count("prefetch")
abandoned_writes = len(cancelled) - abandoned_prefetches
with self._sync_executor_lock:
self._shutdown_drain_state.update(
status="timed_out", abandoned_writes=abandoned_writes,
status="timed_out" if pending else "drained", abandoned_writes=abandoned_writes,
abandoned_prefetches=abandoned_prefetches, active_tasks=active_tasks,
)
if not pending:
return
logger.warning(
"Memory shutdown drain timed out after %.2fs; abandoning %d queued "
"memory write(s) and %d queued prefetch(es); %d active task(s) remain detached",
@@ -897,6 +797,5 @@ class MemoryManager:
if "hermes_home" not in kwargs:
from hermes_constants import get_hermes_home
kwargs["hermes_home"] = str(get_hermes_home())
self._each_provider(
"initialize failed", lambda p: p.initialize(session_id=session_id, **kwargs), level=logging.WARNING,
)
self._each_provider("initialize failed", lambda p: p.initialize(session_id=session_id, **kwargs),
level=logging.WARNING)
agent/message_sanitization.py
+75 -125
View File
@@ -26,9 +26,7 @@ _MESSAGE_CORE_KEYS = frozenset({"content", "name", "tool_calls", "role"})
def _sanitize_surrogates(text: str) -> str:
"""Replace lone surrogate code points with U+FFFD; no-op when none present."""
if _SURROGATE_RE.search(text):
return _SURROGATE_RE.sub('\ufffd', text)
return text
return _SURROGATE_RE.sub('\ufffd', text)
def _strip_non_ascii(text: str) -> str:
@@ -39,12 +37,11 @@ def _strip_non_ascii(text: str) -> str:
def _fix_str_field(container: Any, key: Any, fix: Callable[[str], str]) -> bool:
"""Apply ``fix`` to ``container[key]`` if it is a str; True if it changed."""
value = container.get(key) if isinstance(container, dict) else container[key]
if isinstance(value, str):
fixed = fix(value)
if fixed != value:
container[key] = fixed
return True
return False
fixed = fix(value) if isinstance(value, str) else value
if fixed == value:
return False
container[key] = fixed
return True
def _sanitize_structure(payload: Any, fix: Callable[[str], str]) -> bool:
@@ -63,36 +60,25 @@ def _sanitize_structure(payload: Any, fix: Callable[[str], str]) -> bool:
def _sanitize_messages(messages: list, fix: Callable[[str], str], *, deep: bool) -> bool:
"""Apply ``fix`` to the string fields of every message dict in-place.
Covers content / content-part text, name, tool_call arguments, and every non-core
top-level str field. ``deep=True`` additionally covers tool_call ids, function names,
and NESTED non-core fields (``reasoning_details`` arrays from byte-level reasoning models).
"""
"""Apply ``fix`` to the string fields of every message dict in-place (content / part text,
name, tool_call arguments, non-core top-level str fields). ``deep=True`` adds tool_call ids,
function names, and NESTED non-core fields (``reasoning_details`` from byte-level models)."""
found = False
for msg in messages:
if not isinstance(msg, dict):
continue
content = msg.get("content")
if isinstance(content, list):
for part in content:
if isinstance(part, dict):
found |= _fix_str_field(part, "text", fix)
else:
found |= _fix_str_field(msg, "content", fix)
found |= _fix_str_field(msg, "name", fix)
parts = [(p, "text") for p in content if isinstance(p, dict)] if isinstance(content, list) else None
fields = parts if parts is not None else [(msg, "content")]
fields.append((msg, "name"))
tool_calls = msg.get("tool_calls")
for tc in tool_calls if isinstance(tool_calls, list) else ():
if not isinstance(tc, dict):
continue
fn = tc.get("function")
fn_fields = [(fn, "name"), (fn, "arguments")] if isinstance(fn, dict) else []
for container, key in [(tc, "id")] + fn_fields:
if deep or key == "arguments":
found |= _fix_str_field(container, key, fix)
for key, value in list(msg.items()):
if key in _MESSAGE_CORE_KEYS:
continue
fn = tc.get("function") if isinstance(tc, dict) else None
fields += [(tc, "id")] if deep and isinstance(tc, dict) else []
fields += ([(fn, "name")] if deep else []) + [(fn, "arguments")] if isinstance(fn, dict) else []
for container, key in fields:
found |= _fix_str_field(container, key, fix)
for key, value in [kv for kv in msg.items() if kv[0] not in _MESSAGE_CORE_KEYS]:
if isinstance(value, str):
found |= _fix_str_field(msg, key, fix)
elif deep and isinstance(value, (dict, list)):
@@ -169,14 +155,12 @@ def _repair_tool_call_arguments(raw_args: str, tool_name: str = "?") -> str:
fixed += '}' * max(0, fixed.count('{') - fixed.count('}'))
fixed += ']' * max(0, fixed.count('[') - fixed.count(']'))
for _ in range(50):
if _loads_ok(fixed):
break
if (fixed.endswith('}') and fixed.count('}') > fixed.count('{')) or (
fixed.endswith(']') and fixed.count(']') > fixed.count('[')
if _loads_ok(fixed) or not (
(fixed.endswith('}') and fixed.count('}') > fixed.count('{'))
or (fixed.endswith(']') and fixed.count(']') > fixed.count('['))
):
fixed = fixed[:-1]
else:
break
fixed = fixed[:-1]
if _loads_ok(fixed):
logger.warning("Repaired malformed tool_call arguments for %s: %s → %s", tool_name, raw_stripped[:80], fixed[:80])
@@ -199,12 +183,10 @@ def _repair_tool_call_arguments(raw_args: str, tool_name: str = "?") -> str:
def close_interrupted_tool_sequence(messages: list, final_response: Any = None) -> bool:
"""Append a synthetic assistant turn when an interrupted tail is a tool result.
A transcript ending on a raw ``tool`` message makes the next user message land as
``tool → user`` — an alternation violation strict providers (Gemini, Claude) answer by
hallucinating a continuation. Mutates in place; True if a closing turn was appended.
"""
"""Append a synthetic assistant turn when an interrupted tail is a tool result: a transcript
ending on a raw ``tool`` message makes the next user message land as ``tool → user``, an
alternation violation strict providers (Gemini, Claude) answer by hallucinating a
continuation. Mutates in place; True if a closing turn was appended."""
last = messages[-1] if messages else None
if not isinstance(last, dict) or last.get("role") != "tool":
return False
@@ -216,12 +198,9 @@ 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 prices images at a flat cost, so
it cannot score recovery from an image-dominated 413. Non-serializable values fall
back to ``str()`` so a malformed message can never crash recovery.
"""
"""Exact serialized byte size of ``messages`` (HTTP 413 is a BYTE-size error the token
estimator, pricing images flat, cannot score). 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
try:
@@ -236,19 +215,16 @@ _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).
``tool`` messages and assistant messages carrying ``tool_calls`` whose content was
entirely images get a placeholder, NOT deleted (deleting orphans the paired
``tool_call_id`` → HTTP 400); other now-empty messages are dropped. Rewritten messages
lose their ``api_content`` sidecar (it carries the images being removed).
``tool`` / ``tool_calls`` messages left empty get a placeholder, NOT deleted (deleting
orphans the paired ``tool_call_id`` → HTTP 400); other now-empty messages are dropped.
Rewritten messages lose their ``api_content`` sidecar (it carries the removed images).
"""
from agent.turn_context import drop_stale_api_content
found = False
to_delete = []
for i, msg in enumerate(messages):
if not isinstance(msg, dict):
continue
content = msg.get("content")
content = msg.get("content") if isinstance(msg, dict) else None
if not isinstance(content, list):
continue
new_parts = [p for p in content if not (isinstance(p, dict) and p.get("type") in _IMAGE_PART_TYPES)]
@@ -326,10 +302,7 @@ def _tc_field(tc: Any, key: str) -> Any:
def _tc_set(tc: Any, key: str, value: Any) -> None:
if isinstance(tc, dict):
tc[key] = value
else:
setattr(tc, key, value)
tc.__setitem__(key, value) if isinstance(tc, dict) else setattr(tc, key, value)
def deterministic_call_id(fn_name: str, arguments: str, index: int = 0) -> str:
@@ -344,10 +317,8 @@ def _expand_tool_id_variants(values: tuple[Any, ...]) -> frozenset[str]:
variants: set[str] = set()
for raw in values:
value = raw.strip() if isinstance(raw, str) else ""
if not value:
continue
variants.add(value)
if "|" in value:
if value:
variants.add(value)
variants.update(p for p in (part.strip() for part in value.split("|")) if p)
return frozenset(variants)
@@ -363,11 +334,9 @@ def tool_result_id_variants(tool_call_id: Any) -> frozenset[str]:
def coalesce_tool_call_id(tc: Any) -> str:
"""Effective call id of a tool_call entry (dict or object); ``""`` when none.
Codex Responses calls carry ``call_id`` (authoritative pairing key), Chat Completions
carry ``id`` only, and bridge ids may be ``call_id|response_item_id``.
"""
"""Effective call id of a tool_call entry (dict or object); ``""`` when none. Codex Responses
carry ``call_id`` (authoritative pairing key), Chat Completions ``id`` only, and bridge ids
may be ``call_id|response_item_id``."""
for raw in (_tc_field(tc, "call_id"), _tc_field(tc, "id")):
value = raw.strip() if isinstance(raw, str) else ""
if value:
@@ -379,10 +348,10 @@ 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 a batch; the pre-API sanitizer then keeps only the
first call/result pair per id and strict providers reject duplicates. 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). Blank ids are left for the deterministic fallback in ``build_assistant_message``.
first call/result pair per id and strict providers reject duplicates. Later collisions
get a deterministic ``<id>_d<n>`` suffix (never uuid4 — cache-prefix stability). Mutates
entries (SDK models / SimpleNamespace / dicts) in place. Blank ids are left for the
deterministic fallback in ``build_assistant_message``.
"""
seen: set = set()
for tc in tool_calls or []:
@@ -396,18 +365,14 @@ def uniquify_tool_call_ids(tool_calls: list) -> list:
if cid not in seen:
seen.add(cid)
continue
n = 2
while f"{cid}_d{n}" in seen:
n += 1
new_id = f"{cid}_d{n}"
# range is bounded: at most len(seen) suffixes can already be taken.
new_id = next(f"{cid}_d{n}" for n in range(2, len(seen) + 3) if f"{cid}_d{n}" not in seen)
seen.add(new_id)
def _renamed(value):
# Keep a composite id's response-item half so the provider's fc_/item id survives.
return f"{new_id}|{value.split('|', 1)[1]}" if isinstance(value, str) and "|" in value else new_id
try:
_tc_set(tc, "id", _renamed(_tc_field(tc, "id")))
# Keep a composite id's response-item half so the provider's fc_/item id survives.
old = _tc_field(tc, "id")
_tc_set(tc, "id", f"{new_id}|{old.split('|', 1)[1]}" if isinstance(old, str) and "|" in old else new_id)
if _tc_field(tc, "call_id"):
_tc_set(tc, "call_id", new_id)
except Exception:
@@ -415,22 +380,17 @@ def uniquify_tool_call_ids(tool_calls: list) -> list:
continue
_fn_name = _tc_field(_tc_field(tc, "function"), "name") or "?"
logger.warning(
"Model reused tool call id %s within one turn; renamed the "
"duplicate to %s (tool=%s) to keep call/result pairing "
"lossless.", cid, new_id, _fn_name,
"Model reused tool call id %s within one turn; renamed the duplicate to %s (tool=%s) to keep "
"call/result pairing lossless.", cid, new_id, _fn_name,
)
return tool_calls
# -- reasoning_content policy: single owner of strip-vs-re-pad; adapters keep only SYNTAX --
# require-side (echo-back enforced; replays 400 without the field):
# kimi — provider kimi-coding/kimi-coding-cn, or host api.kimi.com / moonshot.ai /
# moonshot.cn. Host-driven on purpose: aggregators re-exporting kimi reject it.
# deepseek — provider "deepseek", model contains "deepseek", or host api.deepseek.com.
# V4 rejects empty-string pads → " " single space.
# mimo — provider "xiaomi", model contains "mimo", or host *.xiaomimimo.com.
# strict side (field rejected 400/422 "Extra inputs are not permitted"): everyone else —
# Mistral, Cerebras, Groq, SambaNova, … Strip the key entirely, even a one-space pad.
# Require side (echo-back enforced; replays 400 without the field): the families below. Kimi
# is host-driven on purpose (aggregators re-exporting kimi reject it); DeepSeek V4 rejects
# empty-string pads → " ". Strict side (400/422 "Extra inputs are not permitted"): everyone
# else — Mistral, Cerebras, Groq, SambaNova, … Strip the key entirely, even a one-space pad.
_REASONING_ECHO_RULES: tuple = (
# (family, exact providers (raw), exact providers (lowered), model substrings (lowered), hosts)
@@ -449,20 +409,16 @@ def matches_reasoning_echo_family(family: str, provider: Any, model: Any, base_u
_, raw_providers, lowered_providers, model_subs, hosts = _REASONING_ECHO_RULE_BY_FAMILY[family]
model_lower = (model or "").lower()
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)
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":
"""``"kimi"`` / ``"deepseek"`` / ``"mimo"`` (first match in table order) when the
endpoint enforces reasoning_content echo-back, else ``None`` (strip side)."""
return next(
(rule[0] for rule in _REASONING_ECHO_RULES if matches_reasoning_echo_family(rule[0], provider, model, base_url)),
None,
)
families = (rule[0] for rule in _REASONING_ECHO_RULES)
return next((f for f in families if matches_reasoning_echo_family(f, provider, model, base_url)), None)
def needs_reasoning_echo(provider: Any, model: Any, base_url: Any) -> bool:
@@ -471,14 +427,12 @@ def needs_reasoning_echo(provider: Any, model: Any, base_url: Any) -> bool:
def stale_thinking_reaches_wire(api_mode: Any, provider: Any, model: Any, base_url: Any) -> bool:
"""True when stale assistant ``reasoning``/``reasoning_content`` text is actually replayed
on the wire for the active route.
"""True when stale assistant reasoning text is actually replayed on the wire for the route.
The single wire-truth predicate the compaction TRIGGER estimator and the tail-budget
walks must share: if they disagree, a reasoning-heavy session can look over-threshold
to preflight yet fully tail-protected to the walk — an infinite compaction loop.
``codex_responses`` never reads the text keys (continuity rides the encrypted sidecar);
echo-back families replay stored ``reasoning_content`` verbatim; everyone else strips.
``codex_responses`` never reads the text keys (continuity rides the encrypted sidecar).
"""
return (api_mode or "") != "codex_responses" and needs_reasoning_echo(provider, model, base_url)
@@ -494,41 +448,37 @@ def apply_reasoning_content_policy(source_msg: dict, api_msg: dict, needs_thinki
# non-string value (None after compaction): never pass null to the API.
api_msg.pop("reasoning_content", None)
return
existing = source_msg.get("reasoning_content")
existing, reasoning = source_msg.get("reasoning_content"), source_msg.get("reasoning")
if isinstance(existing, str):
# Explicit value: preserve verbatim, upgrading legacy "" to " " (DeepSeek V4 400s on "").
api_msg["reasoning_content"] = existing or " "
return
reasoning = source_msg.get("reasoning")
if isinstance(reasoning, str) and reasoning and not source_msg.get("tool_calls"):
elif isinstance(reasoning, str) and reasoning and not source_msg.get("tool_calls"):
# Healthy session: promote internal 'reasoning' → 'reasoning_content'.
api_msg["reasoning_content"] = reasoning
return
# tool_calls + 'reasoning' but no 'reasoning_content' means the reasoning came from
# ANOTHER provider (DeepSeek's own build pins reasoning_content for tool-call turns):
# pad without leaking foreign CoT. No reasoning at all: every assistant turn still needs
# the field; " " (not "") because DeepSeek V4 rejects empty string.
api_msg["reasoning_content"] = " "
else:
# tool_calls + 'reasoning' but no 'reasoning_content' means the reasoning came from
# ANOTHER provider (DeepSeek's own build pins reasoning_content for tool-call turns):
# pad without leaking foreign CoT. No reasoning at all: every assistant turn still needs
# the field; " " (not "") because DeepSeek V4 rejects empty string.
api_msg["reasoning_content"] = " "
def reapply_reasoning_echo(api_messages: list, needs_thinking_pad: bool) -> int:
"""Re-pad (or strip) assistant turns' reasoning_content for the ACTIVE provider.
``api_messages`` is built once under the primary provider; a mid-conversation
fallback can switch providers, so baked-in fields must be reconciled: TO a
require-side provider re-applies the pad (else 400), TO a strict provider strips it
(else 422). Idempotent. Returns the number of assistant turns changed.
``api_messages`` is built once under the primary provider; a mid-conversation fallback
can switch providers, so baked-in fields must be reconciled: TO a require-side provider
re-applies the pad (else 400), TO a strict one strips it (else 422). Idempotent.
Returns the number of assistant turns changed.
"""
changed = 0
for api_msg in api_messages:
if api_msg.get("role") != "assistant":
continue
if needs_thinking_pad:
if api_msg.get("reasoning_content"):
continue
apply_reasoning_content_policy(api_msg, api_msg, needs_thinking_pad)
if api_msg.get("reasoning_content"):
changed += 1
if not api_msg.get("reasoning_content"):
apply_reasoning_content_policy(api_msg, api_msg, needs_thinking_pad)
changed += 1 if api_msg.get("reasoning_content") else 0
elif "reasoning_content" in api_msg:
api_msg.pop("reasoning_content", None)
changed += 1
agent/micro_compaction.py
+77 -142
View File
@@ -37,60 +37,49 @@ class MicroCompactionMixin:
"""Rolling micro-compaction; host must be a ``ContextCompressor``."""
def _resolve_compact_cursor(self, messages: List[Dict[str, Any]], head_end: int, tail_start: int) -> int:
"""Index of the first message not yet absorbed into the rolling summary.
Uses the in-memory cursor when valid; otherwise scans for the last summary marker.
"""
"""Index of the first message not yet absorbed into the rolling summary: the in-memory
cursor when valid, else just past the last summary marker."""
if head_end < self._micro_compact_cursor < tail_start:
return self._micro_compact_cursor
last_summary_idx = max(
(idx for idx in range(head_end, tail_start) if self._is_context_summary_message(messages[idx])),
default=-1,
)
cursor = head_end
if last_summary_idx >= head_end:
cursor = last_summary_idx + 1
summaries = (i for i in range(head_end, tail_start) if self._is_context_summary_message(messages[i]))
last = max(summaries, default=-1)
cursor = last + 1 if last >= head_end else head_end
if last >= head_end:
# Resumed session: rehydrate the rolling summary from the surviving marker so the next
# pass merges, not replaces.
recovered = "" if self._micro_compact_rolling_summary.strip() else (
self._rolling_summary_from_marker(messages[last_summary_idx].get("content"))
self._rolling_summary_from_marker(messages[last].get("content"))
)
if recovered:
self._micro_compact_rolling_summary = recovered
# Rehydration proves containment: this marker (batch or micro) becomes
# supersede/defrag-eligible; unabsorbed markers never get the key.
messages[last_summary_idx][_cc().MICRO_COMPACT_MARKER_KEY] = True
logger.info(
"Micro-compaction: recovered rolling summary from "
"transcript (%d chars)", len(recovered),
)
messages[last][_cc().MICRO_COMPACT_MARKER_KEY] = True
logger.info("Micro-compaction: recovered rolling summary from transcript (%d chars)", len(recovered))
self._micro_compact_cursor = cursor
return cursor
def _find_one_exchange(
self, messages: List[Dict[str, Any]], start: int, tail_start: int,
) -> Optional[tuple[int, int]]:
"""Find the next complete exchange (full agent turn) starting at *start*.
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).
"""
"""Find the next complete exchange (full agent turn) starting at *start*; 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))
def _real_assistant(msg: Dict[str, Any]) -> bool:
return msg.get("role") == "assistant" and not self._is_context_summary_message(msg)
def _turn_row(idx: int, roles: tuple) -> bool:
return messages[idx].get("role") in roles and not self._is_context_summary_message(messages[idx])
# Skip user messages and (assistant-role) summary markers to reach a real assistant message;
# otherwise a rehydrated cursor could absorb the marker itself.
idx = start
while idx < limit and not _real_assistant(messages[idx]):
while idx < limit and not _turn_row(idx, ("assistant",)):
idx += 1
if idx >= limit:
return None
exchange_start = idx
idx += 1
while idx < limit and messages[idx].get("role") in ("assistant", "tool") and not self._is_context_summary_message(messages[idx]):
while idx < limit and _turn_row(idx, ("assistant", "tool")):
idx += 1
# Boundary must close the turn: a mid-turn stop at tail_start would put the assistant marker
@@ -138,11 +127,8 @@ class MicroCompactionMixin:
call_kwargs["model"] = self.summary_model
if self.model:
call_kwargs.setdefault("main_runtime", {
"model": self.model,
"provider": self.provider or "",
"base_url": self.base_url or "",
"api_key": self.api_key or "",
"api_mode": getattr(self, "api_mode", "") or "",
"model": self.model, "provider": self.provider or "", "base_url": self.base_url or "",
"api_key": self.api_key or "", "api_mode": getattr(self, "api_mode", "") or "",
})
try:
@@ -155,8 +141,7 @@ class MicroCompactionMixin:
# A length stop means a partial merge; leave the exchange unabsorbed so a later pass retries.
if _cc()._response_finish_reason(response) == "length":
logger.warning(
"micro-summarization output hit the token cap "
"(finish_reason=length) — discarding partial summary",
"micro-summarization output hit the token cap (finish_reason=length) — discarding partial summary",
)
return None
@@ -176,33 +161,29 @@ class MicroCompactionMixin:
def _defrag_rolling_summary(self, messages: List[Dict[str, Any]]) -> bool:
"""Re-summarize the rolling summary text and rewrite the marker in place.
Transcript-shape-neutral (no splice, no cursor move). Returns True when it rewrote.
"""
Transcript-shape-neutral (no splice, no cursor move). Returns True when it rewrote."""
old_summary = self._micro_compact_rolling_summary
if not old_summary.strip():
return False
# Empty base turns the merge prompt into a rewrite-compactly instruction.
self._micro_compact_rolling_summary = ""
fresh_summary = self._micro_summarize_one(old_summary)
self._micro_compact_rolling_summary = fresh_summary or old_summary
if not fresh_summary:
self._micro_compact_rolling_summary = old_summary
return False
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 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. An
# in-place pop on a live dict would be identity-skipped by the bounded flush scan;
# flag the finalizer.
entry.pop(_cc()._DB_PERSISTED_MARKER, None)
self._flush_scan_cursor_invalidated = True
break
entry = next((e for e in reversed(messages) if _is_micro_marker(e)), None)
if entry is not None:
entry["content"] = self._render_micro_marker_content(fresh_summary)
# Content changed: clear the persisted stamp so the DB sync rewrites the row. An
# in-place pop on a live dict would be identity-skipped by the bounded flush scan;
# flag the finalizer.
entry.pop(_cc()._DB_PERSISTED_MARKER, None)
self._flush_scan_cursor_invalidated = True
logger.info(
"Micro-compaction defrag: rolling summary re-summarized "
"(%d -> %d chars)", len(old_summary), len(fresh_summary),
"Micro-compaction defrag: rolling summary re-summarized (%d -> %d chars)",
len(old_summary), len(fresh_summary),
)
return True
@@ -211,11 +192,9 @@ class MicroCompactionMixin:
self._micro_compact_last_failure_cursor = -1
def _micro_compact(self, messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Run one round of micro-compaction; public entry point from ``finalize_turn()``.
Returns the (possibly modified) list and syncs the session DB via ``archive_and_compact``
(the append-only flush alone would double-load on resume).
"""
"""Run one round of micro-compaction (entry point from ``finalize_turn()``). Returns the
(possibly modified) list and syncs the session DB via ``archive_and_compact`` (the
append-only flush alone would double-load on resume)."""
if not self._micro_compact_enabled:
return messages
@@ -229,9 +208,7 @@ class MicroCompactionMixin:
self._micro_compact_turns_since_pass = 0
n_messages = len(messages)
if n_messages < 4:
return messages
exchange = self._next_exchange(messages)
exchange = self._next_exchange(messages) if n_messages >= 4 else None
if exchange is None:
return messages
exchange_start, exchange_end = exchange
@@ -253,10 +230,8 @@ class MicroCompactionMixin:
if defragged:
self._sync_micro_compact_to_db(messages)
self._reset_micro_failure_tracking()
_telemetry(
"defrag" if defragged else "defrag_failed", messages,
tokens_after=estimate_messages_tokens_rough(messages),
)
outcome = "defrag" if defragged else "defrag_failed"
_telemetry(outcome, messages, tokens_after=estimate_messages_tokens_rough(messages))
return messages
# Cumulative iff it subsumes an earlier marker; captured before summarizing.
@@ -278,24 +253,20 @@ class MicroCompactionMixin:
self._micro_compact_cursor = self._cursor_after_splice(result, exchange_start + 1)
self._sync_micro_compact_to_db(result)
_telemetry(
"absorbed", result,
tokens_after=estimate_messages_tokens_rough(result), exchange_tokens=_exchange_tokens,
"absorbed", result, tokens_after=estimate_messages_tokens_rough(result), exchange_tokens=_exchange_tokens,
)
return result
def _record_micro_failure(self, exchange_start: int, exchange_end: int) -> str:
"""Count a summarize failure at this cursor; skip the exchange after too many in a row."""
# Track consecutive failures at the same cursor to avoid busy-looping every turn.
if exchange_start == self._micro_compact_last_failure_cursor:
self._micro_compact_consecutive_failures += 1
else:
self._micro_compact_consecutive_failures = 1
self._micro_compact_last_failure_cursor = exchange_start
same_cursor = exchange_start == self._micro_compact_last_failure_cursor
self._micro_compact_consecutive_failures = self._micro_compact_consecutive_failures + 1 if same_cursor else 1
self._micro_compact_last_failure_cursor = exchange_start
if self._micro_compact_consecutive_failures < _cc()._MICRO_COMPACT_MAX_CONSECUTIVE_FAILURES:
return "summarize_failed"
logger.info(
"Micro-compaction: skipping exchange at cursor %d "
"after %d consecutive failures",
"Micro-compaction: skipping exchange at cursor %d after %d consecutive failures",
exchange_start, self._micro_compact_consecutive_failures,
)
# Skip the stuck exchange; it stays in the transcript for batch compression/defrag.
@@ -310,9 +281,7 @@ class MicroCompactionMixin:
if compress_start >= compress_end:
return None
cursor = self._resolve_compact_cursor(messages, compress_start, compress_end)
if cursor >= compress_end:
return None
return self._find_one_exchange(messages, cursor, compress_end)
return None if cursor >= compress_end else self._find_one_exchange(messages, cursor, compress_end)
@staticmethod
def _rolling_summary_from_marker(content: Any) -> str:
@@ -320,70 +289,44 @@ class MicroCompactionMixin:
cc = _cc()
if not isinstance(content, str) or not content.strip():
return ""
body = content
# rfind: SUMMARY_PREFIX itself mentions the heading, so the first hit is in the preamble.
idx = body.rfind(cc.HISTORICAL_TASK_HEADING)
if idx != -1:
body = body[idx + len(cc.HISTORICAL_TASK_HEADING):]
idx = content.rfind(cc.HISTORICAL_TASK_HEADING)
body = content[idx + len(cc.HISTORICAL_TASK_HEADING):] if idx != -1 else content
end = body.find(cc._SUMMARY_END_MARKER)
if end != -1:
body = body[:end]
return body.strip()
return (body[:end] if end != -1 else body).strip()
def _cursor_after_splice(self, result: List[Dict[str, Any]], fallback: int) -> int:
"""Cursor position just past the summary marker in *result*.
Must derive from the SPLICED list: a splice collapses several rows into one marker
(and may drop a superseded one), so pre-splice indices land inside a later exchange
and silently skip it.
"""
for idx in range(len(result) - 1, -1, -1):
if _is_summary_marker(result[idx]):
return idx + 1
return fallback
"""Cursor position just past the summary marker in *result*. Must derive from the SPLICED
list: a splice collapses several rows into one marker (and may drop a superseded one), so
pre-splice indices land inside a later exchange and silently skip it."""
return next((idx + 1 for idx in range(len(result) - 1, -1, -1) if _is_summary_marker(result[idx])), fallback)
def _emit_micro_compaction_telemetry(
self, *, outcome: str, messages_before: int, messages_after: int,
tokens_before: int | None, tokens_after: int | None,
exchange_tokens: int | None = None, duration_ms: int | None = None,
self, *, outcome: str, messages_before: int, messages_after: int, tokens_before: int | None,
tokens_after: int | None, exchange_tokens: int | None = None, duration_ms: int | None = None,
) -> None:
"""Emit one content-free JSON log line for a micro-compaction pass.
``tokens_delta`` < 0 means the pass shrank the transcript; ``*_total`` fields accumulate.
"""
``tokens_delta`` < 0 means the pass shrank the transcript; ``*_total`` fields accumulate."""
_safe_int = _cc()._safe_int
try:
delta = None
if tokens_before is not None and tokens_after is not None:
delta = tokens_after - tokens_before
self._micro_compact_tokens_saved_total -= delta
delta = tokens_after - tokens_before if tokens_before is not None and tokens_after is not None else None
self._micro_compact_tokens_saved_total -= delta or 0
self._micro_compact_passes += 1
# Cached reads only: the lazy properties can fire a synchronous /models probe.
threshold = self._threshold_tokens
occupancy = None
if threshold and tokens_after is not None and threshold > 0:
occupancy = round(tokens_after / threshold * 100, 1)
has_occupancy = threshold and tokens_after is not None and threshold > 0
occupancy = round(tokens_after / threshold * 100, 1) if has_occupancy else None
payload = {
"event": "micro_compaction",
"session_id": getattr(self, "_session_id", "") or "",
"outcome": outcome,
"messages_before": messages_before,
"messages_after": messages_after,
"tokens_before": _safe_int(tokens_before),
"tokens_after": _safe_int(tokens_after),
"tokens_delta": _safe_int(delta),
"exchange_tokens": _safe_int(exchange_tokens),
"event": "micro_compaction", "session_id": getattr(self, "_session_id", "") or "", "outcome": outcome,
"messages_before": messages_before, "messages_after": messages_after,
"tokens_before": _safe_int(tokens_before), "tokens_after": _safe_int(tokens_after),
"tokens_delta": _safe_int(delta), "exchange_tokens": _safe_int(exchange_tokens),
"rolling_summary_tokens": estimate_tokens_rough(self._micro_compact_rolling_summary),
"cursor": _safe_int(self._micro_compact_cursor),
"passes_total": self._micro_compact_passes,
"tokens_saved_total": self._micro_compact_tokens_saved_total,
"duration_ms": _safe_int(duration_ms),
"cursor": _safe_int(self._micro_compact_cursor), "passes_total": self._micro_compact_passes,
"tokens_saved_total": self._micro_compact_tokens_saved_total, "duration_ms": _safe_int(duration_ms),
# Headroom: how full the window is being kept.
"threshold_tokens": _safe_int(threshold),
"context_limit": _safe_int(self._resolved_context_length),
"occupancy_pct": occupancy,
"main_model": self.model or "",
"aux_model": self.summary_model or "",
"threshold_tokens": _safe_int(threshold), "context_limit": _safe_int(self._resolved_context_length),
"occupancy_pct": occupancy, "main_model": self.model or "", "aux_model": self.summary_model or "",
}
logger.info("micro compaction telemetry: %s", json.dumps(payload, sort_keys=True, separators=(",", ":")))
except Exception as exc:
@@ -391,10 +334,8 @@ class MicroCompactionMixin:
def _sync_micro_compact_to_db(self, compacted_messages: List[Dict[str, Any]]) -> None:
"""Persist the micro-compacted set to the session DB atomically and stamp rows persisted.
Without this the old exchange rows stay ``active=1`` and a resume double-loads
both the summary and the originals.
"""
Without this the old exchange rows stay ``active=1`` and a resume double-loads both the
summary and the originals."""
session_db, session_id = getattr(self, "_session_db", None), getattr(self, "_session_id", "")
if not session_db or not session_id:
return
@@ -413,17 +354,14 @@ class MicroCompactionMixin:
self, messages: List[Dict[str, Any]], splice_start: int, splice_end: int, supersede: bool = True,
) -> List[Dict[str, Any]]:
"""Replace *messages[splice_start:splice_end]* with an assistant-role summary marker.
Merges user turns left adjacent by a superseded marker so the result is alternation-valid.
"""
Merges user turns left adjacent by a superseded marker so the result is alternation-valid."""
cc = _cc()
summary_text = self._micro_compact_rolling_summary
if not summary_text.strip():
return messages
summary_msg = {
"role": "assistant",
"content": self._render_micro_marker_content(summary_text),
"role": "assistant", "content": self._render_micro_marker_content(summary_text),
cc.COMPRESSED_SUMMARY_METADATA_KEY: True,
# Micro marker: eligible for supersede/defrag; batch markers never carry this key.
cc.MICRO_COMPACT_MARKER_KEY: True,
@@ -434,10 +372,9 @@ 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 _is_micro_marker(m)]
if len(marker_idxs) > 1:
result = self._merge_adjacent_user_turns([m for i, m in enumerate(result) if i not in marker_idxs[:-1]])
stale = [i for i, m in enumerate(result) if _is_micro_marker(m)][:-1] if supersede else []
if stale:
result = self._merge_adjacent_user_turns([m for i, m in enumerate(result) if i not in stale])
# Deliberately no _strip_persistence_markers: micro archives in place under the same session
# id, so stamps stay accurate and a failed archive keeps the append-only flush idempotent.
@@ -451,17 +388,15 @@ class MicroCompactionMixin:
@staticmethod
def _merge_adjacent_user_turns(result: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Merge consecutive plain-text real user turns left by a supersede.
Same ``\\n\\n`` join as ``repair_message_sequence`` pass 2, done here so the marker
and cursor are never collateral damage of the downstream repair. Lists untouched.
"""
"""Merge consecutive plain-text real user turns left by a supersede. Same ``\\n\\n`` join as
``repair_message_sequence`` pass 2, done here so the marker and cursor are never collateral
damage of the downstream repair. Lists untouched."""
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 _is_summary_marker(m) and isinstance(m.get("content"), str)
isinstance(m, dict) and m.get("role") == "user" and not _is_summary_marker(m)
and isinstance(m.get("content"), str)
)
merged: List[Dict[str, Any]] = []
agent/moa_loop.py
+33 -58
View File
@@ -202,10 +202,9 @@ def _slot_label(slot: dict[str, Any]) -> str:
def _slot_reasoning_config(slot: dict[str, Any]) -> dict[str, Any] | None:
"""Translate optional per-MoA-slot reasoning_effort into runtime config."""
effort = slot.get("reasoning_effort")
try:
from hermes_constants import parse_reasoning_effort
return parse_reasoning_effort(effort)
return parse_reasoning_effort(slot.get("reasoning_effort"))
except Exception: # pragma: no cover - bad config must not break MoA
return None
@@ -246,11 +245,10 @@ def _slot_runtime(slot: dict[str, Any]) -> dict[str, Any]:
from hermes_cli.runtime_provider import resolve_runtime_provider
rt = resolve_runtime_provider(requested=provider, target_model=model)
out.update({k: rt[k] for k in ("base_url", "api_key", "api_mode") if rt.get(k)})
request_overrides = rt.get("request_overrides")
if isinstance(request_overrides, dict):
extra_body = request_overrides.get("extra_body")
if isinstance(extra_body, dict) and extra_body:
out["extra_body"] = dict(extra_body)
overrides = rt.get("request_overrides")
extra_body = overrides.get("extra_body") if isinstance(overrides, dict) else None
if isinstance(extra_body, dict) and extra_body:
out["extra_body"] = dict(extra_body)
except Exception as exc: # pragma: no cover - defensive
logger.debug("MoA slot runtime resolution failed for %s: %s", _slot_label(slot), exc)
return out
@@ -261,27 +259,21 @@ def _slot_runtime(slot: dict[str, Any]) -> dict[str, Any]:
def _merge_slot_extra_body(slot_extra_body: Any, caller_extra_body: Any) -> Any:
"""Merge slot defaults with a caller override (caller wins) for ``call_llm``."""
if isinstance(slot_extra_body, dict) and slot_extra_body:
if isinstance(caller_extra_body, dict):
return {**slot_extra_body, **caller_extra_body}
if caller_extra_body:
return caller_extra_body
return dict(slot_extra_body)
return caller_extra_body
if not (isinstance(slot_extra_body, dict) and slot_extra_body):
return caller_extra_body
if isinstance(caller_extra_body, dict):
return {**slot_extra_body, **caller_extra_body}
return caller_extra_body or dict(slot_extra_body)
def _agent_cache_opts(agent: Any) -> tuple[Any, Any]:
"""The live agent's ``(_cache_disabled, _cache_ttl)``; ``(None, None)`` without an agent."""
if agent is None:
return None, None
return getattr(agent, "_cache_disabled", None), getattr(agent, "_cache_ttl", None)
return (None, None) if agent is None else (getattr(agent, "_cache_disabled", None), getattr(agent, "_cache_ttl", None))
def _with_cache_disabled(runtime: dict[str, Any], cache_disabled: Any) -> dict[str, Any]:
"""Pin the live agent's cache disable onto a runtime snapshot (None is a no-op)."""
if cache_disabled is None:
return runtime
return {**runtime, "_cache_disabled": cache_disabled}
return runtime if cache_disabled is None else {**runtime, "_cache_disabled": cache_disabled}
def _maybe_apply_moa_cache_control(
@@ -697,9 +689,7 @@ def _preset_temperature(preset: dict[str, Any], key: str) -> float | None:
def _hash_messages(msgs: list[dict[str, Any]]) -> str:
return hashlib.sha256(
"\u0000".join(f"{m.get('role')}:{m.get('content')}" for m in msgs).encode("utf-8", "replace")
).hexdigest()
return hashlib.sha256("\u0000".join(f"{m.get('role')}:{m.get('content')}" for m in msgs).encode("utf-8", "replace")).hexdigest()
def _is_failed_reference(text: str) -> bool:
@@ -777,10 +767,8 @@ def aggregate_moa_context(
logger.debug("MoA privacy filter check failed", exc_info=True)
agg_refs, degraded, all_failed = _guidance_inputs(reference_outputs, privacy_full, degraded_reference_policy)
joined = _join_reference_outputs(agg_refs, degraded)
# Every reference failed: skip the aggregator (synthesizing over nothing can block
# for the full provider timeout) and return only the sanitized notice.
if all_failed:
# Skip the aggregator: synthesizing over nothing can block for the full provider timeout.
logger.warning("MoA: all %d reference(s) failed — skipping aggregator synthesis", len(reference_outputs))
return (
"[Mixture of Agents context — all reference models failed. "
@@ -806,14 +794,12 @@ def aggregate_moa_context(
# Same cache_control decoration as the advisor calls; this synthesis call is
# a third independent MoA call path that otherwise re-bills its full input.
agg_messages = _maybe_apply_moa_cache_control(
[{"role": "user", "content": synth_prompt}],
_with_cache_disabled(agg_runtime, cache_disabled), cache_ttl=cache_ttl,
[{"role": "user", "content": synth_prompt}], _with_cache_disabled(agg_runtime, cache_disabled), cache_ttl=cache_ttl,
)
response = call_llm(
synthesis = _extract_text(call_llm(
task="moa_aggregator", messages=agg_messages, temperature=aggregator_temperature,
reasoning_config=_aggregator_reasoning_config(aggregator), **agg_runtime,
)
synthesis = _extract_text(response)
))
except Exception as exc:
logger.warning("MoA aggregator model %s failed: %s", agg_label, exc)
synthesis = ""
@@ -838,9 +824,7 @@ def _completed_response_as_stream_chunk(response: Any) -> Any:
if isinstance(raw_tool_calls, (list, tuple)) and raw_tool_calls:
tool_call_deltas = [
SimpleNamespace(
index=getattr(tc, "index", index),
id=getattr(tc, "id", None),
type=getattr(tc, "type", None) or "function",
index=getattr(tc, "index", index), id=getattr(tc, "id", None), type=getattr(tc, "type", None) or "function",
function=SimpleNamespace(
name=getattr(getattr(tc, "function", None), "name", None),
arguments=getattr(getattr(tc, "function", None), "arguments", None),
@@ -853,12 +837,10 @@ def _completed_response_as_stream_chunk(response: Any) -> Any:
**{k: getattr(message, k, None) for k in ("reasoning_content", "reasoning", "reasoning_details")},
)
choice = SimpleNamespace(
index=getattr(first_choice, "index", 0), delta=delta,
finish_reason=getattr(first_choice, "finish_reason", None) or "stop",
index=getattr(first_choice, "index", 0), delta=delta, finish_reason=getattr(first_choice, "finish_reason", None) or "stop",
)
return SimpleNamespace(
id=getattr(response, "id", None), model=getattr(response, "model", None), choices=[choice],
usage=getattr(response, "usage", None),
id=getattr(response, "id", None), model=getattr(response, "model", None), choices=[choice], usage=getattr(response, "usage", None),
)
@@ -871,15 +853,13 @@ def _attach_reference_guidance(agg_messages: list[dict[str, Any]], guidance: str
consecutive user turns would be rejected by strict providers).
"""
last = agg_messages[-1] if agg_messages else None
if last is not None and last.get("role") == "user":
last_content = last.get("content")
if isinstance(last_content, str):
last["content"] = last_content + "\n\n" + guidance
return
if isinstance(last_content, list):
last["content"] = [*last_content, {"type": "text", "text": "\n\n" + guidance}]
return
agg_messages.append({"role": "user", "content": guidance})
last_content = last.get("content") if last is not None and last.get("role") == "user" else None
if isinstance(last_content, str):
last["content"] = last_content + "\n\n" + guidance
elif isinstance(last_content, list):
last["content"] = [*last_content, {"type": "text", "text": "\n\n" + guidance}]
else:
agg_messages.append({"role": "user", "content": guidance})
def peel_reference_guidance(messages: list[dict[str, Any]], guidance: Any) -> list[dict[str, Any]]:
@@ -892,26 +872,21 @@ def peel_reference_guidance(messages: list[dict[str, Any]], guidance: Any) -> li
if not isinstance(last, dict) or last.get("role") != "user":
return messages
content = last.get("content")
if content == guidance_text:
# Attach shape (c): guidance was appended as its own user message.
if content == guidance_text: # shape (c): guidance was its own user message
return list(messages[:-1])
suffix = "\n\n" + guidance_text
if isinstance(content, str) and content.endswith(suffix):
# Attach shape (a): merged into a trailing string user turn.
if isinstance(content, str) and content.endswith(suffix): # shape (a): merged into a string turn
return [*messages[:-1], {**last, "content": content[: -len(suffix)]}]
if isinstance(content, list) and content:
last_part = content[-1]
if isinstance(last_part, dict) and last_part.get("type", "text") == "text":
text = last_part.get("text") or ""
if text in (suffix, guidance_text):
# Attach shape (b): guidance rode as its own trailing part. Guidance as
# the only content drops the whole message (mirrors shape c).
if len(content) == 1:
return list(messages[:-1])
return [*messages[:-1], {**last, "content": list(content[:-1])}]
# Shape (b): guidance rode as its own trailing part. Guidance as the
# only content drops the whole message (mirrors shape c).
return list(messages[:-1]) if len(content) == 1 else [*messages[:-1], {**last, "content": list(content[:-1])}]
if text.endswith(suffix):
new_part = {**last_part, "text": text[: -len(suffix)]}
return [*messages[:-1], {**last, "content": [*content[:-1], new_part]}]
return [*messages[:-1], {**last, "content": [*content[:-1], {**last_part, "text": text[: -len(suffix)]}]}]
return messages