fix(compression): prevent stale-budget retry loops

This commit is contained in:
cucurigoo 2026-07-21 20:19:54 +00:00 committed by kshitij
parent 3e953ed815
commit 377244f7c8
8 changed files with 482 additions and 4 deletions

View file

@ -1757,8 +1757,9 @@ def init_agent(
)
_config_context_length = None
# Resolve custom_providers list once for reuse below (startup
# context-length override and plugin context-engine init).
# Resolve custom_providers once before route-scoping a global context pin:
# a named custom provider may keep its base URL only in this list rather
# than repeating it under ``model``.
try:
from hermes_cli.config import get_compatible_custom_providers
_custom_providers = get_compatible_custom_providers(_agent_cfg)
@ -1767,6 +1768,79 @@ def init_agent(
if not isinstance(_custom_providers, list):
_custom_providers = []
# ``model.context_length`` describes the configured default model. A
# process launched directly with ``--model`` / ``-m`` has already replaced
# ``agent.model`` before this initializer loads config, so carrying the
# default model's explicit window into that different runtime is stale. The
# live switch/fallback paths already clear this override; keep direct-start
# overrides consistent with them and let provider metadata resolve the
# active model's window instead.
if _config_context_length is not None and isinstance(_model_cfg, dict):
_configured_default_model = str(_model_cfg.get("default") or "").strip()
_configured_default_runtime_model = _configured_default_model
_active_runtime_model = agent.model
if _configured_default_model:
try:
from hermes_cli.model_normalize import normalize_model_for_provider
_configured_default_runtime_model = normalize_model_for_provider(
_configured_default_model, agent.provider
)
_active_runtime_model = normalize_model_for_provider(
agent.model, agent.provider
)
except Exception:
pass
_configured_provider = str(_model_cfg.get("provider") or "").strip()
_configured_base_url = str(_model_cfg.get("base_url") or "").rstrip("/")
if not _configured_base_url and _configured_provider.lower().startswith("custom:"):
_configured_custom_name = _configured_provider.split(":", 1)[1].lower()
for _provider_entry in _custom_providers:
if not isinstance(_provider_entry, dict):
continue
if str(_provider_entry.get("name") or "").strip().lower() != _configured_custom_name:
continue
_configured_base_url = str(
_provider_entry.get("base_url") or ""
).rstrip("/")
break
_active_base_url = str(agent.base_url or "").rstrip("/")
_route_mismatch = bool(
_configured_base_url
and _active_base_url
and _configured_base_url != _active_base_url
)
if not _configured_base_url:
_active_provider = str(agent.provider or "").strip()
try:
from hermes_cli.models import normalize_provider
_configured_provider = normalize_provider(_configured_provider)
_active_provider = normalize_provider(_active_provider)
except Exception:
_configured_provider = _configured_provider.lower()
_active_provider = _active_provider.lower()
_route_mismatch = bool(
_configured_provider
and _active_provider
and _configured_provider != _active_provider
)
_model_mismatch = bool(
_configured_default_runtime_model
and _configured_default_runtime_model != _active_runtime_model
)
if _model_mismatch or _route_mismatch:
_ra().logger.debug(
"Ignoring model.context_length=%s for startup runtime %s at %s "
"(configured default is %s at %s)",
_config_context_length,
agent.model,
_active_base_url or agent.provider,
_configured_default_model,
_configured_base_url or _model_cfg.get("provider"),
)
_config_context_length = None
# Store for reuse by _check_compression_model_feasibility (auxiliary
# compression model context-length detection needs the same list).
agent._custom_providers = _custom_providers

View file

@ -436,6 +436,19 @@ def check_compression_model_feasibility(agent: Any) -> None:
old_threshold = threshold
new_threshold = aux_context
agent.context_compressor.threshold_tokens = new_threshold
# ``tail_token_budget`` is derived from the trigger threshold, not
# directly from the model window. Keep it in lockstep with this
# just-in-time correction exactly as ContextCompressor.update_model()
# does. Leaving the old budget behind can make the tail's 1.5x soft
# ceiling wider than the lowered trigger, so compression preserves
# nearly the entire request and repeatedly re-fires.
summary_target_ratio = getattr(
agent.context_compressor, "summary_target_ratio", None
)
if isinstance(summary_target_ratio, (int, float)):
agent.context_compressor.tail_token_budget = int(
new_threshold * summary_target_ratio
)
# Keep threshold_percent in sync so future main-model
# context_length changes (update_model) re-derive from a
# sensible number rather than the original too-high value.

View file

@ -33,6 +33,7 @@ from agent.display import KawaiiSpinner
from agent.error_classifier import FailoverReason, classify_api_error
from agent.iteration_budget import IterationBudget
from agent.turn_context import (
_compression_warrants_another_preflight_pass,
build_turn_context,
compose_user_api_content,
reanchor_current_turn_user_idx,
@ -684,6 +685,8 @@ def run_conversation(
truncated_tool_call_retries = 0
truncated_response_parts: List[str] = []
compression_attempts = 0
_last_preflight_pressure: Optional[int] = None
_preflight_compression_blocked = _ctx.preflight_compression_blocked
_turn_exit_reason = "unknown" # Diagnostic: why the loop ended
# Last composed answer intentionally held back by a verification gate. If
# that continuation consumes the remaining budget, this is the best
@ -1125,6 +1128,37 @@ def run_conversation(
# LLM cooldown + anti-thrash guards (#11529). compression_attempts is a
# hard per-turn backstop shared with the overflow error handlers.
_compressor = agent.context_compressor
_preflight_threshold = int(
getattr(_compressor, "threshold_tokens", 0) or 0
)
# A previous mid-turn preflight pass deliberately continued the loop so
# API-only context and all sanitization could be rebuilt. Compare that
# fully assembled request with the fully assembled request that caused
# the pass. Raw ``messages`` are not equivalent here: they omit
# api_content/plugin injections, prefills, MoA context, and ephemeral
# system text.
_previous_preflight_pressure = _last_preflight_pressure
_last_preflight_pressure = None
if (
_previous_preflight_pressure is not None
and request_pressure_tokens >= _preflight_threshold
and not _compression_warrants_another_preflight_pass(
_previous_preflight_pressure,
request_pressure_tokens,
_preflight_threshold,
)
):
# Stop proactive retries for this turn without consuming the
# shared overflow-recovery budget. If the provider proves the
# request truly does not fit, its error handler may still compact
# with that stronger signal.
_preflight_compression_blocked = True
logger.warning(
"Pre-API compression made insufficient progress: ~%s -> "
"~%s request tokens; skipping additional preflight passes",
f"{_previous_preflight_pressure:,}",
f"{request_pressure_tokens:,}",
)
_defer_preflight = getattr(
_compressor, "should_defer_preflight_to_real_usage", lambda _t: False
)
@ -1135,6 +1169,7 @@ def run_conversation(
agent.compression_enabled
and len(messages) > 1
and compression_attempts < 3
and not _preflight_compression_blocked
and not _defer_preflight(request_pressure_tokens)
and not _compression_cooldown
and _compressor.should_compress(request_pressure_tokens)
@ -1153,6 +1188,7 @@ def run_conversation(
f"📦 Pre-API compression: ~{request_pressure_tokens:,} tokens "
f"near the context/output limit. Compacting before the next model call."
)
_last_preflight_pressure = request_pressure_tokens
messages, active_system_prompt = agent._compress_context(
messages,
system_message,

View file

@ -210,6 +210,23 @@ def _compression_made_progress(
return orig_tokens > 0 and new_tokens < orig_tokens * 0.95
def _compression_warrants_another_preflight_pass(
orig_tokens: int, new_tokens: int, threshold_tokens: int
) -> bool:
"""Whether an over-threshold request merits another immediate summary.
Row-count progress is enough to prove that a compression boundary was real,
but not enough to justify another expensive pass before trying the provider.
Continue only when the request remains over threshold *and* the previous pass
materially reduced its estimated token pressure (>5%).
"""
return (
new_tokens >= threshold_tokens
and orig_tokens > 0
and new_tokens < orig_tokens * 0.95
)
def _should_run_preflight_estimate(
messages: List[Dict[str, Any]],
protect_first_n: int,
@ -263,6 +280,8 @@ class TurnContext:
plugin_user_context: str = ""
# External-memory prefetch result, reused across loop iterations.
ext_prefetch_cache: str = ""
# Turn-start preflight already proved an immediate retry ineffective.
preflight_compression_blocked: bool = False
def build_turn_context(
@ -565,6 +584,7 @@ def build_turn_context(
# See ``_should_run_preflight_estimate`` for the OR semantics that fix
# issue #27405 (a few very large messages slipping past the count gate).
_preflight_compressed = False
_preflight_compression_blocked = False
if agent.compression_enabled and _should_run_preflight_estimate(
messages,
agent.context_compressor.protect_first_n,
@ -664,6 +684,7 @@ def build_turn_context(
if not _compression_made_progress(
_orig_len, len(messages), _orig_tokens, _preflight_tokens
):
_preflight_compression_blocked = True
break # Cannot compress further: neither rows nor tokens moved
conversation_history = conversation_history_after_compression(
agent, messages
@ -675,6 +696,19 @@ def build_turn_context(
agent._mute_post_response = False
if not _compressor.should_compress(_preflight_tokens):
break
if not _compression_warrants_another_preflight_pass(
_orig_tokens,
_preflight_tokens,
_compressor.threshold_tokens,
):
_preflight_compression_blocked = True
logger.warning(
"Preflight compression made insufficient progress: "
"~%s -> ~%s request tokens; skipping additional passes",
f"{_orig_tokens:,}",
f"{_preflight_tokens:,}",
)
break
if _preflight_compressed:
# Compression rebuilt the list (tail messages are fresh compaction
@ -899,4 +933,5 @@ def build_turn_context(
should_review_memory=should_review_memory,
plugin_user_context=plugin_user_context,
ext_prefetch_cache=ext_prefetch_cache,
preflight_compression_blocked=_preflight_compression_blocked,
)

View file

@ -14,7 +14,10 @@ progress.
from __future__ import annotations
from agent.turn_context import _compression_made_progress
from agent.turn_context import (
_compression_made_progress,
_compression_warrants_another_preflight_pass,
)
class TestCompressionMadeProgress:
@ -84,3 +87,26 @@ class TestCompressionMadeProgress:
assert _compression_made_progress(
orig_len=10, new_len=10, orig_tokens=0, new_tokens=0
) is False
class TestCompressionWarrantsAnotherPreflightPass:
def test_material_reduction_above_threshold_allows_another_pass(self):
assert _compression_warrants_another_preflight_pass(
orig_tokens=400_000,
new_tokens=350_000,
threshold_tokens=272_000,
) is True
def test_marginal_reduction_above_threshold_stops(self):
assert _compression_warrants_another_preflight_pass(
orig_tokens=350_000,
new_tokens=345_000,
threshold_tokens=272_000,
) is False
def test_clearing_threshold_needs_no_additional_pass(self):
assert _compression_warrants_another_preflight_pass(
orig_tokens=280_000,
new_tokens=250_000,
threshold_tokens=272_000,
) is False

View file

@ -987,9 +987,17 @@ class TestPreflightCompression:
with (
patch("agent.turn_context.estimate_request_tokens_rough", return_value=144_669),
patch("agent.conversation_loop.estimate_request_tokens_rough", return_value=144_669),
patch(
"agent.conversation_loop.estimate_messages_tokens_rough",
return_value=144_669,
),
# Compression no-ops (returns input unchanged) — mirrors an aux
# summary-model timeout where the messages can't be reduced.
patch.object(agent, "_compress_context", side_effect=lambda msgs, *a, **k: (msgs, agent._cached_system_prompt)),
patch.object(
agent,
"_compress_context",
side_effect=lambda msgs, *a, **k: (msgs, agent._cached_system_prompt),
) as mock_compress,
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
@ -997,6 +1005,10 @@ class TestPreflightCompression:
result = agent.run_conversation("hello", conversation_history=big_history)
assert result["completed"] is True
# A no-op pass cannot become more effective by immediately summarizing
# the same request twice more. Proceed to the provider/recovery path
# after one attempt instead of spending the full three-pass budget.
assert mock_compress.call_count == 1
# The display token count was revised up to the fresh preflight estimate,
# not left at the stale 74_400.
assert agent.context_compressor.last_prompt_tokens == 144_669
@ -1030,6 +1042,43 @@ class TestPreflightCompression:
# Smaller estimate must not overwrite the larger tracked value.
assert agent.context_compressor.last_prompt_tokens == 160_000
def test_preflight_stops_after_marginal_compression(self, agent):
"""Do not spend three summary calls removing one row per pass."""
agent.compression_enabled = True
agent.context_compressor.context_length = 200_000
agent.context_compressor.threshold_tokens = 130_000
big_history = []
for i in range(20):
big_history.append({"role": "user", "content": f"Message {i} padded text"})
big_history.append({"role": "assistant", "content": f"Response {i} padded text"})
ok_resp = _mock_response(content="After marginal preflight", finish_reason="stop")
agent.client.chat.completions.create.side_effect = [ok_resp]
def _drop_one_row(messages, *_args, **_kwargs):
return messages[:-1], agent._cached_system_prompt
with (
patch("agent.turn_context.estimate_request_tokens_rough", return_value=144_669),
patch("agent.conversation_loop.estimate_request_tokens_rough", return_value=144_669),
patch(
"agent.conversation_loop.estimate_messages_tokens_rough",
return_value=144_669,
),
patch.object(
agent, "_compress_context", side_effect=_drop_one_row
) as mock_compress,
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
result = agent.run_conversation("hello", conversation_history=big_history)
assert result["completed"] is True
assert result["final_response"] == "After marginal preflight"
assert mock_compress.call_count == 1
class TestToolResultPreflightCompression:
"""Compression should trigger when tool results push context past the threshold."""
@ -1072,6 +1121,59 @@ class TestToolResultPreflightCompression:
mock_compress.assert_called_once()
assert result["completed"] is True
def test_mid_turn_retry_compares_fully_assembled_requests(self, agent):
"""API-only context must not make marginal compression look effective."""
agent.compression_enabled = True
agent.context_compressor.context_length = 200_000
agent.context_compressor.threshold_tokens = 130_000
tc = SimpleNamespace(
id="tc1", type="function",
function=SimpleNamespace(name="web_search", arguments='{"query":"test"}'),
)
tool_resp = _mock_response(
content="",
finish_reason="stop",
tool_calls=[tc],
)
ok_resp = _mock_response(
content="Done after one compression", finish_reason="stop"
)
agent.client.chat.completions.create.side_effect = [tool_resp, ok_resp]
# First provider request is small. The tool result pushes the fully
# assembled request over threshold; rebuilding after compression only
# trims it from 150K to 148K. Raw-message estimation is much smaller,
# which previously made the no-op pass look successful and allowed two
# more immediate summaries.
assembled_estimates = iter(
[1_000, 150_000, 148_000, 148_000, 148_000]
)
with (
patch(
"agent.conversation_loop.estimate_messages_tokens_rough",
side_effect=lambda *_a, **_k: next(assembled_estimates),
),
patch("run_agent.handle_function_call", return_value="x" * 100_000),
patch.object(
agent,
"_compress_context",
side_effect=lambda msgs, *_a, **_k: (
msgs,
agent._cached_system_prompt,
),
) as mock_compress,
patch.object(agent, "_persist_session"),
patch.object(agent, "_save_trajectory"),
patch.object(agent, "_cleanup_task_resources"),
):
result = agent.run_conversation("hello")
assert result["completed"] is True
assert result["final_response"] == "Done after one compression"
assert mock_compress.call_count == 1
def test_anthropic_prompt_too_long_safety_net(self, agent):
"""Anthropic 'prompt is too long' error triggers compression as safety net."""
err_400 = Exception(

View file

@ -57,6 +57,10 @@ def _make_agent(
compressor = MagicMock(spec=ContextCompressor)
compressor.context_length = main_context
compressor.threshold_tokens = int(main_context * threshold_percent)
compressor.summary_target_ratio = 0.20
compressor.tail_token_budget = int(
compressor.threshold_tokens * compressor.summary_target_ratio
)
agent.context_compressor = compressor
return agent
@ -96,6 +100,11 @@ def test_auto_corrects_threshold_when_aux_context_below_threshold(mock_get_clien
assert agent._compression_warning is not None
# Threshold on the live compressor was actually lowered to aux_context.
assert agent.context_compressor.threshold_tokens == 80_000
# Every threshold-derived budget must move with it. Keeping the original
# 20K tail here would protect 25% of the lowered threshold instead of the
# configured 20%, and larger real-world mismatches can make the tail's 1.5x
# soft ceiling wider than the entire compression trigger.
assert agent.context_compressor.tail_token_budget == 16_000
@patch("agent.model_metadata.get_model_context_length", return_value=32_768)

View file

@ -6,6 +6,41 @@ from run_agent import AIAgent
from agent.context_compressor import ContextCompressor
class _StubStartupCompressor:
def __init__(self, *args, **kwargs):
self.context_length = kwargs.get("config_context_length") or 272_000
self.config_context_length = kwargs.get("config_context_length")
self.threshold_tokens = int(self.context_length * 0.95)
self.threshold_percent = 0.95
def get_tool_schemas(self):
return []
def on_session_start(self, *args, **kwargs):
return None
def _make_direct_start_agent(
cfg: dict, *, model: str, provider: str, base_url: str
) -> AIAgent:
with (
patch("hermes_cli.config.load_config", return_value=cfg),
patch("run_agent.get_tool_definitions", return_value=[]),
patch("run_agent.check_toolset_requirements", return_value={}),
patch("run_agent.OpenAI"),
patch("agent.agent_init.ContextCompressor", new=_StubStartupCompressor),
):
return AIAgent(
model=model,
provider=provider,
api_key="fake-test-token",
base_url=base_url,
quiet_mode=True,
skip_context_files=True,
skip_memory=True,
)
def _make_agent_with_compressor(config_context_length=None) -> AIAgent:
"""Build a minimal AIAgent with a context_compressor, skipping __init__."""
agent = AIAgent.__new__(AIAgent)
@ -73,3 +108,151 @@ def test_switch_model_without_config_context_length():
mock_ctx_len.assert_called_once()
call_kwargs = mock_ctx_len.call_args.kwargs
assert call_kwargs.get("config_context_length") is None
def test_direct_start_model_override_does_not_inherit_profile_context_length():
"""A CLI ``--model`` startup override must not inherit another model's window."""
cfg = {
"model": {
"default": "kimi-k3",
"provider": "custom:kimi-coding-1m",
"base_url": "https://api.kimi.com/coding",
"context_length": 1_048_576,
},
"custom_providers": [
{
"name": "kimi-coding-1m",
"base_url": "https://api.kimi.com/coding",
"models": {"kimi-k3": {"context_length": 1_048_576}},
}
],
}
agent = _make_direct_start_agent(
cfg,
model="gpt-5.6-sol",
provider="openai-codex",
base_url="https://chatgpt.com/backend-api/codex",
)
assert agent.context_compressor.config_context_length is None
assert agent.context_compressor.context_length == 272_000
def test_direct_start_preserves_context_for_normalized_default_model_alias():
"""Equivalent vendor-prefixed defaults still own their explicit window."""
cfg = {
"model": {
"default": "openai/gpt-5.6-sol",
"provider": "openai-codex",
"base_url": "https://chatgpt.com/backend-api/codex",
"context_length": 272_000,
}
}
agent = _make_direct_start_agent(
cfg,
model="gpt-5.6-sol",
provider="openai-codex",
base_url="https://chatgpt.com/backend-api/codex",
)
assert agent.context_compressor.config_context_length == 272_000
assert agent.context_compressor.context_length == 272_000
def test_direct_start_same_model_on_different_route_drops_context_override():
"""Context pins are route-specific even when the model slug is unchanged."""
cfg = {
"model": {
"default": "gpt-5.6-sol",
"provider": "custom:large-sol-route",
"base_url": "https://large-sol.example/v1",
"context_length": 1_048_576,
}
}
agent = _make_direct_start_agent(
cfg,
model="gpt-5.6-sol",
provider="openai-codex",
base_url="https://chatgpt.com/backend-api/codex",
)
assert agent.context_compressor.config_context_length is None
assert agent.context_compressor.context_length == 272_000
def test_direct_start_preserves_context_for_bare_aggregator_model():
"""Aggregator normalization must compare both sides, not rewrite one side."""
cfg = {
"model": {
"default": "gpt-5.4",
"provider": "openrouter",
"context_length": 1_000_000,
}
}
agent = _make_direct_start_agent(
cfg,
model="gpt-5.4",
provider="openrouter",
base_url="https://openrouter.ai/api/v1",
)
assert agent.context_compressor.config_context_length == 1_000_000
def test_direct_start_preserves_context_for_provider_alias():
"""Canonical provider aliases identify the same route when no URL is pinned."""
cfg = {
"model": {
"default": "gemini-2.5-pro",
"provider": "google",
"context_length": 1_000_000,
}
}
agent = _make_direct_start_agent(
cfg,
model="gemini-2.5-pro",
provider="gemini",
base_url="https://generativelanguage.googleapis.com/v1beta/openai",
)
assert agent.context_compressor.config_context_length == 1_000_000
def test_direct_start_named_custom_route_resolves_configured_base_url():
"""Named custom providers must not collapse to one generic custom route."""
cfg = {
"model": {
"default": "shared-model",
"provider": "custom:large-route",
"context_length": 1_048_576,
},
"custom_providers": [
{
"name": "large-route",
"base_url": "https://large.example/v1",
}
],
}
agent = _make_direct_start_agent(
cfg,
model="shared-model",
provider="custom",
base_url="https://small.example/v1",
)
assert agent.context_compressor.config_context_length is None
assert agent.context_compressor.context_length == 272_000
matching_agent = _make_direct_start_agent(
cfg,
model="shared-model",
provider="custom",
base_url="https://large.example/v1",
)
assert matching_agent.context_compressor.config_context_length == 1_048_576