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

refactor(agent): structural pass-2 on message_sanitization + micro_compaction (-100 LOC, zero behavior change)

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:05:07 -07:00
parent adb241dfc4
commit f6c6bbcb1d
2 changed files with 130 additions and 229 deletions
Download Patch File Download Diff File Expand all files Collapse all files

179
agent/message_sanitization.py
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:
@@ -74,25 +71,17 @@ def _sanitize_messages(messages: list, fix: Callable[[str], str], *, deep: bool)
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 +158,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])
@@ -216,12 +203,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 +220,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 +307,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 +322,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)
@@ -379,10 +355,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 []:
@@ -391,23 +367,17 @@ def uniquify_tool_call_ids(tool_calls: list) -> list:
raw = raw.strip() if isinstance(raw, str) else ""
# Composite Responses ids ("call_x|fc_y") collide on the call half — the pairing key.
cid = raw.split("|", 1)[0]
if not cid:
if not cid or cid not in seen:
seen.update((cid,) if cid else ())
continue
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 +385,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 +414,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 +432,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 +453,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

180
agent/micro_compaction.py
View File

@@ -43,27 +43,21 @@ class MicroCompactionMixin:
"""
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
@@ -77,20 +71,19 @@ class MicroCompactionMixin:
"""
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 +131,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 +145,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 +165,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
@@ -229,9 +214,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 +236,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 +259,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 +287,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,15 +295,11 @@ 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*.
@@ -337,53 +308,34 @@ class MicroCompactionMixin:
(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
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 +343,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 +363,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 +381,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.
@@ -460,8 +406,8 @@ class MicroCompactionMixin:
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]] = []