diff --git a/CHANGELOG.md b/CHANGELOG.md index 46ef2c3..1d4c351 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,17 @@ Every entry reports millisecond measurement error and a confusion matrix. See `d ## [Unreleased] +### Added +- **`hotato candidate`: a measured candidate-identity binding.** `candidate + hash --provider vapi --assistant ` fetches the candidate's configuration, + canonicalizes it (dropping volatile fields like ids and timestamps), and + prints a deterministic content hash to pass to `prove --candidate-config-hash`. + `candidate verify --expect ` re-fetches after the run and refuses + (exit 1) if the configuration drifted, so a Candidate Revision proof cannot + survive a swapped candidate. This makes the strongest proof scope produced by + a real, drift-checked flow rather than an operator-asserted string; the hash + is a measured binding of the configuration, not authentication of the runner. + ## [1.15.1] - 2026-07-22 ### Changed diff --git a/docs/LIFECYCLE.md b/docs/LIFECYCLE.md index 4f99890..2455a43 100644 --- a/docs/LIFECYCLE.md +++ b/docs/LIFECYCLE.md @@ -106,8 +106,25 @@ hotato prove --contracts contracts/ --before before/ --after after/ --out .hotat `pass` means every lane passed. Any failure fails the proof; a lane that could not support its claim is `inconclusive`, and CI never reads "could not tell" as -green. The deeper machinery is there when you need it: `hotato fix trial` binds -a specific patch to its before/after evidence, `hotato apply --clone` stages a +green. + +The Candidate Revision binding is measured, not asserted. `hotato candidate +hash --provider vapi --assistant ` fetches the candidate's configuration, +canonicalizes it (dropping volatile fields like timestamps and ids), and prints +its content hash. Re-run `hotato candidate verify --provider vapi --assistant + --expect ` after the before/after calls: it refuses (exit 1) if the +configuration drifted mid-run, so a Candidate Revision proof cannot survive a +swapped candidate. The end to end flow: + +```bash +hotato candidate hash --provider vapi --assistant asst_123 # -> sha256:... +hotato drive contracts/ID.hotato --stack vapi --assistant asst_123 --out after/ +hotato candidate verify --provider vapi --assistant asst_123 --expect sha256:... # exit 1 = drifted, void +hotato prove --before before/ --after after/ --candidate-config-hash sha256:... --provider vapi +``` + +The deeper machinery is there when you need it: `hotato fix trial` binds a +specific patch to its before/after evidence, `hotato apply --clone` stages a candidate without touching production, and `hotato record render` turns any failure into a share-safe card. diff --git a/llms-full.txt b/llms-full.txt index e10409e..e784ce6 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -9910,8 +9910,25 @@ hotato prove --contracts contracts/ --before before/ --after after/ --out .hotat `pass` means every lane passed. Any failure fails the proof; a lane that could not support its claim is `inconclusive`, and CI never reads "could not tell" as -green. The deeper machinery is there when you need it: `hotato fix trial` binds -a specific patch to its before/after evidence, `hotato apply --clone` stages a +green. + +The Candidate Revision binding is measured, not asserted. `hotato candidate +hash --provider vapi --assistant ` fetches the candidate's configuration, +canonicalizes it (dropping volatile fields like timestamps and ids), and prints +its content hash. Re-run `hotato candidate verify --provider vapi --assistant + --expect ` after the before/after calls: it refuses (exit 1) if the +configuration drifted mid-run, so a Candidate Revision proof cannot survive a +swapped candidate. The end to end flow: + +```bash +hotato candidate hash --provider vapi --assistant asst_123 # -> sha256:... +hotato drive contracts/ID.hotato --stack vapi --assistant asst_123 --out after/ +hotato candidate verify --provider vapi --assistant asst_123 --expect sha256:... # exit 1 = drifted, void +hotato prove --before before/ --after after/ --candidate-config-hash sha256:... --provider vapi +``` + +The deeper machinery is there when you need it: `hotato fix trial` binds a +specific patch to its before/after evidence, `hotato apply --clone` stages a candidate without touching production, and `hotato record render` turns any failure into a share-safe card. diff --git a/src/hotato/candidate.py b/src/hotato/candidate.py new file mode 100644 index 0000000..3f3f009 --- /dev/null +++ b/src/hotato/candidate.py @@ -0,0 +1,281 @@ +"""Candidate-identity binding: the hermetic half of the candidate-bound proof path. + +``hotato prove --candidate-config-hash H --provider vapi`` raises a before/after +proof to claim_scope ``candidate_revision`` -- but only when ``H`` actually binds +the candidate that ran. This module is what PRODUCES a legitimate ``H``: it +fetches the provider's live assistant config, strips the volatile identity noise +that changes without a semantic change, and computes a deterministic sha256 over +what remains. The same config hashes identically on every machine and every run; +a real config change (a moved interruption threshold, a new system prompt) moves +the hash, and nothing else does. + +The flow the operator runs: + + hotato candidate hash --provider vapi --assistant # BEFORE + # ... drive the before/after calls against the candidate ... + hotato candidate verify --provider vapi --assistant --expect H # AFTER + hotato prove --before ... --after ... --candidate-config-hash H --provider vapi + +``candidate verify`` is the refuse-on-drift gate: it re-fetches, recomputes, and +exits non-zero when the config no longer hashes to ``H``. A drifted candidate +means the before/after calls and the config no longer describe the same thing, so +a ``candidate_revision`` proof over that run is void. + +Scope of the binding (documented plainly, matching prove's honesty walls): the +config hash is a MEASURED binding of the candidate's CONFIGURATION -- it says +"these before/after calls ran against a config that hashes to H" -- it is NOT an +authentication of the runner. ``prove``'s evidence_authority stays ``measured``; +this binding never claims ``runner_authenticated``. + +Zero third-party dependencies: stdlib ``hashlib`` / ``json`` / ``os`` for the +pure hashing path, and the vendored ``capture`` HTTP primitives (stdlib +``urllib`` under the hood, with the credential-safe redirect handler and the +default-deny SSRF guard) for the one network function. +""" + +from __future__ import annotations + +import hashlib +import json +import os +from typing import Any, Dict, Optional, Tuple + +from .errors import load_json_file as _load_json_file + +__all__ = [ + "VOLATILE_KEYS", + "PROVIDERS", + "canonicalize_config", + "config_hash", + "fetch_vapi_assistant", + "fetch_assistant", + "resolve_api_key", + "hash_from_config_file", + "hash_from_provider", + "verify_from_provider", +] + +# Config keys that change WITHOUT a semantic change, so two fetches of the SAME +# logical candidate config must hash identically regardless of them. Stripped at +# EVERY nesting level (a nested ``updatedAt`` is dropped too). This set is +# deliberately CONSERVATIVE and EXPLICIT: anything not listed here is kept, so a +# genuine semantic field is never silently dropped from the binding. +# * id / orgId -- server-assigned identity of the assistant/org. +# * createdAt / updatedAt -- server-set timestamps that move on every read +# that touches the record, with no config change. +# * isServerUrlSecretSet -- a server-set secret-PRESENCE flag (whether a +# server-url secret exists), not the config itself. +VOLATILE_KEYS = frozenset({ + "id", + "orgId", + "createdAt", + "updatedAt", + "isServerUrlSecretSet", +}) + +# Providers whose live assistant config `candidate hash`/`verify` can fetch. Only +# vapi is implemented; the dispatch refuses an unknown provider as a usage error +# so more can be added without changing the surface. +PROVIDERS = ("vapi",) + +_PROVIDER_ENV = {"vapi": "VAPI_API_KEY"} +_PROVIDER_BASE_URL = {"vapi": "https://api.vapi.ai"} + + +# ========================================================================= +# pure path: canonicalize + hash (stdlib-only, deterministic, no network) +# ========================================================================= + +def _strip_volatile(value: Any) -> Any: + """Recursively drop every :data:`VOLATILE_KEYS` key from ``value``. Recurses + into nested dicts AND lists so a volatile key nested anywhere is removed.""" + if isinstance(value, dict): + return { + key: _strip_volatile(val) + for key, val in value.items() + if key not in VOLATILE_KEYS + } + if isinstance(value, list): + return [_strip_volatile(item) for item in value] + return value + + +def canonicalize_config(config: Dict[str, Any]) -> Dict[str, Any]: + """Return ``config`` with the volatile identity-noise keys stripped at every + level, so two fetches of the same logical config canonicalize identically. + + Only the keys in :data:`VOLATILE_KEYS` are removed; every semantic field + (model, voice, firstMessage, transcriber, the start/stopSpeakingPlan + interruption settings, the system prompt, tools, ...) is kept verbatim.""" + if not isinstance(config, dict): + raise ValueError( + "a candidate config must be a JSON object (the provider's assistant " + f"config), got {type(config).__name__}" + ) + return _strip_volatile(config) + + +def config_hash(config: Dict[str, Any]) -> str: + """``"sha256:" + sha256`` over the canonical JSON of + :func:`canonicalize_config` -- sorted keys, no insignificant whitespace, + ``ensure_ascii`` -- so the same logical config yields the same hash on every + machine and run, and a semantic change (and only a semantic change) moves it. + ``allow_nan=False`` refuses a digest over a value that cannot round-trip + through a standard JSON reader.""" + canonical = canonicalize_config(config) + blob = json.dumps( + canonical, sort_keys=True, separators=(",", ":"), + ensure_ascii=True, allow_nan=False, + ).encode("utf-8") + return "sha256:" + hashlib.sha256(blob).hexdigest() + + +# ========================================================================= +# network path: fetch a live assistant config (the ONLY network function) +# ========================================================================= + +def fetch_vapi_assistant( + assistant_id: str, + api_key: str, + base_url: str = "https://api.vapi.ai", + timeout: int = 30, +) -> Dict[str, Any]: + """``GET {base_url}/assistant/{assistant_id}`` with ``Authorization: Bearer + `` and return the parsed assistant config object. + + Thin and separately testable: it does nothing but fetch and parse. The + response is treated as untrusted data -- parsed as JSON (never eval'd) and + required to be a JSON object. Network / HTTP / non-JSON failures surface as + the tool's standard clean usage ``ValueError`` (via the shared capture HTTP + primitives, which also carry the credential-safe redirect handler and the + default-deny SSRF guard), never a raw traceback.""" + from urllib.parse import quote + + from . import capture as _capture + + aid = str(assistant_id or "").strip() + if not aid: + raise ValueError( + "an assistant id is required to fetch a Vapi assistant config" + ) + key = str(api_key or "").strip() + if not key: + raise ValueError( + "a Vapi API key is required to fetch an assistant config" + ) + url = f"{str(base_url).rstrip('/')}/assistant/{quote(aid, safe='')}" + obj = _capture._http_get_json( + url, + headers={"Authorization": f"Bearer {key}", "Accept": "application/json"}, + timeout=timeout, + ) + return _capture._require_json_object(obj, f"Vapi assistant {aid!r}") + + +def _require_known_provider(provider: Optional[str]) -> str: + prov = (provider or "").strip().lower() + if prov not in PROVIDERS: + raise ValueError( + f"unknown provider {provider!r}; candidate binding supports: " + + ", ".join(PROVIDERS) + ) + return prov + + +def fetch_assistant( + provider: str, + assistant_id: str, + api_key: str, + *, + base_url: Optional[str] = None, + timeout: int = 30, +) -> Dict[str, Any]: + """Provider dispatch for the live-config fetch. Only ``vapi`` is implemented; + an unknown provider is a clean usage error so more can be added later.""" + prov = _require_known_provider(provider) + if prov == "vapi": + return fetch_vapi_assistant( + assistant_id, api_key, + base_url=base_url or _PROVIDER_BASE_URL["vapi"], + timeout=timeout, + ) + # unreachable: _require_known_provider gates PROVIDERS, kept as a guard so a + # newly-listed provider without a branch fails loudly rather than silently. + raise ValueError( # pragma: no cover + f"provider {prov!r} is listed but has no fetch implementation" + ) + + +def resolve_api_key(provider: str, api_key: Optional[str]) -> str: + """The API key for ``provider``: the explicit ``--api-key`` if given, else the + provider's environment variable (``VAPI_API_KEY`` for vapi). Raises a clean + usage error when neither is present.""" + if api_key and str(api_key).strip(): + return str(api_key).strip() + prov = (provider or "").strip().lower() + env = _PROVIDER_ENV.get(prov) + if env: + val = os.environ.get(env, "") + if val and val.strip(): + return val.strip() + raise ValueError( + f"a {prov or 'provider'} API key is required: pass --api-key or set the " + f"{env or 'provider'} environment variable" + ) + + +# ========================================================================= +# orchestration: the two things the CLI does (kept here so both are testable) +# ========================================================================= + +def hash_from_config_file(path: str) -> Tuple[str, Dict[str, Any]]: + """Hash a LOCAL config file (no network, no key). Returns + ``(config_hash, canonicalized_config)``.""" + config = _load_json_file(path, label=f"candidate config {path!r}") + return config_hash(config), canonicalize_config(config) + + +def hash_from_provider( + provider: str, + assistant: str, + api_key: Optional[str], + *, + base_url: Optional[str] = None, + timeout: int = 30, +) -> Tuple[str, Dict[str, Any]]: + """Fetch the live config and hash it. Returns + ``(config_hash, canonicalized_config)``.""" + prov = _require_known_provider(provider) + key = resolve_api_key(prov, api_key) + config = fetch_assistant(prov, assistant, key, base_url=base_url, timeout=timeout) + return config_hash(config), canonicalize_config(config) + + +def verify_from_provider( + provider: str, + assistant: str, + expect: str, + api_key: Optional[str], + *, + base_url: Optional[str] = None, + timeout: int = 30, +) -> Dict[str, Any]: + """Re-fetch, recompute, and compare to ``expect``. Returns + ``{provider, expected, actual, held}`` where ``held`` is True only when the + recomputed hash equals ``expect`` (the candidate is unchanged).""" + expected = str(expect or "").strip() + if not expected: + raise ValueError( + "candidate verify needs --expect HASH: the config hash you recorded " + "before the change (from `hotato candidate hash`)" + ) + prov = _require_known_provider(provider) + key = resolve_api_key(prov, api_key) + config = fetch_assistant(prov, assistant, key, base_url=base_url, timeout=timeout) + actual = config_hash(config) + return { + "provider": prov, + "expected": expected, + "actual": actual, + "held": actual == expected, + } diff --git a/src/hotato/cli.py b/src/hotato/cli.py index fae6ca4..0da2395 100644 --- a/src/hotato/cli.py +++ b/src/hotato/cli.py @@ -621,6 +621,26 @@ def _atomic_write_json(path: str, obj) -> None: "inconclusive and none failed) -- 'could not tell' is never " "green"), ), + "candidate": ( + (2, "no subcommand given (see hotato candidate hash/verify --help)"), + ), + "candidate hash": ( + (0, "computed the candidate config hash (printed; optionally saved the " + "canonicalized config); pass it to hotato prove " + "--candidate-config-hash"), + (2, "usage error (neither --config nor --provider/--assistant, an " + "unknown provider, a missing API key, an unreadable config file) or " + "a network/HTTP error fetching the live config"), + ), + "candidate verify": ( + (0, "the candidate configuration is unchanged: its recomputed hash " + "matches --expect, so a candidate_revision proof over this run " + "holds"), + (1, "drift: the recomputed hash does not match --expect; a " + "candidate_revision proof over this run is void"), + (2, "usage error (an unknown provider, a missing API key, a missing " + "--expect) or a network/HTTP error fetching the live config"), + ), "loop": ( (0, "advanced the loop and persisted state (or re-reported where it " "left off)"), @@ -2150,6 +2170,97 @@ def _cmd_prove(args) -> int: return proof["exit_code"] +def _cmd_candidate_hash(args) -> int: + from . import candidate as _candidate + + if args.config: + # Local path: hash a config FILE, no network, no provider, no key. + if args.assistant or args.provider or args.api_key: + raise ValueError( + "candidate hash --config hashes a LOCAL config file and takes no " + "--provider / --assistant / --api-key (those fetch a live config " + "instead). Pass either --config FILE or " + "--provider NAME --assistant ID." + ) + chash, canonical = _candidate.hash_from_config_file(args.config) + provider = None + else: + if not args.provider: + raise ValueError( + "candidate hash needs either --config FILE (hash a local config) " + "or --provider NAME --assistant ID (fetch and hash a live config)" + ) + if not args.assistant: + raise ValueError( + "candidate hash --provider NAME needs --assistant ID to fetch" + ) + chash, canonical = _candidate.hash_from_provider( + args.provider, args.assistant, args.api_key, base_url=args.base_url, + ) + provider = args.provider.strip().lower() + if args.out: + # The canonicalized config (volatile keys stripped), written + # deterministically. It is share-safe: the API key is never in the + # provider's config response, and no absolute path is embedded. + _atomic_write_text( + args.out, + json.dumps(canonical, indent=2, sort_keys=True, ensure_ascii=True) + + "\n", + ) + print(f"wrote canonical config -> {args.out}", file=sys.stderr) + if args.format == "json": + result = {"tool": "hotato", "command": "candidate hash", + "config_hash": chash} + if provider: + result["provider"] = provider + print(_errors.safe_json_dumps(result, indent=2)) + else: + print("hotato [candidate hash]") + if provider: + print(f" provider: {provider}") + print(f" config_hash: {chash}") + prov_txt = provider or "vapi" + print( + " pass to prove: hotato prove --before ... --after ... " + f"--candidate-config-hash {chash} --provider {prov_txt}" + ) + print( + " note: a MEASURED binding of the candidate's configuration, not an " + "authentication of the runner (evidence_authority stays 'measured')" + ) + return 0 + + +def _cmd_candidate_verify(args) -> int: + from . import candidate as _candidate + + result = _candidate.verify_from_provider( + args.provider, args.assistant, args.expect, args.api_key, + base_url=args.base_url, + ) + held = result["held"] + if args.format == "json": + print(_errors.safe_json_dumps({ + "tool": "hotato", "command": "candidate verify", + "provider": result["provider"], + "expected": result["expected"], "actual": result["actual"], + "held": held, "exit_code": 0 if held else 1, + }, indent=2)) + else: + print("hotato [candidate verify]") + print(f" provider: {result['provider']}") + if held: + print(f" config_hash: {result['actual']}") + print(" HELD: the candidate configuration is unchanged; a " + "candidate_revision proof over this run is valid") + else: + print(f" expected (before): {result['expected']}") + print(f" actual (after): {result['actual']}") + print(" DRIFTED: the candidate configuration changed; a " + "candidate_revision proof over this run is void") + return 0 if held else 1 + + def _cmd_loop(args) -> int: from . import loop as _loop @@ -9079,6 +9190,119 @@ def build_parser() -> argparse.ArgumentParser: _add_format_arg(pv) pv.set_defaults(func=_cmd_prove) + # --- candidate: compute + verify the candidate-config binding hash ------ + cand = sub.add_parser( + "candidate", + help="compute and verify the candidate-config binding hash that raises " + "a before/after proof to Candidate Revision " + "(hotato candidate hash/verify)", + description=( + "Candidate-identity binding: the hermetic half of the " + "candidate-bound proof path. `candidate hash` fetches a provider's " + "live assistant config, strips the volatile identity noise that " + "changes without a semantic change (id, orgId, createdAt, " + "updatedAt, isServerUrlSecretSet), and prints a deterministic " + "sha256 over the remaining semantic fields (model, voice, " + "firstMessage, transcriber, the start/stopSpeakingPlan interruption " + "settings, the system prompt, tools). `candidate verify` " + "re-fetches, recomputes, and REFUSES on drift. The flow: " + "candidate hash (before the change) -> drive the before/after " + "calls -> candidate verify --expect (after) -> hotato prove " + "--before ... --after ... --candidate-config-hash --provider " + "vapi. The hash is a MEASURED binding of the candidate's " + "configuration, not an authentication of the runner: " + "evidence_authority stays 'measured' and the proof never claims " + "runner_authenticated." + ), + epilog=_exit_codes_epilog("candidate"), + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + candsub = cand.add_subparsers(dest="candidate_command", required=True, + metavar="hash|verify") + + ch = candsub.add_parser( + "hash", + help="compute the candidate config hash from a live provider config " + "(or a local --config file)", + description=( + "Fetch a provider's live assistant config, strip the volatile " + "identity noise, and print a deterministic config hash to pass to " + "hotato prove --candidate-config-hash. With --config it hashes a " + "LOCAL config file instead (no network, no key), so the pure hashing " + "path is usable without a provider account. --out saves the " + "canonicalized config (volatile keys stripped) for inspection; it " + "carries no API key (the key is never in the provider's config " + "response). The hash is a MEASURED binding of the configuration, not " + "an authentication of the runner." + ), + epilog=( + _exit_codes_epilog("candidate hash") + "\n\n" + "Examples:\n" + " hotato candidate hash --provider vapi --assistant asst_123\n" + " VAPI_API_KEY=... hotato candidate hash --provider vapi " + "--assistant asst_123 --out candidate-config.json\n" + " hotato candidate hash --config candidate-config.json" + ), + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + ch.add_argument("--provider", default=None, metavar="NAME", + help="the candidate's provider/runner (vapi); with " + "--assistant it fetches and hashes the live config") + ch.add_argument("--assistant", default=None, metavar="ID", + help="the assistant id to fetch (with --provider)") + ch.add_argument("--api-key", dest="api_key", default=None, metavar="KEY", + help="the provider API key; falls back to VAPI_API_KEY " + "(vapi) in the environment") + ch.add_argument("--config", default=None, metavar="FILE", + help="hash a LOCAL config file instead of fetching (no " + "network, no key); mutually exclusive with " + "--provider/--assistant") + ch.add_argument("--base-url", dest="base_url", default=None, metavar="URL", + help="override the provider API base URL (default " + "https://api.vapi.ai for vapi)") + ch.add_argument("--out", default=None, metavar="FILE", + help="also save the canonicalized config (volatile keys " + "stripped) as JSON") + _add_format_arg(ch) + ch.set_defaults(func=_cmd_candidate_hash) + + cvf = candsub.add_parser( + "verify", + help="re-fetch the live config, recompute its hash, and REFUSE on drift " + "from --expect", + description=( + "The refuse-on-drift gate. Re-fetch the provider's live assistant " + "config, recompute its hash, and compare it to --expect (the hash " + "you recorded with `candidate hash` before the change). Exit 0 when " + "the config is unchanged (the candidate held), exit 1 when it " + "DRIFTED (the candidate_revision claim over this run is void), exit 2 " + "on a usage or network error. Run it after the before/after calls: " + "if it fails, the proof cannot honestly claim candidate_revision." + ), + epilog=( + _exit_codes_epilog("candidate verify") + "\n\n" + "Examples:\n" + " hotato candidate verify --provider vapi --assistant asst_123 " + "--expect 'sha256:THE_HASH'" + ), + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + cvf.add_argument("--provider", default=None, metavar="NAME", + help="the candidate's provider/runner (vapi)") + cvf.add_argument("--assistant", default=None, metavar="ID", + help="the assistant id to re-fetch") + cvf.add_argument("--expect", required=True, metavar="HASH", + help="the config hash recorded before the change (from " + "hotato candidate hash); drift from it is refused") + cvf.add_argument("--api-key", dest="api_key", default=None, metavar="KEY", + help="the provider API key; falls back to VAPI_API_KEY " + "(vapi) in the environment") + cvf.add_argument("--base-url", dest="base_url", default=None, metavar="URL", + help="override the provider API base URL (default " + "https://api.vapi.ai for vapi)") + _add_format_arg(cvf) + cvf.set_defaults(func=_cmd_candidate_verify) + # --- loop: one-command orchestration of the closed loop, with memory ---- lp = sub.add_parser( "loop", diff --git a/tests/test_candidate.py b/tests/test_candidate.py new file mode 100644 index 0000000..e08160a --- /dev/null +++ b/tests/test_candidate.py @@ -0,0 +1,322 @@ +"""``hotato candidate``: the candidate-identity binding -- the hermetic half of +the candidate-bound proof path. + +Pinned here (all hermetic, NO network -- the one network function is +monkeypatched out): + + * canonicalize strips the volatile identity noise (id/orgId/createdAt/ + updatedAt/isServerUrlSecretSet), including NESTED, and keeps every semantic + field (model/voice/startSpeakingPlan/...); + * config_hash is deterministic: two fetches of the same logical config (with + the volatile fields differing) hash identically, and a real semantic change + (a moved stopSpeakingPlan threshold) moves the hash -- drift detection fires + on real changes and not on noise; + * `candidate hash --config FILE` prints the expected hash with no network; + * `candidate verify` is the refuse-on-drift gate: unchanged -> exit 0, + semantically changed -> exit 1, unknown provider / missing key -> exit 2; + * the hash `candidate hash` emits, fed to `prove --candidate-config-hash`, + yields claim_scope candidate_revision; + * share-safety: the saved/canonical config and the printed output carry no + API key and no absolute path. +""" + +from __future__ import annotations + +import json + +import pytest + +from hotato import candidate as _candidate +from hotato import cli + +# --- a fixture Vapi assistant config (the shape `candidate hash` fetches) --- +# Two logically-identical fetches differ ONLY in volatile fields; a third is a +# genuine semantic change (a moved stopSpeakingPlan interruption threshold). + +def _assistant_config(*, updated_at="2026-07-20T10:00:00.000Z", + num_words=0, server_secret_set=False): + return { + "id": "asst_" + "0" * 24, + "orgId": "org_" + "1" * 24, + "createdAt": "2026-07-01T09:00:00.000Z", + "updatedAt": updated_at, + "isServerUrlSecretSet": server_secret_set, + "name": "support-v3", + "firstMessage": "Hi, how can I help?", + "model": { + "provider": "openai", + "model": "gpt-4o", + "messages": [ + {"role": "system", "content": "You are a support agent."}, + ], + # A nested volatile key: must be stripped at depth too. + "updatedAt": "2026-07-19T00:00:00.000Z", + }, + "voice": {"provider": "11labs", "voiceId": "burt"}, + "transcriber": {"provider": "deepgram", "model": "nova-2"}, + "startSpeakingPlan": {"waitSeconds": 0.4}, + "stopSpeakingPlan": {"numWords": num_words, "voiceSeconds": 0.2}, + } + + +# ========================================================================= +# 1. canonicalize: strips volatile (incl. nested), keeps semantic +# ========================================================================= + +def test_canonicalize_strips_volatile_including_nested_and_keeps_semantic(): + canon = _candidate.canonicalize_config(_assistant_config()) + # volatile identity noise gone at the top level ... + for key in ("id", "orgId", "createdAt", "updatedAt", "isServerUrlSecretSet"): + assert key not in canon + # ... and nested (the updatedAt inside model) gone too + assert "updatedAt" not in canon["model"] + # every semantic field kept + assert canon["model"]["model"] == "gpt-4o" + assert canon["model"]["messages"][0]["content"] == "You are a support agent." + assert canon["voice"] == {"provider": "11labs", "voiceId": "burt"} + assert canon["transcriber"]["model"] == "nova-2" + assert canon["startSpeakingPlan"] == {"waitSeconds": 0.4} + assert canon["stopSpeakingPlan"] == {"numWords": 0, "voiceSeconds": 0.2} + assert canon["firstMessage"] == "Hi, how can I help?" + + +def test_canonicalize_does_not_mutate_the_input(): + original = _assistant_config() + _candidate.canonicalize_config(original) + assert "id" in original and "updatedAt" in original["model"] + + +def test_canonicalize_rejects_a_non_object(): + with pytest.raises(ValueError): + _candidate.canonicalize_config(["not", "an", "object"]) + + +# ========================================================================= +# 2. config_hash: deterministic on noise, moves on a real semantic change +# ========================================================================= + +def test_config_hash_is_stable_across_volatile_only_differences(): + a = _candidate.config_hash( + _assistant_config(updated_at="2026-07-20T10:00:00.000Z", + server_secret_set=False)) + b = _candidate.config_hash( + _assistant_config(updated_at="2026-07-22T23:59:59.000Z", + server_secret_set=True)) + assert a == b + assert a.startswith("sha256:") + + +def test_config_hash_moves_on_a_semantic_change(): + base = _candidate.config_hash(_assistant_config(num_words=0)) + changed = _candidate.config_hash(_assistant_config(num_words=3)) + assert base != changed + + +def test_config_hash_is_reproducible_byte_for_byte(): + cfg = _assistant_config() + assert _candidate.config_hash(cfg) == _candidate.config_hash(cfg) + + +# ========================================================================= +# 3. `candidate hash --config FILE`: the pure path, no network +# ========================================================================= + +def _write_config(tmp_path, cfg, name="candidate-config.json"): + path = tmp_path / name + path.write_text(json.dumps(cfg), encoding="utf-8") + return path + + +def test_candidate_hash_config_file_prints_the_expected_hash(tmp_path, capsys): + cfg = _assistant_config() + path = _write_config(tmp_path, cfg) + expected = _candidate.config_hash(cfg) + + rc = cli.main(["candidate", "hash", "--config", str(path)]) + assert rc == 0 + out = capsys.readouterr().out + assert expected in out + + +def test_candidate_hash_config_file_json_format(tmp_path, capsys): + cfg = _assistant_config() + path = _write_config(tmp_path, cfg) + expected = _candidate.config_hash(cfg) + + rc = cli.main(["candidate", "hash", "--config", str(path), "--format", "json"]) + assert rc == 0 + payload = json.loads(capsys.readouterr().out) + assert payload["config_hash"] == expected + assert payload["command"] == "candidate hash" + + +def test_candidate_hash_config_with_provider_is_a_usage_error(tmp_path, capsys): + path = _write_config(tmp_path, _assistant_config()) + rc = cli.main(["candidate", "hash", "--config", str(path), + "--provider", "vapi"]) + assert rc == 2 + + +# ========================================================================= +# 4. `candidate verify`: refuse-on-drift (fetch monkeypatched, no network) +# ========================================================================= + +def _patch_fetch(monkeypatch, config): + """Inject the live-config fetch so no network is touched. Fails loudly if a + real API key is ever required (there must be none in a hermetic test).""" + def _fake(provider, assistant_id, api_key, *, base_url=None, timeout=30): + assert api_key, "the resolved API key must reach the fetch" + return config + monkeypatch.setattr(_candidate, "fetch_assistant", _fake) + + +def test_candidate_verify_holds_when_unchanged(monkeypatch, capsys): + cfg = _assistant_config() + _patch_fetch(monkeypatch, cfg) + expected = _candidate.config_hash(cfg) + + rc = cli.main(["candidate", "verify", "--provider", "vapi", + "--assistant", "asst_123", "--expect", expected, + "--api-key", "vapi-key"]) + assert rc == 0 + assert "HELD" in capsys.readouterr().out + + +def test_candidate_verify_refuses_on_drift(monkeypatch, capsys): + before = _assistant_config(num_words=0) + after = _assistant_config(num_words=3) # a real semantic change + expected = _candidate.config_hash(before) + _patch_fetch(monkeypatch, after) # the live config drifted + + rc = cli.main(["candidate", "verify", "--provider", "vapi", + "--assistant", "asst_123", "--expect", expected, + "--api-key", "vapi-key"]) + assert rc == 1 + out = capsys.readouterr().out + assert "DRIFTED" in out + assert expected in out # before hash shown + assert _candidate.config_hash(after) in out # after hash shown + + +def test_candidate_verify_holds_across_volatile_only_drift(monkeypatch, capsys): + # A server-touched updatedAt / secret flag is NOT drift: the gate must not + # fire on noise, only on a real config change. + before = _assistant_config(updated_at="2026-07-20T10:00:00.000Z") + after = _assistant_config(updated_at="2026-07-23T11:11:11.000Z", + server_secret_set=True) + expected = _candidate.config_hash(before) + _patch_fetch(monkeypatch, after) + + rc = cli.main(["candidate", "verify", "--provider", "vapi", + "--assistant", "asst_123", "--expect", expected, + "--api-key", "vapi-key"]) + assert rc == 0 + + +def test_candidate_verify_unknown_provider_is_exit_2(monkeypatch, capsys): + _patch_fetch(monkeypatch, _assistant_config()) + rc = cli.main(["candidate", "verify", "--provider", "nope", + "--assistant", "asst_123", "--expect", "sha256:x", + "--api-key", "k"]) + assert rc == 2 + + +def test_candidate_verify_missing_key_is_exit_2(monkeypatch, capsys): + _patch_fetch(monkeypatch, _assistant_config()) + monkeypatch.delenv("VAPI_API_KEY", raising=False) + rc = cli.main(["candidate", "verify", "--provider", "vapi", + "--assistant", "asst_123", "--expect", "sha256:x"]) + assert rc == 2 + + +def test_candidate_hash_unknown_provider_is_exit_2(monkeypatch, capsys): + _patch_fetch(monkeypatch, _assistant_config()) + rc = cli.main(["candidate", "hash", "--provider", "nope", + "--assistant", "asst_123", "--api-key", "k"]) + assert rc == 2 + + +def test_candidate_hash_missing_key_is_exit_2(monkeypatch, capsys): + _patch_fetch(monkeypatch, _assistant_config()) + monkeypatch.delenv("VAPI_API_KEY", raising=False) + rc = cli.main(["candidate", "hash", "--provider", "vapi", + "--assistant", "asst_123"]) + assert rc == 2 + + +def test_candidate_hash_api_key_falls_back_to_env(monkeypatch, capsys): + cfg = _assistant_config() + _patch_fetch(monkeypatch, cfg) + monkeypatch.setenv("VAPI_API_KEY", "env-key") + rc = cli.main(["candidate", "hash", "--provider", "vapi", + "--assistant", "asst_123"]) + assert rc == 0 + assert _candidate.config_hash(cfg) in capsys.readouterr().out + + +# ========================================================================= +# 5. the emitted hash feeds prove -> claim_scope candidate_revision +# ========================================================================= + +def _empty_dir(tmp_path, name): + d = tmp_path / name + d.mkdir() + return d + + +def test_hash_feeds_prove_and_yields_candidate_revision( + tmp_path, monkeypatch, capsys): + monkeypatch.setenv("HOTATO_HOME", str(tmp_path / "home")) + cfg = _assistant_config() + path = _write_config(tmp_path, cfg) + + # 1. compute the hash the pure way (no network). + rc = cli.main(["candidate", "hash", "--config", str(path), "--format", "json"]) + assert rc == 0 + the_hash = json.loads(capsys.readouterr().out)["config_hash"] + + # 2. feed it to prove alongside a before/after lane (the test_prove.py + # empty-dir pattern) -> the proof reads Candidate Revision. + before = _empty_dir(tmp_path, "before") + after = _empty_dir(tmp_path, "after") + out = tmp_path / "proofout" + cli.main(["prove", "--before", str(before), "--after", str(after), + "--candidate-config-hash", the_hash, "--provider", "vapi", + "--out", str(out)]) + with open(out / "proof.json", encoding="utf-8") as fh: + proof = json.load(fh) + assert proof["claim_scope"] == "candidate_revision" + assert proof["evidence"]["candidate_config_hash"] == the_hash + assert proof["evidence"]["provider"] == "vapi" + # the runner is not authenticated in this release + assert proof["evidence_authority"] == "measured" + + +# ========================================================================= +# 6. share-safety: no API key, no absolute path in output or the saved config +# ========================================================================= + +def test_saved_canonical_config_and_output_carry_no_secret_or_path( + tmp_path, monkeypatch, capsys): + cfg = _assistant_config() + _patch_fetch(monkeypatch, cfg) + saved = tmp_path / "canon.json" + + rc = cli.main(["candidate", "hash", "--provider", "vapi", + "--assistant", "asst_123", "--api-key", "super-secret-key", + "--out", str(saved)]) + assert rc == 0 + + # the saved canonical config: no API key, no absolute path + saved_text = saved.read_text(encoding="utf-8") + assert "super-secret-key" not in saved_text + assert str(tmp_path) not in saved_text + # it is the canonicalized config (volatile stripped, semantic kept) + saved_obj = json.loads(saved_text) + assert "id" not in saved_obj + assert saved_obj["model"]["model"] == "gpt-4o" + + # the machine surface (stdout): the hash, no API key + stdout = capsys.readouterr().out + assert "super-secret-key" not in stdout + assert _candidate.config_hash(cfg) in stdout diff --git a/tests/test_describe_cli.py b/tests/test_describe_cli.py index 6b1865a..4deed25 100644 --- a/tests/test_describe_cli.py +++ b/tests/test_describe_cli.py @@ -50,7 +50,8 @@ "simulate", "compare", "scan", "synth", "battery", "battery robustness", "gauntlet", "gauntlet badge", "trust", - "ingest", "analyze", "verify", "fix", "fix trial", "prove", "loop", + "ingest", "analyze", "verify", "fix", "fix trial", "prove", + "candidate", "candidate hash", "candidate verify", "loop", "investigate", "investigate label", "describe", "init", "init webhook", "init starter", "init ci", "issue", "issue create",