test: pin the wire every provider receives, real error bodies, and the prompt-cache prefix contract

Three regression nets, each built from production evidence:

- tests/agent/transports/test_provider_wire_snapshot.py + tests/fixtures/provider_wire_snapshot.json:
  the request every registered provider is sent (51 providers x reasoning unset/high/off), built
  through the production assembly path (build_api_request: cache decoration, sanitization,
  transport kwargs, Codex preflight, middleware). A shared-code edit that changes what any
  provider receives fails naming the provider and the first diverging key; a new provider fails
  until its wire is recorded. Network and DNS are blocked so capability probes (Ollama
  /api/show, LM Studio, catalogs) record their offline answer, not a live server's.
  Mutation check: +1 on the shared timeout fails 38 providers.

- tests/agent/test_error_classifier_observed.py: error bodies Hermes actually received (relay
  traces and agent.log, scrubbed), raised as the real openai/httpx exception types. Two known
  gaps are run-time xfails that flip to passing when their fix lands: #49769 (OpenRouter
  "can only afford N" 402 classified as terminal billing) and #85005 ("not a multimodal model"
  not recognised as an image rejection).

- tests/agent/test_prompt_cache_prefix_contract.py: a real AIAgent on a caching route against
  the recording fake provider; every request must re-send the previous request's content and
  only append. Pins a live cache bust as a run-time xfail: the #81867 skill-scaffold split is
  applied only while the skill message is inside the last-N marker window, so request 3 of a
  skill session re-sends the scaffold as a plain string and re-writes it (~20k tokens on call 2
  of skill-invoked sessions in agent.log).

tests/agent/test_error_classifier_observed.py

