diff --git a/agent/context_compressor.py b/agent/context_compressor.py index 069a5b65e15e..faffa59605c5 100644 --- a/agent/context_compressor.py +++ b/agent/context_compressor.py @@ -38,10 +38,15 @@ # Minimum tokens for the summary output _MIN_SUMMARY_TOKENS = 2000 +# Lower floor for the more aggressive repeated-compaction handoff. +_MIN_HEAVY_SUMMARY_TOKENS = 1000 # Proportion of compressed content to allocate for summary _SUMMARY_RATIO = 0.20 +_HEAVY_SUMMARY_RATIO = 0.12 # Absolute ceiling for summary tokens (even on very large context windows) _SUMMARY_TOKENS_CEILING = 12_000 +_HEAVY_REPEAT_THRESHOLD = 2 +_HEAVY_SEVERE_PRESSURE_RATIO = 1.35 # Placeholder used when pruning old tool results _PRUNED_TOOL_PLACEHOLDER = "[Old tool output cleared to save context space]" @@ -95,6 +100,7 @@ def __init__( threshold_percent: float = 0.50, protect_first_n: int = 3, protect_last_n: int = 20, + protect_recent_n: int = 3, summary_target_ratio: float = 0.20, quiet_mode: bool = False, summary_model_override: str = None, @@ -110,6 +116,7 @@ def __init__( self.threshold_percent = threshold_percent self.protect_first_n = protect_first_n self.protect_last_n = protect_last_n + self.protect_recent_n = max(3, protect_recent_n) self.summary_target_ratio = max(0.10, min(summary_target_ratio, 0.80)) self.quiet_mode = quiet_mode @@ -191,12 +198,7 @@ def _prune_old_tool_results( min_protect = min(protect_tail_count, len(result) - 1) for i in range(len(result) - 1, -1, -1): msg = result[i] - content_len = len(msg.get("content") or "") - msg_tokens = content_len // _CHARS_PER_TOKEN + 10 - for tc in msg.get("tool_calls") or []: - if isinstance(tc, dict): - args = tc.get("function", {}).get("arguments", "") - msg_tokens += len(args) // _CHARS_PER_TOKEN + msg_tokens = self._estimate_message_tokens(msg) if accumulated + msg_tokens > protect_tail_tokens and (len(result) - i) >= min_protect: boundary = i break @@ -224,7 +226,9 @@ def _prune_old_tool_results( # Summarization # ------------------------------------------------------------------ - def _compute_summary_budget(self, turns_to_summarize: List[Dict[str, Any]]) -> int: + def _compute_summary_budget( + self, turns_to_summarize: List[Dict[str, Any]], *, heavy_mode: bool = False, + ) -> int: """Scale summary token budget with the amount of content being compressed. The maximum scales with the model's context window (5% of context, @@ -232,11 +236,26 @@ def _compute_summary_budget(self, turns_to_summarize: List[Dict[str, Any]]) -> i richer summaries instead of being hard-capped at 8K tokens. """ content_tokens = estimate_messages_tokens_rough(turns_to_summarize) + if heavy_mode: + budget = int(content_tokens * _HEAVY_SUMMARY_RATIO) + max_budget = max(_MIN_HEAVY_SUMMARY_TOKENS, int(self.max_summary_tokens * 0.60)) + return max(_MIN_HEAVY_SUMMARY_TOKENS, min(budget, max_budget)) 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* + def _should_use_heavy_compaction(self, current_tokens: int | None = None) -> bool: + """Escalate to a shorter Layer-3 handoff after repeated or severe pressure.""" + if self.compression_count >= _HEAVY_REPEAT_THRESHOLD: + return True + if current_tokens is None or self.threshold_tokens <= 0: + return False + severe_threshold = min( + int(self.context_length * 0.90), + int(self.threshold_tokens * _HEAVY_SEVERE_PRESSURE_RATIO), + ) + return current_tokens >= severe_threshold + + # Truncation limits for summarizer input. The budget is for 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 @@ -244,6 +263,38 @@ def _compute_summary_budget(self, turns_to_summarize: List[Dict[str, Any]]) -> i _TOOL_ARGS_MAX = 1500 # tool call argument chars _TOOL_ARGS_HEAD = 1200 # kept from the start of tool args + def _truncate_summary_text( + self, + text: str, + *, + limit: int = _CONTENT_MAX, + head: int = _CONTENT_HEAD, + tail: int = _CONTENT_TAIL, + ) -> str: + """Keep enough leading and trailing text for summary generation.""" + if len(text) <= limit: + return text + return text[:head] + "\n...[truncated]...\n" + text[-tail:] + + @staticmethod + def _estimate_tool_call_tokens(tool_calls: list[Any]) -> int: + """Estimate the serialized size of tool-call arguments.""" + tokens = 0 + for tc in tool_calls or []: + if isinstance(tc, dict): + args = tc.get("function", {}).get("arguments", "") + tokens += len(args) // _CHARS_PER_TOKEN + return tokens + + def _estimate_message_tokens(self, msg: Dict[str, Any]) -> int: + """Estimate message size for pruning and tail-budget decisions.""" + content = msg.get("content") or "" + return ( + len(content) // _CHARS_PER_TOKEN + + 10 + + self._estimate_tool_call_tokens(msg.get("tool_calls")) + ) + def _serialize_for_summary(self, turns: List[Dict[str, Any]]) -> str: """Serialize conversation turns into labeled text for the summarizer. @@ -259,15 +310,13 @@ def _serialize_for_summary(self, turns: List[Dict[str, Any]]) -> str: # 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:] + content = self._truncate_summary_text(content) 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:] + content = self._truncate_summary_text(content) tool_calls = msg.get("tool_calls", []) if tool_calls: tc_parts = [] @@ -276,7 +325,6 @@ def _serialize_for_summary(self, turns: List[Dict[str, Any]]) -> str: fn = tc.get("function", {}) name = fn.get("name", "?") args = 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})") @@ -289,13 +337,14 @@ def _serialize_for_summary(self, turns: List[Dict[str, Any]]) -> str: continue # User and other roles - if len(content) > self._CONTENT_MAX: - content = content[:self._CONTENT_HEAD] + "\n...[truncated]...\n" + content[-self._CONTENT_TAIL:] + content = self._truncate_summary_text(content) parts.append(f"[{role.upper()}]: {content}") return "\n\n".join(parts) - def _generate_summary(self, turns_to_summarize: List[Dict[str, Any]]) -> Optional[str]: + def _generate_summary( + self, turns_to_summarize: List[Dict[str, Any]], *, heavy_mode: bool = False, + ) -> Optional[str]: """Generate a structured summary of conversation turns. Uses a structured template (Goal, Progress, Decisions, Files, Next Steps) @@ -314,10 +363,86 @@ def _generate_summary(self, turns_to_summarize: List[Dict[str, Any]]) -> Optiona ) return None - summary_budget = self._compute_summary_budget(turns_to_summarize) + summary_budget = self._compute_summary_budget( + turns_to_summarize, + heavy_mode=heavy_mode, + ) content_to_summarize = self._serialize_for_summary(turns_to_summarize) - if self._previous_summary: + if heavy_mode and self._previous_summary: + prompt = f"""Create a shorter, higher-level handoff summary for a later assistant. This conversation has already been compacted multiple times, so compress aggressively. + +PREVIOUS SUMMARY: +{self._previous_summary} + +NEW TURNS TO INCORPORATE: +{content_to_summarize} + +Rewrite everything into a concise durable handoff. Keep only the current goal, current state, constraints, key decisions, important files, open risks, next steps, and critical values. Do NOT preserve turn-by-turn chronology, exhaustive tool logs, or transcript-like detail unless it is necessary to continue correctly. + +Use this exact structure: + +## Goal +[What the user is trying to accomplish now] + +## Current State +[Current implementation/work status after the repeated compactions] + +## Constraints & Preferences +[Durable user constraints, preferences, and non-negotiables] + +## Key Decisions +[Important technical decisions and why they matter] + +## Relevant Files +[Only the files that still matter, with brief notes] + +## Open Risks +[Active risks, blockers, or uncertainties] + +## Next Steps +[What should happen next] + +## Critical Data +[Specific values, error messages, commands, IDs, or config details that must not be lost] + +Target ~{summary_budget} tokens. Prefer durable state over chronology. Write only the summary body. Do not include any preamble or prefix.""" + elif heavy_mode: + prompt = f"""Create a shorter, higher-level handoff summary for a later assistant. This conversation is being compacted aggressively to recover context space. + +TURNS TO SUMMARIZE: +{content_to_summarize} + +Keep only durable information needed to continue correctly. Do NOT preserve turn-by-turn chronology, exhaustive tool logs, or transcript-like detail unless it is necessary to continue the work safely. + +Use this exact structure: + +## Goal +[What the user is trying to accomplish now] + +## Current State +[Current implementation/work status] + +## Constraints & Preferences +[Durable user constraints, preferences, and non-negotiables] + +## Key Decisions +[Important technical decisions and why they matter] + +## Relevant Files +[Only the files that still matter, with brief notes] + +## Open Risks +[Active risks, blockers, or uncertainties] + +## Next Steps +[What should happen next] + +## Critical Data +[Specific values, error messages, commands, IDs, or config details that must not be lost] + +Target ~{summary_budget} tokens. Prefer durable state over chronology. Write only the summary body. Do not include any preamble or prefix.""" + elif self._previous_summary: # Iterative update: preserve existing info, add new progress prompt = f"""You are updating a context compaction summary. A previous compaction produced the summary below. New conversation turns have occurred since then and need to be incorporated. @@ -569,32 +694,27 @@ def _find_tail_cut_by_tokens( derived from ``summary_target_ratio * context_length``, so it scales automatically with the model's context window. - Token budget is the primary criterion. A hard minimum of 3 messages - is always protected, but the budget is allowed to exceed by up to - 1.5x to avoid cutting inside an oversized message (tool output, file - read, etc.). If even the minimum 3 messages exceed 1.5x the budget - the cut is placed right after the head so compression still runs. + Token budget is the primary criterion. A hard minimum of + ``self.protect_recent_n`` recent messages is always protected, but the + budget is allowed to exceed by up to 1.5x to avoid cutting inside an + oversized message (tool output, file read, etc.). If even that minimum + tail exceeds 1.5x the budget the cut is placed right after the head so + compression still runs. Never cuts inside a tool_call/result group. """ if token_budget is None: token_budget = self.tail_token_budget n = len(messages) - # Hard minimum: always keep at least 3 messages in the tail - min_tail = min(3, n - head_end - 1) if n - head_end > 1 else 0 + # Hard minimum: always keep at least protect_recent_n messages in the tail + min_tail = min(self.protect_recent_n, n - head_end - 1) if n - head_end > 1 else 0 soft_ceiling = int(token_budget * 1.5) accumulated = 0 cut_idx = n # start from beyond the end for i in range(n - 1, head_end - 1, -1): msg = messages[i] - content = msg.get("content") or "" - msg_tokens = len(content) // _CHARS_PER_TOKEN + 10 # +10 for role/metadata - # Include tool call arguments in estimate - for tc in msg.get("tool_calls") or []: - if isinstance(tc, dict): - args = tc.get("function", {}).get("arguments", "") - msg_tokens += len(args) // _CHARS_PER_TOKEN + msg_tokens = self._estimate_message_tokens(msg) # Stop once we exceed the soft ceiling (unless we haven't hit min_tail yet) if accumulated + msg_tokens > soft_ceiling and (n - i) >= min_tail: break @@ -634,8 +754,9 @@ def compress(self, messages: List[Dict[str, Any]], current_tokens: int = None) - up so the API never receives mismatched IDs. """ n_messages = len(messages) - # Only need head + 3 tail messages minimum (token budget decides the real tail size) - _min_for_compress = self.protect_first_n + 3 + 1 + # Only need head + the protected recent tail + 1 middle message. + # The token budget still decides the real tail size beyond this floor. + _min_for_compress = self.protect_first_n + self.protect_recent_n + 1 if n_messages <= _min_for_compress: if not self.quiet_mode: logger.warning( @@ -645,11 +766,15 @@ def compress(self, messages: List[Dict[str, Any]], current_tokens: int = None) - return messages display_tokens = current_tokens if current_tokens else self.last_prompt_tokens or estimate_messages_tokens_rough(messages) + heavy_mode = self._should_use_heavy_compaction(current_tokens=display_tokens) + prune_tail_count = self.protect_recent_n if heavy_mode else self.protect_last_n + prune_tail_tokens = None if heavy_mode else self.tail_token_budget + tail_token_budget = 0 if heavy_mode else self.tail_token_budget # Phase 1: Prune old tool results (cheap, no LLM call) messages, pruned_count = self._prune_old_tool_results( - messages, protect_tail_count=self.protect_last_n, - protect_tail_tokens=self.tail_token_budget, + messages, protect_tail_count=prune_tail_count, + protect_tail_tokens=prune_tail_tokens, ) if pruned_count and not self.quiet_mode: logger.info("Pre-compression: pruned %d old tool result(s)", pruned_count) @@ -659,7 +784,11 @@ def compress(self, messages: List[Dict[str, Any]], current_tokens: int = None) - compress_start = self._align_boundary_forward(messages, compress_start) # Use token-budget tail protection instead of fixed message count - compress_end = self._find_tail_cut_by_tokens(messages, compress_start) + compress_end = self._find_tail_cut_by_tokens( + messages, + compress_start, + token_budget=tail_token_budget, + ) if compress_start >= compress_end: return messages @@ -680,16 +809,17 @@ def compress(self, messages: List[Dict[str, Any]], current_tokens: int = None) - ) tail_msgs = n_messages - compress_end logger.info( - "Summarizing turns %d-%d (%d turns), protecting %d head + %d tail messages", + "Summarizing turns %d-%d (%d turns), protecting %d head + %d tail messages%s", compress_start + 1, compress_end, len(turns_to_summarize), compress_start, tail_msgs, + " [heavy]" if heavy_mode else "", ) # Phase 3: Generate structured summary - summary = self._generate_summary(turns_to_summarize) + summary = self._generate_summary(turns_to_summarize, heavy_mode=heavy_mode) # Phase 4: Assemble compressed message list compressed = [] diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 89606edc2e36..79f6441370b6 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -357,7 +357,8 @@ def _ensure_hermes_home_managed(home: Path): "enabled": True, "threshold": 0.50, # compress when context usage exceeds this ratio "target_ratio": 0.20, # fraction of threshold to preserve as recent tail - "protect_last_n": 20, # minimum recent messages to keep uncompressed + "protect_last_n": 20, # minimum recent messages protected from tool-result pruning + "protect_recent_n": 5, # minimum newest messages kept verbatim during compaction "summary_model": "", # empty = use main configured model "summary_provider": "auto", "summary_base_url": None, @@ -2629,7 +2630,8 @@ def show_config(): if enabled: print(f" Threshold: {compression.get('threshold', 0.50) * 100:.0f}%") print(f" Target ratio: {compression.get('target_ratio', 0.20) * 100:.0f}% of threshold preserved") - print(f" Protect last: {compression.get('protect_last_n', 20)} messages") + print(f" Protect last: {compression.get('protect_last_n', 20)} messages (tool pruning floor)") + print(f" Protect recent: {compression.get('protect_recent_n', 5)} messages (kept verbatim)") _sm = compression.get('summary_model', '') or '(main model)' print(f" Model: {_sm}") comp_provider = compression.get('summary_provider', 'auto') diff --git a/run_agent.py b/run_agent.py index aef1a3b151e1..668750387eca 100644 --- a/run_agent.py +++ b/run_agent.py @@ -1231,6 +1231,7 @@ def __init__( compression_summary_model = _compression_cfg.get("summary_model") or None compression_target_ratio = float(_compression_cfg.get("target_ratio", 0.20)) compression_protect_last = int(_compression_cfg.get("protect_last_n", 20)) + compression_protect_recent = int(_compression_cfg.get("protect_recent_n", 5)) # Read explicit context_length override from model config _model_cfg = _agent_cfg.get("model", {}) @@ -1267,7 +1268,7 @@ def __init__( except (TypeError, ValueError): pass break - + # Select context engine: config-driven (like memory providers). # 1. Check config.yaml context.engine setting # 2. Check plugins/context_engine// directory (repo-shipped) @@ -1316,6 +1317,7 @@ def __init__( threshold_percent=compression_threshold, protect_first_n=3, protect_last_n=compression_protect_last, + protect_recent_n=compression_protect_recent, summary_target_ratio=compression_target_ratio, summary_model_override=compression_summary_model, quiet_mode=self.quiet_mode, @@ -6367,6 +6369,58 @@ def flush_memories(self, messages: list = None, min_turns: int = None): if messages and messages[-1].get("_flush_sentinel") == _sentinel: messages.pop() + def _estimate_request_tokens(self, messages: list, system_prompt: str = "") -> int: + """Roughly estimate the next request size including tools and system prompt.""" + return estimate_request_tokens_rough( + messages, + system_prompt=system_prompt or "", + tools=self.tools or None, + ) + + def _micro_compact_messages(self, messages: list, *, reason: str) -> tuple[list, int]: + """Cheap Layer-1 compaction: prune older tool outputs before a request. + + This keeps assistant/user turns intact, preserves tool call/result pairing, + and respects both the pruning tail floor (protect_last_n) and the recent + verbatim-retention floor (protect_recent_n). + """ + if not messages or not self.compression_enabled: + return messages, 0 + + compressor = getattr(self, "context_compressor", None) + if compressor is None: + return messages, 0 + + protect_tail_count = max(compressor.protect_last_n, compressor.protect_recent_n) + compacted, pruned_count = compressor._prune_old_tool_results( + messages, + protect_tail_count=protect_tail_count, + protect_tail_tokens=compressor.tail_token_budget, + ) + if pruned_count: + logger.info( + "micro-compact (%s): pruned %d old tool result(s) " + "(tool floor=%d, recent floor=%d, tail budget=%d)", + reason, + pruned_count, + compressor.protect_last_n, + compressor.protect_recent_n, + compressor.tail_token_budget, + ) + return compacted, pruned_count + + def _prepare_messages_for_request( + self, + messages: list, + system_prompt: str = "", + *, + reason: str, + ) -> tuple[list, int, int]: + """Apply micro-compact and return the estimated next request size.""" + compacted, pruned_count = self._micro_compact_messages(messages, reason=reason) + estimated_tokens = self._estimate_request_tokens(compacted, system_prompt) + return compacted, pruned_count, estimated_tokens + def _compress_context(self, messages: list, system_message: str, *, approx_tokens: int = None, task_id: str = "default") -> tuple: """Compress conversation context and split the session in SQLite. @@ -7571,14 +7625,12 @@ def run_conversation( if ( self.compression_enabled and len(messages) > self.context_compressor.protect_first_n - + self.context_compressor.protect_last_n + 1 + + self.context_compressor.protect_recent_n + 1 ): - # Include tool schema tokens — with many tools these can add - # 20-30K+ tokens that the old sys+msg estimate missed entirely. - _preflight_tokens = estimate_request_tokens_rough( + messages, _preflight_pruned, _preflight_tokens = self._prepare_messages_for_request( messages, - system_prompt=active_system_prompt or "", - tools=self.tools or None, + active_system_prompt or "", + reason="preflight", ) if _preflight_tokens >= self.context_compressor.threshold_tokens: @@ -7610,11 +7662,11 @@ def run_conversation( # skipping them because conversation_history is still the # pre-compression length. conversation_history = None - # Re-estimate after compression - _preflight_tokens = estimate_request_tokens_rough( + # Re-run micro-compact + estimate after compression + messages, _preflight_pruned, _preflight_tokens = self._prepare_messages_for_request( messages, - system_prompt=active_system_prompt or "", - tools=self.tools or None, + active_system_prompt or "", + reason=f"preflight-pass-{_pass + 1}", ) if _preflight_tokens < self.context_compressor.threshold_tokens: break # Under threshold @@ -7741,6 +7793,11 @@ def run_conversation( # Note: Reasoning is embedded in content via tags for trajectory storage. # However, providers like Moonshot AI require a separate 'reasoning_content' field # on assistant messages with tool_calls. We handle both cases here. + messages, _turn_pruned, _ = self._prepare_messages_for_request( + messages, + active_system_prompt or "", + reason=f"loop-{api_call_count}", + ) api_messages = [] for idx, msg in enumerate(messages): api_msg = msg.copy() @@ -9517,7 +9574,10 @@ def _stop_spinner(): + _compressor.last_completion_tokens ) else: - _real_tokens = estimate_messages_tokens_rough(messages) + _real_tokens = self._estimate_request_tokens( + messages, + active_system_prompt or "", + ) # ── Context pressure warnings (user-facing only) ────────── # Notify the user (NOT the LLM) as context approaches the