From f6c6bbcb1da7af93fdf5c1f05e8fe1d1422ff29d Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 22:05:07 -0700 Subject: [PATCH] refactor(agent): structural pass-2 on message_sanitization + micro_compaction (-100 LOC, zero behavior change) --- agent/message_sanitization.py | 179 +++++++++++++-------------------- agent/micro_compaction.py | 180 ++++++++++++---------------------- 2 files changed, 130 insertions(+), 229 deletions(-) diff --git a/agent/message_sanitization.py b/agent/message_sanitization.py index 12a9e7f3d1..b929ad7aad 100644 --- a/agent/message_sanitization.py +++ b/agent/message_sanitization.py @@ -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 ``_d`` 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 ``_d`` 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 diff --git a/agent/micro_compaction.py b/agent/micro_compaction.py index 78abdd0037..697e6f361d 100644 --- a/agent/micro_compaction.py +++ b/agent/micro_compaction.py @@ -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]] = []