From bac905227e65138ab3ee6d68c794f6c30444124f Mon Sep 17 00:00:00 2001 From: teknium1 <127238744+teknium1@users.noreply.github.com> Date: Wed, 23 Sep 2026 02:33:29 -0700 Subject: [PATCH] test: secrets-gated LIVE provider canary suite behind a new `live` marker Classes: C9 provider wire-format drift, C17 prompt-cache hits, C11 real auth / credential routing / `/models` parse. Mocks encode our own belief about each vendor's wire format; vendor-side schema changes, reasoning-replay rules, streaming shape changes and cache behaviour are only observable against the real APIs. This drives the REAL AIAgent + real adapters (chat_completions, anthropic_messages via OpenRouter, codex_responses for xAI/OpenAI, GeminiNativeClient) in a temp HERMES_HOME through a scripted 3-turn conversation with a deterministic registered tool: turn 1 one forced tool call; valid JSON args; result round-trips turn 2 two tool calls in ONE assistant message; both results round-trip turn 3 no tools; answer from replayed history (tool results + reasoning replay) under a cache breakpoint Invariants per turn: no 4xx on any agent-loop request (429 excepted), the turn completes, no /DSML/control-token/raw tool JSON in user text, the credential resolved from env is the one sent and ONLY to the provider's host, tool results are persisted to state.db. Anthropic-family routes also assert 1..4 well-formed breakpoints (none on role:tool / inside tool_result.content[]) and cache-read tokens > 0 on turn 3 (warm once). A `/models` listing check per provider uses the live fetchers without the curated fallback (Gemini is strict-xfail on open #62259). Spend: cheap models, <= 3 turns, max_tokens 2048, max_iterations 6, a hard per-test token and $ guard; usage + estimated $ printed per test and appended to $HERMES_LIVE_USAGE_FILE. Every case skips cleanly without its key. `live` is excluded by default addopts; select with `-m live`. (cherry picked from commit 3bd7de6991bcb03bfe5d11945de6f2317c534b7d) --- pyproject.toml | 3 +- tests/e2e/core/live/__init__.py | 0 tests/e2e/core/live/_helpers.py | 379 +++++++++++++++++++++ tests/e2e/core/live/test_live_providers.py | 286 ++++++++++++++++ 4 files changed, 667 insertions(+), 1 deletion(-) create mode 100644 tests/e2e/core/live/__init__.py create mode 100644 tests/e2e/core/live/_helpers.py create mode 100644 tests/e2e/core/live/test_live_providers.py diff --git a/pyproject.toml b/pyproject.toml index fad1079011..094c46e06e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -626,6 +626,7 @@ tools = ["bot_desktop/launcher.sh", "bot_desktop/wallpaper.png"] testpaths = ["tests"] markers = [ "integration: marks tests requiring external services (API keys, Modal, etc.)", + "live: secrets-gated canaries against REAL LLM provider APIs (spends money; select with -m live)", "real_concurrent_gate: opt out of the autouse stub that disables _detect_concurrent_hermes_instances", "real_post_swap_handoff: opt out of the autouse stub that runs the update post-swap tail in-process", "real_agent_prewarm: opt out of the autouse stub that disables the tui_gateway deferred agent pre-warm timer", @@ -638,7 +639,7 @@ markers = [ "windows_only: exercises native-Windows behaviour; skipped on other hosts", ] # integration tests take way too long to run in the normal CI environments -addopts = "-m 'not integration'" +addopts = "-m 'not integration and not live'" [tool.ty.environment] python-version = "3.13" diff --git a/tests/e2e/core/live/__init__.py b/tests/e2e/core/live/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/e2e/core/live/_helpers.py b/tests/e2e/core/live/_helpers.py new file mode 100644 index 0000000000..13bab2a1bb --- /dev/null +++ b/tests/e2e/core/live/_helpers.py @@ -0,0 +1,379 @@ +"""Shared machinery for the secrets-gated LIVE provider canary (``-m live``). + +The live lane drives the REAL ``AIAgent`` + real provider adapters against real +vendor endpoints on cheap models. Only the tool is synthetic (a deterministic +``live_lookup`` registered in-process) so every assertion is about the wire: + +* C9 wire-format drift: every inference request is accepted (no 4xx), tool calls + come back parseable (valid JSON args), parallel tool calls and their results + round-trip, history (incl. reasoning replay) is re-sent and accepted, and no + reasoning/tool markup leaks into user-visible text. +* C17 prompt cache: on cache-capable routes the follow-up turn reads the cache and + the outgoing request carries well-formed breakpoints. +* C11 real auth/routing: the credential resolved from the environment is the one + sent, and it is only ever sent to the provider's own host; ``/models`` parses. + +Credential values are snapshotted at import (the root conftest blanks every +credential-shaped env var per test) and are never printed, logged or written. +""" + +from __future__ import annotations + +import json +import os +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Callable + +import pytest + +# Snapshot BEFORE tests/conftest.py::_hermetic_environment strips credentials. +LIVE_KEY_VARS = ( + "OPENROUTER_API_KEY", "ANTHROPIC_API_KEY", "NOUS_API_KEY", + "OPENAI_API_KEY", "GEMINI_API_KEY", "XAI_API_KEY", +) +_LIVE_KEYS: dict[str, str] = { + name: os.environ[name].strip() for name in LIVE_KEY_VARS if os.environ.get(name, "").strip() +} + +LOOKUP_VALUES = {"alpha": "ZEBRA-7731", "beta": "OTTER-2204", "gamma": "HERON-5918"} +TOOL_NAME = "live_lookup" +TOOLSET = "live_probe" + +TURN1 = ( + f'Call the {TOOL_NAME} tool with key "alpha", then reply with only the value it returned.' +) +TURN2 = ( + f'Call {TOOL_NAME} for key "beta" AND for key "gamma". Issue BOTH calls in parallel in a ' + "single response (two tool calls in one message), then reply with only the two values." +) +TURN3 = ( + "Do not call any tools. From this conversation's history, list the three values you " + "looked up, in the order alpha, beta, gamma, separated by spaces." +) + +# Text that must never reach the user: reasoning tags, DeepSeek DSML, chat-template +# control tokens, raw tool-call JSON / XML. +LEAK_MARKERS = ( + "", "", "", "", "DSML", "<|", "|>", + "", " 0 on the follow-up turn + pad_tokens: int = 0 # grow the prefix past the vendor's minimum cacheable length + native_anthropic: bool = False + + +# Cheapest tool-capable model per family (verified live Sep 2026). The first +# preference present in the provider's live /models listing is used, so a +# retired slug fails the listing test loudly instead of 404ing mid-conversation. +# Override with HERMES_LIVE_MODEL_ (upper-cased, '-' -> '_'). +LIVE_CASES: tuple[LiveCase, ...] = ( + LiveCase("openrouter-openai", "openrouter", "OPENROUTER_API_KEY", + ("openai/gpt-4.1-nano", "openai/gpt-5-nano"), ("openrouter.ai",), 0.10, 0.40), + LiveCase("openrouter-anthropic", "openrouter", "OPENROUTER_API_KEY", + ("anthropic/claude-haiku-4.5", "anthropic/claude-3-haiku"), ("openrouter.ai",), 1.0, 5.0, + cache_expected=True, pad_tokens=4600), + LiveCase("openrouter-google", "openrouter", "OPENROUTER_API_KEY", + ("google/gemini-2.5-flash-lite", "google/gemini-3.1-flash-lite"), ("openrouter.ai",), 0.10, 0.40), + LiveCase("openrouter-xai", "openrouter", "OPENROUTER_API_KEY", + ("x-ai/grok-4.3", "x-ai/grok-build-0.1", "x-ai/grok-4.20"), ("openrouter.ai",), 1.25, 2.50), + LiveCase("openrouter-deepseek", "openrouter", "OPENROUTER_API_KEY", + ("deepseek/deepseek-v4-flash", "deepseek/deepseek-v4.1-flash", "deepseek/deepseek-chat-v3.1"), + ("openrouter.ai",), 0.10, 0.50), + LiveCase("openrouter-qwen", "openrouter", "OPENROUTER_API_KEY", + ("qwen/qwen3.7-flash", "qwen/qwen3.5-flash-02-23", "qwen/qwen3-235b-a22b-2507"), + ("openrouter.ai",), 0.10, 0.40), + LiveCase("nous-portal", "nous", "NOUS_API_KEY", + ("deepseek/deepseek-v4-flash", "qwen/qwen3.7-flash", "google/gemini-2.5-flash-lite", + "openai/gpt-4.1-nano"), + ("inference-api.nousresearch.com", "portal.nousresearch.com"), 0.30, 1.20), + LiveCase("anthropic-direct", "anthropic", "ANTHROPIC_API_KEY", + ("claude-haiku-4-5", "claude-haiku-4-5-20251001"), ("api.anthropic.com",), 1.0, 5.0, + cache_expected=True, pad_tokens=4600, native_anthropic=True), + LiveCase("openai-direct", "openai", "OPENAI_API_KEY", + ("gpt-5-nano", "gpt-5.4-nano", "gpt-4.1-nano"), ("api.openai.com",), 0.05, 0.40), + LiveCase("gemini-direct", "gemini", "GEMINI_API_KEY", + ("gemini-3.1-flash-lite", "gemini-2.5-flash-lite", "gemini-flash-lite-latest"), + ("generativelanguage.googleapis.com",), 0.25, 1.50), + LiveCase("xai-direct", "xai", "XAI_API_KEY", + ("grok-4.3", "grok-4.20-0309-reasoning", "grok-build-0.1"), ("api.x.ai",), 1.25, 2.50), +) + +LISTING_PROVIDERS: dict[str, str] = {c.provider: c.key_env for c in LIVE_CASES} + + +def live_key(env_name: str) -> str: + """The snapshotted credential, or skip the test cleanly when it is absent.""" + value = _LIVE_KEYS.get(env_name, "") + if not value: + pytest.skip(f"{env_name} not set; live canary skipped") + return value + + +def model_override(case: LiveCase) -> str | None: + return os.environ.get("HERMES_LIVE_MODEL_" + case.id.upper().replace("-", "_")) or None + + +# HTTP wire recorder ----------------------------------------------------------- + +_INFERENCE_SUFFIXES = ("/chat/completions", "/responses", "/messages", ":streamGenerateContent", + ":generateContent") + + +@dataclass +class WireRecord: + method: str + host: str + path: str + status: int + carries_key: bool + body: Any = None + error: str = "" # redacted, truncated response body for status >= 400 + + @property + def is_inference(self) -> bool: + return self.method == "POST" and self.path.endswith(_INFERENCE_SUFFIXES) + + +@dataclass +class WireLog: + """Every HTTP exchange made through httpx (all SDKs Hermes uses sit on it).""" + + secret: str + records: list[WireRecord] = field(default_factory=list) + + def inference(self, start: int = 0) -> list[WireRecord]: + return [r for r in self.records[start:] if r.is_inference] + + def main_turn(self, start: int = 0) -> list[WireRecord]: + """Inference requests of the agent loop itself (they carry the test tool), + excluding auxiliary calls such as title generation.""" + return [r for r in self.inference(start) + if isinstance(r.body, dict) and TOOL_NAME in json.dumps(r.body.get("tools") or "")] + + def describe(self, recs: list[WireRecord]) -> str: + return ", ".join(f"{r.method} {r.host}{r.path} -> {r.status}" + (f" {r.error}" if r.error else "") + for r in recs) + + def status_counts(self) -> dict[str, int]: + counts: dict[str, int] = {} + for r in self.inference(): + kind = "main" if r in self.main_turn() else "aux" + counts[f"{kind}:{r.status}"] = counts.get(f"{kind}:{r.status}", 0) + 1 + return counts + + +def install_wire_recorder(monkeypatch: pytest.MonkeyPatch, secret: str) -> WireLog: + import httpx + + log = WireLog(secret=secret) + + def _carries(request: "httpx.Request") -> bool: + if secret in str(request.url): + return True + return any(secret in v for v in request.headers.values()) + + def _body(request: "httpx.Request") -> Any: + if request.method != "POST": + return None + try: + return json.loads(request.content or b"null") + except Exception: + return None + + def _record(request: "httpx.Request", status: int, body: Any, error: str = "") -> None: + log.records.append(WireRecord(request.method, request.url.host, request.url.path, status, + _carries(request), body, error.replace(secret, "")[:400])) + + def _error_text(resp: "httpx.Response") -> str: + # Error bodies are small and read eagerly by every SDK anyway; cache them on + # the response so the caller still sees the same content. + if resp.status_code < 400: + return "" + try: + return resp.read().decode("utf-8", "replace") + except Exception as exc: + return f"" + + orig_sync = httpx.HTTPTransport.handle_request + orig_async = httpx.AsyncHTTPTransport.handle_async_request + + def handle_request(self, request): # noqa: ANN001 + body = _body(request) + try: + resp = orig_sync(self, request) + except Exception: + _record(request, -1, body) + raise + _record(request, resp.status_code, body, _error_text(resp)) + return resp + + async def handle_async_request(self, request): # noqa: ANN001 + body = _body(request) + try: + resp = await orig_async(self, request) + except Exception: + _record(request, -1, body) + raise + err = "" + if resp.status_code >= 400: + try: + err = (await resp.aread()).decode("utf-8", "replace") + except Exception as exc: + err = f"" + _record(request, resp.status_code, body, err) + return resp + + monkeypatch.setattr(httpx.HTTPTransport, "handle_request", handle_request) + monkeypatch.setattr(httpx.AsyncHTTPTransport, "handle_async_request", handle_async_request) + return log + + +# Deterministic test tool -------------------------------------------------------- + +def register_lookup_tool() -> Callable[[], None]: + from tools.registry import registry + + def handler(args: dict, **_kw: Any) -> str: + key = str(args.get("key", "")) + return json.dumps({"key": key, "value": LOOKUP_VALUES.get(key, "UNKNOWN")}) + + registry.register( + name=TOOL_NAME, toolset=TOOLSET, + schema={ + "name": TOOL_NAME, + "description": "Look up the secret value stored under a key.", + "parameters": { + "type": "object", + "properties": {"key": {"type": "string", "description": "alpha, beta or gamma"}}, + "required": ["key"], + }, + }, + handler=handler, + ) + return lambda: registry.deregister(TOOL_NAME) + + +# HERMES_HOME wiring ------------------------------------------------------------- + +def write_live_home(hermes_home: Path, provider: str, model: str) -> None: + """Minimal real config: the provider under test, eager tools (no tool_search + bridge), bounded retries. No credential is ever written to disk.""" + hermes_home.mkdir(parents=True, exist_ok=True) + (hermes_home / "config.yaml").write_text( + "model:\n" + f" provider: {provider}\n" + f" default: {model}\n" + "agent:\n" + " api_max_retries: 2\n" + "tools:\n" + " tool_search:\n" + " enabled: \"off\"\n", + encoding="utf-8", + ) + + +def pad_text(tokens: int) -> str: + """~``tokens`` tokens of inert reference material (≈8 tokens per row).""" + if tokens <= 0: + return "" + rows = [f"R{i:04d} amber quartz lantern {i * 7 % 1000:03d}" for i in range(tokens // 8)] + return "\n\nReference table (ignore unless asked):\n" + "\n".join(rows) + + +# Assertions helpers ---------------------------------------------------------------- + +def assistant_tool_calls(messages: list[dict]) -> list[list[dict]]: + """Tool-call groups (one list per assistant message) in order.""" + return [list(m["tool_calls"]) for m in messages + if m.get("role") == "assistant" and m.get("tool_calls")] + + +def parse_tool_args(call: dict) -> dict: + fn = call.get("function") or {} + raw = fn.get("arguments") + args = json.loads(raw) if isinstance(raw, str) else raw + assert isinstance(args, dict), f"tool args are not a JSON object: {raw!r}" + return args + + +def leaked_markers(text: str) -> list[str]: + return [m for m in LEAK_MARKERS if m in (text or "")] + + +def walk(obj: Any, path: str = "") -> Any: + """Yield (path, key, value) for every dict entry in a nested JSON value.""" + if isinstance(obj, dict): + for k, v in obj.items(): + yield path, k, v + yield from walk(v, f"{path}.{k}") + elif isinstance(obj, list): + for i, v in enumerate(obj): + yield from walk(v, f"{path}[{i}]") + + +def conversation_part(body: Any) -> Any: + """The history part of a request body across wire formats.""" + if not isinstance(body, dict): + return None + for key in ("messages", "input", "contents"): + if key in body: + return body[key] + return None + + +REASONING_KEYS = {"reasoning_content", "reasoning", "reasoning_details", "thinking", + "thought_signature", "thoughtSignature", "encrypted_content", "signature"} + + +def reasoning_replay_keys(body: Any) -> set[str]: + return {k for _p, k, _v in walk(conversation_part(body)) if k in REASONING_KEYS} + + +def cache_markers(body: Any) -> list[str]: + return [p for p, k, _v in walk(body) if k == "cache_control"] + + +def usage_line(case: LiveCase, model: str, agent: Any, wire: WireLog) -> dict: + inp = int(agent.session_input_tokens or 0) + out = int(agent.session_output_tokens or 0) + cr = int(agent.session_cache_read_tokens or 0) + cw = int(agent.session_cache_write_tokens or 0) + # Upper bound: cache reads at full input price, cache writes at 1.25x (Anthropic's rate). + ceiling = ((inp + cr + 1.25 * cw) * case.price_in + out * case.price_out) / 1_000_000 + return { + "case": case.id, "model": model, "api_calls": int(agent.session_api_calls or 0), + "http_inference_calls": len(wire.inference()), + "http_statuses": wire.status_counts(), + "http_errors": sorted({f"{r.status} {r.error[:160]}" for r in wire.inference() if r.status >= 400})[:5], + "input": inp, "output": out, "cache_read": cr, "cache_write": cw, + "reasoning": int(getattr(agent, "session_reasoning_tokens", 0) or 0), + "hermes_est_usd": round(float(agent.session_estimated_cost_usd or 0.0), 6), + "hermes_cost_status": str(getattr(agent, "session_cost_status", "")), + "list_price_ceiling_usd": round(ceiling, 6), + } + + +def emit_usage(line: dict) -> None: + print("LIVE-USAGE " + json.dumps(line, sort_keys=True), flush=True) + target = os.environ.get("HERMES_LIVE_USAGE_FILE") + if target: + with open(target, "a", encoding="utf-8") as fh: + fh.write(json.dumps(line, sort_keys=True) + "\n") diff --git a/tests/e2e/core/live/test_live_providers.py b/tests/e2e/core/live/test_live_providers.py new file mode 100644 index 0000000000..71a11b4083 --- /dev/null +++ b/tests/e2e/core/live/test_live_providers.py @@ -0,0 +1,286 @@ +"""LIVE provider canary: real AIAgent + real adapters against real vendor APIs. + +Secrets-gated and excluded from the default run (``-m live`` to select). Each +case skips cleanly when its credential env var is absent. Classes covered: +C9 (provider wire-format drift), C17 (prompt-cache hits), C11 (real auth + +credential routing + ``/models`` parse). See ``_helpers.py`` for the matrix. + +Spend: cheap models only, three scripted turns (plus at most one cache warm-up +retry on cache-capable routes), bounded max_tokens/iterations and a hard +per-test token + dollar guard. Usage and estimated cost are printed per test +(``LIVE-USAGE {...}``) and appended to ``$HERMES_LIVE_USAGE_FILE`` when set. +""" + +from __future__ import annotations + +from pathlib import Path +from urllib.parse import urlparse + +import pytest + +from tests.e2e.core.live._helpers import ( + LISTING_PROVIDERS, + LIVE_CASES, + LOOKUP_VALUES, + MAX_TOKENS_PER_TEST, + MAX_USD_PER_TEST, + TOOL_NAME, + TURN1, + TURN2, + TURN3, + LiveCase, + assistant_tool_calls, + cache_markers, + emit_usage, + install_wire_recorder, + leaked_markers, + live_key, + model_override, + pad_text, + parse_tool_args, + reasoning_replay_keys, + register_lookup_tool, + usage_line, + write_live_home, +) + +pytestmark = pytest.mark.live + +_LISTING_CACHE: dict[str, list[str]] = {} + + +@pytest.fixture(scope="module", autouse=True) +def _lookup_tool(): + remove = register_lookup_tool() + yield + remove() + + +@pytest.fixture() +def live_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + # HOME too (a sibling, not the parent: state.db's live-system guard treats + # $HOME/.hermes as production), so nothing can resolve the developer's real home. + home = tmp_path / "home" + hermes_home = tmp_path / "hermes_home" + home.mkdir() + hermes_home.mkdir() + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + return hermes_home + + +def _resolve(provider: str, model: str | None) -> dict: + from hermes_cli.runtime_provider import resolve_runtime_provider + + return resolve_runtime_provider(requested=provider, target_model=model) + + +def _live_listing(provider: str, runtime: dict) -> list[str]: + """The same live fetchers the /model picker uses, WITHOUT the curated fallback + (a silent fallback is exactly what hides a broken listing).""" + api_key, base_url = runtime.get("api_key") or "", runtime.get("base_url") or "" + if provider == "openai": + from hermes_cli.models import fetch_api_models + + return list(fetch_api_models(api_key, base_url, timeout=20.0) or []) + if provider == "nous": + from hermes_cli.auth import fetch_nous_models + + return list(fetch_nous_models(inference_base_url=base_url, api_key=api_key) or []) + from providers import get_provider_profile + + profile = get_provider_profile(provider) + assert profile is not None, f"no provider profile registered for {provider!r}" + return list(profile.fetch_models(api_key=api_key, base_url=base_url, timeout=20.0) or []) + + +def _normalize_id(model_id: str) -> str: + return model_id.split("/", 1)[1] if model_id.startswith("models/") else model_id + + +def _listing(provider: str) -> list[str]: + if provider not in _LISTING_CACHE: + runtime = _resolve(provider, None) + _LISTING_CACHE[provider] = [_normalize_id(str(m)) for m in _live_listing(provider, runtime)] + return _LISTING_CACHE[provider] + + +def _pick_model(case: LiveCase) -> str: + override = model_override(case) + if override: + return override + listed = set(_listing(case.provider)) + if not listed: + # A broken listing is test_models_listing_parses' verdict; still run the wire canary. + return case.model_prefs[0] + for pref in case.model_prefs: + if pref in listed: + return pref + pytest.fail( + f"{case.id}: none of {case.model_prefs} is in {case.provider}'s live /models listing " + f"({len(listed)} ids) — update LIVE_CASES in tests/e2e/core/live/_helpers.py" + ) + + +def _host_ok(host: str, allowed: tuple[str, ...]) -> bool: + return any(host == h or host.endswith("." + h) for h in allowed) + + +# /models ------------------------------------------------------------------------ + + +_KNOWN_LISTING_BUGS = { + # Native /v1beta/models rejects a Bearer AI-Studio key (401) and the picker silently + # falls back to the curated list. strict: flips red once the fix lands (drop the entry). + "gemini": "#62259 Gemini live model discovery sends Bearer auth (fix PRs #62267/#116509)", +} + + +@pytest.mark.parametrize("provider", [ + pytest.param(p, marks=pytest.mark.xfail(reason=_KNOWN_LISTING_BUGS[p], strict=True)) + if p in _KNOWN_LISTING_BUGS else p + for p in sorted(LISTING_PROVIDERS) +]) +def test_models_listing_parses(provider: str, live_home: Path, monkeypatch: pytest.MonkeyPatch) -> None: + key_env = LISTING_PROVIDERS[provider] + secret = live_key(key_env) + monkeypatch.setenv(key_env, secret) + wire = install_wire_recorder(monkeypatch, secret) + case = next(c for c in LIVE_CASES if c.provider == provider) + write_live_home(live_home, provider, case.model_prefs[0]) + + runtime = _resolve(provider, None) + assert runtime.get("api_key") == secret, f"{provider}: resolver did not pick the {key_env} credential" + assert _host_ok(urlparse(runtime["base_url"]).hostname or "", case.hosts), runtime["base_url"] + + models = [_normalize_id(str(m)) for m in _live_listing(provider, runtime)] + assert models, f"{provider}: live /models returned nothing ({wire.describe(wire.records)})" + assert all(isinstance(m, str) and m.strip() for m in models) + wanted = {pref for c in LIVE_CASES if c.provider == provider for pref in c.model_prefs} + assert wanted & set(models), f"{provider}: none of the canary models {sorted(wanted)} are listed" + leaks = [r for r in wire.records if r.carries_key and not _host_ok(r.host, case.hosts)] + assert not leaks, f"{provider}: credential sent to a foreign host: {wire.describe(leaks)}" + print(f"LIVE-MODELS provider={provider} count={len(models)}", flush=True) + + +# 3-turn canary -------------------------------------------------------------------- + + +def _run_turn(agent, text: str, history, wire, label: str): + start = len(wire.records) + result = agent.run_conversation(text, conversation_history=history) + calls = wire.inference(start) + main = wire.main_turn(start) + assert main, f"{label}: no agent-loop request carried the test tool ({wire.describe(calls)})" + # Agent-loop requests must never be rejected: a 4xx (other than 429) is wire drift. + # Auxiliary calls (title generation) may take a designed 400->adapt retry (e.g. + # "reasoning is mandatory"); the test end asserts the last aux attempt succeeded. + bad = [r for r in main if 400 <= r.status < 500 and r.status != 429] + assert not bad, f"{label}: provider rejected the request (wire drift): {wire.describe(bad)}" + assert not result.get("failed"), f"{label}: turn failed: {result.get('error')} ({wire.describe(calls)})" + assert main[-1].status == 200, f"{label}: final request did not succeed: {wire.describe(main)}" + final = result.get("final_response") or "" + assert final.strip(), f"{label}: empty final response" + assert not leaked_markers(final), f"{label}: markup leaked into final text: {leaked_markers(final)} in {final!r}" + return result, main + + +@pytest.mark.parametrize("case", LIVE_CASES, ids=[c.id for c in LIVE_CASES]) +def test_three_turn_tool_conversation(case: LiveCase, live_home: Path, monkeypatch: pytest.MonkeyPatch) -> None: + secret = live_key(case.key_env) + monkeypatch.setenv(case.key_env, secret) + wire = install_wire_recorder(monkeypatch, secret) + model = _pick_model(case) + write_live_home(live_home, case.provider, model) + + runtime = _resolve(case.provider, model) + assert runtime.get("api_key") == secret, f"{case.id}: resolver did not pick the {case.key_env} credential" + + from hermes_constants import parse_reasoning_effort + from hermes_state import SessionDB + from run_agent import AIAgent + + db = SessionDB(live_home / "state.db") + agent = AIAgent( + provider=runtime.get("provider"), api_mode=runtime.get("api_mode"), + base_url=runtime.get("base_url"), api_key=runtime.get("api_key"), + credential_pool=runtime.get("credential_pool"), model=model, + session_db=db, session_id=f"live-{case.id}", quiet_mode=True, platform="cli", + enabled_toolsets=["live_probe"], skip_context_files=True, skip_memory=True, + skip_background_review=True, max_tokens=2048, max_iterations=6, + reasoning_config=parse_reasoning_effort("low"), run_budget_seconds=240, + ) + usage: dict = {} + try: + # Turn 1: one forced tool call; the tool result must round-trip into the answer. + r1, _ = _run_turn(agent, TURN1 + pad_text(case.pad_tokens), None, wire, f"{case.id} turn1") + groups = assistant_tool_calls(r1["messages"]) + assert groups, f"{case.id} turn1: model made no tool call" + args = [parse_tool_args(c) for g in groups for c in g] + assert {"key": "alpha"} in [{"key": a.get("key")} for a in args], args + assert all(c["function"]["name"] == TOOL_NAME for g in groups for c in g) + tool_msgs = [m for m in r1["messages"] if m.get("role") == "tool"] + assert any(LOOKUP_VALUES["alpha"] in str(m.get("content")) for m in tool_msgs) + assert LOOKUP_VALUES["alpha"] in r1["final_response"], r1["final_response"] + + # Turn 2: two tool calls in ONE assistant message; both results round-trip. + before = len(r1["messages"]) + r2, _ = _run_turn(agent, TURN2, r1["messages"], wire, f"{case.id} turn2") + new_groups = assistant_tool_calls(r2["messages"][before:]) + keys = {parse_tool_args(c).get("key") for g in new_groups for c in g} + assert {"beta", "gamma"} <= keys, f"{case.id} turn2: tool calls {new_groups}" + assert any(len(g) >= 2 for g in new_groups), ( + f"{case.id} turn2: expected parallel tool calls in one message, got {[len(g) for g in new_groups]}") + for key in ("beta", "gamma"): + assert LOOKUP_VALUES[key] in r2["final_response"], r2["final_response"] + + # Turn 3: history reuse (tool results + reasoning replay) under a cache breakpoint. + cache_before = agent.session_cache_read_tokens + r3, calls3 = _run_turn(agent, TURN3, r2["messages"], wire, f"{case.id} turn3") + first3 = calls3[0].body + sent = str(first3) + for value in LOOKUP_VALUES.values(): + assert value in sent, f"{case.id} turn3: tool result {value} missing from replayed history" + assert value in r3["final_response"], f"{case.id} turn3: {r3['final_response']!r}" + replay = sorted(reasoning_replay_keys(first3)) + cache_read_turn3 = agent.session_cache_read_tokens - cache_before + + if case.cache_expected: + markers = cache_markers(first3) + assert 1 <= len(markers) <= 4, f"{case.id} turn3: cache breakpoints {markers}" + for m in (first3.get("messages") or []): + if m.get("role") == "tool" and not case.native_anthropic: + assert "cache_control" not in m, "top-level cache_control on role:tool (OpenRouter hangs)" + content = m.get("content") + for part in content if isinstance(content, list) else []: + if isinstance(part, dict) and part.get("type") == "tool_result": + inner = part.get("content") + assert not (isinstance(inner, list) and cache_markers(inner)), ( + "cache_control inside tool_result.content[] (#89886: non-retryable 400)") + if cache_read_turn3 <= 0: + # Retry-tolerant: the first write can land on another replica; warm once. + cache_before = agent.session_cache_read_tokens + _run_turn(agent, TURN3, r3["messages"], wire, f"{case.id} turn3-warm") + cache_read_turn3 = agent.session_cache_read_tokens - cache_before + assert cache_read_turn3 > 0, f"{case.id}: no prompt-cache read on the follow-up turn" + + # Persisted == sent: the three tool results are in state.db for this session. + rows = db.get_messages(agent.session_id) + persisted = " ".join(str(r.get("content")) for r in rows if r.get("role") == "tool") + for value in LOOKUP_VALUES.values(): + assert value in persisted, f"{case.id}: tool result {value} not persisted" + + leaks = [r for r in wire.records if r.carries_key and not _host_ok(r.host, case.hosts)] + assert not leaks, f"{case.id}: credential sent to a foreign host: {wire.describe(leaks)}" + main_ids = {id(r) for r in wire.main_turn()} + aux = [r for r in wire.inference() if id(r) not in main_ids] + assert not aux or 200 <= aux[-1].status < 300, f"{case.id}: auxiliary call never recovered: {wire.describe(aux)}" + usage = {"turn3_cache_read": cache_read_turn3, "turn3_replay_keys": replay} + finally: + line = usage_line(case, model, agent, wire) | usage + emit_usage(line) + if hasattr(agent, "close"): + agent.close() + total = line["input"] + line["output"] + line["cache_read"] + line["cache_write"] + assert total <= MAX_TOKENS_PER_TEST, f"{case.id}: spend guard tripped ({total} tokens)" + assert line["list_price_ceiling_usd"] <= MAX_USD_PER_TEST, f"{case.id}: spend guard tripped {line}"