@@ -0,0 +1,93 @@
+"""Provider error bodies observed in real sessions, through the real SDK exception types.
+
+Every row is a body Hermes actually received (relay traces + agent.log, scrubbed of request ids
+and account data), raised as the exception class the OpenAI SDK / httpx raises for it, so a
+classifier change is checked against what providers really send rather than hand-written
+strings. Add a row whenever a new production body shows up in a bug report or trace.
+
+The last two tests are bodies main still mishandles; each is a run-time xfail keyed on the current
+wrong outcome, so it turns into a passing regression test the moment its fix lands:
+
+* #49769 — OpenRouter's "can only afford N tokens" 402 is recoverable by lowering max_tokens,
+  but the main loop classifies it as terminal billing and abandons the provider.
+* #85005 — "<model> is not a multimodal model" (vLLM/text-only endpoints) is not recognised as
+  an image rejection, so the image is never stripped and every retry fails identically.
+"""
+
+from __future__ import annotations
+
+import httpx
+import openai
+import pytest
+
+from agent.error_classifier import FailoverReason, classify_api_error
+from agent.message_sanitization import _looks_like_image_content_rejection
+from tests.e2e.core._pending_fixes import known_failure
+
+_REQUEST = httpx.Request("POST", "https://example.invalid/v1/chat/completions")
+
+
+def _status(cls: type, code: int, body: dict) -> Exception:
+    return cls(f"Error code: {code} - {body}", response=httpx.Response(code, request=_REQUEST, json=body), body=body)
+
+
+AFFORD_402 = {"error": {"message": (
+    "This request requires more credits, or fewer max_tokens. You requested up to 128000 tokens, but can only "
+    "afford 81664. To increase, visit https://openrouter.ai/settings/credits and upgrade to a paid account"),
+    "code": 402}}
+NOT_MULTIMODAL_400 = {"error": {"message": "glm-5.2-fp8 is not a multimodal model", "type": "BadRequestError",
+                                "param": None, "code": 400}}
+
+# (id, provider, exception, expected reason, retryable, should_fallback)
+OBSERVED = [
+    ("connection-error", "custom", openai.APIConnectionError(request=_REQUEST), FailoverReason.timeout, True, False),
+    ("broken-pipe", "nous", httpx.ReadError("[Errno 32] Broken pipe", request=_REQUEST),
+     FailoverReason.timeout, True, False),
+    ("peer-closed-mid-body", "custom", httpx.RemoteProtocolError(
+        "peer closed connection without sending complete message body (incomplete chunked read)", request=_REQUEST),
+     FailoverReason.timeout, True, False),
+    ("request-timeout", "nous", openai.APITimeoutError(request=_REQUEST), FailoverReason.timeout, True, False),
+    ("nous-404-no-credits", "nous", _status(openai.NotFoundError, 404, {"status": 404, "message": (
+        "Model 'anthropic/claude-opus-5' requires available credits. Your account balance is too low to use paid "
+        "models — add credits at https://portal.nousresearch.com or pick a free model.")}),
+     FailoverReason.billing, False, True),
+    ("nous-401-invalid-key", "nous", _status(openai.AuthenticationError, 401, {"status": 401, "message": (
+        "Your API key is invalid, blocked or out of funds. Please go visit the portal to sort that out: "
+        "https://portal.nousresearch.com ")}),
+     FailoverReason.auth, False, True),
+    ("cloudflare-502", "nous", _status(openai.InternalServerError, 502, {
+        "type": "https://developers.cloudflare.com/support/troubleshooting/http-status-codes/cloudflare-5xx-errors/"
+                "error-502/", "title": "Error 502: Bad gateway", "status": 502,
+        "detail": "The origin web server returned an invalid or incomplete response to Cloudflare.",
+        "error_code": 502, "error_name": "origin_bad_gateway", "error_category": "origin"}),
+     FailoverReason.server_error, True, False),
+    ("nous-500-unexpected", "nous", _status(openai.InternalServerError, 500, {"status": 500, "message": (
+        "Something unexpected happened while processing your request. Please try again in a moment, or contact us "
+        "if the issue persists.")}),
+     FailoverReason.server_error, True, False),
+    ("router-no-workers", "custom", _status(openai.InternalServerError, 500, {"error": {
+        "message": "No available workers (all circuits open or unhealthy)"}}),
+     FailoverReason.server_error, True, False),
+]
+
+
+@pytest.mark.parametrize("provider,error,reason,retryable,fallback", [r[1:] for r in OBSERVED],
+                         ids=[r[0] for r in OBSERVED])
+def test_observed_body_routes_to_its_recovery(provider, error, reason, retryable, fallback):
+    result = classify_api_error(error, provider=provider, model="test-model")
+    assert (result.reason, result.retryable, result.should_fallback) == (reason, retryable, fallback)
+
+
+def test_affordable_402_is_retried_on_the_same_provider():
+    result = classify_api_error(_status(openai.APIStatusError, 402, AFFORD_402), provider="openrouter",
+                                model="anthropic/claude-opus-5.5")
+    outcome = f"reason={result.reason.name} retryable={result.retryable}"
+    with known_failure(r"reason=billing retryable=False", "#49769: affordable-tokens 402 treated as terminal billing"):
+        assert result.retryable and result.reason != FailoverReason.billing, (
+            f"402 'can only afford N' must be retried on this provider, not abandoned as billing: {outcome}")
+
+
+def test_not_a_multimodal_model_is_an_image_rejection():
+    body = str(_status(openai.BadRequestError, 400, NOT_MULTIMODAL_400))
+    with known_failure(r"not recognised", "#85005: text-only endpoint image rejection not recognised"):
+        assert _looks_like_image_content_rejection(body), f"image rejection not recognised: {body[:120]}"

tests/agent/test_prompt_cache_prefix_contract.py

