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