hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

test: secrets-gated LIVE provider canary suite behind a new live marker

Browse Source 
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 <think>/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)
This commit is contained in:
teknium1
2026-09-23 02:33:29 -07:00
committed by Teknium
parent cfdfa4a696
commit bac905227e
4 changed files with 667 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
pyproject.toml
View File

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

0
tests/e2e/core/live/__init__.py Normal file
View File

379
tests/e2e/core/live/_helpers.py Normal file
View File

@@ -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 = (
"<think>", "</think>", "<thinking>", "</thinking>", "DSML", "<|", "|>",
"<tool_call", "</tool_call>", "<function", "</function", '"tool_calls"',
'"arguments"', '{"key"', "<invoke", "antml",
)
# Per-test spend guard (tokens, all buckets). A regression that loops or re-sends
# a huge prompt fails here instead of burning money.
MAX_TOKENS_PER_TEST = 90_000
MAX_USD_PER_TEST = 0.12
@dataclass(frozen=True)
class LiveCase:
id: str
provider: str # hermes provider id passed to resolve_runtime_provider
key_env: str
model_prefs: tuple[str, ...]
hosts: tuple[str, ...] # the only hosts allowed to see the credential
price_in: float # $/M input tokens, list price (ceiling estimate)
price_out: float # $/M output tokens
cache_expected: bool = False # assert cache-read > 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_<ID> (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, "<redacted>")[: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"<unreadable: {type(exc).__name__}>"
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"<unreadable: {type(exc).__name__}>"
_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")

286
tests/e2e/core/live/test_live_providers.py Normal file
View File

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