@@ -0,0 +1,111 @@
+"""Prompt-cache prefix contract on caching routes, through a real AIAgent and a recording fake provider.
+
+Provider prompt caches key on the request content up to each ``cache_control`` breakpoint, so every
+request in a session must re-send the previous request's content unchanged and only append; a
+request that rewrites earlier content re-bills the whole prefix. Marker placement is NOT part of the
+key: the rolling breakpoint window moves every request, and a single text part with a marker is
+the same content as the plain string it decorates (production telemetry: plain sessions reuse the
+full previous prefix on 99.5% of consecutive calls). Splitting one string into several parts is a
+content change.
+
+The cells mirror what real sessions do on the two cache layouts Hermes emits (OpenAI-wire envelope
+markers: OpenRouter / Nous Portal / custom relays; native Anthropic Messages content-block markers)
+and were chosen from per-call cache telemetry (``agent.log`` ``cache=R/T write=W``): consecutive
+calls in one session reuse the whole previous prefix except where these cells say otherwise.
+
+Gap pinned here (run-time xfail, turns into a passing regression test when fixed): a skill turn's
+first user message is split into [marked scaffold, unmarked tail] only while it sits inside the
+last-N marker window (#81867). Once the tool round pushes it out, the same message is re-sent as a
+plain string, so the provider's cached scaffold bytes stop matching and request 3 re-writes the
+whole skill body — the ~20k-token re-write seen on call 2 of skill-invoked sessions.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from tests.e2e.core._pending_fixes import known_failure
+from tests.e2e.core.history._helpers import NO_BACKGROUND_REVIEW, OFFLINE_CONFIG, canon, prefix_breaks
+from tests.fakes.fake_llm_provider import FakeLLMServer, Text, ToolCall, write_hermes_home
+
+SKILL_BODY = "Follow these steps carefully. " * 300
+
+
+def _cache_content(value: Any) -> Any:
+    """What the cache keys on: markers dropped, a lone text part folded back to its string."""
+    if isinstance(value, list):
+        value = [_cache_content(v) for v in value]
+        if len(value) == 1 and isinstance(value[0], dict) and set(value[0]) == {"type", "text"}:
+            return value[0]["text"]
+        return value
+    if isinstance(value, dict):
+        return {k: _cache_content(v) for k, v in value.items() if k != "cache_control"}
+    return value
+
+
+def _prefix_breaks(requests: list[dict[str, Any]]) -> list[tuple[int, str]]:
+    """``prefix_breaks`` over what the cache keys on, after asserting markers reached the wire."""
+    assert any("cache_control" in canon(r) for r in requests), "caching route sent no cache_control markers"
+    return prefix_breaks([_cache_content(r) for r in requests])
+
+
+@pytest.fixture
+def caching_route():
+    """A custom OpenAI-wire relay declared ``prompt_caching: true`` -> envelope cache layout."""
+    home = Path(os.environ["HERMES_HOME"])
+    skill = home / "skills" / "demo"
+    skill.mkdir(parents=True)
+    (skill / "SKILL.md").write_text(f"---\nname: demo\ndescription: Demo skill.\n---\n\n# Demo\n\n{SKILL_BODY}\n")
+    srv = FakeLLMServer([]).__enter__()
+    route = (f"custom_providers:\n  - name: cacherelay\n    base_url: {srv.base_url}\n    api_key: sk-fake-e2e\n"
+             "    models:\n      fake-model:\n        prompt_caching: true\n")
+    write_hermes_home(home, srv.base_url, extra_config=OFFLINE_CONFIG + NO_BACKGROUND_REVIEW + route)
+    yield srv, home
+    srv.__exit__(None, None, None)
+
+
+def _agent(srv, home):
+    from hermes_state import SessionDB
+    from run_agent import AIAgent
+
+    agent = AIAgent(provider="custom:cacherelay", base_url=srv.base_url, api_key="sk-fake-e2e", model="fake-model",
+                    session_db=SessionDB(db_path=home / "state.db"), session_id="cache-contract", quiet_mode=True,
+                    platform="cli")
+    assert (agent._use_prompt_caching, agent._use_native_cache_layout) == (True, False)
+    return agent
+
+
+def test_tool_rounds_and_follow_up_turns_only_append(caching_route):
+    srv, home = caching_route
+    srv.push(ToolCall("terminal", {"command": "echo one"}), ToolCall("terminal", {"command": "echo two"}),
+             Text("first done"), ToolCall("terminal", {"command": "echo three"}), Text("second done"))
+    agent = _agent(srv, home)
+    history: list[dict[str, Any]] = []
+    for text in ("run two commands", "and one more"):
+        history = agent.run_conversation(text, conversation_history=history, task_id="t")["messages"]
+    agent.close()
+    main = srv.main_requests()
+    assert len(main) == 5
+    assert not _prefix_breaks(main), _prefix_breaks(main)
+
+
+def test_skill_turn_keeps_its_scaffold_bytes_across_the_tool_loop(caching_route):
+    from agent.skill_commands import build_skill_invocation_message, scan_skill_commands
+
+    srv, home = caching_route
+    srv.push(ToolCall("terminal", {"command": "echo one"}), ToolCall("terminal", {"command": "echo two"}),
+             Text("done"))
+    scan_skill_commands()
+    message = build_skill_invocation_message("/demo", "do the thing", task_id="t")
+    assert message and SKILL_BODY.strip() in message
+    agent = _agent(srv, home)
+    agent.run_conversation(message, conversation_history=[], task_id="t")
+    agent.close()
+    main = srv.main_requests()
+    assert len(main) == 3
+    with known_failure(r"messages\[1\] \(user\) changed", "#81867 split dropped outside the marker window"):
+        assert not _prefix_breaks(main), f"skill scaffold re-sent with different bytes: {_prefix_breaks(main)}"

tests/agent/transports/test_provider_wire_snapshot.py

@@ -0,0 +1,197 @@
+"""Golden wire snapshot: the exact request every registered provider is sent.
+
+Each cell builds one request through the production assembly path
+(``agent.turn_api_request.build_api_request``: cache decoration, sanitization, transport
+kwargs, Codex preflight, request middleware) for one provider x three reasoning settings,
+and compares it against ``tests/fixtures/provider_wire_snapshot.json``.
+
+A change to shared request code that alters what ANY provider receives fails here, naming
+the provider and the first diverging key, before a user on that provider finds out. A new
+provider fails ``test_snapshot_covers_every_provider`` until its wire is reviewed and recorded.
+
+Intentional wire changes: regenerate and review the JSON diff in the same PR::
+
+    HERMES_UPDATE_WIRE_SNAPSHOT=1 .venv/bin/python -m pytest tests/agent/transports/test_provider_wire_snapshot.py
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import socket
+from pathlib import Path
+from typing import Any
+from unittest.mock import patch
+
+import pytest
+
+SNAPSHOT = Path(__file__).resolve().parents[2] / "fixtures" / "provider_wire_snapshot.json"
+UPDATE = os.environ.get("HERMES_UPDATE_WIRE_SNAPSHOT") == "1"
+
+REASONING = {
+    "unset": None,
+    "high": {"enabled": True, "effort": "high"},
+    "off": {"enabled": False},
+}
+
+TOOLS = [{"type": "function", "function": {
+    "name": "terminal", "description": "Run a shell command.",
+    "parameters": {"type": "object", "properties": {"command": {"type": "string"}}, "required": ["command"]}}}]
+
+# One completed tool round, then a fresh user turn: role mapping, tool-call/result pairing
+# and every cache breakpoint slot.
+MESSAGES = [
+    {"role": "system", "content": "You are a test agent."},
+    {"role": "user", "content": "list files"},
+    {"role": "assistant", "content": "", "tool_calls": [{"id": "call_1", "type": "function", "function": {
+        "name": "terminal", "arguments": "{\"command\": \"ls\"}"}}]},
+    {"role": "tool", "tool_call_id": "call_1", "content": "a.txt\nb.txt"},
+    {"role": "user", "content": "thanks, now summarize"},
+]
+
+# A representative model per provider (else the profile's first curated model).
+MODELS = {
+    "nous": "anthropic/claude-opus-5.5", "openai-codex": "gpt-5.5", "openai-api": "gpt-5.5",
+    "xai-oauth": "grok-4", "qwen-oauth": "qwen3-coder-plus", "lmstudio": "qwen/qwen3-8b",
+    "copilot": "gpt-5.5", "gemini": "gemini-3-pro", "kimi-coding": "kimi-k3", "kimi-coding-cn": "kimi-k2.5",
+    "stepfun": "step-3", "arcee": "trinity-large", "actual": "actual-1", "minimax": "MiniMax-M3",
+    "minimax-oauth": "MiniMax-M3", "minimax-cn": "MiniMax-M3", "anthropic": "claude-opus-5.5",
+    "alibaba": "qwen3.5-plus", "alibaba-coding-plan": "qwen3-coder-plus", "xai": "grok-4",
+    "ai-gateway": "anthropic/claude-opus-5.5", "opencode-zen": "claude-opus-5.5", "opencode-go": "kimi-k2.5",
+    "kilocode": "anthropic/claude-opus-5.5", "xiaomi": "mimo-v2-pro", "tencent-tokenhub": "hunyuan-t1",
+    "tencent-tokenplan": "hunyuan-t1", "ollama-cloud": "gpt-oss:120b",
+    "bedrock": "us.anthropic.claude-opus-5-5-v1:0", "vertex": "gemini-3-pro", "azure-foundry": "gpt-5.5",
+    "alibaba-cn": "qwen3.5-plus", "alibaba-token-plan": "qwen3.5-plus", "alibaba-token-plan-cn": "qwen3.5-plus",
+    "alibaba-coding-plan-cn": "qwen3-coder-plus", "deepinfra": "deepseek-ai/DeepSeek-V4",
+    "router": "openai/gpt-5.5", "openrouter": "anthropic/claude-opus-5.5", "custom": "my-local-model",
+}
+# Providers with a user-configured endpoint, or outside the auth registry.
+BASE_URLS = {
+    "vertex": "https://aiplatform.googleapis.com/v1/projects/p/locations/global/endpoints/openapi",
+    "azure-foundry": "https://example.openai.azure.com/openai/v1",
+    "openrouter": "https://openrouter.ai/api/v1",
+    "custom": "http://127.0.0.1:8080/v1",
+}
+
+# Values that legitimately differ per run or per release; everything else compares exactly.
+_VOLATILE = (
+    (re.compile(r"client=hermes-client-v[^\"]*"), "client=hermes-client-v<VERSION>"),
+    (re.compile(r'"promptId": "[0-9a-f-]{36}"'), '"promptId": "<UUID>"'),
+)
+
+
+def _cells() -> list[tuple[str, str, str]]:
+    from hermes_cli.auth import PROVIDER_REGISTRY
+    from providers import get_provider_profile
+
+    seen: set[str] = set()
+    cells = []
+    for pid, cfg in PROVIDER_REGISTRY.items():
+        profile = get_provider_profile(pid)
+        name = profile.name if profile else pid
+        if name in seen or cfg.auth_type == "external_process":
+            continue
+        seen.add(name)
+        fallback = profile.fallback_models[0] if profile and profile.fallback_models else "test-model"
+        cells.append((name, MODELS.get(name, fallback), BASE_URLS.get(name) or cfg.inference_base_url))
+    for name in ("openrouter", "custom"):
+        if name not in seen:
+            cells.append((name, MODELS[name], BASE_URLS[name]))
+    return cells
+
+
+@pytest.fixture
+def offline(monkeypatch):
+    """No network and no ambient AWS profile. Capability probes (Ollama ``/api/show``, LM Studio,
+    catalogs) must take their offline answer, or the recorded wire depends on a live server.
+    Name resolution is blocked too: some clients connect through paths ``socket.connect`` misses."""
+    real_connect = socket.socket.connect
+
+    def guarded(sock, address, *args, **kwargs):
+        if sock.family in (socket.AF_INET, socket.AF_INET6):
+            raise OSError(f"network disabled in wire snapshot: {address}")
+        return real_connect(sock, address, *args, **kwargs)
+
+    def no_dns(*_args, **_kwargs):
+        raise socket.gaierror("DNS disabled in wire snapshot")
+
+    monkeypatch.setattr(socket.socket, "connect", guarded)
+    monkeypatch.setattr(socket, "getaddrinfo", no_dns)
+    monkeypatch.setenv("AWS_REGION", "us-east-1")
+    monkeypatch.setenv("AWS_CONFIG_FILE", os.devnull)
+    monkeypatch.setenv("AWS_SHARED_CREDENTIALS_FILE", os.devnull)
+
+
+def _wire(provider: str, model: str, base_url: str) -> dict[str, Any]:
+    from agent.turn_api_request import build_api_request
+    from agent.turn_context import _PER_TURN_RESET_STATE
+    from run_agent import AIAgent
+
+    with patch("model_tools.get_tool_definitions", return_value=TOOLS), \
+         patch("model_tools.check_toolset_requirements", return_value={}), \
+         patch("agent.process_bootstrap.OpenAI"):
+        agent = AIAgent(provider=provider, model=model, base_url=base_url, api_key="sk-test-0000000000",
+                        quiet_mode=True, skip_context_files=True, skip_memory=True, session_id="wire-snapshot")
+    for name, value in _PER_TURN_RESET_STATE:
+        setattr(agent, name, value)
+    out: dict[str, Any] = {"api_mode": agent.api_mode}
+    for label, reasoning in REASONING.items():
+        agent.reasoning_config = reasoning
+        messages = json.loads(json.dumps(MESSAGES))
+        built = build_api_request(
+            agent, api_messages=messages, _moa_prepared_request=None, tools_for_api=agent.tools,
+            system_message=messages[0]["content"], messages=messages,
+            original_user_message=messages[-1]["content"], approx_tokens=0, total_chars=0, retry_count=0,
+            api_call_count=1, api_request_id="req-1", api_start_time=0.0, effective_task_id="task-1",
+            turn_id="turn-1",
+        )
+        text = json.dumps(built.api_kwargs, sort_keys=True, default=repr)
+        for pattern, replacement in _VOLATILE:
+            text = pattern.sub(replacement, text)
+        out[label] = json.loads(text)
+    return out
+
+
+def _first_divergence(expected: Any, actual: Any, path: str = "") -> str:
+    if isinstance(expected, dict) and isinstance(actual, dict):
+        for key in sorted(set(expected) | set(actual)):
+            if key not in actual:
+                return f"{path}.{key}: missing (expected {json.dumps(expected[key])[:200]})"
+            if key not in expected:
+                return f"{path}.{key}: unexpected {json.dumps(actual[key])[:200]}"
+            if expected[key] != actual[key]:
+                return _first_divergence(expected[key], actual[key], f"{path}.{key}")
+    if isinstance(expected, list) and isinstance(actual, list) and len(expected) == len(actual):
+        for i, (e, a) in enumerate(zip(expected, actual)):
+            if e != a:
+                return _first_divergence(e, a, f"{path}[{i}]")
+    return f"{path}: expected {json.dumps(expected)[:200]} got {json.dumps(actual)[:200]}"
+
+
+def _load() -> dict[str, Any]:
+    return json.loads(SNAPSHOT.read_text(encoding="utf-8")) if SNAPSHOT.exists() else {}
+
+
+CELLS = _cells()
+
+
+@pytest.mark.parametrize("provider,model,base_url", CELLS, ids=[c[0] for c in CELLS])
+def test_provider_wire_matches_snapshot(offline, provider, model, base_url):
+    actual = _wire(provider, model, base_url)
+    if UPDATE:
+        golden = _load()
+        golden[provider] = {"model": model, "base_url": base_url, **actual}
+        SNAPSHOT.write_text(json.dumps(dict(sorted(golden.items())), indent=1, sort_keys=True) + "\n",
+                            encoding="utf-8")
+        return
+    expected = _load().get(provider)
+    assert expected is not None, f"{provider} has no recorded wire; regenerate the snapshot (module docstring)"
+    expected = {k: v for k, v in expected.items() if k not in ("model", "base_url")}
+    assert actual == expected, f"{provider} wire changed — {_first_divergence(expected, actual)}"
+
+
+def test_snapshot_covers_every_provider():
+    recorded, current = set(_load()), {c[0] for c in CELLS}
+    assert current <= recorded, f"providers without a recorded wire: {sorted(current - recorded)}"
+    assert recorded <= current, f"snapshot entries for providers that no longer exist: {sorted(recorded - current)}"