From 9a5cf8354102b28e3c24bcb7fa3da17658f59fb8 Mon Sep 17 00:00:00 2001 From: webtecnica Date: Wed, 12 Aug 2026 16:51:43 -0300 Subject: [PATCH] fix(agent): propagate prompt-cache TTL to MoA/aux, clamp Qwen 1h, re-preflight on failover (#84733) --- agent/agent_runtime_helpers.py | 19 ++ agent/conversation_loop.py | 89 +++++++- agent/moa_loop.py | 42 +++- agent/prompt_caching.py | 36 +++ tests/agent/test_moa_slot_api_mode.py | 4 +- tests/agent/test_moa_slot_max_tokens.py | 6 +- .../test_prompt_cache_ttl_propagation.py | 207 ++++++++++++++++++ tests/agent/test_prompt_caching.py | 38 ++++ 8 files changed, 421 insertions(+), 20 deletions(-) create mode 100644 tests/agent/test_prompt_cache_ttl_propagation.py diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py index d7d5c39ebd427..4128da8a1b269 100644 --- a/agent/agent_runtime_helpers.py +++ b/agent/agent_runtime_helpers.py @@ -2015,6 +2015,8 @@ def plan_cache_sections_for_destination( api_mode: str, model: str, cache_disabled: Optional[bool] = None, + cache_ttl: Optional[str] = None, + static_system_prefix: Optional[str] = None, ) -> Tuple[list, list]: """Plan request-local cache sections for one resolved destination. @@ -2032,9 +2034,18 @@ def plan_cache_sections_for_destination( disable into the blank policy stub. When omitted, the live config is consulted so MoA/auxiliary paths cannot re-enable markers after the user turned caching off (#76085). + + ``cache_ttl`` threads the operator's configured tier (default ``5m``) + into the destination plan so MoA/auxiliary requests stop regressing to + the 5m default while the main loop honors ``1h`` (#84733); it is + clamped per-destination by :func:`effective_cache_ttl` (Qwen → 5m). + ``static_system_prefix`` threads the builder-declared stable prefix so + the destination system prompt receives the same early breakpoint the + main loop applies instead of marking the whole prompt as a breakpoint. """ from agent.prompt_caching import ( build_prompt_cache_plan, + effective_cache_ttl, strip_anthropic_cache_control, strip_anthropic_tool_cache_control, ) @@ -2054,7 +2065,15 @@ def plan_cache_sections_for_destination( plan = build_prompt_cache_plan( messages, tools, + cache_ttl=effective_cache_ttl( + cache_ttl or "5m", + provider=provider, + model=model, + ), native_anthropic=native_layout, + static_system_prefix=( + static_system_prefix if isinstance(static_system_prefix, str) else None + ), direct_native_tool_cache=_direct_native_anthropic_tool_cache_capability( stub, provider=provider, diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index 95d48421c12fc..3cdac5a1f2d28 100644 --- a/agent/conversation_loop.py +++ b/agent/conversation_loop.py @@ -76,6 +76,7 @@ from agent.model_metadata import ( from agent.process_bootstrap import _install_safe_stdio from agent.prompt_caching import ( build_prompt_cache_plan, + effective_cache_ttl, strip_anthropic_cache_control, strip_anthropic_tool_cache_control, ) @@ -1276,7 +1277,13 @@ def _redecorate_prompt_cache_for_provider( plan = build_prompt_cache_plan( messages, planned_tools, - cache_ttl=agent._cache_ttl, + # Clamp per-destination: a configured 1h regresses to 5m on + # Qwen/Alibaba routes, whose context cache is 5m-only (#84733). + cache_ttl=effective_cache_ttl( + agent._cache_ttl, + provider=agent.provider, + model=agent.model, + ), native_anthropic=agent._use_native_cache_layout, static_system_prefix=static if isinstance(static, str) else None, direct_native_tool_cache=direct_tool_cache, @@ -2119,7 +2126,13 @@ def run_conversation( _initial_cache_plan = build_prompt_cache_plan( api_messages, tools_for_api, - cache_ttl=agent._cache_ttl, + # Clamp per-destination: a configured 1h regresses to 5m on + # Qwen/Alibaba routes, whose context cache is 5m-only (#84733). + cache_ttl=effective_cache_ttl( + agent._cache_ttl, + provider=agent.provider, + model=agent.model, + ), native_anthropic=agent._use_native_cache_layout, static_system_prefix=( _static_system_prefix @@ -2458,7 +2471,13 @@ def run_conversation( retry_count = 0 compression_attempts = 0 _retry.primary_recovery_attempted = False - continue + # Failover shrank the compressor's context window to + # the fallback's; restart the outer iteration so the + # pre-API preflight re-runs against the new threshold + # before the first fallback call (#84733). + _preflight_compression_blocked = False + _retry.restart_with_rebuilt_messages = True + break # No fallback available — surface buffered context # so user sees the rate-limit message that led here. agent._flush_status_buffer() @@ -2910,7 +2929,13 @@ def run_conversation( retry_count = 0 compression_attempts = 0 _retry.primary_recovery_attempted = False - continue + # Failover shrank the compressor's context window to + # the fallback's; restart the outer iteration so the + # pre-API preflight re-runs against the new threshold + # before the first fallback call (#84733). + _preflight_compression_blocked = False + _retry.restart_with_rebuilt_messages = True + break # Check for error field in response (some providers include this) error_msg = "Unknown" @@ -2983,7 +3008,13 @@ def run_conversation( retry_count = 0 compression_attempts = 0 _retry.primary_recovery_attempted = False - continue + # Failover shrank the compressor's context window to + # the fallback's; restart the outer iteration so the + # pre-API preflight re-runs against the new threshold + # before the first fallback call (#84733). + _preflight_compression_blocked = False + _retry.restart_with_rebuilt_messages = True + break # Terminal — flush buffered retry trace so user sees what happened. agent._flush_status_buffer() agent._emit_status(f"❌ Max retries ({max_retries}) exceeded for invalid responses. Giving up.") @@ -3160,7 +3191,13 @@ def run_conversation( retry_count = 0 compression_attempts = 0 _retry.primary_recovery_attempted = False - continue + # Failover shrank the compressor's context window to + # the fallback's; restart the outer iteration so the + # pre-API preflight re-runs against the new threshold + # before the first fallback call (#84733). + _preflight_compression_blocked = False + _retry.restart_with_rebuilt_messages = True + break agent._flush_status_buffer() _refusal_log = ( @@ -4861,7 +4898,13 @@ def run_conversation( retry_count = 0 compression_attempts = 0 _retry.primary_recovery_attempted = False - continue + # Failover shrank the compressor's context window to + # the fallback's; restart the outer iteration so the + # pre-API preflight re-runs against the new threshold + # before the first fallback call (#84733). + _preflight_compression_blocked = False + _retry.restart_with_rebuilt_messages = True + break # ── Auth-failure provider failover ─────────────────────── # A 401/403 that survives the per-provider credential-refresh @@ -4894,7 +4937,13 @@ def run_conversation( retry_count = 0 compression_attempts = 0 _retry.primary_recovery_attempted = False - continue + # Failover shrank the compressor's context window to + # the fallback's; restart the outer iteration so the + # pre-API preflight re-runs against the new threshold + # before the first fallback call (#84733). + _preflight_compression_blocked = False + _retry.restart_with_rebuilt_messages = True + break # ── Nous Portal: record rate limit & skip retries ───── # When Nous returns a 429 that is a genuine account- @@ -5501,7 +5550,13 @@ def run_conversation( retry_count = 0 compression_attempts = 0 _retry.primary_recovery_attempted = False - continue + # Failover shrank the compressor's context window to + # the fallback's; restart the outer iteration so the + # pre-API preflight re-runs against the new threshold + # before the first fallback call (#84733). + _preflight_compression_blocked = False + _retry.restart_with_rebuilt_messages = True + break if api_kwargs is not None: agent._dump_api_request_debug( api_kwargs, reason="non_retryable_client_error", error=api_error, @@ -5724,7 +5779,13 @@ def run_conversation( retry_count = 0 compression_attempts = 0 _retry.primary_recovery_attempted = False - continue + # Failover shrank the compressor's context window to + # the fallback's; restart the outer iteration so the + # pre-API preflight re-runs against the new threshold + # before the first fallback call (#84733). + _preflight_compression_blocked = False + _retry.restart_with_rebuilt_messages = True + break # Terminal — flush buffered retry/fallback trace. agent._flush_status_buffer() _final_summary = agent._summarize_api_error(api_error) @@ -7247,7 +7308,13 @@ def run_conversation( "now using %s on %s", agent.model, agent.provider, ) - continue + # Failover shrank the compressor's context window to + # the fallback's; restart the outer iteration so the + # pre-API preflight re-runs against the new threshold + # before the first fallback call (#84733). + _preflight_compression_blocked = False + _retry.restart_with_rebuilt_messages = True + break # Exhausted retries and fallback chain (or no # fallback configured). Fall through to the diff --git a/agent/moa_loop.py b/agent/moa_loop.py index 0e15492a8fca1..a34808106ebdc 100644 --- a/agent/moa_loop.py +++ b/agent/moa_loop.py @@ -410,6 +410,7 @@ def _maybe_apply_moa_cache_control( runtime: dict[str, Any], *, cache_disabled: bool | None = None, + cache_ttl: str | None = None, ) -> list[dict[str, Any]]: """Decorate an advisor or aggregator request with cache_control when its route honors it. @@ -427,13 +428,20 @@ def _maybe_apply_moa_cache_control( ``cache_disabled`` (or the live config when omitted) is stamped onto the policy stub so ``prompt_caching.cache_ttl: off`` is not bypassed by the blank-agent pattern (#76085). + + ``cache_ttl`` threads the agent's configured tier (default ``5m``) so + advisor/synthesis requests stop regressing 1h → 5m; it is clamped + per-destination by :func:`effective_cache_ttl` (Qwen → 5m, #84733). """ try: from agent.agent_runtime_helpers import ( anthropic_prompt_cache_policy, blank_cache_policy_stub, ) - from agent.prompt_caching import apply_anthropic_cache_control + from agent.prompt_caching import ( + apply_anthropic_cache_control, + effective_cache_ttl, + ) # Prefer an explicit kwarg, then a snapshot on the runtime dict # (threaded from the live agent), else config via the stub factory. @@ -454,7 +462,13 @@ def _maybe_apply_moa_cache_control( if not should_cache: return messages return apply_anthropic_cache_control( - messages, native_anthropic=native_layout + messages, + cache_ttl=effective_cache_ttl( + cache_ttl or "5m", + provider=runtime.get("provider") or "", + model=runtime.get("model") or "", + ), + native_anthropic=native_layout, ) except Exception as exc: # pragma: no cover - decoration must never break a call logger.debug("MoA cache_control decoration skipped: %s", exc) @@ -470,6 +484,7 @@ def _run_reference( reference_timeout: float | None = None, context_length_cache: Any = None, cache_disabled: bool | None = None, + cache_ttl: str | None = None, ) -> tuple[str, str, Any]: """Call one reference model and return ``(label, text, accounting)``. @@ -536,7 +551,9 @@ def _run_reference( cache_runtime = runtime if cache_disabled is not None: cache_runtime = {**runtime, "_cache_disabled": cache_disabled} - messages = _maybe_apply_moa_cache_control(messages, cache_runtime) + messages = _maybe_apply_moa_cache_control( + messages, cache_runtime, cache_ttl=cache_ttl + ) # Per-slot max_tokens takes precedence over the preset-level # reference_max_tokens passed in by the caller. This lets each # reference model have its own output cap independently. @@ -843,6 +860,10 @@ def _run_references_parallel( cache_disabled = ( getattr(agent, "_cache_disabled", None) if agent is not None else None ) + # Thread the agent's configured cache TTL into every advisor request so + # the fan-out stops regressing 1h → 5m (#84733); the destination-aware + # clamp (Qwen → 5m) runs inside _maybe_apply_moa_cache_control. + cache_ttl = getattr(agent, "_cache_ttl", None) if agent is not None else None try: for idx, slot in enumerate(reference_models): if slot.get("provider") == "moa": @@ -862,6 +883,7 @@ def _run_references_parallel( reference_timeout=reference_timeout, context_length_cache=_ctx_len_cache, cache_disabled=cache_disabled, + cache_ttl=cache_ttl, ) ] = idx @@ -1322,6 +1344,9 @@ def aggregate_moa_context( **agg_runtime, "_cache_disabled": _agg_cache_disabled, } + # Thread the agent's configured cache TTL into the synthesis decoration + # so the one-shot /moa path stops regressing 1h → 5m (#84733). + _agg_cache_ttl = getattr(agent, "_cache_ttl", None) if agent is not None else None try: # Same cache_control decoration as _run_reference's advisor calls # (see _maybe_apply_moa_cache_control) — this synthesis call is a @@ -1334,7 +1359,9 @@ def aggregate_moa_context( # breakpoints, even when the resolved aggregator slot is a # cache-honoring route (e.g. Claude on OpenRouter/native Anthropic). agg_messages = _maybe_apply_moa_cache_control( - [{"role": "user", "content": synth_prompt}], agg_cache_runtime + [{"role": "user", "content": synth_prompt}], + agg_cache_runtime, + cache_ttl=_agg_cache_ttl, ) response = call_llm( task="moa_aggregator", @@ -1751,6 +1778,13 @@ class MoAChatCompletions: api_mode=agg_runtime.get("api_mode") or "", model=agg_runtime.get("model") or "", cache_disabled=_cache_disabled, + # Thread the agent's configured TTL and stable system prefix + # into the aggregator plan so MoA stops regressing 1h → 5m and + # marking the whole system prompt as one breakpoint (#84733). + cache_ttl=getattr(_agent, "_cache_ttl", None), + static_system_prefix=getattr( + _agent, "_cached_system_prompt_static", None + ), ) if guidance: _attach_reference_guidance(agg_messages, str(guidance)) diff --git a/agent/prompt_caching.py b/agent/prompt_caching.py index a9488345c6959..b56591ea5adfa 100644 --- a/agent/prompt_caching.py +++ b/agent/prompt_caching.py @@ -120,6 +120,42 @@ def _build_marker(ttl: str) -> Dict[str, str]: return marker +# Routes whose context cache documents a five-minute window (renewed on +# hit) and rejects the Anthropic 1h tier. Kept in parity with the +# alibaba-family set in agent_runtime_helpers.anthropic_prompt_cache_policy. +_QWEN_1H_UNSUPPORTED_PROVIDERS = frozenset({ + "opencode", + "opencode-zen", + "opencode-go", + "alibaba", +}) + + +def effective_cache_ttl( + ttl: str | None, + *, + model: str = "", + provider: str = "", +) -> str: + """Clamp a requested cache TTL to what the destination route supports. + + Qwen/Alibaba context caching documents an explicit five-minute window + (renewed on hit); the Anthropic ``1h`` tier is ignored/rejected there, + so a configured ``1h`` regresses to ``5m`` instead of shipping a marker + the provider drops and creating a false 1h-cache expectation (#84733). + All other caching routes keep the requested TTL. + + ``None`` (caching active with no explicit tier) resolves to ``5m``. + """ + if ttl != "1h": + return ttl or "5m" + if "qwen" in (model or "").lower(): + return "5m" + if (provider or "").lower() in _QWEN_1H_UNSUPPORTED_PROVIDERS: + return "5m" + return "1h" + + def _apply_system_cache_markers( message: dict, cache_marker: dict, diff --git a/tests/agent/test_moa_slot_api_mode.py b/tests/agent/test_moa_slot_api_mode.py index b709a4a2fcd6e..494891b2d8372 100644 --- a/tests/agent/test_moa_slot_api_mode.py +++ b/tests/agent/test_moa_slot_api_mode.py @@ -79,7 +79,7 @@ def test_run_reference_passes_slot_extra_body(monkeypatch): }, ) monkeypatch.setattr(moa_loop, "call_llm", fake_call_llm) - monkeypatch.setattr(moa_loop, "_maybe_apply_moa_cache_control", lambda messages, runtime: messages) + monkeypatch.setattr(moa_loop, "_maybe_apply_moa_cache_control", lambda messages, runtime, **kwargs: messages) label, text, _usage = moa_loop._run_reference( {"provider": "dashscope", "model": "qwen3.7-max"}, @@ -178,7 +178,7 @@ def test_one_shot_aggregate_moa_context_passes_slot_extra_body(monkeypatch): ) monkeypatch.setattr(moa_loop, "call_llm", fake_call_llm) monkeypatch.setattr( - moa_loop, "_maybe_apply_moa_cache_control", lambda messages, runtime: messages + moa_loop, "_maybe_apply_moa_cache_control", lambda messages, runtime, **kwargs: messages ) result = moa_loop.aggregate_moa_context( diff --git a/tests/agent/test_moa_slot_max_tokens.py b/tests/agent/test_moa_slot_max_tokens.py index 1b3a6813c3006..19127af0e8bcf 100644 --- a/tests/agent/test_moa_slot_max_tokens.py +++ b/tests/agent/test_moa_slot_max_tokens.py @@ -32,7 +32,7 @@ class TestRunReferenceSlotMaxTokens: with patch("agent.moa_loop._slot_runtime", return_value={"provider": "openrouter", "model": "deepseek/deepseek-v4-pro"}), \ patch("agent.moa_loop.call_llm", side_effect=fake_call_llm), \ - patch("agent.moa_loop._maybe_apply_moa_cache_control", side_effect=lambda msgs, rt: msgs): + patch("agent.moa_loop._maybe_apply_moa_cache_control", side_effect=lambda msgs, rt, **kwargs: msgs): _run_reference(slot, [{"role": "user", "content": "hi"}], max_tokens=2000) assert captured_kwargs.get("max_tokens") == 600 @@ -54,7 +54,7 @@ class TestRunReferenceSlotMaxTokens: with patch("agent.moa_loop._slot_runtime", return_value={"provider": "openrouter", "model": "deepseek/deepseek-v4-pro"}), \ patch("agent.moa_loop.call_llm", side_effect=fake_call_llm), \ - patch("agent.moa_loop._maybe_apply_moa_cache_control", side_effect=lambda msgs, rt: msgs): + patch("agent.moa_loop._maybe_apply_moa_cache_control", side_effect=lambda msgs, rt, **kwargs: msgs): _run_reference(slot, [{"role": "user", "content": "hi"}], max_tokens=2000) assert captured_kwargs.get("max_tokens") == 2000 @@ -76,7 +76,7 @@ class TestRunReferenceSlotMaxTokens: with patch("agent.moa_loop._slot_runtime", return_value={"provider": "openrouter", "model": "deepseek/deepseek-v4-pro"}), \ patch("agent.moa_loop.call_llm", side_effect=fake_call_llm), \ - patch("agent.moa_loop._maybe_apply_moa_cache_control", side_effect=lambda msgs, rt: msgs): + patch("agent.moa_loop._maybe_apply_moa_cache_control", side_effect=lambda msgs, rt, **kwargs: msgs): _run_reference(slot, [{"role": "user", "content": "hi"}], max_tokens=None) assert captured_kwargs.get("max_tokens") is None diff --git a/tests/agent/test_prompt_cache_ttl_propagation.py b/tests/agent/test_prompt_cache_ttl_propagation.py new file mode 100644 index 0000000000000..5a9c3e206501c --- /dev/null +++ b/tests/agent/test_prompt_cache_ttl_propagation.py @@ -0,0 +1,207 @@ +"""#84733: prompt-cache TTL/prefix propagation into MoA/aux paths + failover re-preflight. + +The main loop threads ``agent._cache_ttl`` and the stable system prefix into +``build_prompt_cache_plan``, but the MoA/aux helper only accepted +``cache_disabled`` — so a configured ``1h`` regressed to the 5m default and +the destination system prompt was marked as one whole breakpoint. These +tests pin the threaded parameters (TTL + static prefix) on +``plan_cache_sections_for_destination`` and the MoA decoration helper, the +per-destination Qwen clamp (1h -> 5m), and the failover re-preflight +contract (every fallback activation must restart the outer iteration so the +pre-API preflight re-runs against the fallback's context window). +""" + +import ast +import inspect + + +def _collect_cache_controls(obj): + """Return every ``cache_control`` marker dict reachable in ``obj``.""" + markers = [] + if isinstance(obj, dict): + if "cache_control" in obj: + markers.append(obj["cache_control"]) + for value in obj.values(): + markers.extend(_collect_cache_controls(value)) + elif isinstance(obj, list): + for value in obj: + markers.extend(_collect_cache_controls(value)) + return markers + + +class TestPlanCacheSectionsThreadsTtlAndPrefix: + def test_cache_ttl_1h_reaches_markers(self): + from agent.agent_runtime_helpers import plan_cache_sections_for_destination + + messages = [ + {"role": "system", "content": "You are helpful."}, + {"role": "user", "content": "hello"}, + ] + out_msgs, _ = plan_cache_sections_for_destination( + messages, + None, + provider="anthropic", + base_url="https://api.anthropic.com", + api_mode="anthropic_messages", + model="claude-opus-4.8", + cache_disabled=False, + cache_ttl="1h", + ) + markers = _collect_cache_controls(out_msgs) + assert markers, "expected cache_control markers on a caching route" + assert all(m.get("ttl") == "1h" for m in markers), ( + "the configured 1h tier must reach the destination plan markers" + ) + + def test_static_system_prefix_gets_early_breakpoint(self): + from agent.agent_runtime_helpers import plan_cache_sections_for_destination + + messages = [ + {"role": "system", "content": "stable prefix\nvolatile suffix"}, + {"role": "user", "content": "hello"}, + ] + out_msgs, _ = plan_cache_sections_for_destination( + messages, + None, + provider="anthropic", + base_url="https://api.anthropic.com", + api_mode="anthropic_messages", + model="claude-opus-4.8", + cache_disabled=False, + cache_ttl="5m", + static_system_prefix="stable prefix", + ) + system_content = out_msgs[0]["content"] + assert isinstance(system_content, list) and len(system_content) == 2, ( + "the destination system prompt must split into [static, volatile] " + "parts instead of marking the whole prompt as one breakpoint" + ) + assert system_content[0]["text"] == "stable prefix" + assert system_content[1]["text"] == "\nvolatile suffix" + + def test_qwen_1h_clamped_to_5m(self): + from agent.agent_runtime_helpers import plan_cache_sections_for_destination + + messages = [ + {"role": "system", "content": "You are helpful."}, + {"role": "user", "content": "hello"}, + ] + out_msgs, _ = plan_cache_sections_for_destination( + messages, + None, + provider="opencode", + base_url="https://api.opencode.ai", + api_mode="chat_completions", + model="qwen3.6-plus", + cache_disabled=False, + cache_ttl="1h", + ) + markers = _collect_cache_controls(out_msgs) + assert markers, "opencode+qwen is a cache-honoring route" + assert all("ttl" not in m for m in markers), ( + "Qwen's 5-minute-only context cache must clamp a configured 1h" + ) + + +class TestMoACacheControlThreadsTtl: + def test_moa_decoration_uses_threaded_1h(self): + from agent.moa_loop import _maybe_apply_moa_cache_control + + messages = [ + {"role": "system", "content": "sys"}, + {"role": "user", "content": "q1"}, + {"role": "assistant", "content": "a1"}, + {"role": "user", "content": "q2"}, + ] + runtime = { + "provider": "anthropic", + "model": "claude-opus-4.8", + "base_url": "", + "api_mode": "anthropic_messages", + } + out = _maybe_apply_moa_cache_control( + messages, runtime, cache_disabled=False, cache_ttl="1h" + ) + markers = _collect_cache_controls(out) + assert markers, "expected MoA decoration on a caching route" + assert all(m.get("ttl") == "1h" for m in markers), ( + "the agent's 1h tier must stop regressing to 5m on MoA advisor calls" + ) + # Caller messages must stay undecorated. + assert not _collect_cache_controls(messages) + + def test_moa_qwen_1h_clamped_to_5m(self): + from agent.moa_loop import _maybe_apply_moa_cache_control + + messages = [ + {"role": "system", "content": "sys"}, + {"role": "user", "content": "q1"}, + ] + runtime = { + "provider": "opencode", + "model": "qwen3.6-plus", + "base_url": "", + "api_mode": "chat_completions", + } + out = _maybe_apply_moa_cache_control( + messages, runtime, cache_disabled=False, cache_ttl="1h" + ) + markers = _collect_cache_controls(out) + assert markers, "opencode+qwen is a cache-honoring MoA route" + assert all("ttl" not in m for m in markers), ( + "MoA decoration must clamp 1h to 5m on Qwen destinations" + ) + + def test_moa_decoration_defaults_to_5m_without_ttl(self): + from agent.moa_loop import _maybe_apply_moa_cache_control + + messages = [ + {"role": "system", "content": "sys"}, + {"role": "user", "content": "q1"}, + ] + runtime = { + "provider": "anthropic", + "model": "claude-opus-4.8", + "base_url": "", + "api_mode": "anthropic_messages", + } + out = _maybe_apply_moa_cache_control( + messages, runtime, cache_disabled=False + ) + markers = _collect_cache_controls(out) + assert markers + assert all("ttl" not in m for m in markers) + + +class TestFailoverRestartsPreflight: + """#84733: a fallback provider switch must re-run the pre-API preflight. + + ``_try_activate_fallback`` already shrinks the compressor's context + window to the fallback's; the old ``continue`` re-fired the request + without re-running the pre-API pressure check. Every activation site + must instead ``break`` to the ``restart_with_rebuilt_messages`` handler, + which restarts the outer iteration (and its preflight) via a budget + refund. Source-level guard: importing the module and parsing the + function is cheap, and the assertion encodes the bug class — a new + failover site added with ``continue`` fails here on purpose. + """ + + def test_every_fallback_activation_breaks_to_repreflight(self): + from agent import conversation_loop + + tree = ast.parse(inspect.getsource(conversation_loop.run_conversation)) + fallback_ifs = [ + node + for node in ast.walk(tree) + if isinstance(node, ast.If) + and isinstance(node.test, ast.Call) + and isinstance(node.test.func, ast.Attribute) + and node.test.func.attr == "_try_activate_fallback" + ] + assert fallback_ifs, "expected _try_activate_fallback sites in run_conversation" + for node in fallback_ifs: + assert any(isinstance(stmt, ast.Break) for stmt in node.body), ( + "fallback activation must break to the restart-with-rebuilt-" + "messages handler so the pre-API preflight re-runs against " + "the fallback's context window (#84733)" + ) diff --git a/tests/agent/test_prompt_caching.py b/tests/agent/test_prompt_caching.py index 72067935e454b..59a253edbae1d 100644 --- a/tests/agent/test_prompt_caching.py +++ b/tests/agent/test_prompt_caching.py @@ -7,6 +7,7 @@ from agent.prompt_caching import ( _can_carry_marker, apply_anthropic_cache_control, build_prompt_cache_plan, + effective_cache_ttl, strip_anthropic_cache_control, strip_anthropic_tool_cache_control, ) @@ -465,5 +466,42 @@ class TestStripAnthropicCacheControl: assert content[1]["type"] == "image_url" +class TestEffectiveCacheTtl: + """#84733: Qwen/Alibaba routes document a 5-minute context cache only. + + ``effective_cache_ttl`` clamps a requested ``1h`` tier down to ``5m`` on + those routes so the marker the provider would ignore/reject is never + shipped and no false 1h-cache expectation survives. + """ + + def test_none_resolves_to_default_5m(self): + assert effective_cache_ttl(None) == "5m" + assert effective_cache_ttl(None, provider="anthropic", model="claude-x") == "5m" + + def test_5m_passthrough_everywhere(self): + assert effective_cache_ttl("5m") == "5m" + assert effective_cache_ttl("5m", provider="opencode", model="qwen3.6-plus") == "5m" + + def test_1h_preserved_on_non_qwen_routes(self): + assert effective_cache_ttl("1h", provider="anthropic", model="claude-opus-4.8") == "1h" + assert effective_cache_ttl("1h", provider="openrouter", model="claude-3-5-sonnet") == "1h" + assert effective_cache_ttl("1h", provider="", model="") == "1h" + + def test_1h_clamped_for_qwen_model_on_any_route(self): + assert effective_cache_ttl("1h", provider="openrouter", model="qwen3.6-plus") == "5m" + assert effective_cache_ttl("1h", provider="anthropic", model="Qwen-Max") == "5m" + + def test_1h_clamped_for_alibaba_family_providers(self): + for provider in ("opencode", "opencode-zen", "opencode-go", "alibaba"): + assert effective_cache_ttl("1h", provider=provider, model="qwen-max") == "5m", provider + assert effective_cache_ttl("1h", provider=provider.upper(), model="claude-x") == "5m", provider + + def test_marker_built_from_clamped_ttl_has_no_1h_key(self): + marker = _build_marker(effective_cache_ttl("1h", provider="opencode", model="qwen3.6-plus")) + assert marker == {"type": "ephemeral"} + marker = _build_marker(effective_cache_ttl("1h", provider="anthropic", model="claude-x")) + assert marker == {"type": "ephemeral", "ttl": "1h"} + +