RomanNum3ral
/
hermes-agent
mirror of https://github.com/NousResearch/hermes-agent.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix(agent): propagate prompt-cache TTL to MoA/aux, clamp Qwen 1h, re-preflight on failover (#84733)

This commit is contained in:
webtecnica 2026-08-12 16:51:43 -03:00 committed by kshitij
parent 7060ac7bed
commit 9a5cf83541
8 changed files with 421 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
agent/agent_runtime_helpers.py
View File

@ -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,

89
agent/conversation_loop.py
View File

@ -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

42
agent/moa_loop.py
View File

@ -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))

36
agent/prompt_caching.py
View File

@ -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,

4
tests/agent/test_moa_slot_api_mode.py
View File

@ -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(

6
tests/agent/test_moa_slot_max_tokens.py
View File

@ -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

207
tests/agent/test_prompt_cache_ttl_propagation.py Normal file
View File

@ -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)"
)

38
tests/agent/test_prompt_caching.py
View File

@ -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"}