From 2a5926cade7f91bbab816c22184bcadc8fe0679d Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 5 Oct 2026 18:51:43 +0000 Subject: [PATCH] Add a live probe for the Decisions API contract The jev decision adapter was written against a partly confirmed contract: the fields inside one question and one answer were inferred. scripts/probe-jev-decisions.py settles it against a live endpoint. - Sends one decision through OpenRouterJevDecider itself, so the request body, headers, deadline and answer reader are production's, recorded at the transport so the raw response can be shown beside what Cerberus read from it. - The decision state is synthetic: a made-up three-model pool described by the real state builder, and a fixed or --prompt user message. Nothing from a configuration or a real request is sent; the key is never printed. - Asks the route and effort questions by default (--no-effort for the route alone), then names the outcome: accepted and read, request rejected (with the error body), unreadable response, answer not read (pointing at _answer), or answered outside the offered options. - Exit 0 only when the request is accepted and every answer is an offered option; 1 otherwise; 2 when it cannot run (no key). - tests/test_probe_jev_decisions.py drives it offline against a mocked endpoint for each outcome. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01P4Nm4MVSMmG6JyZi8WumtK --- README.md | 9 ++ docs/architecture.md | 5 +- scripts/probe-jev-decisions.py | 255 ++++++++++++++++++++++++++++++ tests/test_probe_jev_decisions.py | 118 ++++++++++++++ 4 files changed, 386 insertions(+), 1 deletion(-) create mode 100644 scripts/probe-jev-decisions.py create mode 100644 tests/test_probe_jev_decisions.py diff --git a/README.md b/README.md index a76b646..6b3de39 100644 --- a/README.md +++ b/README.md @@ -116,6 +116,15 @@ that model first through its own failover loop, the rest in configured order. candidates that already set one, and a caller's own effort still wins. - Streams, fails over and cools down like any dispatch alias. +To check a real Decisions API against what a jev alias sends, run the live +probe. It sends one synthetic decision through Cerberus's own adapter, prints +the raw response beside what Cerberus read from it, and exits non-zero when the +request is rejected or an answer cannot be read (one decision call is billed): + +```sh +OPENROUTER_API_KEY=... uv run --frozen python scripts/probe-jev-decisions.py +``` + ## Dispatch Dispatch resolves a stable Cerberus alias through a validated, identity-scoped diff --git a/docs/architecture.md b/docs/architecture.md index 6d041d7..9fae7f8 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -128,7 +128,10 @@ The confirmed Decisions API contract covers the endpoint, the model, the `model`/`state`/`questions` request and `answers` keyed by question id; the fields inside one question and one answer are not yet confirmed, so the answer is read tolerantly and a misread degrades to configured order rather than to an -unvetted model. +unvetted model. `scripts/probe-jev-decisions.py` settles it against a live +endpoint: it sends one synthetic decision through the production adapter and +reports whether the request was accepted and each answer read as an offered +option. ## Dispatch control plane diff --git a/scripts/probe-jev-decisions.py b/scripts/probe-jev-decisions.py new file mode 100644 index 0000000..0aceec8 --- /dev/null +++ b/scripts/probe-jev-decisions.py @@ -0,0 +1,255 @@ +"""Live probe: does OpenRouter's Decisions API accept what a jev alias sends? + +The jev decision adapter (src/cerberus/jev/backend.py) was written against a +partly confirmed contract: the endpoint, the model and the model/state/questions +envelope are confirmed; the fields inside one question and one answer are not. +This sends one real decision through that adapter's own code path — the same +request body, headers, deadline and answer reader production uses — and reports +the raw response beside what Cerberus made of it. + + OPENROUTER_API_KEY=... uv run --frozen python scripts/probe-jev-decisions.py + +The decision state is synthetic: three made-up pool models and a fixed prompt, +or one you pass with --prompt. Nothing from a configuration or a real request is +sent, and the key is never printed. One decision call is billed. + +Exit status: 0 when the request is accepted and every question asked comes back +as one of the options offered; 1 when not; 2 when the probe cannot run. +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import os +import ssl +import sys +from dataclasses import dataclass, field +from typing import Any + +import httpx + +from cerberus.jev.backend import ( + EFFORT_QUESTION, + EFFORT_QUESTION_ID, + ROUTE_QUESTION, + ROUTE_QUESTION_ID, + DecisionError, + DecisionRequest, + OpenRouterJevDecider, + Question, +) +from cerberus.jev.state import Option, decision_state +from cerberus.registry.schema import CerberusConfig, JevPolicy + +DEFAULTS = JevPolicy.model_validate({"decider": {"provider": "probe", "credential": "probe"}}) +DEFAULT_PROMPT = "Rename the variable `tmp` to `total` in this three-line Python function." +EFFORTS = ["low", "medium", "high"] + +# A synthetic registry: the decision is described from it exactly as a real +# pool would be, but nothing in it belongs to a deployment. +_PROBE_CONFIG = CerberusConfig.model_validate( + { + "metadata": {"version": "cerberus-2026-10-05.1"}, + "providers": { + "probe": { + "base_url": "https://probe.invalid/v1", + "credentials": {"probe": {"api_key_env": "PROBE_UNUSED"}}, + "models": { + "probe/small-local": { + "cost_tier": "free", "strength": "standard", "context_window": 32768, + "capabilities": ["chat", "coding"], "description": "Small local model; fast, free.", + }, + "probe/mid": { + "cost_tier": "free", "strength": "strong", "context_window": 131072, + "capabilities": ["chat", "coding"], + }, + "probe/frontier": { + "cost_tier": "paid", "strength": "frontier", "context_window": 200000, + "capabilities": ["chat", "coding", "tools"], "description": "Strongest; most expensive.", + }, + }, + } + }, + "aliases": { + "cerberus/probe": { + "mode": "dispatch", + "candidates": [{"provider": "probe", "credential": "probe", "model": "probe/small-local"}], + } + }, + } +) +_OPTIONS = [ + Option(id=f"m{i}", provider="probe", model=model) + for i, model in enumerate(_PROBE_CONFIG.providers["probe"].models, start=1) +] + + +class _Recording(httpx.AsyncBaseTransport): + """Passes the call through unchanged and keeps what crossed the wire.""" + + def __init__(self, inner: httpx.AsyncBaseTransport) -> None: + self._inner = inner + self.status: int | None = None + self.body: bytes | None = None + + async def handle_async_request(self, request: httpx.Request) -> httpx.Response: + response = await self._inner.handle_async_request(request) + content = await response.aread() + self.status, self.body = response.status_code, content + return httpx.Response(response.status_code, headers=response.headers, content=content, request=request) + + async def aclose(self) -> None: + await self._inner.aclose() + + +@dataclass +class Report: + request_body: dict[str, Any] + status: int | None = None + raw: Any = None # the parsed response, or its text when not JSON + latency_ms: float | None = None + failure: str | None = None # why the adapter raised, if it did + answers: dict[str, dict[str, Any]] = field(default_factory=dict) + usage: dict[str, float] | None = None + + @property + def ok(self) -> bool: + return self.failure is None and all(a["offered"] for a in self.answers.values()) + + +async def run_probe( + *, + api_key: str, + endpoint: str = str(DEFAULTS.endpoint), + model: str = DEFAULTS.model, + prompt: str = DEFAULT_PROMPT, + ask_effort: bool = True, + timeout_seconds: float = 20.0, + transport: httpx.AsyncBaseTransport, +) -> Report: + questions = [Question(ROUTE_QUESTION_ID, ROUTE_QUESTION, [o.id for o in _OPTIONS])] + if ask_effort: + questions.append(Question(EFFORT_QUESTION_ID, EFFORT_QUESTION, list(EFFORTS))) + request = DecisionRequest( + endpoint=endpoint, + api_key=api_key, + model=model, + state=decision_state( + {"messages": [{"role": "user", "content": prompt}]}, + _OPTIONS, + _PROBE_CONFIG, + input_mode="last_user_message", + max_chars=DEFAULTS.max_input_chars, + ), + questions=questions, + timeout_seconds=timeout_seconds, + ) + report = Report(request_body=OpenRouterJevDecider.build_body(request)) + recording = _Recording(transport) + decider = OpenRouterJevDecider(recording) + try: + decision = await decider.decide(request) + except DecisionError as exc: + report.failure, report.latency_ms = exc.reason, exc.latency_ms + else: + report.latency_ms, report.usage = decision.latency_ms, decision.usage + for question in questions: + answer = decision.answers[question.id] + report.answers[question.id] = { + "choice": answer.choice, + "confidence": answer.confidence, + "offered": answer.choice in question.options, + } + finally: + await decider.aclose() + report.status = recording.status + if recording.body is not None: + try: + report.raw = json.loads(recording.body) + except ValueError: + report.raw = recording.body.decode("utf-8", errors="replace")[:2000] + return report + + +def render(report: Report, *, endpoint: str, model: str) -> str: + lines = [f"POST {endpoint}", f"model: {model}", "", "request body (synthetic; no key):"] + lines.append(json.dumps(report.request_body, indent=2)) + lines += ["", f"response: HTTP {report.status if report.status is not None else '—'}" + + (f" in {report.latency_ms:.0f} ms" if report.latency_ms is not None else "")] + if report.raw is not None: + lines.append(json.dumps(report.raw, indent=2) if not isinstance(report.raw, str) else report.raw) + lines += ["", "Cerberus reads:"] + if report.failure is not None: + lines.append(f" adapter failed: {report.failure}") + for question_id, answer in report.answers.items(): + mark = "ok" if answer["offered"] else "NOT AN OFFERED OPTION" + lines.append( + f" {question_id}: choice={answer['choice']!r} confidence={answer['confidence']} ({mark})" + ) + if report.usage is not None: + lines.append(f" usage: {report.usage}") + lines += ["", "verdict: " + verdict(report)] + return "\n".join(lines) + + +def verdict(report: Report) -> str: + if report.ok: + return "OK. The request is accepted and every answer reads as an offered option." + if report.failure == "http_error": + return ( + f"REQUEST REJECTED (HTTP {report.status}). Compare the error body with the request body; " + "the question shape is built in OpenRouterJevDecider.build_body." + ) + if report.failure in ("timeout", "unreachable"): + return f"NO ANSWER ({report.failure}). Check the endpoint, network and key, then retry." + if report.failure == "invalid_response": + return "UNREADABLE RESPONSE. The body above is not a JSON object." + unread = [qid for qid, a in report.answers.items() if a["choice"] is None] + if unread: + return ( + f"ANSWER NOT READ for {', '.join(unread)}. The request was accepted; find each answer in " + "the response above and adjust _answer in src/cerberus/jev/backend.py to read it." + ) + return "ANSWERED OUTSIDE THE OPTIONS. An answer was read but is not one Cerberus offered." + + +def _transport() -> httpx.AsyncBaseTransport: + # what an AsyncClient would have taken from the environment by itself + proxy = os.environ.get("HTTPS_PROXY") or os.environ.get("https_proxy") + cafile = os.environ.get("SSL_CERT_FILE") + return httpx.AsyncHTTPTransport(proxy=proxy, verify=ssl.create_default_context(cafile=cafile) if cafile else True) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + parser.add_argument("--key-env", default="OPENROUTER_API_KEY", help="environment variable holding the key") + parser.add_argument("--endpoint", default=str(DEFAULTS.endpoint)) + parser.add_argument("--model", default=DEFAULTS.model) + parser.add_argument("--prompt", default=DEFAULT_PROMPT, help="the synthetic user message to decide on") + parser.add_argument("--no-effort", action="store_true", help="ask only which model, not the effort") + parser.add_argument("--timeout", type=float, default=20.0) + args = parser.parse_args(argv) + + api_key = os.environ.get(args.key_env, "").strip() + if not api_key: + print(f"probe: set {args.key_env} to an OpenRouter API key", file=sys.stderr) + return 2 + report = asyncio.run( + run_probe( + api_key=api_key, + endpoint=args.endpoint, + model=args.model, + prompt=args.prompt, + ask_effort=not args.no_effort, + timeout_seconds=args.timeout, + transport=_transport(), + ) + ) + print(render(report, endpoint=args.endpoint, model=args.model)) + return 0 if report.ok else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_probe_jev_decisions.py b/tests/test_probe_jev_decisions.py new file mode 100644 index 0000000..b62b384 --- /dev/null +++ b/tests/test_probe_jev_decisions.py @@ -0,0 +1,118 @@ +"""The live Decisions API probe, run offline against a mocked endpoint. + +The probe exists to tell an operator whether a real Decisions API accepts what a +jev alias sends and whether Cerberus reads the answers back. These prove it +reports each outcome truthfully, sends nothing from a real configuration, and +never prints the key. +""" + +import importlib.util +import json +import sys +from pathlib import Path + +import httpx +import pytest + +_PATH = Path(__file__).resolve().parents[1] / "scripts" / "probe-jev-decisions.py" +_spec = importlib.util.spec_from_file_location("probe_jev_decisions", _PATH) +probe = importlib.util.module_from_spec(_spec) +# dataclasses resolve their module through sys.modules, so register before running +sys.modules[_spec.name] = probe +_spec.loader.exec_module(probe) + +KEY = "sk-or-probe-secret" + + +def transport(status=200, payload=None, content=None, capture=None): + def respond(request): + if capture is not None: + capture.append(request) + if content is not None: + return httpx.Response(status, content=content) + return httpx.Response(status, json=payload) + + return httpx.MockTransport(respond) + + +def answered(route="m1", effort="low"): + return { + "answers": { + "route": {"choice": route, "probabilities": {route: 0.84}}, + "effort": {"choice": effort, "probabilities": {effort: 0.66}}, + }, + "usage": {"prompt_tokens": 210, "completion_tokens": 2, "total_tokens": 212}, + } + + +async def run(t, **kwargs): + return await probe.run_probe(api_key=KEY, transport=t, **kwargs) + + +@pytest.mark.asyncio +async def test_the_probe_sends_the_production_request_shape(): + sent = [] + report = await run(transport(payload=answered(), capture=sent)) + + assert len(sent) == 1 + request = sent[0] + assert str(request.url) == "https://openrouter.ai/api/alpha/decisions" + assert request.headers["authorization"] == f"Bearer {KEY}" + body = json.loads(request.content) + assert body == report.request_body + assert body["model"] == "typesafe/jev-1.13" + assert [q["id"] for q in body["questions"]] == ["route", "effort"] + assert body["questions"][0]["options"] == ["m1", "m2", "m3"] + assert body["questions"][1]["options"] == ["low", "medium", "high"] + # synthetic state only: the probe's own made-up pool and prompt + assert "probe/small-local" in body["state"] and probe.DEFAULT_PROMPT in body["state"] + + +@pytest.mark.asyncio +async def test_an_accepted_request_with_readable_answers_passes(): + report = await run(transport(payload=answered("m3", "high"))) + assert report.ok + assert report.answers["route"] == {"choice": "m3", "confidence": 0.84, "offered": True} + assert report.answers["effort"] == {"choice": "high", "confidence": 0.66, "offered": True} + assert report.usage == {"prompt_tokens": 210, "completion_tokens": 2, "total_tokens": 212} + assert probe.verdict(report).startswith("OK") + + +@pytest.mark.asyncio +async def test_no_effort_asks_only_the_route(): + sent = [] + report = await run(transport(payload=answered(), capture=sent), ask_effort=False) + assert [q["id"] for q in json.loads(sent[0].content)["questions"]] == ["route"] + assert list(report.answers) == ["route"] and report.ok + + +@pytest.mark.parametrize( + ("t", "start", "status"), + [ + (transport(400, {"error": {"message": "questions[0].options: unknown field"}}), "REQUEST REJECTED (HTTP 400)", 400), + (transport(200, content=b""), "UNREADABLE RESPONSE", 200), + (transport(200, {"answers": {"route": {"label": "m1"}}}), "ANSWER NOT READ for route, effort", 200), + (transport(200, answered("anthropic/claude", "extreme")), "ANSWERED OUTSIDE THE OPTIONS", 200), + ], + ids=["rejected", "not-json", "unread", "outside"], +) +@pytest.mark.asyncio +async def test_each_failure_is_reported_as_what_it_is(t, start, status): + report = await run(t) + assert not report.ok + assert report.status == status + assert probe.verdict(report).startswith(start) + + +@pytest.mark.asyncio +async def test_the_rendered_report_shows_the_raw_response_and_never_the_key(): + report = await run(transport(400, {"error": {"message": "bad question"}})) + text = probe.render(report, endpoint="https://openrouter.ai/api/alpha/decisions", model="typesafe/jev-1.13") + assert "bad question" in text and "HTTP 400" in text + assert KEY not in text + + +def test_without_a_key_the_probe_refuses_to_run(monkeypatch, capsys): + monkeypatch.delenv("OPENROUTER_API_KEY", raising=False) + assert probe.main([]) == 2 + assert "OPENROUTER_API_KEY" in capsys.readouterr().err