"""Automatic context window compression for long conversations. Self-contained class with its own OpenAI client for summarization. Uses auxiliary model (cheap/fast) to summarize middle turns while protecting head and tail context. Improvements over v2: - Structured summary template with Resolved/Pending question tracking - Filter-safe summarizer preamble that treats prior turns as source material - Historical (reference-only) section headings replace "Next Steps"/"Remaining Work" to avoid reading as active instructions - Clear separator when summary merges into tail message - Iterative summary updates (preserves info across multiple compactions) - Token-budget tail protection instead of fixed message count - Tool output pruning before LLM summarization (cheap pre-pass) - Scaled summary budget (proportional to compressed content) - Richer tool call/result detail in summarizer input """ import hashlib import json import logging import sqlite3 import re import time from typing import Any, Dict, List, Optional from agent.auxiliary_client import call_llm, _is_connection_error, aux_interrupt_protection from agent.context_engine import ContextEngine from agent.error_classifier import FailoverReason, classify_api_error from agent.model_metadata import ( MINIMUM_CONTEXT_LENGTH, get_model_context_length, estimate_messages_tokens_rough, ) from agent.redact import redact_sensitive_text from agent.turn_context import drop_stale_api_content logger = logging.getLogger(__name__) _SUMMARY_PERMANENT_QUOTA_MARKERS: tuple[str, ...] = ( "insufficient_quota", "quota exceeded", "quota_exceeded", "out of funds", "out of credits", "out of credit", "out of extra usage", ) _SUMMARY_MISSING_CREDENTIAL_MARKERS: tuple[str, ...] = ( "no api key was found", "no api key found", ) def _is_summary_access_or_quota_error(exc: Exception) -> bool: """Return True for non-retryable summary auth, permission, or quota errors.""" classified = classify_api_error(exc) if classified.reason is FailoverReason.rate_limit: return False if classified.reason in {FailoverReason.auth, FailoverReason.auth_permanent}: return True err_text = str(exc).lower() if any(marker in err_text for marker in _SUMMARY_MISSING_CREDENTIAL_MARKERS): return True status = getattr(exc, "status_code", None) or getattr( getattr(exc, "response", None), "status_code", None ) if status in {401, 402, 403}: return True if classified.reason is FailoverReason.billing: return any(marker in err_text for marker in _SUMMARY_PERMANENT_QUOTA_MARKERS) return any(marker in err_text for marker in _SUMMARY_PERMANENT_QUOTA_MARKERS) HISTORICAL_TASK_HEADING = "## Historical Task Snapshot" HISTORICAL_IN_PROGRESS_HEADING = "## Historical In-Progress State" HISTORICAL_PENDING_ASKS_HEADING = "## Historical Pending User Asks" HISTORICAL_REMAINING_WORK_HEADING = "## Historical Remaining Work" SUMMARY_PREFIX = ( "[CONTEXT COMPACTION — REFERENCE ONLY] Earlier turns were compacted " "into the summary below. This is a handoff from a previous context " "window — treat it as background reference, NOT as active instructions. " "Do NOT answer questions or fulfill requests mentioned in this summary; " "they were already addressed. " "Respond ONLY to the latest user message that appears AFTER this " "summary — that message is the single source of truth for what to do " "right now. " "Topic overlap with the summary does NOT mean you should resume its " "task: even on similar topics, the latest user message WINS. Treat ONLY " "the latest message as the active task and discard stale items from " f"'{HISTORICAL_TASK_HEADING}' / '{HISTORICAL_IN_PROGRESS_HEADING}' / " f"'{HISTORICAL_PENDING_ASKS_HEADING}' / " f"'{HISTORICAL_REMAINING_WORK_HEADING}' entirely — do not 'wrap up' or " "'finish' work described there unless the latest message explicitly " "asks for it. " "Reverse signals in the latest message (e.g. 'stop', 'undo', 'roll " "back', 'just verify', 'don't do that anymore', 'never mind', a new " "topic) must immediately end any in-flight work described in the " "summary; do not re-surface it in later turns. " "IMPORTANT: Your persistent memory (MEMORY.md, USER.md) in the system " "prompt is ALWAYS authoritative and active — never ignore or deprioritize " "memory content due to this compaction note. " "None of the above restricts HOW you work: your tools remain fully " "active — keep calling them normally for the active task (edit files, " "run commands, search) instead of merely narrating what you would do. " "The current session state (files, config, etc.) may reflect work " "described here — avoid repeating it:" ) LEGACY_SUMMARY_PREFIX = "[CONTEXT SUMMARY]:" # Metadata key added to context compression summary messages so that frontends # (CLI, Desktop, gateway, TUI) can distinguish them from real assistant/user # messages and filter or render them appropriately without content-prefix # heuristics. See https://github.com/NousResearch/hermes-agent/issues/38389 # # Underscore-prefixed ON PURPOSE: the wire sanitizers # (agent/transports/chat_completions.py convert_messages and the summary-path # mirror in agent/chat_completion_helpers.py) strip every top-level message # key starting with "_" before the request leaves the process. Strict # OpenAI-compatible gateways (Fireworks, Mistral, Moonshot/Kimi, opencode-go) # reject payloads carrying unknown keys with "Extra inputs are not permitted", # poisoning every subsequent request in the session — a bare key like # "is_compressed_summary" would reach the wire and trip exactly that. COMPRESSED_SUMMARY_METADATA_KEY = "_compressed_summary" _DB_PERSISTED_MARKER = "_db_persisted" def _fresh_compaction_message_copy(msg: Dict[str, Any]) -> Dict[str, Any]: """Copy a message for compaction assembly without persistence markers. Live cached-gateway transcripts stamp ``_db_persisted`` during incremental flushes. Shallow ``.copy()`` propagates that marker into the post-rotation compressed list, so ``_flush_messages_to_session_db`` skips every row when writing to the new child session (#57491). This strips at the copy site (clearest intent, and cheap), but the authoritative guarantee is the single terminal sweep in ``compress()`` (``_strip_persistence_markers``): no message may leave ``compress()`` carrying ``_db_persisted`` regardless of how many intermediate copy sites a future refactor adds. """ fresh = msg.copy() fresh.pop(_DB_PERSISTED_MARKER, None) return fresh def _strip_persistence_markers(messages: List[Dict[str, Any]]) -> None: """Enforce the compaction invariant: no assembled message carries a session-store persistence marker. ``compress()`` copies protected head/tail messages out of the live cached-gateway transcript, which stamps ``_db_persisted`` on every message over the life of the session. If any copied dict keeps that marker, the rotation flush to the child session skips it and the compacted transcript is lost from ``state.db`` (#57491). Stripping at each copy site is necessary but *positional* — a copy site added after the assembly loops would re-leak. This single terminal sweep makes the guarantee structural instead: run it once on the fully-assembled list so the invariant holds no matter where the copies happened. Mutates in place (the dicts are compaction-local copies). """ for msg in messages: if isinstance(msg, dict): msg.pop(_DB_PERSISTED_MARKER, None) # Appended to every standalone summary message (and to the merged-into-tail # prefix) so the model has an unambiguous "summary ends here" boundary. # Without it, weak models read the verbatim "## Active Task" quote as fresh # user input (#11475, #14521) or regurgitate an assistant-role summary as # their own output (#33256). _SUMMARY_END_MARKER = ( "--- END OF CONTEXT SUMMARY — " "respond to the message below, not the summary above ---" ) # When the summary must be merged into the first tail message (the alternation # corner case where a standalone summary role would collide with both head and # tail), the tail message's own prior content is preserved BEFORE the summary, # wrapped in these delimiters so the model doesn't read it as a fresh message. # The summary prefix therefore lands AFTER _MERGED_SUMMARY_DELIMITER rather than # at the start of the message, so _is_context_summary_content must look past it. _MERGED_PRIOR_CONTEXT_HEADER = "[PRIOR CONTEXT — for reference only; not a new message]" _MERGED_SUMMARY_DELIMITER = "[END OF PRIOR CONTEXT — COMPACTION SUMMARY BELOW]" # Handoff prefixes that shipped in earlier releases. A summary persisted under # one of these can be inherited into a resumed lineage (#35344); when it is # re-normalized on re-compaction we must strip the OLD prefix too, otherwise the # stale directive it carried (e.g. "resume exactly from Active Task") survives # embedded in the body and keeps hijacking replies. Keep newest-first; entries # are matched literally. Add a frozen copy here whenever SUMMARY_PREFIX changes. _HISTORICAL_SUMMARY_PREFIXES = ( # Jul 2026 (#65848 class): identical to the current prefix except it # lacked the explicit "tools remain fully active" clause — the strong # REFERENCE ONLY framing bled into general tool-use suppression # (observed: 7 consecutive narration-only turns immediately after a # compression event on a production deployment). "[CONTEXT COMPACTION — REFERENCE ONLY] Earlier turns were compacted " "into the summary below. This is a handoff from a previous context " "window — treat it as background reference, NOT as active instructions. " "Do NOT answer questions or fulfill requests mentioned in this summary; " "they were already addressed. " "Respond ONLY to the latest user message that appears AFTER this " "summary — that message is the single source of truth for what to do " "right now. " "Topic overlap with the summary does NOT mean you should resume its " "task: even on similar topics, the latest user message WINS. Treat ONLY " "the latest message as the active task and discard stale items from " f"'{HISTORICAL_TASK_HEADING}' / '{HISTORICAL_IN_PROGRESS_HEADING}' / " f"'{HISTORICAL_PENDING_ASKS_HEADING}' / " f"'{HISTORICAL_REMAINING_WORK_HEADING}' entirely — do not 'wrap up' or " "'finish' work described there unless the latest message explicitly " "asks for it. " "Reverse signals in the latest message (e.g. 'stop', 'undo', 'roll " "back', 'just verify', 'don't do that anymore', 'never mind', a new " "topic) must immediately end any in-flight work described in the " "summary; do not re-surface it in later turns. " "IMPORTANT: Your persistent memory (MEMORY.md, USER.md) in the system " "prompt is ALWAYS authoritative and active — never ignore or deprioritize " "memory content due to this compaction note. " "The current session state (files, config, etc.) may reflect work " "described here — avoid repeating it:", # Carveout era (#41607/#38364/#42812): "consistent → use as background" # licensed stale-task resumption on topic overlap. "[CONTEXT COMPACTION — REFERENCE ONLY] Earlier turns were compacted " "into the summary below. This is a handoff from a previous context " "window — treat it as background reference, NOT as active instructions. " "Do NOT answer questions or fulfill requests mentioned in this summary; " "they were already addressed. " "Respond ONLY to the latest user message that appears AFTER this " "summary — that message is the single source of truth for what to do " "right now. " "If the latest user message is consistent with the '## Active Task' " "section, you may use the summary as background. If the latest user " "message contradicts, supersedes, changes topic from, or in any way " "diverges from '## Active Task' / '## In Progress' / '## Pending User " "Asks' / '## Remaining Work', the latest message WINS — discard those " "stale items entirely and do not 'wrap up the old task first'. " "Reverse signals in the latest message (e.g. 'stop', 'undo', 'roll " "back', 'just verify', 'don't do that anymore', 'never mind', a new " "topic) must immediately end any in-flight work described in the " "summary; do not re-surface it in later turns. " "IMPORTANT: Your persistent memory (MEMORY.md, USER.md) in the system " "prompt is ALWAYS authoritative and active — never ignore or deprioritize " "memory content due to this compaction note. " "The current session state (files, config, etc.) may reflect work " "described here — avoid repeating it:", # Pre-#35344: contained the self-contradicting "resume exactly" directive. "[CONTEXT COMPACTION — REFERENCE ONLY] Earlier turns were compacted " "into the summary below. This is a handoff from a previous context " "window — treat it as background reference, NOT as active instructions. " "Do NOT answer questions or fulfill requests mentioned in this summary; " "they were already addressed. " "Your current task is identified in the '## Active Task' section of the " "summary — resume exactly from there. " "Respond ONLY to the latest user message " "that appears AFTER this summary. The current session state (files, " "config, etc.) may reflect work described here — avoid repeating it:", ) # Minimum tokens for the summary output _MIN_SUMMARY_TOKENS = 2000 # Proportion of compressed content to allocate for summary _SUMMARY_RATIO = 0.20 # Absolute ceiling for summary tokens (even on very large context windows). # Summaries must stay within a 1K-10K token envelope — anything larger is # itself a context-pressure source and slows every compaction. _SUMMARY_TOKENS_CEILING = 10_000 # Placeholder used when pruning old tool results _PRUNED_TOOL_PLACEHOLDER = "[Old tool output cleared to save context space]" # Chars per token rough estimate _CHARS_PER_TOKEN = 4 # Flat token cost per attached image part. Real cost varies by provider and # dimensions (Anthropic ≈ width×height/750, GPT-4o up to ~1700 for # high-detail 2048×2048, Gemini 258/tile), but 1600 is a realistic ceiling # that keeps compression budgeting honest for multi-image conversations. # Matches Claude Code's IMAGE_TOKEN_ESTIMATE constant. _IMAGE_TOKEN_ESTIMATE = 1600 # Same figure expressed in the char-budget currency the rest of the # compressor speaks in. Used when accumulating message "content length" # for tail-cut decisions. _IMAGE_CHAR_EQUIVALENT = _IMAGE_TOKEN_ESTIMATE * _CHARS_PER_TOKEN _SUMMARY_FAILURE_COOLDOWN_SECONDS = 600 # Hard ceiling for the deterministic summary-failure handoff. The fallback is # only meant to preserve continuity anchors from the dropped window, not to # become another unbounded transcript copy after the LLM summarizer failed. _FALLBACK_SUMMARY_MAX_CHARS = 8_000 _FALLBACK_TURN_MAX_CHARS = 700 _AUTO_FOCUS_MAX_TURNS = 3 _AUTO_FOCUS_TURN_MAX_CHARS = 260 _AUTO_FOCUS_MAX_CHARS = 700 _ACTIVE_TASK_MAX_CHARS = 1400 # Keep a short run of recent messages verbatim even when the token budget is # already exhausted. The public ``protect_last_n`` default is intentionally # high for small/light tails, but using all 20 as a hard floor here would bring # back the old large-tool-output case where nothing can be compacted. _MAX_TAIL_MESSAGE_FLOOR = 8 # Models with context windows below this get their compression threshold # floored at ``_SMALL_CTX_THRESHOLD_PERCENT`` (raise-only — an explicitly # higher user/model threshold always wins). At the default 50% trigger a # 128K-262K model compacts with only ~64-131K consumed; the incompressible # floor (system prompt + tool schemas + protected tail + rolling summary) # eats most of the reclaimed headroom, so compaction re-fires every 1-2 # turns and the session spends most of its wall-clock summarizing. _SMALL_CTX_WINDOW_LIMIT = 512_000 _SMALL_CTX_THRESHOLD_PERCENT = 0.75 _PATH_MENTION_RE = re.compile(r"(?:/|~/?|[A-Za-z]:\\)[^\s`'\")\]}<>]+") # MEDIA delivery directives must not reach the summarizer — if one leaks into # the summary, the downstream model may re-emit it as an active directive on # the next turn, triggering bogus attachment sends (#14665). _MEDIA_DIRECTIVE_RE = re.compile(r"MEDIA:\S+") _HISTORICAL_TASK_SECTION_RE = re.compile( rf"(?ms)^{re.escape(HISTORICAL_TASK_HEADING)}\s*\n.*?(?=^## |\Z)" ) def _dedupe_append(items: list[str], value: str, *, limit: int) -> None: value = value.strip() if value and value not in items and len(items) < limit: items.append(value) def _extract_tool_call_name_and_args(tool_call: Any) -> tuple[str, str]: """Return a best-effort ``(name, arguments)`` pair for dict/object tool calls.""" if isinstance(tool_call, dict): fn = tool_call.get("function") or {} return str(fn.get("name") or "unknown"), str(fn.get("arguments") or "") fn = getattr(tool_call, "function", None) if fn is None: return "unknown", "" return str(getattr(fn, "name", None) or "unknown"), str(getattr(fn, "arguments", None) or "") def _extract_tool_call_id(tool_call: Any) -> str: if isinstance(tool_call, dict): return str(tool_call.get("id") or "") return str(getattr(tool_call, "id", "") or "") def _collect_path_mentions(text: str, relevant_files: list[str], *, limit: int = 12) -> None: for match in _PATH_MENTION_RE.findall(text): _dedupe_append(relevant_files, match.rstrip(".,:;"), limit=limit) def _content_length_for_budget(raw_content: Any) -> int: """Return the effective char-length of a message's content for token budgeting. Plain strings: ``len(content)``. Multimodal lists: sum of text-part ``len(text)`` plus a flat ``_IMAGE_CHAR_EQUIVALENT`` per image part (``image_url`` / ``input_image`` / Anthropic-style ``image``). This keeps the compressor from treating a turn with 5 attached images as near-zero tokens just because the text part is empty. """ if isinstance(raw_content, str): return len(raw_content) if not isinstance(raw_content, list): return len(str(raw_content or "")) total = 0 for p in raw_content: if isinstance(p, str): total += len(p) continue if not isinstance(p, dict): total += len(str(p)) continue ptype = p.get("type") if ptype in {"image_url", "input_image", "image"}: total += _IMAGE_CHAR_EQUIVALENT else: # text / input_text / tool_result-with-text / anything else with # a text field. Ignore the raw base64 payload inside image_url # dicts — dimensions don't matter, only whether it's an image. total += len(p.get("text", "") or "") return total def _serialized_length_for_budget(value: Any) -> int: """Return a stable char-length for non-content replay/metadata fields.""" if value is None or value == "": return 0 if isinstance(value, str): return len(value) try: return len(json.dumps(value, ensure_ascii=False, sort_keys=True, default=str)) except (TypeError, ValueError): return len(str(value)) # Provider replay/metadata fields that ride the wire on every request but are # invisible to ``msg["content"]``/``msg["tool_calls"]`` accounting. Codex # Responses sessions in particular carry ``codex_reasoning_items`` blobs of # ``encrypted_content`` that can dominate the serialized session (a measured # 214-turn session held ~115K tokens / 27% of its payload there — #55572). _REPLAY_BUDGET_KEYS = ( "reasoning", "reasoning_content", "reasoning_details", "codex_reasoning_items", "codex_message_items", ) def _estimate_msg_budget_tokens(msg: dict) -> int: """Token estimate for one message in the tail-protection budget walks. Counts the message content plus the **full** ``tool_call`` envelope — ``id``, ``type``, ``function.name`` and JSON structure — not just ``function.arguments``. Counting only the arguments string undercounted assistant turns that fan out into parallel tool calls by 2-15x (a 4-tool-call turn measures ~73 vs ~1,090 real tokens), so the protected tail overshot ``tail_token_budget`` and compression became ineffective. See issue #28053. Also counts provider replay fields (``codex_reasoning_items`` etc. — see ``_REPLAY_BUDGET_KEYS``). The preflight "should I compress?" estimator sees the full message shape, so the tail walk must use the same size class; otherwise an assistant message with tiny visible content but large hidden replay blobs is protected as if it were small, the post-compression session stays near the context limit, and compaction re-fires continuously (#55572). Accounting-only: replay fields are never mutated or pruned here. """ content_len = _content_length_for_budget(msg.get("content") or "") tokens = content_len // _CHARS_PER_TOKEN + 10 # +10 for role/key overhead for tc in msg.get("tool_calls") or []: if isinstance(tc, dict): tokens += len(str(tc)) // _CHARS_PER_TOKEN for key in _REPLAY_BUDGET_KEYS: tokens += _serialized_length_for_budget(msg.get(key)) // _CHARS_PER_TOKEN return tokens def _content_text_for_contains(content: Any) -> str: """Return a best-effort text view of message content. Used only for substring checks when we need to know whether we've already appended a note to a message. Keeps multimodal lists intact elsewhere. """ if content is None: return "" if isinstance(content, str): return content if isinstance(content, list): parts: list[str] = [] for item in content: if isinstance(item, str): parts.append(item) elif isinstance(item, dict): text = item.get("text") if isinstance(text, str): parts.append(text) return "\n".join(part for part in parts if part) return str(content) def _append_text_to_content(content: Any, text: str, *, prepend: bool = False) -> Any: """Append or prepend plain text to message content safely. Compression sometimes needs to add a note or merge a summary into an existing message. Message content may be plain text or a multimodal list of blocks, so direct string concatenation is not always safe. """ if content is None: return text if isinstance(content, str): return text + content if prepend else content + text if isinstance(content, list): text_block = {"type": "text", "text": text} return [text_block, *content] if prepend else [*content, text_block] rendered = str(content) return text + rendered if prepend else rendered + text def _strip_image_parts_from_parts(parts: Any) -> Any: """Strip image parts from an OpenAI-style content-parts list. Returns a new list with image_url / image / input_image parts replaced by a text placeholder, or None if the list had no images (callers skip the replacement in that case). Used by the compressor to prune old computer_use screenshots. """ if not isinstance(parts, list): return None had_image = False out = [] for part in parts: if not isinstance(part, dict): out.append(part) continue ptype = part.get("type") if ptype in {"image", "image_url", "input_image"}: had_image = True out.append({"type": "text", "text": "[screenshot removed to save context]"}) else: out.append(part) return out if had_image else None def _truncate_tool_call_args_json(args: str, head_chars: int = 200) -> str: """Shrink long string values inside a tool-call arguments JSON blob while preserving JSON validity. The ``function.arguments`` field on a tool call is a JSON-encoded string passed through to the LLM provider; downstream providers strictly validate it and return a non-retryable 400 when it is not well-formed. An earlier implementation sliced the raw JSON at a fixed byte offset and appended ``...[truncated]`` — which routinely produced strings like:: {"path": "/foo/bar", "content": "# long markdown ...[truncated] i.e. an unterminated string and a missing closing brace. MiniMax, for example, rejects this with ``invalid function arguments json string`` and the session gets stuck re-sending the same broken history on every turn. See issue #11762 for the observed loop. This helper parses the arguments, shrinks long string leaves inside the parsed structure, and re-serialises. Non-string values (paths, ints, booleans) are preserved intact. If the arguments are not valid JSON to begin with — some model backends use non-JSON tool arguments — the original string is returned unchanged rather than replaced with something neither we nor the backend can parse. """ try: parsed = json.loads(args) except (ValueError, TypeError): return args def _shrink(obj: Any) -> Any: if isinstance(obj, str): if len(obj) > head_chars: return obj[:head_chars] + "...[truncated]" return obj if isinstance(obj, dict): return {k: _shrink(v) for k, v in obj.items()} if isinstance(obj, list): return [_shrink(v) for v in obj] return obj shrunken = _shrink(parsed) # ensure_ascii=False preserves CJK/emoji instead of bloating with \uXXXX return json.dumps(shrunken, ensure_ascii=False) _IMAGE_PART_TYPES = frozenset({"image_url", "input_image", "image"}) def _is_image_part(part: Any) -> bool: """True if ``part`` is a multimodal image content block. Recognizes all three shapes the agent handles: - OpenAI chat.completions: ``{"type": "image_url", "image_url": ...}`` - OpenAI Responses API: ``{"type": "input_image", "image_url": "..."}`` - Anthropic native: ``{"type": "image", "source": {...}}`` """ if not isinstance(part, dict): return False return part.get("type") in _IMAGE_PART_TYPES def _content_has_images(content: Any) -> bool: """True if a message's ``content`` is a multimodal list with image parts.""" if not isinstance(content, list): return False return any(_is_image_part(p) for p in content) def _strip_images_from_content(content: Any) -> Any: """Return a copy of ``content`` with every image part replaced by a short text placeholder. - String content is returned unchanged. - Non-list, non-string content is returned unchanged. - List content: image parts become ``{"type": "text", "text": "[Attached image — stripped after compression]"}``; other parts are preserved as-is. Input is never mutated. """ if not isinstance(content, list): return content if not any(_is_image_part(p) for p in content): return content new_parts: List[Any] = [] for p in content: if _is_image_part(p): new_parts.append({ "type": "text", "text": "[Attached image — stripped after compression]", }) else: new_parts.append(p) return new_parts def _strip_historical_media(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Replace image parts in older messages with placeholder text. The anchor is the *last* user message that has any image content. Every message before that anchor gets its image parts replaced with a short placeholder so the outgoing request stops re-shipping the same multi-MB base-64 image blobs on every turn. If no user message carries images, the list is returned unchanged. If the only user message with images is the very first one (nothing earlier to strip), the list is returned unchanged. Shallow copies of touched messages only; input is never mutated. Port of Kilo-Org/kilocode#9434 (adapted for the OpenAI-style message shape the hermes compressor emits). """ if not messages: return messages # Find the newest user message that carries at least one image part. # We anchor on image-bearing user messages (not all user messages) so # a plain text follow-up after a big-image turn still strips the old # image — matching the problem kilocode#9434 set out to solve. anchor = -1 for i in range(len(messages) - 1, -1, -1): msg = messages[i] if not isinstance(msg, dict): continue if msg.get("role") != "user": continue if _content_has_images(msg.get("content")): anchor = i break if anchor <= 0: # No image-bearing user message, or it's the very first message — # nothing before it to strip. return messages changed = False result: List[Dict[str, Any]] = [] for i, msg in enumerate(messages): if i >= anchor or not isinstance(msg, dict): result.append(msg) continue content = msg.get("content") if not _content_has_images(content): result.append(msg) continue new_msg = msg.copy() new_msg["content"] = _strip_images_from_content(content) # Content rewritten → the api_content sidecar (exact bytes previously # sent) is stale; drop it so replay can't resend the pre-rewrite bytes. drop_stale_api_content(new_msg) result.append(new_msg) changed = True return result if changed else messages def _image_part_label(part: Dict[str, Any]) -> str: """Render a multimodal image part as a short text label for the summarizer. Keeps a real, referenceable URL when the image lives at an http(s) address — the summary can then preserve the handle so the agent (or a later vision_analyze call) can still reach the image after compaction. Base64 ``data:`` URLs carry no reusable reference and would flood the summarizer input, so they collapse to ``[image]``. """ url = "" if isinstance(part.get("image_url"), dict): url = str(part["image_url"].get("url") or "") elif isinstance(part.get("image_url"), str): url = part["image_url"] elif isinstance(part.get("url"), str): url = part["url"] if url.startswith(("http://", "https://")): return f"[image: {url}]" return "[image]" def _str_arg(args: dict, key: str, default: str = "") -> str: """Safely get a string argument from parsed tool args. LLMs sometimes return non-string parameter values (e.g. bool, int) for tool calls. Calling ``len()`` / ``.count()`` / slicing on those causes ``TypeError`` / ``AttributeError`` which crashes context compression. This helper coerces any value to ``str`` so downstream code can assume a string is always returned. """ val = args.get(key, default) if isinstance(val, str): return val return str(val) if val is not None else default def _summarize_tool_result(tool_name: str, tool_args: str, tool_content: str) -> str: """Create an informative 1-line summary of a tool call + result. Used during the pre-compression pruning pass to replace large tool outputs with a short but useful description of what the tool did, rather than a generic placeholder that carries zero information. Returns strings like:: [terminal] ran `npm test` -> exit 0, 47 lines output [read_file] read config.py from line 1 (1,200 chars) [search_files] content search for 'compress' in agent/ -> 12 matches Never raises: models sometimes emit non-string argument values (bool, int, None) and the args here come from persisted session history, so a single malformed historical call must not crash compression — which retries on the same history and would crash-loop. Individual branches coerce the values they slice/measure (keeping summaries informative); this wrapper is the backstop for anything they miss. """ try: return _summarize_tool_result_unguarded(tool_name, tool_args, tool_content) except Exception as exc: # noqa: BLE001 — a summary must never crash compression logger.debug("Tool-result summary failed for %s: %s", tool_name, exc) _len = len(tool_content) if isinstance(tool_content, str) else 0 return f"[{tool_name}] ({_len:,} chars result)" def _summarize_tool_result_unguarded(tool_name: str, tool_args: str, tool_content: str) -> str: """Build the summary line (unguarded; see ``_summarize_tool_result``).""" try: args = json.loads(tool_args) if tool_args else {} except (json.JSONDecodeError, TypeError): args = {} if not isinstance(args, dict): args = {} content = tool_content or "" content_len = len(content) line_count = content.count("\n") + 1 if content.strip() else 0 if tool_name == "terminal": cmd = _str_arg(args, "command") if len(cmd) > 80: cmd = cmd[:77] + "..." exit_match = re.search(r'"exit_code"\s*:\s*(-?\d+)', content) exit_code = exit_match.group(1) if exit_match else "?" return f"[terminal] ran `{cmd}` -> exit {exit_code}, {line_count} lines output" if tool_name == "read_file": path = args.get("path", "?") offset = args.get("offset", 1) return f"[read_file] read {path} from line {offset} ({content_len:,} chars)" if tool_name == "write_file": path = args.get("path", "?") written_lines = _str_arg(args, "content").count("\n") + 1 if args.get("content") else "?" return f"[write_file] wrote to {path} ({written_lines} lines)" if tool_name == "search_files": pattern = args.get("pattern", "?") path = args.get("path", ".") target = args.get("target", "content") match_count = re.search(r'"total_count"\s*:\s*(\d+)', content) count = match_count.group(1) if match_count else "?" return f"[search_files] {target} search for '{pattern}' in {path} -> {count} matches" if tool_name == "patch": path = args.get("path", "?") mode = args.get("mode", "replace") return f"[patch] {mode} in {path} ({content_len:,} chars result)" if tool_name in {"browser_navigate", "browser_click", "browser_snapshot", "browser_type", "browser_scroll", "browser_vision"}: url = args.get("url", "") ref = args.get("ref", "") detail = f" {url}" if url else (f" ref={ref}" if ref else "") return f"[{tool_name}]{detail} ({content_len:,} chars)" if tool_name == "web_search": query = args.get("query", "?") return f"[web_search] query='{query}' ({content_len:,} chars result)" if tool_name == "web_extract": urls = args.get("urls", []) first = urls[0] if isinstance(urls, list) and urls else "?" # web_search results are dicts ({"url"/"href": ...}) and models often # forward them straight into web_extract. Unwrap to the URL string so # the summary stays readable and the ``+=`` below never hits the # ``dict + str`` TypeError that would abort pre-compression pruning. if isinstance(first, dict): first = first.get("url") or first.get("href") or "?" elif not isinstance(first, str): first = "?" url_desc = first if isinstance(urls, list) and len(urls) > 1: url_desc += f" (+{len(urls) - 1} more)" return f"[web_extract] {url_desc} ({content_len:,} chars)" if tool_name == "delegate_task": goal = _str_arg(args, "goal") if len(goal) > 60: goal = goal[:57] + "..." return f"[delegate_task] '{goal}' ({content_len:,} chars result)" if tool_name == "execute_code": code_str = _str_arg(args, "code") code_preview = code_str[:60].replace("\n", " ") if len(code_str) > 60: code_preview += "..." return f"[execute_code] `{code_preview}` ({line_count} lines output)" if tool_name in {"skill_view", "skills_list", "skill_manage"}: name = args.get("name", "?") return f"[{tool_name}] name={name} ({content_len:,} chars)" if tool_name == "vision_analyze": question = _str_arg(args, "question")[:50] return f"[vision_analyze] '{question}' ({content_len:,} chars)" if tool_name == "memory": action = args.get("action", "?") target = args.get("target", "?") return f"[memory] {action} on {target}" if tool_name == "todo": return "[todo] updated task list" if tool_name == "clarify": return "[clarify] asked user a question" if tool_name == "text_to_speech": return f"[text_to_speech] generated audio ({content_len:,} chars)" if tool_name == "cronjob": action = args.get("action", "?") return f"[cronjob] {action}" if tool_name == "process": action = args.get("action", "?") sid = args.get("session_id", "?") return f"[process] {action} session={sid}" # Generic fallback first_arg = "" for k, v in list(args.items())[:2]: sv = str(v)[:40] first_arg += f" {k}={sv}" return f"[{tool_name}]{first_arg} ({content_len:,} chars result)" class ContextCompressor(ContextEngine): """Default context engine — compresses conversation context via lossy summarization. Algorithm: 1. Prune old tool results (cheap, no LLM call) 2. Protect head messages (system prompt + first exchange) 3. Protect tail messages by token budget (most recent ~20K tokens) 4. Summarize middle turns with structured LLM prompt 5. On subsequent compactions, iteratively update the previous summary """ @property def name(self) -> str: return "compressor" def on_session_reset(self) -> None: """Reset all per-session state for /new or /reset.""" super().on_session_reset() self._context_probed = False self._context_probe_persistable = False self._previous_summary = None self._last_summary_error = None self._consecutive_timeout_failures = 0 self._last_summary_dropped_count = 0 self._last_summary_fallback_used = False self._last_aux_model_failure_error = None self._last_aux_model_failure_model = None self._last_compression_savings_pct = 100.0 self._ineffective_compression_count = 0 self._fallback_compression_streak = 0 self._verify_compaction_cleared_threshold = False self._last_compression_made_progress = False self._summary_failure_cooldown_until = 0.0 # transient errors must not block a fresh session self._cooldown_persist_failed = False self._last_summary_error = None self._last_compress_aborted = False self.last_real_prompt_tokens = 0 self.last_compression_rough_tokens = 0 self.last_rough_tokens_when_real_prompt_fit = 0 self.awaiting_real_usage_after_compression = False def on_session_end(self, session_id: str, messages: List[Dict[str, Any]]) -> None: """Clear all per-session compaction state at a real session boundary. Session end (CLI exit, gateway expiry, session-id rotation) goes through this method rather than ``on_session_reset()`` (/new, /reset). The original fix (#38788) only cleared ``_previous_summary``, but the same cross-session contamination risk applies to every per-session variable that ``on_session_reset()`` clears: stale ``_ineffective_compression_count`` can suppress compression in a subsequent live session; ``_summary_failure_cooldown_until`` can block summary generation; ``_last_compress_aborted`` can make callers think compression is still aborted; ``_last_aux_model_failure_*`` can surface stale error warnings; ``_last_summary_dropped_count`` / ``_last_summary_fallback_used`` can produce misleading user warnings. ``compress()`` already guards ``_previous_summary`` leakage at the point of use; this is defense-in-depth that resets the full per-session surface the moment the owning session ends. """ self._previous_summary = None self._last_summary_error = None self._consecutive_timeout_failures = 0 self._last_summary_dropped_count = 0 self._last_summary_fallback_used = False self._last_aux_model_failure_error = None self._last_aux_model_failure_model = None self._last_compression_savings_pct = 100.0 self._ineffective_compression_count = 0 self._fallback_compression_streak = 0 self._verify_compaction_cleared_threshold = False self._last_compression_made_progress = False self._summary_failure_cooldown_until = 0.0 self._cooldown_persist_failed = False self._last_compress_aborted = False self._context_probed = False self._context_probe_persistable = False self.last_real_prompt_tokens = 0 self.last_compression_rough_tokens = 0 self.last_rough_tokens_when_real_prompt_fit = 0 self.awaiting_real_usage_after_compression = False def bind_session_state(self, session_db: Any = None, session_id: str = "") -> None: """Bind the current session row so durable cooldowns can round-trip.""" self._session_db = session_db self._session_id = session_id or "" self._summary_failure_cooldown_until = 0.0 self._cooldown_persist_failed = False self._last_summary_error = None self._consecutive_timeout_failures = 0 self._fallback_compression_streak = 0 self.get_active_compression_failure_cooldown() self._load_fallback_compression_streak() def on_session_start(self, session_id: str, **kwargs) -> None: """Bind session-scoped compression state for a new or resumed session.""" super().on_session_start(session_id, **kwargs) boundary_reason = kwargs.get("boundary_reason") old_session_id = kwargs.get("old_session_id") session_db = kwargs.get("session_db", getattr(self, "_session_db", None)) previous_fallback_streak = self._fallback_compression_streak if boundary_reason == "compression" and old_session_id: getter = getattr(session_db, "get_compression_fallback_streak", None) if callable(getter): try: stored_streak = getter(old_session_id) if isinstance(stored_streak, (int, float, str)): previous_fallback_streak = max(0, int(stored_streak)) except (TypeError, ValueError, sqlite3.Error) as exc: logger.debug("compression parent fallback streak lookup failed: %s", exc) except Exception as exc: logger.debug( "compression parent fallback streak lookup failed (non-sqlite): %s", exc, ) self.bind_session_state(session_db, session_id) if boundary_reason == "compression": # Rotation creates a fresh child row before this callback. Preserve # the logical conversation's streak until boundary bookkeeping # persists the updated value onto the child row. self._fallback_compression_streak = previous_fallback_streak def _load_fallback_compression_streak(self) -> None: session_db = getattr(self, "_session_db", None) session_id = getattr(self, "_session_id", "") getter = getattr(session_db, "get_compression_fallback_streak", None) if not session_id or not callable(getter): return try: stored_streak = getter(session_id) self._fallback_compression_streak = max( 0, int(stored_streak) if isinstance(stored_streak, (int, float, str)) else 0, ) except (TypeError, ValueError, sqlite3.Error) as exc: logger.debug("compression fallback streak lookup failed: %s", exc) except Exception as exc: logger.debug("compression fallback streak lookup failed (non-sqlite): %s", exc) def _persist_fallback_compression_streak(self) -> None: session_db = getattr(self, "_session_db", None) session_id = getattr(self, "_session_id", "") setter = getattr(session_db, "set_compression_fallback_streak", None) if not session_id or not callable(setter): return try: setter(session_id, self._fallback_compression_streak) except sqlite3.Error as exc: logger.debug("compression fallback streak persist failed: %s", exc) except Exception as exc: logger.debug("compression fallback streak persist failed (non-sqlite): %s", exc) def record_completed_compaction(self, *, used_fallback: bool = False) -> None: """Record one completed boundary and its summary quality.""" self._verify_compaction_cleared_threshold = True if used_fallback: self._fallback_compression_streak += 1 if not self.quiet_mode: logger.warning( "Compaction completed with a deterministic fallback summary. " "fallback_compression_streak=%d", self._fallback_compression_streak, ) elif self._fallback_compression_streak: self._fallback_compression_streak = 0 self._persist_fallback_compression_streak() def get_active_compression_failure_cooldown( self, *, refresh: bool = False, ) -> Optional[Dict[str, Any]]: """Return the live compression-failure cooldown for the bound session.""" now_mono = time.monotonic() local_state = None if self._summary_failure_cooldown_until > now_mono: local_state = { "cooldown_until": time.time() + ( self._summary_failure_cooldown_until - now_mono ), "remaining_seconds": self._summary_failure_cooldown_until - now_mono, "error": self._last_summary_error, } if not refresh: return local_state session_db = getattr(self, "_session_db", None) session_id = getattr(self, "_session_id", "") if not session_db or not session_id: return local_state getter = getattr(session_db, "get_compression_failure_cooldown", None) if getter is None: return local_state try: state = getter(session_id) except sqlite3.Error as exc: logger.debug("compression failure cooldown lookup failed: %s", exc) return local_state except Exception: return local_state if not state: if refresh: if local_state is not None and self._cooldown_persist_failed: # The live local cooldown never made it to the DB (persist # failed), so the empty row is not evidence that another # agent cleared it. Honouring the DB here would re-enable # auto-compress mid-cooldown and reopen the #11529 thrash # window. Keep the local timer authoritative until it # expires or a successful DB read supersedes it. return local_state self._summary_failure_cooldown_until = 0.0 self._last_summary_error = None return None remaining_seconds = float(state.get("remaining_seconds") or 0.0) if remaining_seconds <= 0: if refresh: if local_state is not None and self._cooldown_persist_failed: return local_state self._summary_failure_cooldown_until = 0.0 self._last_summary_error = None return None self._summary_failure_cooldown_until = now_mono + remaining_seconds self._last_summary_error = state.get("error") self._cooldown_persist_failed = False return { "cooldown_until": float(state.get("cooldown_until") or 0.0), "remaining_seconds": remaining_seconds, "error": self._last_summary_error, } def _record_compression_failure_cooldown( self, cooldown_seconds: float, error: Optional[str], ) -> None: cooldown_until = time.time() + cooldown_seconds self._summary_failure_cooldown_until = time.monotonic() + cooldown_seconds self._last_summary_error = error session_db = getattr(self, "_session_db", None) session_id = getattr(self, "_session_id", "") if not session_db or not session_id: return recorder = getattr(session_db, "record_compression_failure_cooldown", None) if recorder is None: self._cooldown_persist_failed = True return try: recorder(session_id, cooldown_until, error) self._cooldown_persist_failed = False except sqlite3.Error as exc: self._cooldown_persist_failed = True logger.debug("compression failure cooldown persist failed: %s", exc) except Exception as exc: self._cooldown_persist_failed = True logger.debug("compression failure cooldown persist failed (non-sqlite): %s", exc) def _clear_compression_failure_cooldown(self) -> None: self._summary_failure_cooldown_until = 0.0 self._last_summary_error = None self._consecutive_timeout_failures = 0 self._cooldown_persist_failed = False session_db = getattr(self, "_session_db", None) session_id = getattr(self, "_session_id", "") if not session_db or not session_id: return clearer = getattr(session_db, "clear_compression_failure_cooldown", None) if clearer is None: return try: clearer(session_id) except sqlite3.Error as exc: logger.debug("compression failure cooldown clear failed: %s", exc) except Exception as exc: logger.debug("compression failure cooldown clear failed (non-sqlite): %s", exc) def update_model( self, model: str, context_length: int, base_url: str = "", api_key: Any = "", provider: str = "", api_mode: str = "", max_tokens: int | None = None, ) -> None: """Update model info after a model switch or fallback activation.""" runtime_changed = any(( model != self.model, provider != self.provider, base_url != self.base_url, api_mode != self.api_mode, )) self.model = model self.base_url = base_url self.api_key = api_key self.provider = provider self.api_mode = api_mode self.context_length = context_length # Re-apply the small-context threshold floor for the NEW window, # starting from the originally-configured percent (not the possibly # floored live value) so a small -> large switch drops back to the # configured threshold and a large -> small switch gains the floor. # Guard with getattr: compressors unpickled/constructed before this # attribute existed fall back to the live value. _configured_pct = getattr( self, "_configured_threshold_percent", self.threshold_percent, ) self.threshold_percent = self._effective_threshold_percent( context_length, _configured_pct, ) # max_tokens=None here means "caller didn't specify" → keep the existing # output reservation. A switch that genuinely changes the output budget # passes the new value explicitly. (#43547) if max_tokens is not None: self.max_tokens = self._coerce_max_tokens(max_tokens) self.threshold_tokens = self._compute_threshold_tokens( context_length, self.threshold_percent, self.max_tokens, ) # Recalculate token budgets for the new context length so the # compressor stays calibrated after a model switch (e.g. 200K → 32K). target_tokens = int(self.threshold_tokens * self.summary_target_ratio) self.tail_token_budget = target_tokens self.max_summary_tokens = min( int(context_length * 0.05), _SUMMARY_TOKENS_CEILING, ) # Reset cross-call calibration state captured under the PREVIOUS model. # These fields encode "the provider proved this prompt fit" / "preflight # can be deferred" decisions that are only valid for the model that # produced them. Carrying them across a switch to a smaller-context # model would let should_defer_preflight_to_real_usage() suppress a # preflight compression the new model actually needs — the exact # oversized-send-after-switch failure in #23767. The new model's first # response repopulates them via update_from_response(). Setting # last_prompt_tokens to 0 (NOT -1) is deliberate: 0 is the documented # "no real usage yet -> use the rough estimate" state, so the post- # response should_compress path falls back to estimate_request_tokens_rough # rather than skipping compression. -1 is a different sentinel # (#36718, "compression just ran, await real usage") and must not be set here. self.last_prompt_tokens = 0 self.last_completion_tokens = 0 self.last_total_tokens = 0 self.last_real_prompt_tokens = 0 self.last_rough_tokens_when_real_prompt_fit = 0 self.last_compression_rough_tokens = 0 self.awaiting_real_usage_after_compression = False self._ineffective_compression_count = 0 if runtime_changed: self._fallback_compression_streak = 0 self._persist_fallback_compression_streak() # Failure cooldowns are scoped to the model/provider that failed. # A switch must give the new runtime an immediate summary attempt. self._clear_compression_failure_cooldown() self._verify_compaction_cleared_threshold = False self._last_compression_made_progress = False # When the MINIMUM_CONTEXT_LENGTH floor meets/exceeds a small context # window, compacting at the percentage (50% → 32K of a 64K window) wastes # half the usable context. Trigger near the top of the window instead so a # minimum-context model uses most of its budget before compacting — same # rationale as the gpt-5.5/Codex 85% autoraise. _MIN_CTX_TRIGGER_RATIO = 0.85 @staticmethod def _coerce_max_tokens(value: Any) -> int | None: """Normalize a max_tokens value to a positive int or None. Only a positive integer is a real output reservation. None (provider default), non-numeric values, or <= 0 all mean "no reservation" — this keeps the threshold arithmetic safe from non-int inputs (e.g. a test MagicMock reaching ContextCompressor via a mocked parent agent). """ if value is None: return None try: ivalue = int(value) except (TypeError, ValueError): return None return ivalue if ivalue > 0 else None @staticmethod def _effective_threshold_percent( context_length: int, threshold_percent: float, ) -> float: """Apply the small-context threshold floor (raise-only). Models under ``_SMALL_CTX_WINDOW_LIMIT`` (512K) trigger at no less than ``_SMALL_CTX_THRESHOLD_PERCENT`` (75%) of the window. An explicitly higher threshold (user config or per-model autoraise, e.g. Codex gpt-5.5's 85%) always wins; only lower values are raised. Large-context models keep the configured value — at 512K+ the default 50% trigger already leaves ample post-compaction headroom. """ if context_length and context_length < _SMALL_CTX_WINDOW_LIMIT: return max(threshold_percent, _SMALL_CTX_THRESHOLD_PERCENT) return threshold_percent @staticmethod def _compute_threshold_tokens( context_length: int, threshold_percent: float, max_tokens: int | None = None, ) -> int: """Compute the compaction trigger threshold in tokens. The base value is ``effective_input_budget * threshold_percent``, floored at ``MINIMUM_CONTEXT_LENGTH`` so large-context models don't compress prematurely at 50%. BUT that floor degenerates at small windows: for a model whose ``context_length`` is at/below the minimum (e.g. a 64K local model), ``max(0.5*64000, 64000) == 64000`` makes the threshold equal the ENTIRE window — auto-compression can never fire because the provider rejects the request before usage reaches 100% (#14690). When the floor would meet or exceed the context window, trigger at ``_MIN_CTX_TRIGGER_RATIO`` (85%) of the window — high enough that a small model uses most of its context before compacting, but below 100% so compaction fires before the provider rejects the request. The provider reserves ``max_tokens`` of output space out of the same window, so the usable INPUT budget is ``context_length - max_tokens``. With a large ``max_tokens`` (e.g. 65536 on a custom provider) the input budget is materially smaller than the raw window, and a threshold based on the full window lets the session hit a provider 400 before compaction fires (#43547). The percentage and the degenerate-window check below both operate on the effective input budget. ``max_tokens=None`` (provider default) conservatively assumes no reservation (full window). """ effective_window = context_length - (max_tokens or 0) if effective_window <= 0: effective_window = context_length pct_value = int(effective_window * threshold_percent) floored = max(pct_value, MINIMUM_CONTEXT_LENGTH) # If flooring pushed the threshold to/over the effective window it can # never be reached. Trigger at 85% of the effective input budget so a # minimum-context model rides most of its budget before compacting # instead of wasting half. if effective_window > 0 and floored >= effective_window: return max(1, min(int(effective_window * ContextCompressor._MIN_CTX_TRIGGER_RATIO), effective_window - 1)) return floored def __init__( self, model: str, threshold_percent: float = 0.50, protect_first_n: int = 3, protect_last_n: int = 20, summary_target_ratio: float = 0.20, quiet_mode: bool = False, summary_model_override: str = None, base_url: str = "", api_key: str = "", config_context_length: int | None = None, provider: str = "", api_mode: str = "", abort_on_summary_failure: bool = False, max_tokens: int | None = None, ): self.model = model self.base_url = base_url self.api_key = api_key self.provider = provider self.api_mode = api_mode self.threshold_percent = threshold_percent self.protect_first_n = protect_first_n self.protect_last_n = protect_last_n self.summary_target_ratio = max(0.10, min(summary_target_ratio, 0.80)) self.quiet_mode = quiet_mode # Output-token reservation: the provider carves max_tokens out of the # context window, so the usable input budget is context_length - # max_tokens. None = provider default => assume no reservation. (#43547) # Coerce defensively: only a positive int is a real reservation; any # other value (None, non-numeric, <=0) means "no reservation" so the # threshold arithmetic never sees a non-int (e.g. a test MagicMock). self.max_tokens = self._coerce_max_tokens(max_tokens) # When True, summary-generation failure aborts compression entirely # (returns messages unchanged, sets _last_compress_aborted=True). # When False (default = historical behavior), insert a # deterministic "summary unavailable" handoff and drop the middle window. self.abort_on_summary_failure = abort_on_summary_failure self.context_length = get_model_context_length( model, base_url=base_url, api_key=api_key, config_context_length=config_context_length, provider=provider, ) # Small-context threshold floor: models under 512K trigger at >=75% # so compaction doesn't fire with half the window still free (the # incompressible floor makes 50%-triggered compaction thrash on # 128K-262K models). Raise-only; must run AFTER context_length is # resolved and BEFORE threshold_tokens is derived. The pre-floor # value is kept so update_model() can re-derive for a new window # (switching small -> large must drop back to the configured value). self._configured_threshold_percent = self.threshold_percent self.threshold_percent = self._effective_threshold_percent( self.context_length, self.threshold_percent, ) threshold_percent = self.threshold_percent # Floor: never compress below MINIMUM_CONTEXT_LENGTH tokens even if # the percentage would suggest a lower value. This prevents premature # compression on large-context models at 50% while keeping the % sane # for models right at the minimum. _compute_threshold_tokens also # guards the degenerate case where the floor would equal/exceed the # window (small models), so auto-compression can still fire (#14690). self.threshold_tokens = self._compute_threshold_tokens( self.context_length, threshold_percent, self.max_tokens, ) self.compression_count = 0 # Derive token budgets: ratio is relative to the threshold, not total context target_tokens = int(self.threshold_tokens * self.summary_target_ratio) self.tail_token_budget = target_tokens self.max_summary_tokens = min( int(self.context_length * 0.05), _SUMMARY_TOKENS_CEILING, ) if not quiet_mode: logger.info( "Context compressor initialized: model=%s context_length=%d " "threshold=%d (%.0f%%) target_ratio=%.0f%% tail_budget=%d " "provider=%s base_url=%s", model, self.context_length, self.threshold_tokens, threshold_percent * 100, self.summary_target_ratio * 100, self.tail_token_budget, provider or "none", base_url or "none", ) self._context_probed = False # True after a step-down from context error self.last_prompt_tokens = 0 self.last_completion_tokens = 0 self.last_real_prompt_tokens = 0 self.last_compression_rough_tokens = 0 self.last_rough_tokens_when_real_prompt_fit = 0 self.awaiting_real_usage_after_compression = False self.summary_model = summary_model_override or "" self._session_db: Any = None self._session_id: str = "" # Stores the previous compaction summary for iterative updates self._previous_summary: Optional[str] = None # Anti-thrashing: track whether last compression was effective self._last_compression_savings_pct: float = 100.0 self._ineffective_compression_count: int = 0 # Consecutive completed deterministic-fallback boundaries. Unlike the # real-usage effectiveness counter, ordinary fitting responses must not # reset this breaker; only a healthy completed summary does. self._fallback_compression_streak: int = 0 # Set after a completed compression boundary; consumed by the next # provider-reported prompt count in update_from_response(). self._verify_compaction_cleared_threshold: bool = False # Lets the boundary wrapper distinguish a completed rewrite from a # no-op/abort without inferring progress from message-list length. self._last_compression_made_progress: bool = False self._summary_failure_cooldown_until: float = 0.0 # True while the live local cooldown failed to persist to the DB; # a refresh must then treat an empty durable row as unknown, not # cleared (see get_active_compression_failure_cooldown). self._cooldown_persist_failed: bool = False self._last_summary_error: Optional[str] = None # When summary generation fails and a static fallback is inserted, # record how many turns were unrecoverably dropped so callers # (gateway hygiene, /compress) can surface a visible warning. self._last_summary_dropped_count: int = 0 self._last_summary_fallback_used: bool = False # When summary generation fails we now ABORT compression entirely # and return the original messages unchanged instead of dropping # the middle window with a static placeholder. Callers inspect # this flag to know "compression was attempted but aborted, freeze # the chat until the user manually retries via /compress". self._last_compress_aborted: bool = False # Set True when the summary call failed with an authentication / # permission error (HTTP 401/403). Auth failures are non-recoverable # at the request level — the credential or endpoint is broken — so # compress() must ABORT (preserve the session unchanged) rather than # rotate into a degraded child session with a placeholder summary. # This is independent of the abort_on_summary_failure config flag: # rotating on a broken credential is never the right behavior. self._last_summary_auth_failure: bool = False # Set when summary generation ultimately fails due to a transient # network/connection error (httpx/httpcore connection drop, premature # stream close, etc.) — distinct from auth failures but treated the # same way by compress(): ABORT and preserve the session unchanged # rather than destroy the middle window for a deterministic # "summary unavailable" marker. Retrying once the network recovers is # strictly better than discarding context for a transient blip # (#29559, #25585). Independent of abort_on_summary_failure. self._last_summary_network_failure: bool = False # retrying on the main model, record the failure so gateway / # CLI callers can still warn the user even though compression # succeeded. Silent recovery would hide the broken config. self._last_aux_model_failure_error: Optional[str] = None self._last_aux_model_failure_model: Optional[str] = None def update_from_response(self, usage: Dict[str, Any]): """Update tracked token usage from API response.""" self.last_prompt_tokens = usage.get("prompt_tokens", 0) self.last_completion_tokens = usage.get("completion_tokens", 0) self.last_total_tokens = usage.get("total_tokens", self.last_prompt_tokens + self.last_completion_tokens) if self.last_prompt_tokens > 0: self.last_real_prompt_tokens = self.last_prompt_tokens if self.last_prompt_tokens < self.threshold_tokens: if self.awaiting_real_usage_after_compression and self.last_compression_rough_tokens > 0: self.last_rough_tokens_when_real_prompt_fit = self.last_compression_rough_tokens # Any real provider reading below the trigger proves the prompt # fits again. Clear the real-usage effectiveness latch even # when this response was not immediately after compaction. The # independent fallback streak is boundary-scoped and survives # ordinary fitting responses during context regrowth. self._ineffective_compression_count = 0 else: self.last_rough_tokens_when_real_prompt_fit = 0 # Anti-thrashing verdict, judged HERE because this is the only place # that sees the provider's real prompt count for the just-compacted # conversation. Effectiveness is "did the prompt get under the # threshold?", not "did the message list shrink?": compaction can # only shrink messages, while the system prompt and tool schemas are # an incompressible floor (with 50+ tools, 20-30K tokens — see # #14695). When that floor alone meets the threshold, every pass # shrinks messages by a healthy margin yet leaves the prompt over the # line, so the next turn compacts again, forever. # # It must NOT live in should_compress(): that runs twice per turn # with two different measures (a rough preflight estimate and the # real post-response count, #36718), and the rough one can dip below # the threshold and reset the strike every turn, re-opening the loop. # Keying on real usage compares like with like and fires exactly once # per compaction. if self._verify_compaction_cleared_threshold: if self.last_prompt_tokens >= self.threshold_tokens: self._ineffective_compression_count += 1 if not self.quiet_mode: logger.warning( "Compaction did not clear the threshold: %d real " "tokens still >= %d. The incompressible prompt " "(system prompt + tool schemas) may already exceed " "it, in which case shrinking messages cannot help. " "ineffective_compression_count=%d", self.last_prompt_tokens, self.threshold_tokens, self._ineffective_compression_count, ) else: self._ineffective_compression_count = 0 # Consume the pending-verification flag once real usage arrives, whether # or not prompt_tokens was reported, so a usage-less response can't leave # it armed for a later, unrelated reading. self._verify_compaction_cleared_threshold = False self.awaiting_real_usage_after_compression = False def should_defer_preflight_to_real_usage(self, rough_tokens: int) -> bool: """Return True when a high rough preflight estimate is known-noisy. ``estimate_request_tokens_rough(..., tools=...)`` intentionally overestimates schema-heavy requests so Hermes compresses before a provider rejects the payload. After a successful compressed API call, though, provider ``prompt_tokens`` are a better signal than repeating compaction from the same rough schema overhead. Defer only while the rough estimate has grown modestly since a request the provider proved fit under the threshold. """ if rough_tokens < self.threshold_tokens: return False # Immediately after a compaction the post-compression path sets # ``awaiting_real_usage_after_compression`` and parks # ``last_prompt_tokens = -1``, but ``last_real_prompt_tokens`` still # holds the STALE pre-compression value (above threshold — that's why # compaction fired). Without this guard that stale value defeats the # ``last_real_prompt_tokens >= threshold_tokens`` check below, so # preflight fires a SECOND compaction before the provider has reported # real token usage for the now-shorter conversation. Defer for exactly # one turn; update_from_response() clears the flag when real usage # arrives. (#36718) if self.awaiting_real_usage_after_compression: return True if self.last_real_prompt_tokens <= 0: return False if self.last_real_prompt_tokens >= self.threshold_tokens: return False baseline = self.last_rough_tokens_when_real_prompt_fit or self.last_compression_rough_tokens if baseline <= 0: return False growth = max(0, rough_tokens - baseline) tolerated_growth = max(4096, int(self.threshold_tokens * 0.05)) if growth > tolerated_growth: return False self.last_rough_tokens_when_real_prompt_fit = max(baseline, rough_tokens) return True def should_compress(self, prompt_tokens: int = None) -> bool: """Check if context exceeds the compression threshold. Includes anti-thrashing protection: if the last two compressions each saved less than 10%, skip compression to avoid infinite loops where each pass removes only 1-2 messages. """ tokens = prompt_tokens if prompt_tokens is not None else self.last_prompt_tokens if tokens < self.threshold_tokens: return False return not self._automatic_compression_blocked() def _refresh_durable_guards(self) -> None: """Re-read durable cooldown + fallback-streak state from the DB. Cheap, best-effort, and only called when a gate is about to say "blocked": another agent on the same session may have cleared the durable rows (successful boundary, forced retry) after this compressor was bound, and a fallback streak has no timer — without a re-read the stale in-memory snapshot blocks forever. """ try: self.get_active_compression_failure_cooldown(refresh=True) except Exception as exc: logger.debug("compression cooldown refresh failed: %s", exc) try: self._load_fallback_compression_streak() except Exception as exc: logger.debug("compression fallback-streak refresh failed: %s", exc) def _automatic_compression_blocked(self) -> bool: """Return whether automatic compaction is in cooldown or tripped.""" if not self._automatic_compression_blocked_locally(): return False # Blocked on the in-memory snapshot. Durable guard rows may have # been cleared by another agent since bind_session_state(); refresh # and re-evaluate so a stale local block cannot outlive the durable # state that justified it. The unblocked hot path above never pays # for the DB reads. if ( self._summary_failure_cooldown_until <= time.monotonic() and self._fallback_compression_streak < 2 ): # Blocked solely by the in-memory ineffective-compression # counter, which is not durable — there is nothing in the DB # that could unblock it, so skip the refresh (otherwise this # branch would re-read the DB on every gate check for the rest # of the session). return True self._refresh_durable_guards() return self._automatic_compression_blocked_locally() def _automatic_compression_blocked_locally(self) -> bool: """Evaluate the automatic-compaction gate on in-memory state only.""" # Do not trigger compression while the summary LLM is in cooldown. # On a 429/transient failure _generate_summary() sets a cooldown and # returns None; compress() then inserts a static fallback marker and # returns. Tokens stay above threshold, so without this guard every # subsequent turn re-fires _compress_context() — re-inserting the # marker and re-entering the loop, making the CLI appear frozen until # the cooldown expires (issue #11529). Manual /compress passes # force=True, which clears this cooldown in compress() before running, # so it still retries immediately. _cooldown_remaining = self._summary_failure_cooldown_until - time.monotonic() if _cooldown_remaining > 0: if not self.quiet_mode: logger.debug( "Compression deferred — summary LLM in cooldown for %.0fs more", _cooldown_remaining, ) return True # Anti-thrashing: back off if recent compressions were ineffective if ( self._ineffective_compression_count >= 2 or self._fallback_compression_streak >= 2 ): if not self.quiet_mode: logger.warning( "Compression skipped — repeated compaction attempts did not " "restore healthy context. ineffective=%d fallback=%d. " "Consider /new to start fresh, or /compress for " "focused compression.", self._ineffective_compression_count, self._fallback_compression_streak, ) return True return False # ------------------------------------------------------------------ # Tool output pruning (cheap pre-pass, no LLM call) # ------------------------------------------------------------------ def _prune_old_tool_results( self, messages: List[Dict[str, Any]], protect_tail_count: int, protect_tail_tokens: int | None = None, ) -> tuple[List[Dict[str, Any]], int]: """Replace old tool result contents with informative 1-line summaries. Instead of a generic placeholder, generates a summary like:: [terminal] ran `npm test` -> exit 0, 47 lines output [read_file] read config.py from line 1 (3,400 chars) Also deduplicates identical tool results (e.g. reading the same file 5x keeps only the newest full copy) and truncates large tool_call arguments in assistant messages outside the protected tail. Walks backward from the end, protecting the most recent messages that fall within ``protect_tail_tokens`` (when provided) OR the last ``protect_tail_count`` messages (backward-compatible default). When both are given, the token budget takes priority and the message count acts as a hard minimum floor. Returns (pruned_messages, pruned_count). """ if not messages: return messages, 0 result = [m.copy() for m in messages] pruned = 0 # Build index: tool_call_id -> (tool_name, arguments_json) call_id_to_tool: Dict[str, tuple] = {} for msg in result: if msg.get("role") == "assistant": for tc in msg.get("tool_calls") or []: if isinstance(tc, dict): cid = tc.get("id", "") fn = tc.get("function", {}) call_id_to_tool[cid] = (fn.get("name", "unknown"), fn.get("arguments", "")) else: cid = getattr(tc, "id", "") or "" fn = getattr(tc, "function", None) name = getattr(fn, "name", "unknown") if fn else "unknown" args_str = getattr(fn, "arguments", "") if fn else "" call_id_to_tool[cid] = (name, args_str) # Determine the prune boundary if protect_tail_tokens is not None and protect_tail_tokens > 0: # Token-budget approach: walk backward accumulating tokens accumulated = 0 boundary = len(result) min_protect = min(protect_tail_count, len(result)) for i in range(len(result) - 1, -1, -1): msg = result[i] msg_tokens = _estimate_msg_budget_tokens(msg) if accumulated + msg_tokens > protect_tail_tokens and (len(result) - i) >= min_protect: boundary = i break accumulated += msg_tokens boundary = i # Translate the budget walk into a "protected count", apply the # floor in count-space (where `max` reads naturally: protect at # least `min_protect` messages or whatever the budget reserved, # whichever is more), then convert back to a prune boundary. # Doing this in index-space with `max` would invert the direction # (smaller index = MORE protected), so a generous budget would # silently get truncated back down to `min_protect`. budget_protect_count = len(result) - boundary protected_count = max(budget_protect_count, min_protect) prune_boundary = len(result) - protected_count else: prune_boundary = len(result) - protect_tail_count # Pass 1: Deduplicate identical tool results. # When the same file is read multiple times, keep only the most recent # full copy and replace older duplicates with a back-reference. content_hashes: dict = {} # hash -> (index, tool_call_id) for i in range(len(result) - 1, -1, -1): msg = result[i] if msg.get("role") != "tool": continue content = msg.get("content") or "" # Multimodal content — dedupe by the text summary if available. if isinstance(content, list): continue if not isinstance(content, str): # Multimodal dict envelopes ({_multimodal: True, content: [...]}) and # other non-string tool-result shapes can't be hashed/deduped by text. continue if len(content) < 200: continue h = hashlib.md5(content.encode("utf-8", errors="replace")).hexdigest()[:12] if h in content_hashes: # This is an older duplicate — replace with back-reference result[i] = {**msg, "content": "[Duplicate tool output — same content as a more recent call]"} pruned += 1 else: content_hashes[h] = (i, msg.get("tool_call_id", "?")) # Pass 2: Replace old tool results with informative summaries for i in range(prune_boundary): msg = result[i] if msg.get("role") != "tool": continue content = msg.get("content", "") # Multimodal content (base64 screenshots etc.): strip the image # payload — keep a lightweight text placeholder in its place. # Without this, an old computer_use screenshot (~1MB base64 + # ~1500 real tokens) survives every compression pass forever. if isinstance(content, list): stripped = _strip_image_parts_from_parts(content) if stripped is not None: result[i] = {**msg, "content": stripped} pruned += 1 continue if isinstance(content, dict) and content.get("_multimodal"): summary = content.get("text_summary") or "[screenshot removed to save context]" result[i] = {**msg, "content": f"[screenshot removed] {summary[:200]}"} pruned += 1 continue if not isinstance(content, str): continue if not content or content == _PRUNED_TOOL_PLACEHOLDER: continue # Skip already-deduplicated or previously-summarized results if content.startswith("[Duplicate tool output"): continue # Only prune if the content is substantial (>200 chars) if len(content) > 200: call_id = msg.get("tool_call_id", "") tool_name, tool_args = call_id_to_tool.get(call_id, ("unknown", "")) summary = _summarize_tool_result(tool_name, tool_args, content) result[i] = {**msg, "content": summary} pruned += 1 # Pass 3: Truncate large tool_call arguments in assistant messages # outside the protected tail. write_file with 50KB content, for # example, survives pruning entirely without this. # # The shrinking is done inside the parsed JSON structure so the # result remains valid JSON — otherwise downstream providers 400 # on every subsequent turn until the broken call falls out of # the window. See ``_truncate_tool_call_args_json`` docstring. for i in range(prune_boundary): msg = result[i] if msg.get("role") != "assistant" or not msg.get("tool_calls"): continue new_tcs = [] modified = False for tc in msg["tool_calls"]: if isinstance(tc, dict): args = tc.get("function", {}).get("arguments", "") if len(args) > 500: new_args = _truncate_tool_call_args_json(args) if new_args != args: tc = {**tc, "function": {**tc["function"], "arguments": new_args}} modified = True new_tcs.append(tc) if modified: result[i] = {**msg, "tool_calls": new_tcs} return result, pruned # ------------------------------------------------------------------ # Summarization # ------------------------------------------------------------------ def _compute_summary_budget(self, turns_to_summarize: List[Dict[str, Any]]) -> int: """Scale summary token budget with the amount of content being compressed. The maximum scales with the model's context window (5% of context, capped at ``_SUMMARY_TOKENS_CEILING``) so large-context models get richer summaries instead of being hard-capped at 8K tokens. """ content_tokens = estimate_messages_tokens_rough(turns_to_summarize) budget = int(content_tokens * _SUMMARY_RATIO) return max(_MIN_SUMMARY_TOKENS, min(budget, self.max_summary_tokens)) # Truncation limits for the summarizer input. These bound how much of # each message the summary model sees — the budget is the *summary* # model's context window, not the main model's. _CONTENT_MAX = 6000 # total chars per message body _CONTENT_HEAD = 4000 # chars kept from the start _CONTENT_TAIL = 1500 # chars kept from the end _TOOL_ARGS_MAX = 1500 # tool call argument chars _TOOL_ARGS_HEAD = 1200 # kept from the start of tool args def _serialize_for_summary(self, turns: List[Dict[str, Any]]) -> str: """Serialize conversation turns into labeled text for the summarizer. Includes tool call arguments and result content (up to ``_CONTENT_MAX`` chars per message) so the summarizer can preserve specific details like file paths, commands, and outputs. All content is redacted before serialization to prevent secrets (API keys, tokens, passwords) from leaking into the summary that gets sent to the auxiliary model and persisted across compactions. """ # Lazy import (matches title_generator.py) — agent_runtime_helpers # pulls in heavy transitive imports we don't want at module load. from agent.agent_runtime_helpers import strip_think_blocks parts = [] for msg in turns: role = msg.get("role", "unknown") content = msg.get("content") if isinstance(content, list): text_parts: list[str] = [] for part in content: if isinstance(part, dict): ptype = part.get("type") if ptype == "text": text_parts.append(part.get("text", "")) elif ptype in {"image", "image_url", "input_image"}: text_parts.append(_image_part_label(part)) else: # Unknown part type — keep a marker so the # summarizer knows content existed here. text_parts.append(f"[{ptype or 'attachment'}]") elif isinstance(part, str): text_parts.append(part) content = "\n".join(text_parts) content = redact_sensitive_text(content or "") content = _MEDIA_DIRECTIVE_RE.sub("[media attachment]", content) # Strip inline reasoning blocks (, , etc.) from # assistant content before it reaches the summarizer. Reasoning # traces are transient scratch work — feeding them to the aux # model wastes summarizer context and risks scratch-work # conclusions being preserved as facts in the summary. The native # ``reasoning`` message field is already excluded (only # ``content`` is serialized); this closes the inline-tag path # used when native thinking is disabled or the provider inlines # traces into content. if role == "assistant" and content: content = strip_think_blocks(None, content) # Tool results: keep enough content for the summarizer if role == "tool": tool_id = msg.get("tool_call_id", "") if len(content) > self._CONTENT_MAX: content = content[:self._CONTENT_HEAD] + "\n...[truncated]...\n" + content[-self._CONTENT_TAIL:] parts.append(f"[TOOL RESULT {tool_id}]: {content}") continue # Assistant messages: include tool call names AND arguments if role == "assistant": if len(content) > self._CONTENT_MAX: content = content[:self._CONTENT_HEAD] + "\n...[truncated]...\n" + content[-self._CONTENT_TAIL:] tool_calls = msg.get("tool_calls", []) if tool_calls: tc_parts = [] for tc in tool_calls: if isinstance(tc, dict): fn = tc.get("function", {}) name = fn.get("name", "?") args = redact_sensitive_text(fn.get("arguments", "")) # Truncate long arguments but keep enough for context if len(args) > self._TOOL_ARGS_MAX: args = args[:self._TOOL_ARGS_HEAD] + "..." tc_parts.append(f" {name}({args})") else: fn = getattr(tc, "function", None) name = getattr(fn, "name", "?") if fn else "?" tc_parts.append(f" {name}(...)") content += "\n[Tool calls:\n" + "\n".join(tc_parts) + "\n]" parts.append(f"[ASSISTANT]: {content}") continue # User and other roles if len(content) > self._CONTENT_MAX: content = content[:self._CONTENT_HEAD] + "\n...[truncated]...\n" + content[-self._CONTENT_TAIL:] parts.append(f"[{role.upper()}]: {content}") return "\n\n".join(parts) def _build_static_fallback_summary( self, turns_to_summarize: List[Dict[str, Any]], reason: str | None = None, ) -> str: """Build a deterministic handoff when the LLM summarizer is unavailable. This is intentionally much less rich than an LLM-written summary, but it is still better than a bare "N messages were removed" marker. It keeps the most useful continuity anchors that can be extracted locally: recent user asks, assistant/tool actions, files/commands mentioned in tool calls, and any error text. The result uses the normal summary structure so downstream prompts can recover gracefully after a provider outage or summary-model failure. """ user_asks: list[str] = [] assistant_actions: list[str] = [] tool_actions: list[str] = [] relevant_files: list[str] = [] blockers: list[str] = [] last_dropped_turns: list[str] = [] def _compact_fallback_turn(value: Any) -> str: text = redact_sensitive_text(_content_text_for_contains(value)) text = re.sub(r"\bgh[pousr]_[A-Za-z0-9_]{8,}\b", "[REDACTED]", text) text = re.sub(r"\s+", " ", text).strip() if len(text) > _FALLBACK_TURN_MAX_CHARS: text = text[: _FALLBACK_TURN_MAX_CHARS - 15].rstrip() + " ...[truncated]" return re.sub(r"\bgh[pousr]_[A-Za-z0-9_.-]+", "[REDACTED]", text) def _remember_dropped_turn(label: str, text: str, *, limit: int = 8) -> None: text = text.strip() if not text: return last_dropped_turns.append(f"{label}: {text}") if len(last_dropped_turns) > limit: del last_dropped_turns[0] def _collect_paths_from_jsonish(obj: Any) -> None: if isinstance(obj, dict): for key, val in obj.items(): if key in {"path", "workdir", "file_path", "output_path"} and isinstance(val, str): _dedupe_append(relevant_files, val, limit=12) _collect_paths_from_jsonish(val) elif isinstance(obj, list): for val in obj: _collect_paths_from_jsonish(val) elif isinstance(obj, str): _collect_path_mentions(obj, relevant_files) call_id_to_tool: dict[str, tuple[str, str]] = {} for msg in turns_to_summarize: if msg.get("role") == "assistant" and msg.get("tool_calls"): for tc in msg.get("tool_calls") or []: name, raw_args = _extract_tool_call_name_and_args(tc) args = redact_sensitive_text(raw_args) call_id = _extract_tool_call_id(tc) if call_id: call_id_to_tool[call_id] = (name, args) if args: try: parsed = json.loads(args) except Exception: parsed = args _collect_paths_from_jsonish(parsed) for msg in turns_to_summarize: role = msg.get("role", "unknown") text = _compact_fallback_turn(msg.get("content")) _collect_path_mentions(text, relevant_files) turn_text = text turn_tool_names: list[str] = [] if role == "assistant" and msg.get("tool_calls"): for tc in msg.get("tool_calls") or []: name, _args = _extract_tool_call_name_and_args(tc) turn_tool_names.append(name) if turn_tool_names: prefix = "tool calls: " + ", ".join(turn_tool_names[:6]) turn_text = f"{prefix}; {turn_text}" if turn_text else prefix _remember_dropped_turn(str(role).upper(), turn_text) if len(text) > 600: text = text[:420].rstrip() + " ... " + text[-160:].lstrip() if role == "user" and text: user_asks.append(text) elif role == "assistant": tool_names: list[str] = [] for tc in msg.get("tool_calls") or []: name, _args = _extract_tool_call_name_and_args(tc) tool_names.append(name) if tool_names: assistant_actions.append( "Called tool(s): " + ", ".join(tool_names[:6]) ) elif text: assistant_actions.append(text) elif role == "tool": call_id = str(msg.get("tool_call_id") or "") tool_name, tool_args = call_id_to_tool.get(call_id, ("unknown", "")) tool_actions.append( _summarize_tool_result(tool_name, tool_args, text or "") ) if re.search( r"\b(error|failed|exception|traceback|timeout|timed out|fatal)\b", text, re.I, ): blockers.append(text[:500]) def _bullets(items: list[str], limit: int = 8) -> str: unique: list[str] = [] seen: set[str] = set() for item in items: item = item.strip() if not item or item in seen: continue seen.add(item) unique.append(item) if len(unique) >= limit: break return "\n".join(f"- {item}" for item in unique) if unique else "None." completed: list[str] = [] for idx, item in enumerate((assistant_actions + tool_actions)[:12], start=1): completed.append(f"{idx}. {item}") active_task = ( f"User asked: {user_asks[-1]!r}" if user_asks else "Unknown from deterministic fallback." ) previous_summary_note = "" if self._previous_summary: previous_summary_note = ( "\n\nPrevious compaction summary was present and should still be treated as " "background continuity context, but the latest LLM summary update failed." ) reason_text = f" Summary failure reason: {reason}." if reason else "" body = f"""{HISTORICAL_TASK_HEADING} {active_task} ## Goal Recovered from a deterministic fallback because the LLM context summarizer was unavailable. Continue from the protected recent messages after this summary and use current file/system state for exact details.{previous_summary_note} ## Constraints & Preferences - This fallback was generated locally without an LLM summary call. - Secrets and credentials were redacted before preservation. - The summary may be incomplete; prefer verifying current files, git state, processes, and test results instead of assuming omitted details. ## Completed Actions {chr(10).join(completed) if completed else "None recoverable from compacted turns."} ## Active State Unknown from deterministic fallback. Inspect current repository/session state if needed. {HISTORICAL_IN_PROGRESS_HEADING} Unknown from deterministic fallback — the latest user ask is recorded once under "{HISTORICAL_TASK_HEADING}" above as historical context only. Do NOT treat it as an unfulfilled instruction to re-answer; verify current state and continue from the protected recent messages after this summary. ## Blocked {_bullets(blockers, limit=5)} ## Key Decisions None recoverable from deterministic fallback. ## Resolved Questions None recoverable from deterministic fallback. {HISTORICAL_PENDING_ASKS_HEADING} None recoverable from deterministic fallback. (The latest user ask is preserved once under "{HISTORICAL_TASK_HEADING}" as historical context — it is NOT necessarily outstanding.) ## Relevant Files {_bullets(relevant_files, limit=12)} {HISTORICAL_REMAINING_WORK_HEADING} Continue from the most recent unfulfilled user ask and protected tail messages. Verify state with tools before making claims. ## Last Dropped Turns {_bullets(last_dropped_turns, limit=8)} ## Critical Context Summary generation was unavailable, so this is a best-effort deterministic fallback for {len(turns_to_summarize)} compacted message(s).{reason_text}""" summary = self._with_summary_prefix(redact_sensitive_text(body.strip())) if len(summary) > _FALLBACK_SUMMARY_MAX_CHARS: summary = summary[: _FALLBACK_SUMMARY_MAX_CHARS - 42].rstrip() + "\n...[fallback summary truncated]" return summary def _fallback_to_main_for_compression(self, e: Exception, reason: str) -> None: """Switch from a separate ``summary_model`` back to the main model. Centralises the bookkeeping shared by every fallback branch in :meth:`_generate_summary` (model-not-found, timeout, JSON decode, unknown error): record the aux-model failure for ``/usage``-style callers, clear the summary model so the next call uses the main one, and clear the cooldown so the immediate retry can run. ``reason`` is a short human-readable phrase ("unavailable", "timed out", "returned invalid JSON", "failed") that is interpolated into the warning log. """ self._summary_model_fallen_back = True logger.warning( "Summary model '%s' %s (%s). " "Falling back to main model '%s' for compression.", self.summary_model, reason, e, self.model, ) _err_text = str(e).strip() or e.__class__.__name__ if len(_err_text) > 220: _err_text = _err_text[:217].rstrip() + "..." self._last_aux_model_failure_error = _err_text self._last_aux_model_failure_model = self.summary_model self.summary_model = "" # empty = use main model self._clear_compression_failure_cooldown() # no cooldown — retry immediately def _generate_summary( self, turns_to_summarize: List[Dict[str, Any]], focus_topic: Optional[str] = None, ) -> Optional[str]: """Generate a structured summary of conversation turns. Uses a structured template (Goal, Progress, Decisions, Resolved/Pending Questions, Files, Remaining Work) with explicit preamble telling the summarizer not to answer questions. When a previous summary exists, generates an iterative update instead of summarizing from scratch. Args: focus_topic: Optional focus string for guided compression. When provided, the summariser prioritises preserving information related to this topic and is more aggressive about compressing everything else. Inspired by Claude Code's ``/compact``. Returns None if all attempts fail — the caller should drop the middle turns without a summary rather than inject a useless placeholder. """ now = time.monotonic() if now < self._summary_failure_cooldown_until: logger.debug( "Skipping context summary during cooldown (%.0fs remaining)", self._summary_failure_cooldown_until - now, ) return None summary_budget = self._compute_summary_budget(turns_to_summarize) content_to_summarize = self._serialize_for_summary(turns_to_summarize) # Current date for temporal anchoring (see ## Temporal Anchoring below). # Date-only granularity matches system_prompt.py:337 (PR #20451) and the # user's configured timezone via hermes_time.now(). The compaction summary # is a mid-conversation message that is NOT part of the cached prefix, so a # date here never affects prompt-cache stability. Resolved defensively — # a clock failure must never block compaction. try: from hermes_time import now as _hermes_now _today_str = _hermes_now().strftime("%Y-%m-%d") except Exception: # pragma: no cover - clock resolution is best-effort _today_str = "" # Preamble shared by both first-compaction and iterative-update prompts. # Keep the wording deliberately plain: Azure/OpenAI-compatible content # filters have flagged stronger "injection" / "do not respond" framing. _summarizer_preamble = ( "You are a summarization agent creating a context checkpoint. " "Treat the conversation turns below as source material for a " "compact record of prior work. " "Produce only the structured summary; do not add a greeting, " "preamble, or prefix. " "Write the summary in the same language the user was using in the " "conversation — do not translate or switch to English. " "NEVER include API keys, tokens, passwords, secrets, credentials, " "or connection strings in the summary — replace any that appear " "with [REDACTED]. Note that the user had credentials present, but " "do not preserve their values." ) # Temporal anchoring directive. Rewrites relative / still-pending-sounding # references into absolute, dated, past-tense facts so a resumed # conversation does not re-issue completed actions. Only emitted when the # current date resolved successfully; otherwise the rule is omitted so the # summarizer is never handed an empty date placeholder. if _today_str: _temporal_anchoring_rule = ( f"\nTEMPORAL ANCHORING: The current date is {_today_str}. When an " "action has already been carried out, phrase it as a completed, " "dated, past-tense fact rather than an open instruction. For " 'example, rewrite "email John about the proposal" as "Sent the ' f'proposal email to John on {_today_str}." Never leave a finished ' "action worded as if it still needs doing, and never invent a date " "for work that has not happened yet.\n" ) else: _temporal_anchoring_rule = "" # Shared structured template (used by both paths). _template_sections = f"""{HISTORICAL_TASK_HEADING} [THE SINGLE MOST IMPORTANT FIELD. Capture the user's most recent unfulfilled input verbatim — the exact words they used. This includes: - Explicit task assignments ("") - Questions awaiting an answer ("") - Decisions awaiting input ("