diff --git a/README.md b/README.md index 6d14d04..a11d758 100644 --- a/README.md +++ b/README.md @@ -76,11 +76,31 @@ Gemini answer would look like it cited Google. | `gemini` | `GEMINI_API_KEY` | `gemini-3.5-flash` | Google Search grounding | | `perplexity` | `PERPLEXITY_API_KEY` | `perplexity/sonar` | Responses API `web_search` tool | | `claude` | `ANTHROPIC_API_KEY` | `claude-sonnet-5` | Messages API web search tool | +| `claude-cli` | none, uses your `claude` login | whatever the CLI picks | Claude Code's `WebSearch` and `WebFetch` tools | Keys are read from the environment only. They are sent in request headers and removed from any error message before it is written, so they never reach the results file. An engine with no key is skipped. +`claude-cli` is for people who pay for a Claude subscription and have no API +key. It runs the [Claude Code](https://docs.claude.com/en/docs/claude-code) CLI +(`claude -p`) once per call, with only the web search and fetch tools switched +on. It never joins a run on its own: name it with `--engine claude-cli`. If +`claude` is not on your PATH, the run stops before the first call and says so. +Log in once by running `claude`. On macOS, where the login sits in the Keychain, +run `claude setup-token` and export `CLAUDE_CODE_OAUTH_TOKEN` instead. Its rows +record a cost of 0 and `"subscription": true`. The calls use up your plan's +usage limits and never show on an API bill. A call gets 600 seconds by +default. Change that with `--timeout`, which works for every engine (the API +engines default to 180). `--model claude-cli=sonnet` is passed on to the CLI. +Its sources are the search results and fetched pages from the tool calls, plus +any link in the answer text. + +The CLI's answers are not the API's answers. Claude Code adds its own system +prompt. The model and search limits come from your account and plan. Two people +can get different results from the same question file. Say which one you +used when you publish numbers. + Change a model with `--model claude=claude-opus-5`. Pick the models your buyers actually use in the apps, or say in the report which ones you used. @@ -125,6 +145,17 @@ API. The CLI passed the operator's own git identity to the model, and many brand answers then told the reader the company was probably their own. The API adapters here send only the question and a one-line instruction to cite sources. +The `claude-cli` engine guards against the same leak. Each call runs in a new, +empty temporary folder with no git repository above it. HOME and the config +folder are temporary too, and git is told to read no config at all. Only an +allow-list of environment variables gets through (PATH, locale, proxy and CA +settings), so `GIT_*`, `USER` and any `CLAUDE*` or `ANTHROPIC*` variables are +dropped. Your `CLAUDE.md` files, memory, settings, hooks and MCP servers are +never loaded. Only your login is copied in, and if the CLI refreshes it during +the call, the new one is written back. What cannot be hidden is the account: +the answers still come from your Claude login, on your plan, so results depend +on the local account. + ## Example `examples/synapse-launch-week-2026-09.md` is a real run on one company's own @@ -150,7 +181,8 @@ pip install -e ".[test]" pytest ``` -33 tests. They mock every API, so they spend nothing and need no keys. +46 tests. They mock every API and stand in a fake `claude` for the CLI, so they +spend nothing and need no keys or login. ## Licence diff --git a/src/answer_engine_benchmark/cli.py b/src/answer_engine_benchmark/cli.py index 499f7d5..7a4d026 100644 --- a/src/answer_engine_benchmark/cli.py +++ b/src/answer_engine_benchmark/cli.py @@ -35,6 +35,8 @@ def _engine_flags(p: argparse.ArgumentParser) -> None: p.add_argument("--engine", action="append", choices=sorted(ENGINES), help="only these engines (repeatable)") p.add_argument("--model", action="append", type=_kv, default=[], metavar="ENGINE=MODEL", help="override a model, e.g. --model claude=claude-opus-5") + p.add_argument("--timeout", type=float, metavar="SECONDS", + help="give up on one call after this long (default 180, claude-cli 600)") p.add_argument("--runs", type=int, default=3, help="times to ask each question on each engine (default 3)") p.add_argument("--max-calls", type=int, default=500, help="refuse to start a run that needs more API calls than this (default 500)") @@ -111,19 +113,20 @@ def main(argv: list[str] | None = None) -> int: return 0 q = load(args.questions, dict(args.set), allow_unfilled=getattr(args, "allow_unfilled", False)) - engines = available(only=args.engine, models=dict(args.model)) + engines = available(only=args.engine, models=dict(args.model), timeout=args.timeout) if args.cmd == "check": print(f"ok: {count(q)} questions in {len(q['groups'])} groups; ours = {q['ours']}") missing = [f"{n} ({e.env_key})" for n, e in ENGINES.items() - if n not in engines and (not args.engine or n in args.engine)] + if e.env_key and n not in engines and (not args.engine or n in args.engine)] if missing: print("no key set for: " + ", ".join(missing)) _plan(q, engines, args.runs, args.max_calls) return 0 if not engines: - keys = ", ".join(e.env_key for e in ENGINES.values()) - print(f"aeb: no engine has a key. Set one or more of: {keys}", file=sys.stderr) + keys = ", ".join(e.env_key for e in ENGINES.values() if e.env_key) + print(f"aeb: no engine has a key. Set one or more of: {keys}, " + "or use --engine claude-cli", file=sys.stderr) return 2 if args.cmd == "run": diff --git a/src/answer_engine_benchmark/engines.py b/src/answer_engine_benchmark/engines.py index a691f3b..cd1612f 100644 --- a/src/answer_engine_benchmark/engines.py +++ b/src/answer_engine_benchmark/engines.py @@ -11,13 +11,21 @@ from __future__ import annotations +import json import os import re +import shutil +import signal +import subprocess +import tempfile from collections.abc import Callable from dataclasses import dataclass +from pathlib import Path import requests +from .scoring import URL_RX + TIMEOUT = 180 SYSTEM_HINT = "Answer as you would for any user. Cite your sources with links." @@ -43,7 +51,7 @@ def token_cost(model: str, inp: int, out: int) -> float | None: # ----------------------------------------------------------------------------- OpenAI ---- -def openai_answer(question: str, *, key: str, model: str = "gpt-5-mini") -> dict: +def openai_answer(question: str, *, key: str, model: str = "gpt-5-mini", timeout: float = TIMEOUT) -> dict: body = { "model": model, "tools": [{"type": "web_search"}], @@ -52,7 +60,7 @@ def openai_answer(question: str, *, key: str, model: str = "gpt-5-mini") -> dict "instructions": SYSTEM_HINT, } r = requests.post("https://api.openai.com/v1/responses", - headers={"Authorization": f"Bearer {key}"}, json=body, timeout=TIMEOUT) + headers={"Authorization": f"Bearer {key}"}, json=body, timeout=timeout) r.raise_for_status() d = r.json() text, urls = [], [] @@ -75,13 +83,13 @@ def openai_answer(question: str, *, key: str, model: str = "gpt-5-mini") -> dict # ----------------------------------------------------------------------------- Gemini ---- -def gemini_answer(question: str, *, key: str, model: str = "gemini-3.5-flash") -> dict: +def gemini_answer(question: str, *, key: str, model: str = "gemini-3.5-flash", timeout: float = TIMEOUT) -> dict: body = {"contents": [{"parts": [{"text": question}]}], "tools": [{"google_search": {}}]} # The key goes in a header, not the URL. In the query string it ends up in the text of # any HTTPError, and from there in the results file. r = requests.post( f"https://generativelanguage.googleapis.com/v1beta/models/{model}:generateContent", - headers={"x-goog-api-key": key}, json=body, timeout=TIMEOUT, + headers={"x-goog-api-key": key}, json=body, timeout=timeout, ) r.raise_for_status() d = r.json() @@ -103,7 +111,7 @@ def gemini_answer(question: str, *, key: str, model: str = "gemini-3.5-flash") - # ----------------------------------------------------------------------------- Claude ---- -def anthropic_answer(question: str, *, key: str, model: str = "claude-sonnet-5") -> dict: +def anthropic_answer(question: str, *, key: str, model: str = "claude-sonnet-5", timeout: float = TIMEOUT) -> dict: """Claude through the Messages API with the server-side web search tool. No model fallback is configured on purpose: an answer from a different model would be @@ -126,7 +134,7 @@ def anthropic_answer(question: str, *, key: str, model: str = "claude-sonnet-5") "tools": [{"type": "web_search_20260209", "name": "web_search", "max_uses": 5}], } r = requests.post("https://api.anthropic.com/v1/messages", headers=headers, json=body, - timeout=TIMEOUT) + timeout=timeout) r.raise_for_status() d = r.json() served = d.get("model", served) @@ -153,7 +161,7 @@ def anthropic_answer(question: str, *, key: str, model: str = "claude-sonnet-5") # ------------------------------------------------------------------------- Perplexity ---- -def perplexity_answer(question: str, *, key: str, model: str = "perplexity/sonar") -> dict: +def perplexity_answer(question: str, *, key: str, model: str = "perplexity/sonar", timeout: float = TIMEOUT) -> dict: """Perplexity's Responses API. Web search is OFF unless the tool is passed. Without `tools=[{"type": "web_search"}]` the call still succeeds and returns a fluent @@ -162,7 +170,7 @@ def perplexity_answer(question: str, *, key: str, model: str = "perplexity/sonar """ body = {"model": model, "input": question, "tools": [{"type": "web_search"}]} r = requests.post("https://api.perplexity.ai/v1/responses", - headers={"Authorization": f"Bearer {key}"}, json=body, timeout=TIMEOUT) + headers={"Authorization": f"Bearer {key}"}, json=body, timeout=timeout) r.raise_for_status() d = r.json() text, urls = "", [] @@ -184,10 +192,239 @@ def perplexity_answer(question: str, *, key: str, model: str = "perplexity/sonar } +# ------------------------------------------------------------------------- Claude CLI ---- +# Claude through the locally logged-in Claude Code CLI (`claude -p`), for people with a Claude +# subscription and no API key. The CLI is a coding agent that normally reads a lot about the +# person running it: git identity, CLAUDE.md files, memory, settings, hooks, MCP servers. In an +# earlier run it passed the operator's git username to the model, and brand answers then told +# the reader the company was probably their own. So every call runs as a stranger: HOME, the +# config dir and the working directory are fresh empty temp dirs, git reads no config, and the +# environment is an allow-list. Only the login credentials are copied in. +CLAUDE_CLI_TIMEOUT = 600 +# Everything else is dropped: USER, EMAIL, every GIT_* variable, the calling session's CLAUDE* +# and ANTHROPIC* variables, SSH agents and so on. +_CLI_ENV_KEEP = ("PATH", "LANG", "LANGUAGE", "LC_ALL", "TZ", "TMPDIR", + "HTTP_PROXY", "HTTPS_PROXY", "NO_PROXY", "http_proxy", "https_proxy", "no_proxy", + "SSL_CERT_FILE", "SSL_CERT_DIR", "NODE_EXTRA_CA_CERTS") +# A long-lived token from `claude setup-token`. When set, no credentials file is copied. +_CLI_TOKEN_VAR = "CLAUDE_CODE_OAUTH_TOKEN" +_CLI_TOOLS = "WebSearch,WebFetch" + + +class EngineUnavailable(ValueError): + """An engine was asked for but cannot run on this machine.""" + + +def claude_exe(env: dict | None = None) -> str: + env = os.environ if env is None else env + exe = shutil.which("claude", path=env.get("PATH")) + if not exe: + raise EngineUnavailable("claude-cli needs the Claude Code CLI: no `claude` on PATH. " + "Install it and run `claude` once to log in.") + return exe + + +def _cli_config_dir(env: dict) -> Path: + return Path(env.get("CLAUDE_CONFIG_DIR") or Path(env.get("HOME") or Path.home()) / ".claude") + + +def cli_env(home: str, src: dict | None = None) -> dict: + """The environment of someone who has never met the operator.""" + src = os.environ if src is None else src + env = {k: src[k] for k in (*_CLI_ENV_KEEP, _CLI_TOKEN_VAR) if src.get(k)} + env.update({ + "HOME": home, + "CLAUDE_CONFIG_DIR": os.path.join(home, ".claude"), + "XDG_CONFIG_HOME": os.path.join(home, ".config"), + "XDG_DATA_HOME": os.path.join(home, ".local", "share"), + "XDG_CACHE_HOME": os.path.join(home, ".cache"), + "XDG_STATE_HOME": os.path.join(home, ".local", "state"), + "GIT_CONFIG_GLOBAL": os.devnull, + "GIT_CONFIG_NOSYSTEM": "1", + "DISABLE_AUTOUPDATER": "1", + "USER": "user", + "LOGNAME": "user", + }) + return env + + +def _stage_auth(cfg_dir: str, src: dict) -> Path | None: + """Copy only the login credentials into the empty config dir. Returns the original's path, + or None when the token variable carries the login instead.""" + os.makedirs(cfg_dir, mode=0o700, exist_ok=True) + if src.get(_CLI_TOKEN_VAR): + return None + real = _cli_config_dir(src) / ".credentials.json" + if not real.is_file(): + # macOS keeps the login in the Keychain, where it cannot be copied. + raise EngineUnavailable(f"claude-cli: no login found at {real}. Run `claude setup-token` and " + f"export {_CLI_TOKEN_VAR}, or run `claude` once to log in.") + fd = os.open(os.path.join(cfg_dir, ".credentials.json"), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(fd, "wb") as f: + f.write(real.read_bytes()) + return real + + +def _expires_at(raw: bytes) -> int: + try: + return int(json.loads(raw)["claudeAiOauth"]["expiresAt"]) + except (ValueError, KeyError, TypeError): + return 0 + + +def _write_back_auth(real: Path, staged: str) -> None: + """If the CLI refreshed its login inside the temp dir, hand the new one back. The refresh + token rotates, so keeping the old one could log the operator's own sessions out.""" + try: + new, old = Path(staged).read_bytes(), real.read_bytes() + except OSError: + return + if new == old or _expires_at(new) <= _expires_at(old): + return + tmp = real.with_name(real.name + ".aeb.tmp") + fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(fd, "wb") as f: + f.write(new) + os.replace(tmp, real) + + +def _run_cli(cmd: list[str], prompt: str, *, cwd: str, env: dict, timeout: float) -> subprocess.CompletedProcess: + # The prompt goes in on stdin, so a question that starts with "-" is never read as a flag. + # Own process group, so a timeout kills the CLI and anything it started. + p = subprocess.Popen(cmd, cwd=cwd, env=env, stdin=subprocess.PIPE, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, text=True, start_new_session=True) + try: + out, err = p.communicate(prompt, timeout=timeout) + except subprocess.TimeoutExpired: + try: + os.killpg(p.pid, signal.SIGKILL) + except OSError: + p.kill() + p.communicate() + raise TimeoutError(f"claude-cli: no answer after {timeout:g}s") from None + return subprocess.CompletedProcess(cmd, p.returncode, out, err) + + +def _tool_text(content) -> str: + if isinstance(content, str): + return content + if isinstance(content, list): + return "".join(c.get("text", "") for c in content if isinstance(c, dict)) + return "" + + +def _search_links(text: str) -> list[str]: + """WebSearch results arrive as text with a `Links: [{"title": ..., "url": ...}]` list.""" + i = text.find("Links: [") + if i < 0: + return [] + try: + links, _ = json.JSONDecoder().raw_decode(text, i + len("Links: ")) + except ValueError: + return [] + return [x["url"] for x in links if isinstance(x, dict) and isinstance(x.get("url"), str)] + + +def parse_cli_stream(stdout: str) -> dict: + """Read `--output-format stream-json`. Sources come from the tool stream: the CLI's web + tools run client-side, so the usage counter for server-side searches stays at 0.""" + calls: dict[str, tuple[str, dict]] = {} + tools: list[str] = [] + urls: list[str] = [] + final: dict = {} + for line in stdout.splitlines(): + try: + ev = json.loads(line) + except ValueError: + continue + if not isinstance(ev, dict): + continue + blocks = (ev.get("message") or {}).get("content") or [] + if ev.get("type") == "assistant": + for b in blocks: + if isinstance(b, dict) and b.get("type") == "tool_use" and b.get("name"): + tools.append(b["name"]) + calls[b.get("id", "")] = (b["name"], b.get("input") or {}) + elif ev.get("type") == "user": + for b in blocks: + if not isinstance(b, dict) or b.get("type") != "tool_result" or b.get("is_error"): + continue + name, inp = calls.get(b.get("tool_use_id", ""), ("", {})) + if name == "WebSearch": + urls += _search_links(_tool_text(b.get("content"))) + elif name == "WebFetch" and isinstance(inp.get("url"), str): + urls.append(inp["url"]) + elif ev.get("type") == "result": + final = ev + if not final: + raise RuntimeError("claude-cli: no result in the CLI output") + if final.get("is_error") or final.get("subtype") != "success": + raise RuntimeError(f"claude-cli: {final.get('subtype')}: {str(final.get('result', ''))[:200]}") + text = final.get("result") or "" + mu = final.get("modelUsage") or {} + # The CLI may use a small model for side jobs; the answer comes from the one that wrote most. + model = max(mu, key=lambda m: (mu[m] or {}).get("outputTokens", 0)) if mu else "" + return { + "text": text, + "urls": list(dict.fromkeys(urls + URL_RX.findall(text))), + "model": model, + "usage": { + "turns": final.get("num_turns"), + "tools": tools, + "web_search_requests": tools.count("WebSearch"), + "web_fetch_requests": tools.count("WebFetch"), + "duration_ms": final.get("duration_ms"), + # What the CLI says the call would cost at API prices. Not charged on a subscription. + "api_equivalent_usd": final.get("total_cost_usd"), + }, + } + + +def claude_cli_answer(question: str, *, key: str = "", model: str = "default", + timeout: float = CLAUDE_CLI_TIMEOUT, env: dict | None = None) -> dict: + """Claude through the local Claude Code CLI, with only WebSearch and WebFetch available. + + Uses the operator's Claude subscription, not API credit, so `cost_usd` is 0 and the row + is flagged `subscription`. `key` is unused. `model="default"` lets the CLI pick. + """ + src = os.environ if env is None else env + exe = claude_exe(src) # from the caller's PATH, before the environment is replaced + with tempfile.TemporaryDirectory(prefix="aeb-home-") as home, \ + tempfile.TemporaryDirectory(prefix="aeb-cwd-") as cwd: + cenv = cli_env(home, src) + cenv["GIT_CEILING_DIRECTORIES"] = os.path.dirname(cwd) # never find a repo above the temp dir + real = _stage_auth(cenv["CLAUDE_CONFIG_DIR"], src) + mcp = os.path.join(home, "mcp.json") + with open(mcp, "w", encoding="utf-8") as f: + f.write('{"mcpServers": {}}') + cmd = [exe, "-p", "--output-format", "stream-json", "--verbose", + "--append-system-prompt", SYSTEM_HINT, "--max-turns", "12", + "--tools", _CLI_TOOLS, "--allowedTools", _CLI_TOOLS, + "--strict-mcp-config", "--mcp-config", mcp, "--setting-sources", "user"] + if model and model != "default": + cmd += ["--model", model] + try: + r = _run_cli(cmd, question, cwd=cwd, env=cenv, timeout=timeout) + finally: + if real is not None: + _write_back_auth(real, os.path.join(cenv["CLAUDE_CONFIG_DIR"], ".credentials.json")) + try: + out = parse_cli_stream(r.stdout or "") + except RuntimeError as e: + detail = (r.stderr or "").strip()[:200] + raise RuntimeError(f"{e} (exit {r.returncode}){': ' + detail if detail else ''}") from None + if r.returncode != 0: + raise RuntimeError(f"claude-cli exit {r.returncode}: {(r.stderr or '').strip()[:200]}") + out["model"] = out["model"] or model + out["cost_usd"] = 0.0 + out["subscription"] = True + return out + + @dataclass(frozen=True) class Engine: name: str - env_key: str + env_key: str # empty: the engine needs no key and runs only when asked for by name fn: Callable[..., dict] default_model: str @@ -197,6 +434,7 @@ class Engine: "gemini": Engine("gemini", "GEMINI_API_KEY", gemini_answer, "gemini-3.5-flash"), "claude": Engine("claude", "ANTHROPIC_API_KEY", anthropic_answer, "claude-sonnet-5"), "perplexity": Engine("perplexity", "PERPLEXITY_API_KEY", perplexity_answer, "perplexity/sonar"), + "claude-cli": Engine("claude-cli", "", claude_cli_answer, "default"), } @@ -208,26 +446,37 @@ class Bound: model: str key: str fn: Callable[..., dict] + timeout: float | None = None # None: the adapter's own default (180 s, claude-cli 600 s) def ask(self, question: str) -> dict: - return self.fn(question, key=self.key, model=self.model) + if self.timeout is None: + return self.fn(question, key=self.key, model=self.model) + return self.fn(question, key=self.key, model=self.model, timeout=self.timeout) def __repr__(self) -> str: # keep the key out of tracebacks and debug output return f"Bound(name={self.name!r}, model={self.model!r})" def available(env: dict | None = None, *, only: list[str] | None = None, - models: dict[str, str] | None = None) -> dict[str, Bound]: - """Engines whose key is set. `models` overrides a default model per engine.""" + models: dict[str, str] | None = None, timeout: float | None = None) -> dict[str, Bound]: + """Engines whose key is set, plus keyless engines named in `only`. `models` overrides a + default model per engine and `timeout` the per-call limit in seconds.""" env = os.environ if env is None else env models = models or {} out = {} for name, e in ENGINES.items(): if only and name not in only: continue - key = env.get(e.env_key, "") - if key: - out[name] = Bound(name, models.get(name, e.default_model), key, e.fn) + if e.env_key: + key = env.get(e.env_key, "") + if not key: + continue + elif only: + claude_exe(env) # fail now, with a clear message, rather than on every call + key = "" + else: + continue + out[name] = Bound(name, models.get(name, e.default_model), key, e.fn, timeout) return out diff --git a/src/answer_engine_benchmark/runner.py b/src/answer_engine_benchmark/runner.py index be8f5b4..74d31a6 100644 --- a/src/answer_engine_benchmark/runner.py +++ b/src/answer_engine_benchmark/runner.py @@ -50,6 +50,8 @@ def run(q: dict, engines: dict[str, Bound], *, runs: int = 3, out: Path, "urls": ans.get("urls", []), "usage": ans.get("usage", {}), "cost_usd": ans.get("cost_usd"), + # True when the call used a subscription (claude-cli), so cost_usd is 0. + "subscription": bool(ans.get("subscription")), **score(ans, q["ours"], q["brand_terms"], stale_markers=q["stale_markers"]), } f.write(json.dumps(rec, ensure_ascii=False) + "\n") diff --git a/src/answer_engine_benchmark/summary.py b/src/answer_engine_benchmark/summary.py index f048a93..13af5fd 100644 --- a/src/answer_engine_benchmark/summary.py +++ b/src/answer_engine_benchmark/summary.py @@ -85,6 +85,9 @@ def render(records: list[dict], ours: list[str], *, anonymise: bool = False, top lines.append(f"| {k} | {b['answered']} | {b['errors']} | {_pct(b['mention_rate'])} | " f"{b['cited']} of {b['answered']} ({_pct(b['citation_rate'])}) | " f"{b['avg_position'] or '-'} | {b['cost']:.4f} |") + subs = sorted({r["engine"] for r in records if r.get("subscription")}) + if subs: + lines += ["", f"{', '.join(subs)} ran on a subscription, not API credit, so its cost shows as 0."] engines = sorted({r["engine"] for r in records}) groups = list(dict.fromkeys(r["group"] for r in records)) diff --git a/tests/test_claude_cli.py b/tests/test_claude_cli.py new file mode 100644 index 0000000..940a0e7 --- /dev/null +++ b/tests/test_claude_cli.py @@ -0,0 +1,213 @@ +"""claude-cli engine tests. A fake `claude` script stands in for the real CLI, so nothing +here needs a login or makes a network call.""" + +import json +import os +import stat +import sys +import textwrap +import time + +import pytest + +from answer_engine_benchmark import engines as E +from answer_engine_benchmark.cli import main +from answer_engine_benchmark.runner import run + +SEARCH_RESULT = ('Web search results for query: "acme"\n\nLinks: [{"title": "A [1]", "url": "https://s.test/1"},' + '{"title": "B", "url": "https://acme.test/about"}]\n\nMore text') + +STREAM = [ + {"type": "system", "subtype": "init", "tools": ["WebFetch", "WebSearch"]}, + {"type": "assistant", "message": {"content": [ + {"type": "tool_use", "id": "t1", "name": "WebSearch", "input": {"query": "acme"}}]}}, + {"type": "user", "message": {"content": [{"type": "tool_result", "tool_use_id": "t1", "content": SEARCH_RESULT}]}}, + {"type": "assistant", "message": {"content": [ + {"type": "tool_use", "id": "t2", "name": "WebFetch", "input": {"url": "https://down.test/"}}, + {"type": "tool_use", "id": "t3", "name": "WebFetch", "input": {"url": "https://f.test/page"}}]}}, + {"type": "user", "message": {"content": [ + {"type": "tool_result", "tool_use_id": "t2", "content": "ENOTFOUND", "is_error": True}, + {"type": "tool_result", "tool_use_id": "t3", "content": [{"type": "text", "text": "page"}]}]}}, + {"type": "result", "subtype": "success", "is_error": False, "num_turns": 3, "duration_ms": 1200, + "result": "Acme makes widgets ([About](https://acme.test/about)). See https://t.test/x.", + "total_cost_usd": 0.0421, + "modelUsage": {"claude-haiku-5": {"outputTokens": 40}, "claude-opus-5": {"outputTokens": 900}}}, +] + + +def make_fake(tmp_path, *, stream=None, exit_code=0, stderr="", sleep=0.0): + """Put a fake `claude` on a fresh PATH. It records what it was given, then replies.""" + bindir = tmp_path / "bin" + bindir.mkdir(exist_ok=True) + record = tmp_path / "record.json" + out = "\n".join(json.dumps(e) for e in (STREAM if stream is None else stream)) + script = bindir / "claude" + script.write_text(textwrap.dedent(f"""\ + #!{sys.executable} + import json, os, sys, time + cfg = os.environ.get("CLAUDE_CONFIG_DIR", "") + cred = os.path.join(cfg, ".credentials.json") + json.dump({{"argv": sys.argv[1:], "stdin": sys.stdin.read(), "cwd": os.getcwd(), + "cwd_files": os.listdir("."), "env": dict(os.environ), + "cred": open(cred).read() if os.path.exists(cred) else None}}, + open({str(record)!r}, "w")) + time.sleep({sleep}) + sys.stdout.write({out!r}) + sys.stderr.write({stderr!r}) + sys.exit({exit_code}) + """)) + script.chmod(script.stat().st_mode | stat.S_IXUSR) + return bindir, record + + +@pytest.fixture +def user_home(tmp_path): + """A pretend operator: logged in, with a git identity and a CLAUDE.md.""" + home = tmp_path / "home" + (home / ".claude").mkdir(parents=True) + (home / ".claude" / ".credentials.json").write_text('{"claudeAiOauth": {"expiresAt": 100}}') + (home / ".claude" / "CLAUDE.md").write_text("The user owns Acme.") + (home / ".gitconfig").write_text("[user]\n name = operator-handle\n") + return home + + +def env_for(bindir, home, **extra): + return {"PATH": f"{bindir}{os.pathsep}/usr/bin{os.pathsep}/bin", "HOME": str(home), "USER": "operator-handle", + "GIT_AUTHOR_NAME": "operator-handle", "GIT_DIR": "/somewhere/.git", "ANTHROPIC_API_KEY": "sk-ant-x", + "CLAUDECODE": "1", "LANG": "C.UTF-8", **extra} + + +def test_answer_text_sources_and_zero_cost(tmp_path, user_home): + bindir, _ = make_fake(tmp_path) + out = E.claude_cli_answer("Who makes widgets?", env=env_for(bindir, user_home)) + assert out["text"].startswith("Acme makes widgets") + # Search results, then pages fetched without error, then links in the text; no repeats. + assert out["urls"] == ["https://s.test/1", "https://acme.test/about", "https://f.test/page", "https://t.test/x."] + assert out["model"] == "claude-opus-5" + assert out["cost_usd"] == 0.0 and out["subscription"] is True + assert out["usage"]["web_search_requests"] == 1 and out["usage"]["web_fetch_requests"] == 2 + assert out["usage"]["api_equivalent_usd"] == 0.0421 + + +def test_runs_as_a_stranger(tmp_path, user_home): + bindir, record = make_fake(tmp_path) + E.claude_cli_answer("-rf what is acme?", env=env_for(bindir, user_home)) + rec = json.loads(record.read_text()) + env = rec["env"] + # Fresh empty working dir and HOME, outside any repo, and gone afterwards. + assert rec["cwd_files"] == [] and not os.path.exists(rec["cwd"]) + assert env["HOME"] != str(user_home) and not os.path.exists(env["HOME"]) + assert env["CLAUDE_CONFIG_DIR"].startswith(env["HOME"]) + assert env["GIT_CONFIG_GLOBAL"] == os.devnull and env["GIT_CONFIG_NOSYSTEM"] == "1" + assert env["GIT_CEILING_DIRECTORIES"] == os.path.dirname(rec["cwd"]) + assert not [k for k in env if k.startswith("GIT_") and k not in + ("GIT_CONFIG_GLOBAL", "GIT_CONFIG_NOSYSTEM", "GIT_CEILING_DIRECTORIES")] + assert "operator-handle" not in json.dumps(env) + assert "ANTHROPIC_API_KEY" not in env and "CLAUDECODE" not in env + # Only the login was copied in; the operator's CLAUDE.md was not. + assert json.loads(rec["cred"]) == {"claudeAiOauth": {"expiresAt": 100}} + # The question goes in on stdin, so a leading "-" cannot become a flag. + assert rec["stdin"] == "-rf what is acme?" and "-rf what is acme?" not in rec["argv"] + argv = rec["argv"] + assert argv[:3] == ["-p", "--output-format", "stream-json"] + assert argv[argv.index("--tools") + 1] == "WebSearch,WebFetch" + assert argv[argv.index("--allowedTools") + 1] == "WebSearch,WebFetch" + assert "--strict-mcp-config" in argv and "--model" not in argv + + +def test_model_override_is_passed(tmp_path, user_home): + bindir, record = make_fake(tmp_path) + E.claude_cli_answer("q", model="sonnet", env=env_for(bindir, user_home)) + argv = json.loads(record.read_text())["argv"] + assert argv[argv.index("--model") + 1] == "sonnet" + + +def test_token_variable_replaces_the_credentials_file(tmp_path): + bindir, record = make_fake(tmp_path) + E.claude_cli_answer("q", env=env_for(bindir, tmp_path / "nobody", CLAUDE_CODE_OAUTH_TOKEN="tok")) + rec = json.loads(record.read_text()) + assert rec["env"]["CLAUDE_CODE_OAUTH_TOKEN"] == "tok" and rec["cred"] is None + + +def test_no_login_is_a_clear_error(tmp_path): + bindir, _ = make_fake(tmp_path) + with pytest.raises(E.EngineUnavailable, match="setup-token"): + E.claude_cli_answer("q", env=env_for(bindir, tmp_path / "nobody")) + + +def test_refreshed_login_is_written_back(tmp_path, user_home): + bindir, _ = make_fake(tmp_path) + script = bindir / "claude" + # The fake refreshes its login the way the real CLI does when the token is near expiry. + script.write_text(script.read_text().replace( + "time.sleep(", 'open(cred, "w").write(\'{"claudeAiOauth": {"expiresAt": 200}}\'); time.sleep(')) + E.claude_cli_answer("q", env=env_for(bindir, user_home)) + assert json.loads((user_home / ".claude" / ".credentials.json").read_text())["claudeAiOauth"]["expiresAt"] == 200 + + +def test_timeout(tmp_path, user_home): + bindir, _ = make_fake(tmp_path, sleep=30) + t0 = time.time() + with pytest.raises(TimeoutError, match="after 1s"): + E.claude_cli_answer("q", timeout=1, env=env_for(bindir, user_home)) + assert time.time() - t0 < 10 + + +def test_cli_error_result_and_bad_exit(tmp_path, user_home): + bindir, _ = make_fake(tmp_path, stream=[{"type": "result", "subtype": "error_max_turns", "is_error": True}], + exit_code=1) + with pytest.raises(RuntimeError, match="error_max_turns"): + E.claude_cli_answer("q", env=env_for(bindir, user_home)) + bindir, _ = make_fake(tmp_path, stream=[], exit_code=1, stderr="Invalid API key") + with pytest.raises(RuntimeError, match="Invalid API key"): + E.claude_cli_answer("q", env=env_for(bindir, user_home)) + + +def test_missing_cli_is_a_clear_error(tmp_path): + env = {"PATH": str(tmp_path)} + with pytest.raises(E.EngineUnavailable, match="no `claude` on PATH"): + E.available(env, only=["claude-cli"]) + with pytest.raises(E.EngineUnavailable): + E.claude_cli_answer("q", env=env) + + +def test_selected_only_by_name(tmp_path): + bindir, _ = make_fake(tmp_path) + env = {"PATH": str(bindir), "OPENAI_API_KEY": "sk-1"} + assert list(E.available(env)) == ["openai"] # never joins a run on its own + got = E.available(env, only=["claude-cli"], timeout=42) + assert list(got) == ["claude-cli"] and got["claude-cli"].timeout == 42 + assert E.available(env, only=["claude-cli"])["claude-cli"].timeout is None # the adapter's default + + +def test_bound_passes_timeout_only_when_set(): + seen = [] + + def old_style(question, *, key, model): # an adapter written before --timeout existed + seen.append(model) + return {} + + E.Bound("x", "m", "", old_style).ask("q") + assert seen == ["m"] + got = {} + E.Bound("x", "m", "", lambda q, **kw: got.update(kw) or {}, timeout=5).ask("q") + assert got["timeout"] == 5 + + +def test_run_records_subscription_flag(tmp_path, user_home, monkeypatch): + bindir, _ = make_fake(tmp_path) + for k, v in env_for(bindir, user_home).items(): + monkeypatch.setenv(k, v) + q = {"ours": ["acme.test"], "brand_terms": ["Acme"], "stale_markers": [], "groups": {"brand": ["What is Acme?"]}} + eng = E.available(only=["claude-cli"]) + recs = run(q, eng, runs=1, out=tmp_path / "out" / "answers.jsonl", sleep=0) + assert recs[0]["subscription"] is True and recs[0]["cost_usd"] == 0.0 + assert recs[0]["cited"] and recs[0]["engine"] == "claude-cli" + + +def test_cli_reports_a_missing_claude(tmp_path, monkeypatch, capsys): + qf = tmp_path / "q.yaml" + qf.write_text("ours: [acme.test]\nbrand_terms: [Acme]\ngroups:\n brand:\n - What is Acme?\n") + monkeypatch.setenv("PATH", str(tmp_path)) + assert main(["run", str(qf), "--engine", "claude-cli", "--out", str(tmp_path / "o")]) == 2 + assert "no `claude` on PATH" in capsys.readouterr().err