From a874bfb33b8facf1e9fb43bf9892312d99cb4e22 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 5 Oct 2026 12:14:34 +0000 Subject: [PATCH 1/2] Add jev-router: Jev as a model-selection strategy under Cerberus policy A jev-router alias hands model selection to OpenRouter's hosted Jev Router while Cerberus keeps the policy. Cerberus answers "what may run?"; Jev answers "which of those should run?" for one request. - Schema: new `jev-router` mode with a `jev_router` block (backend, timeout_seconds, allow_paid_pool). The pool is exact registry slugs under one OpenRouter credential; patterns, duplicates, per-candidate reasoning budgets and allow_paid_fallback are refused at validation. - router/pool.py: one strategy-neutral pool filter (cost, then health, cooldown and credential in the failover loop's order), shared by dispatch and /admin/routes and reusable by a native Jev strategy. - Dispatch sends only the filtered pool as the jev-router plugin's include list, never an empty one, and always requests router metadata. A response is accepted only when the jev-router stage is present, reports no list_fallback, and names served and resolved models inside the pool sent; models_ignored and every other miss is withheld with 502 as out_of_policy while its billed usage is still recorded. - Callers may send an allowlist of generation fields and function tools; model lists, plugins, route, preset, provider, server tools, reasoning controls and unknown fields are refused with 400 before any call. - Telemetry schema 6 adds a jev_router record and the out_of_policy, jev_router_unavailable and invalid_request outcomes. The console projects the alias as a pool with readiness and decision evidence. Not streamed: the decision is verified before the answer is returned. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01P4Nm4MVSMmG6JyZi8WumtK --- README.md | 22 ++ config/cerberus.example.yaml | 16 + docs/architecture.md | 49 +++ docs/persistence.md | 2 +- src/cerberus/app.py | 27 +- src/cerberus/control/routes.py | 59 +++ src/cerberus/control/sqlite.py | 16 +- src/cerberus/jev_router/__init__.py | 30 ++ src/cerberus/jev_router/backend.py | 303 +++++++++++++++ src/cerberus/jev_router/dispatch.py | 316 ++++++++++++++++ src/cerberus/registry/schema.py | 84 ++++- src/cerberus/router/dispatch.py | 3 +- src/cerberus/router/pool.py | 59 +++ src/cerberus/telemetry/emitter.py | 12 +- src/cerberus/ui/static/app.js | 86 ++++- tests/test_admin_ui.py | 2 +- tests/test_example_config.py | 7 +- tests/test_jev_router_dispatch.py | 546 ++++++++++++++++++++++++++++ tests/test_schema.py | 90 +++++ tests/test_telemetry.py | 2 +- 20 files changed, 1709 insertions(+), 22 deletions(-) create mode 100644 src/cerberus/jev_router/__init__.py create mode 100644 src/cerberus/jev_router/backend.py create mode 100644 src/cerberus/jev_router/dispatch.py create mode 100644 src/cerberus/router/pool.py create mode 100644 tests/test_jev_router_dispatch.py diff --git a/README.md b/README.md index b9570de..c710d07 100644 --- a/README.md +++ b/README.md @@ -73,6 +73,28 @@ Requirements and caveats: - Per-seat persona prompts from the earlier bundled worker are not available; the panel answers the caller's prompt directly. +## Jev Router + +A `jev-router` alias uses +[OpenRouter's Jev Router](https://openrouter.ai/typesafe/jev-router) +as one model-selection strategy under Cerberus policy. Cerberus owns the pool: +the alias's candidates, exact registry slugs under one OpenRouter credential, +filtered before every request by cost, provider health, cooldowns and credential +presence. Only what survives is sent; Jev picks one model and its reasoning +effort from it. Cerberus then checks the router's own metadata and withholds any +response that left the pool — including OpenRouter's `models_ignored` fallback, +which routes over its whole pool — while still recording the billed usage. + +Requirements and caveats: + +- An OpenRouter API key is required; the pool must share one credential. +- Callers may send generation fields and function tools only. Model lists, + `plugins`, `provider`, server tools, `reasoning_effort` and unknown fields are + refused with 400; Jev chooses the reasoning effort. +- Not streamed: the decision is verified before the answer is returned. +- A withheld response has already been billed by OpenRouter. +- The hosted router cannot choose local or non-OpenRouter providers. + ## Dispatch Dispatch resolves a stable Cerberus alias through a validated, identity-scoped diff --git a/config/cerberus.example.yaml b/config/cerberus.example.yaml index 423775f..ac748f9 100644 --- a/config/cerberus.example.yaml +++ b/config/cerberus.example.yaml @@ -19,6 +19,7 @@ providers: models: openrouter/free: { cost_tier: free, capabilities: [chat, tools] } anthropic/claude-sonnet: { cost_tier: paid, capabilities: [chat, tools, coding] } + google/gemini-pro: { cost_tier: paid, capabilities: [chat, tools, coding] } google: protocol: openai @@ -61,6 +62,21 @@ aliases: judge: { provider: openrouter, credential: primary, model: openrouter/free } require_human_review: true + cerberus/auto: + mode: jev-router + # Jev as one model-selection strategy under Cerberus policy. Before every + # request Cerberus filters this pool by cost, health, cooldown and + # credential; OpenRouter's Jev Router then picks one model (and its + # reasoning effort) from what remains. Exact registry slugs only, all under + # one OpenRouter credential. A response from outside the pool is withheld. + candidates: # the pool; no preference order + - { provider: openrouter, credential: primary, model: anthropic/claude-sonnet } + - { provider: openrouter, credential: primary, model: google/gemini-pro } + jev_router: + backend: openrouter + timeout_seconds: 120 + allow_paid_pool: true # Jev often picks a paid model; opt in explicitly + identities: example-reader: credential_env: CB_KEY_EXAMPLE_READER # holds a cb- client key diff --git a/docs/architecture.md b/docs/architecture.md index 30766ce..703c3bb 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -6,6 +6,9 @@ client → cerberus-api → identity + alias authorization + cost policy → Fusion backend → OpenRouter /chat/completions → openrouter:fusion server tool → panel models → analyst → final response + → filtered pool → Jev Router → OpenRouter /chat/completions + → typesafe/jev-router chooses one pool model + → decision verified against the pool → response ``` Cerberus owns routing policy, identity authorization, configured free/paid @@ -24,6 +27,52 @@ the alias names. A caller cannot select the panel, the analyst, or the tool surface of a fusion request, and cannot reach the backend except through a fusion alias it is allowed to use. Fusion-mode deliberation is not streamed. +## Jev Router + +A `jev-router` alias makes Jev one model-selection strategy beside dedicated +paths and Fusion; it does not hand Cerberus's policy to it. Cerberus answers +"what may run?" and Jev answers "which of those should run?" for one request. + +The pool is the alias's candidates, each an exact registry slug under one +OpenRouter credential. Before every request `router/pool.py` filters it in the +failover loop's own order — cost (a paid member needs `allow_paid_pool`), then +provider health, the most specific active cooldown, and credential presence — +and only the survivors are sent, as the `jev-router` plugin's `models` list on a +`typesafe/jev-router` call. An empty pool is never sent: the plugin treats an +include list that matches nothing as no list and routes over OpenRouter's whole +pool, so Cerberus answers 503 instead. Patterns (`*`, `~family`) are refused at +validation for the same reason. + +Every call asks for router metadata, and the response is accepted only when the +`jev-router` stage is present, reports no `list_fallback`, and names a served +model and resolved models that are all in the pool sent (a slug also matches its +`-YYYYMMDD` revisions, as the plugin documents). Anything else — `models_ignored` +above all — is withheld with 502 and recorded as `out_of_policy`. The call has +already run and been billed, so its usage and reported cost are still recorded. +The metadata is removed from the payload; the router's decision is reported in +the `cerberus` envelope and the `jev_router` telemetry record instead. + +A jev-router request accepts an allowlist of caller fields, not a denylist: +messages, sampling and output controls, and function tools. Anything else is +refused with 400 before any upstream call, including model fallback lists, +`route`, `plugins`, `preset`, `provider`, server tools and any field this +release does not know. `reasoning_effort` and `reasoning` are refused because +Jev chooses the effort; a per-candidate reasoning budget is refused at +validation for the same reason. Jev Router requests are not streamed, because +the decision has to be checked before any of the answer reaches the caller. + +`/admin/routes` projects the alias as a pool with no preference order: each +member is `eligible` (in the pool the next request would send) or `excluded` +with the first filter that refused it, from the same `resolve_pool` dispatch +calls. Readiness is the backend's presence and a non-empty pool. + +The hosted router can only choose among models one OpenRouter credential +reaches; it cannot place a request on a local or non-OpenRouter provider. A +native strategy in which Cerberus asks Jev for a decision over every eligible +backend and executes the winner itself is the intended next step, and would +reuse the same pool filter. There is no per-request cost ceiling: Cerberus does +not hold model prices, and a ceiling checked after a routed call is too late. + ## Dispatch control plane A Cerberus alias is a stable, policy-bearing address. The active SQLite revision diff --git a/docs/persistence.md b/docs/persistence.md index bcbc114..7032135 100644 --- a/docs/persistence.md +++ b/docs/persistence.md @@ -1,7 +1,7 @@ # Persistence review for 0.01 SQLite stores two state classes. Revisioned control state contains service -aliases, bindings, ordered paths, Fusion members, identity policy and credential +aliases, bindings, ordered paths, Fusion and jev-router pool members, identity policy and credential references plus one atomic active-revision pointer. Operational state contains provider health, cooldowns, routing events, usage projections and audit records; with OIDC enabled it also includes pending login and session state. Credential diff --git a/src/cerberus/app.py b/src/cerberus/app.py index b91b273..a170667 100644 --- a/src/cerberus/app.py +++ b/src/cerberus/app.py @@ -47,6 +47,7 @@ from cerberus.fusion import FusionBackend, OpenRouterFusionBackend, fusion_dispatch, fusion_status from cerberus.identity.session_store import InMemorySessionStore, SqliteSessionStore from cerberus.identity.sso import AdminSSO +from cerberus.jev_router import JevRouterBackend, OpenRouterJevRouterBackend, jev_router_dispatch from cerberus.registry import CerberusConfig, ConfigDocument, load_config_document from cerberus.registry.schema import Alias, Candidate from cerberus.router.dispatch import dispatch, shadow_decision_event, unauthorized_event @@ -161,6 +162,7 @@ def create_app( jwks_transport: httpx.AsyncBaseTransport | None = None, fusion_transport: httpx.AsyncBaseTransport | None = None, sso_transport: httpx.AsyncBaseTransport | None = None, + jev_router_transport: httpx.AsyncBaseTransport | None = None, ) -> FastAPI: if config is None: document = load_config_document() @@ -209,6 +211,11 @@ def create_app( # Managed deliberation backends, keyed by FusionPolicy.backend. Fixed at boot # like the telemetry sink; an alias whose backend is absent fails closed. fusion_backends: dict[str, FusionBackend] = {"openrouter": OpenRouterFusionBackend(fusion_transport)} + # Managed model-selection routers, keyed by JevRouterPolicy.backend; fixed at + # boot the same way. + jev_router_backends: dict[str, JevRouterBackend] = { + "openrouter": OpenRouterJevRouterBackend(jev_router_transport) + } if boot_config.admin_sso is not None: # sessions + pending logins share the state DB when one is configured, so # they survive restarts and are consistent across workers; else in-memory @@ -228,6 +235,8 @@ async def lifespan(app: FastAPI): await app.state.http_client.aclose() for backend in fusion_backends.values(): await backend.aclose() + for router in jev_router_backends.values(): + await router.aclose() if verifier is not None: await verifier.aclose() if admin_sso is not None: @@ -643,7 +652,8 @@ async def admin_routes(request: Request) -> Response: with the real one. This answers with the router's own verdicts: the same cost gate, the same health/cooldown/credential order the failover loop walks, the same authorization predicate the inference path applies, and - the same three gates fusion dispatch refuses on. + the same three gates fusion dispatch refuses on, and the same filtered + pool a jev-router request would send. Read-only, and behind the same boundary as /admin/status: the projection names every provider, model and credential reference in the revision, so @@ -660,6 +670,7 @@ async def admin_routes(request: Request) -> Response: control_plane=control_plane, store=store, backend_names=fusion_backends.keys(), + jev_router_backend_names=jev_router_backends.keys(), ), headers=_ADMIN_UI_HEADERS, ) @@ -945,6 +956,20 @@ async def chat_completions(request: Request) -> JSONResponse | StreamingResponse telemetry=telemetry, control_plane=control_plane, ) + if alias.mode == "jev-router": + # Cerberus filters the pool; Jev picks within it. A router outage + # fails only jev-router aliases. + return await jev_router_dispatch( + body=body, + alias_name=alias_name, + alias=alias, + identity_name=context.name if context else None, + document=document, + backends=jev_router_backends, + store=store, + telemetry=telemetry, + control_plane=control_plane, + ) return await dispatch( body=body, alias_name=alias_name, diff --git a/src/cerberus/control/routes.py b/src/cerberus/control/routes.py index 66cbfc8..f8c4f3b 100644 --- a/src/cerberus/control/routes.py +++ b/src/cerberus/control/routes.py @@ -10,6 +10,8 @@ runtime_exclusion() health, then cooldown, then credential presence authorization_error() whether a configured identity may use the alias fusion_readiness() backend, credential and provider gates for fusion + resolve_pool() the filtered pool a jev-router request would send + jev_router_readiness() backend presence and a non-empty pool Fusion is projected differently on purpose. Its panel never enters the failover loop — Cerberus resolves one backend, one credential and one provider and sends @@ -17,6 +19,11 @@ or cooldown state. Calling them "eligible" would describe a loop that does not run. They are a service chain with one readiness verdict over the whole alias. +A jev-router alias is a pool, not a ladder: Jev chooses among its members, so +none is preferred over another and there is no standby. Each member is either in +the pool the next request would send (eligible) or filtered out before Jev sees +it (excluded, with the first filter that refused it). + Read-only: no upstream connection, no cooldown applied, no lifecycle state, and no secret value or credential locator — a path names its credential, never where that credential is read from. @@ -29,10 +36,13 @@ from cerberus.fusion.dispatch import fusion_readiness from cerberus.identity.auth import IdentityContext, authorization_error +from cerberus.jev_router.backend import JEV_ROUTER_MODEL +from cerberus.jev_router.dispatch import jev_router_readiness from cerberus.registry.loader import ConfigDocument from cerberus.registry.schema import Alias, CerberusConfig, Candidate from cerberus.router.availability import runtime_exclusion from cerberus.router.engine import Target, cost_eligible, ordered_targets +from cerberus.router.pool import resolve_pool # a path is eligible when the router would attempt it first, standby when it is # attemptable but a later preference, excluded when the router would skip it @@ -155,12 +165,51 @@ def _fusion( } +def _jev_router( + config: CerberusConfig, + alias_name: str, + alias: Alias, + *, + backend_names: Collection[str], + control_plane: Any | None, + store: Any, + now: float | None, +) -> dict[str, Any]: + assert alias.jev_router is not None + policy = alias.jev_router + members = resolve_pool( + config, alias_name, allow_paid=policy.allow_paid_pool, control_plane=control_plane, store=store, now=now + ) + pool = [ + _path( + ordinal, + member.target, + STATE_EXCLUDED if member.exclusion is not None else STATE_ELIGIBLE, + ( + {"reason": member.exclusion.reason, "scope": member.exclusion.scope, "retry_at": member.exclusion.retry_at} + if member.exclusion is not None + else None + ), + ) + for ordinal, member in enumerate(members) + ] + return { + "backend": policy.backend, + "router_model": JEV_ROUTER_MODEL, + "allow_paid_pool": policy.allow_paid_pool, + "timeout_seconds": policy.timeout_seconds, + "pool": pool, + "readiness": jev_router_readiness(members, policy, backend_names=backend_names), + } + + def route_projection( document: ConfigDocument, *, control_plane: Any | None, store: Any, backend_names: Collection[str] = (), + jev_router_backend_names: Collection[str] = (), now: float | None = None, ) -> dict[str, Any]: """The active revision's routes, with every state decided server-side.""" @@ -177,6 +226,16 @@ def route_projection( projected["fusion"] = _fusion( config, alias, backend_names=backend_names, control_plane=control_plane ) + elif alias.mode == "jev-router": + projected["jev_router"] = _jev_router( + config, + alias_name, + alias, + backend_names=jev_router_backend_names, + control_plane=control_plane, + store=store, + now=now, + ) else: projected["allow_paid_fallback"] = alias.allow_paid_fallback projected["paths"] = _paths( diff --git a/src/cerberus/control/sqlite.py b/src/cerberus/control/sqlite.py index 1ddaa3f..16f3d03 100644 --- a/src/cerberus/control/sqlite.py +++ b/src/cerberus/control/sqlite.py @@ -309,7 +309,7 @@ def _materialize(self, document: ConfigDocument) -> None: ) self._connection.execute( "INSERT INTO alias_bindings VALUES (?, ?, ?)", - (revision, alias_name, "fusion" if alias.mode == "fusion" else "dedicated"), + (revision, alias_name, alias.mode if alias.mode in ("fusion", "jev-router") else "dedicated"), ) for ordinal, candidate in enumerate(alias.candidates): model = config.resolve_model(candidate, context=f"alias {alias_name!r} candidate") @@ -317,10 +317,18 @@ def _materialize(self, document: ConfigDocument) -> None: "INSERT INTO route_paths VALUES (?, ?, ?, ?, ?, ?, ?)", (revision, alias_name, ordinal, candidate.provider, candidate.credential, candidate.model, model.cost_tier), ) - if alias.mode == "fusion": + if alias.mode in ("fusion", "jev-router"): self._connection.execute( - "INSERT INTO route_members VALUES (?, ?, 'panel', ?, ?, ?, ?)", - (revision, alias_name, ordinal, candidate.provider, candidate.credential, candidate.model), + "INSERT INTO route_members VALUES (?, ?, ?, ?, ?, ?, ?)", + ( + revision, + alias_name, + "panel" if alias.mode == "fusion" else "pool", + ordinal, + candidate.provider, + candidate.credential, + candidate.model, + ), ) if alias.fusion is not None: for role, candidate in (("analyst", alias.fusion.judge), ("outer", alias.fusion.outer_model)): diff --git a/src/cerberus/jev_router/__init__.py b/src/cerberus/jev_router/__init__.py new file mode 100644 index 0000000..d0ae035 --- /dev/null +++ b/src/cerberus/jev_router/__init__.py @@ -0,0 +1,30 @@ +"""jev-router mode: OpenRouter's hosted Jev Router as one strategy under Cerberus policy. + +Cerberus decides which models may serve the alias and filters them by cost, +health, cooldown and credential before every request; Jev picks one model from +what survives. Cerberus verifies the router's decision against the pool it sent +and withholds any response that left it. Jev chooses; Cerberus governs. +""" + +from cerberus.jev_router.backend import ( + JEV_ROUTER_MODEL, + JevRouterBackend, + JevRouterError, + JevRouterRequest, + JevRouterResult, + OpenRouterJevRouterBackend, + RouterDecision, +) +from cerberus.jev_router.dispatch import jev_router_dispatch, jev_router_readiness + +__all__ = [ + "JEV_ROUTER_MODEL", + "JevRouterBackend", + "JevRouterError", + "JevRouterRequest", + "JevRouterResult", + "OpenRouterJevRouterBackend", + "RouterDecision", + "jev_router_dispatch", + "jev_router_readiness", +] diff --git a/src/cerberus/jev_router/backend.py b/src/cerberus/jev_router/backend.py new file mode 100644 index 0000000..54024c4 --- /dev/null +++ b/src/cerberus/jev_router/backend.py @@ -0,0 +1,303 @@ +"""Provider-neutral Jev Router backend seam plus the OpenRouter implementation. + +Cerberus resolves which models may serve a jev-router alias, filters them, and +hands the survivors to a ``JevRouterBackend`` as a ``JevRouterRequest``. The +backend makes one routed call and returns a normalized ``JevRouterResult``, +including the router's own account of its decision, or raises +``JevRouterError``. Whether that decision stayed inside the pool is Cerberus +policy and is judged by the caller of this module, not here; nothing outside +this module knows the wire shape of a given backend. +""" + +from __future__ import annotations + +import asyncio +import re +import time +from dataclasses import dataclass, field +from typing import Any, Literal, Protocol + +import httpx + +from cerberus.telemetry import AttemptOutcome + +JevRouterOutcome = Literal["upstream_error", "jev_router_unavailable"] + +# OpenRouter's hosted Jev Router and the plugin that narrows its pool. +JEV_ROUTER_MODEL = "typesafe/jev-router" +JEV_ROUTER_PLUGIN = "jev-router" +# Asks OpenRouter to attach openrouter_metadata, whose jev-router pipeline stage +# is the only evidence of whether the include list was honoured. +METADATA_HEADER = "x-openrouter-metadata" + +# "openai/gpt-6-luna" also matches its dated revisions, "openai/gpt-6-luna-20260922". +_DATED_REVISION = re.compile(r"-\d{8}") + + +@dataclass(frozen=True, slots=True) +class JevRouterRequest: + """Everything a backend needs, already resolved and filtered by Cerberus.""" + + body: dict[str, Any] # caller generation fields Cerberus allowed through + pool_models: list[str] # exact provider-native slugs; never empty, never a pattern + base_url: str + api_key: str + timeout_seconds: float # absolute deadline for the whole call + + +@dataclass(frozen=True, slots=True) +class RouterDecision: + """The router's own report of what it did, exactly as observed.""" + + resolved_models: list[str] | None # models the router selected, in fallback order + list_fallback: str | None # e.g. "models_ignored": the include list was NOT applied + list_tier_cap: str | None + max_fallback: str | None + + def record(self) -> dict[str, Any]: + return { + "resolved_models": self.resolved_models, + "list_fallback": self.list_fallback, + "list_tier_cap": self.list_tier_cap, + "max_fallback": self.max_fallback, + } + + +@dataclass(frozen=True, slots=True) +class JevRouterResult: + payload: dict[str, Any] # OpenAI-compatible completion with router metadata removed + returned_model: str | None # exactly what the response said, or None + generation_id: str | None + decision: RouterDecision | None # None when the response carried no jev-router stage + http_status: int + latency_ms: float + # metadata OBSERVED in the response (e.g. the serving provider); never + # populated from what was requested + metadata: dict[str, Any] = field(default_factory=dict) + + +class JevRouterError(Exception): + """A backend failure with the status and outcome Cerberus reports; fail closed. + + ``http_status`` is what Cerberus answers the caller; ``upstream_status`` is + what the backend actually returned (None when no HTTP response arrived). + """ + + def __init__( + self, + message: str, + *, + http_status: int, + outcome: JevRouterOutcome, + attempt_outcome: AttemptOutcome, + latency_ms: float, + upstream_status: int | None = None, + ) -> None: + super().__init__(message) + self.http_status = http_status + self.upstream_status = upstream_status + self.outcome = outcome + self.attempt_outcome = attempt_outcome + self.latency_ms = latency_ms + + +class JevRouterBackend(Protocol): + name: str + + async def execute(self, request: JevRouterRequest) -> JevRouterResult: ... + + async def aclose(self) -> None: ... + + +def slug_matches(served: str, registered: str) -> bool: + """Whether a served model is the registered slug or one of its dated revisions. + + The jev-router plugin documents that an exact slug matches every dated + revision of it, so the model a response names may carry a date suffix the + registry entry does not. Nothing looser is accepted. + """ + + if served == registered: + return True + suffix = served.removeprefix(registered) + return suffix != served and _DATED_REVISION.fullmatch(suffix) is not None + + +class OpenRouterJevRouterBackend: + """Route through OpenRouter's hosted Jev Router, narrowed to the Cerberus pool. + + One chat-completions call whose ``model`` is ``typesafe/jev-router`` and + whose ``jev-router`` plugin lists the pool as exact slugs. Router metadata + is always requested, because it is the only place OpenRouter says whether + the list was applied; it is parsed here and removed from the payload, so a + caller gets the completion shape it would have had without it. + """ + + name = "openrouter" + + def __init__(self, transport: httpx.AsyncBaseTransport | None = None) -> None: + self._client = httpx.AsyncClient(transport=transport) + + @staticmethod + def build_body(request: JevRouterRequest) -> dict[str, Any]: + # stream_options is only valid with stream: true, and this call never streams + body = { + k: v for k, v in request.body.items() if k not in ("model", "plugins", "stream", "stream_options") + } + body["model"] = JEV_ROUTER_MODEL + body["plugins"] = [{"id": JEV_ROUTER_PLUGIN, "models": list(request.pool_models)}] + return body + + async def execute(self, request: JevRouterRequest) -> JevRouterResult: + if not request.pool_models: + # An empty include list matches nothing, and OpenRouter answers a + # list that matches nothing by routing over its whole pool. + raise ValueError("jev-router request requires a non-empty pool") + started = time.perf_counter() + + def elapsed() -> float: + return (time.perf_counter() - started) * 1000 + + try: + # httpx timeouts are per operation, so the asyncio scope makes + # timeout_seconds an absolute wall-clock budget. Local cancellation + # does not stop upstream work already billed. + async with asyncio.timeout(request.timeout_seconds): + response = await self._client.post( + f"{request.base_url.rstrip('/')}/chat/completions", + json=self.build_body(request), + headers={ + "authorization": f"Bearer {request.api_key}", + "content-type": "application/json", + METADATA_HEADER: "enabled", + }, + timeout=request.timeout_seconds, + ) + except TimeoutError: + raise JevRouterError( + "Jev Router deadline exceeded", + http_status=504, + outcome="jev_router_unavailable", + attempt_outcome="transport_error", + latency_ms=elapsed(), + ) from None + except (httpx.ConnectError, httpx.ConnectTimeout, httpx.ReadTimeout, httpx.WriteTimeout, httpx.PoolTimeout): + raise JevRouterError( + "Jev Router unreachable", + http_status=503, + outcome="jev_router_unavailable", + attempt_outcome="transport_error", + latency_ms=elapsed(), + ) from None + except httpx.HTTPError: + raise JevRouterError( + "Jev Router request failed", + http_status=502, + outcome="upstream_error", + attempt_outcome="transport_error", + latency_ms=elapsed(), + ) from None + + if response.status_code >= 400: + # A 404 here is the plugin saying the lists left no admitted model. + raise JevRouterError( + f"Jev Router returned HTTP {response.status_code}", + http_status=502 if response.status_code < 500 else 503, + outcome="upstream_error" if response.status_code < 500 else "jev_router_unavailable", + attempt_outcome="retryable_status" if response.status_code in (408, 429) or response.status_code >= 500 else "invalid_response", + latency_ms=elapsed(), + upstream_status=response.status_code, + ) + + try: + payload = response.json() + except ValueError: + raise JevRouterError( + "Jev Router returned invalid JSON", + http_status=502, + outcome="upstream_error", + attempt_outcome="invalid_response", + latency_ms=elapsed(), + upstream_status=response.status_code, + ) from None + + if not _has_completion_content(payload): + raise JevRouterError( + "Jev Router returned no completion content", + http_status=502, + outcome="upstream_error", + attempt_outcome="invalid_response", + latency_ms=elapsed(), + upstream_status=response.status_code, + ) + + metadata = {"provider": payload["provider"]} if isinstance(payload.get("provider"), str) else {} + return JevRouterResult( + payload={k: v for k, v in payload.items() if k != "openrouter_metadata"}, + returned_model=payload.get("model") if isinstance(payload.get("model"), str) else None, + generation_id=payload.get("id") if isinstance(payload.get("id"), str) else None, + decision=_router_decision(payload), + http_status=response.status_code, + latency_ms=elapsed(), + metadata=metadata, + ) + + async def aclose(self) -> None: + await self._client.aclose() + + +def _router_decision(payload: dict[str, Any]) -> RouterDecision | None: + """The jev-router stage of ``openrouter_metadata.pipeline``, or None if absent. + + Only the documented fields are kept, and only in the shapes documented for + them, so nothing else the upstream attaches reaches telemetry. + """ + + metadata = payload.get("openrouter_metadata") + pipeline = metadata.get("pipeline") if isinstance(metadata, dict) else None + if not isinstance(pipeline, list): + return None + for stage in pipeline: + if isinstance(stage, dict) and stage.get("name") == JEV_ROUTER_PLUGIN: + data = stage.get("data") + data = data if isinstance(data, dict) else {} + resolved = data.get("resolved_models") + return RouterDecision( + resolved_models=( + [m for m in resolved if isinstance(m, str)] if isinstance(resolved, list) else None + ), + list_fallback=_scalar(data.get("list_fallback")), + list_tier_cap=_scalar(data.get("list_tier_cap")), + max_fallback=_scalar(data.get("max_fallback")), + ) + return None + + +def _scalar(value: Any) -> str | None: + """A documented scalar field as a string; None when absent, null, false or empty.""" + + if value is None or value is False or value == "": + return None + if isinstance(value, bool): + return "true" + if isinstance(value, (str, int, float)): + return str(value) + # present but not a scalar: keep that it was present, never its content + return "unrecognized" + + +def _has_completion_content(payload: Any) -> bool: + if not isinstance(payload, dict): + return False + choices = payload.get("choices") + if not isinstance(choices, list) or not choices: + return False + first = choices[0] + if not isinstance(first, dict): + return False + message = first.get("message") + if not isinstance(message, dict): + return False + content = message.get("content") + # a routed request may legitimately answer with a function-tool call + return (isinstance(content, str) and bool(content)) or bool(message.get("tool_calls")) diff --git a/src/cerberus/jev_router/dispatch.py b/src/cerberus/jev_router/dispatch.py new file mode 100644 index 0000000..2f9783b --- /dev/null +++ b/src/cerberus/jev_router/dispatch.py @@ -0,0 +1,316 @@ +"""Route a jev-router request: Cerberus policy first, Jev's choice second. + +Cerberus resolves the alias's pool and filters it (cost, health, cooldown, +credential — ``router/pool.py``) before Jev sees anything. The survivors go to +the hosted Jev Router as exact slugs; Jev picks one. Cerberus then checks the +router's own account of what it did against the pool it sent, and withholds any +response that left it. A backend outage fails only jev-router aliases, never +dispatch, free or fusion. +""" + +from __future__ import annotations + +import os +import time +import uuid +from collections.abc import Collection +from datetime import datetime, timezone +from typing import Any + +from fastapi.responses import JSONResponse + +# cerberus.router before cerberus.egress: egress.client imports router.engine, and +# the router package imports egress.client in turn, so egress first is a cycle. +from cerberus.router.engine import Target +from cerberus.router.pool import PoolMember, resolve_pool +from cerberus.egress.client import reported_cost, token_usage +from cerberus.jev_router.backend import ( + JEV_ROUTER_MODEL, + JevRouterBackend, + JevRouterError, + JevRouterRequest, + JevRouterResult, + slug_matches, +) +from cerberus.registry.loader import ConfigDocument +from cerberus.registry.schema import Alias, JevRouterPolicy +from cerberus.telemetry import RoutingAttempt, RoutingEvent, TelemetryEmitter + +# What a caller may set on a jev-router request. This is an allowlist, not a +# denylist: anything that decides where a request runs (model fallback lists, +# route, plugins, presets, provider routing, server tools) is Cerberus's to +# construct, and anything not listed here is refused rather than forwarded, so +# a field OpenRouter adds tomorrow cannot widen the pool. reasoning_effort and +# reasoning are absent on purpose: Jev chooses the effort per request, and +# overriding it silently in either direction would hide the substitution. +_CALLER_FIELDS = frozenset( + { + "messages", + "model", # the alias; replaced by the router model + "stream", # removed; jev-router is not streamed + "stream_options", + "temperature", + "top_p", + "n", + "stop", + "max_tokens", + "max_completion_tokens", + "presence_penalty", + "frequency_penalty", + "logit_bias", + "logprobs", + "top_logprobs", + "seed", + "user", + "response_format", + "tools", # function tools only; see caller_field_violations + "tool_choice", + "parallel_tool_calls", + "functions", + "function_call", + } +) +_TOOL_CHOICE_MODES = frozenset({"none", "auto", "required"}) + + +def caller_field_violations(body: dict[str, Any]) -> list[str]: + """The fields a jev-router alias refuses in this body; empty when acceptable. + + Function tools run on the caller's side and are accepted. Any other tool + type is a server tool that executes upstream under Cerberus's credential — + ``openrouter:fusion`` alone names a whole panel — so it is refused. + """ + + found = sorted(key for key in body if key not in _CALLER_FIELDS) + tools = body.get("tools") + if tools is not None and not ( + isinstance(tools, list) and all(isinstance(t, dict) and t.get("type") == "function" for t in tools) + ): + found.append("tools") + choice = body.get("tool_choice") + if choice is not None and not ( + (isinstance(choice, str) and choice in _TOOL_CHOICE_MODES) + or (isinstance(choice, dict) and choice.get("type") == "function") + ): + found.append("tool_choice") + return found + + +def jev_router_readiness( + members: list[PoolMember], policy: JevRouterPolicy, *, backend_names: Collection[str] +) -> dict[str, Any]: + """Whether this alias could route right now, by its own two gates. + + A routed request needs its backend and at least one pool member that + survived Cerberus's filters. ``jev_router_dispatch`` refuses with 503 when + either fails, and ``/admin/routes`` reports the same answer from here. + """ + + backend_present = policy.backend in backend_names + pool_available = sum(1 for member in members if member.exclusion is None) + return { + "backend_present": backend_present, + "pool_available": pool_available, + "pool_size": len(members), + "available": backend_present and pool_available > 0, + } + + +def served_target(model: str, pool: list[Target]) -> Target | None: + """The pool member a model the router named belongs to, or None.""" + + return next((target for target in pool if slug_matches(model, target.model)), None) + + +def pool_violation(result: JevRouterResult, pool: list[Target]) -> str | None: + """Why a routed response left the pool Cerberus sent, or None if it stayed. + + Fails closed. A response is accepted only when the router's own decision is + present, says it applied the include list, and every model it names — the + one that served and any it lined up behind it — is in the pool. + """ + + decision = result.decision + if decision is None: + return "decision_unverifiable" + if decision.list_fallback is not None: + # "models_ignored" means OpenRouter routed over its whole pool + return "models_ignored" if decision.list_fallback == "models_ignored" else "list_fallback" + if result.returned_model is None: + return "returned_model_unknown" + if served_target(result.returned_model, pool) is None: + return "returned_model_outside_pool" + if any(served_target(model, pool) is None for model in decision.resolved_models or []): + return "resolved_model_outside_pool" + return None + + +async def jev_router_dispatch( + *, + body: dict[str, Any], + alias_name: str, + alias: Alias, + identity_name: str | None, + document: ConfigDocument, + backends: dict[str, JevRouterBackend], + store: Any, + telemetry: TelemetryEmitter, + control_plane: Any | None = None, +) -> JSONResponse: + request_id = str(uuid.uuid4()) + started_at = time.perf_counter() + policy = alias.jev_router + assert policy is not None + + members = resolve_pool( + document.config, alias_name, allow_paid=policy.allow_paid_pool, control_plane=control_plane, store=store + ) + anchor = members[0].target # every member shares this provider and credential + pool = [member.target for member in members if member.exclusion is None] + candidates = [f"{member.target.provider_id}/{member.target.model}" for member in members] + exclusions = [ + {**member.target.describe(), **member.exclusion.telemetry()} + for member in members + if member.exclusion is not None + ] + record: dict[str, Any] = { + "backend": policy.backend, + "router_model": JEV_ROUTER_MODEL, + "pool": [target.model for target in pool], + "returned_model": None, + "generation_id": None, + "decision": None, + "violation": None, + "metadata": None, + } + + def emit(*, outcome, http_status, attempts, payload=None, served=None): + telemetry.emit( + RoutingEvent( + request_id=request_id, + alias=alias_name, + provider=anchor.provider_id, + mode=alias.mode, + model=record["returned_model"] or JEV_ROUTER_MODEL, + used_fallback=False, + credential=anchor.credential_id, + cost_tier=served.cost_tier if served is not None else None, + reported_cost=reported_cost(payload) if payload is not None else None, + candidates=candidates, + exclusions=exclusions, + attempts=attempts, + http_status=http_status, + outcome=outcome, + latency_ms=(time.perf_counter() - started_at) * 1000, + token_usage=token_usage(payload) if payload is not None else None, + timestamp=datetime.now(timezone.utc), + streaming=False, + identity=identity_name, + config_version=document.version, + config_checksum=document.checksum, + jev_router=dict(record), + ) + ) + + def error(status: int, message: str, **extra: Any) -> JSONResponse: + return JSONResponse( + status_code=status, content={"error": {"message": message, "request_id": request_id, **extra}} + ) + + def attempt(outcome, latency_ms, http_status=None) -> RoutingAttempt: + # exactly one upstream call: the routed request + return RoutingAttempt( + provider=f"{anchor.provider_id}/{anchor.credential_id}", + pool="jev-router", + model=record["returned_model"] or JEV_ROUTER_MODEL, + used_fallback=False, + outcome=outcome, + latency_ms=latency_ms, + http_status=http_status, + ) + + rejected = caller_field_violations(body) + if rejected: + emit(outcome="invalid_request", http_status=400, attempts=[]) + return error( + 400, + f"jev-router alias does not accept caller-supplied {', '.join(rejected)}", + reason="caller_field_not_allowed", + ) + + readiness = jev_router_readiness(members, policy, backend_names=backends.keys()) + if not readiness["backend_present"]: + emit(outcome="jev_router_unavailable", http_status=503, attempts=[]) + return error(503, "Jev Router backend not configured") + if not pool: + # Never send an empty or partial-by-accident list: OpenRouter answers a + # list that matches nothing by routing over its whole pool. + emit(outcome="routing_exhausted", http_status=503, attempts=[]) + return error(503, "No pool model available", exclusions=exclusions) + + # Non-streaming: the router's decision has to be checked against the pool + # before any of the answer reaches the caller. + request = JevRouterRequest( + body={k: v for k, v in body.items() if k != "stream"}, + pool_models=[target.model for target in pool], + base_url=anchor.base_url, + api_key=os.environ.get(anchor.api_key_env, "").strip(), + timeout_seconds=float(policy.timeout_seconds), + ) + backend = backends[policy.backend] + try: + result = await backend.execute(request) + except JevRouterError as exc: + emit( + outcome=exc.outcome, + http_status=exc.http_status, + attempts=[attempt(exc.attempt_outcome, exc.latency_ms, exc.upstream_status)], + ) + return error(exc.http_status, str(exc)) + + violation = pool_violation(result, pool) + served = served_target(result.returned_model, pool) if result.returned_model is not None else None + record.update( + returned_model=result.returned_model, + generation_id=result.generation_id, + decision=result.decision.record() if result.decision is not None else None, + violation=violation, + metadata=result.metadata or None, + ) + if violation is not None: + # The call ran and is billed, so its usage and cost are still recorded. + # The answer is withheld: it came from outside the pool Cerberus sent, + # or with no evidence that it did not. + emit( + outcome="out_of_policy", + http_status=502, + attempts=[attempt("out_of_policy", result.latency_ms, result.http_status)], + payload=result.payload, + ) + return error(502, f"Jev Router response withheld: {violation}", reason=violation) + + emit( + outcome="success", + http_status=200, + attempts=[attempt("response", result.latency_ms, result.http_status)], + payload=result.payload, + served=served, + ) + assert served is not None and result.decision is not None + payload = { + **result.payload, + "cerberus": { + "request_id": request_id, + "alias": alias_name, + "mode": alias.mode, + "identity": identity_name, + "backend": policy.backend, + "router": JEV_ROUTER_MODEL, + "pool": [f"{target.provider_id}/{target.model}" for target in pool], + "model": f"{served.provider_id}/{result.returned_model}", + "decision": result.decision.record(), + "config_version": document.version, + "config_checksum": document.checksum, + }, + } + return JSONResponse(content=payload, status_code=200, headers={"x-request-id": request_id}) diff --git a/src/cerberus/registry/schema.py b/src/cerberus/registry/schema.py index 81af93c..e6fbf64 100644 --- a/src/cerberus/registry/schema.py +++ b/src/cerberus/registry/schema.py @@ -12,9 +12,13 @@ from pydantic import BaseModel, ConfigDict, Field, HttpUrl, model_validator CostTier = Literal["free", "paid"] -AliasMode = Literal["direct", "dispatch", "free", "fusion"] +AliasMode = Literal["direct", "dispatch", "free", "fusion", "jev-router"] # Managed deliberation backends Cerberus can hand a fusion request to. FusionBackendName = Literal["openrouter"] +# Managed model-selection routers Cerberus can hand a jev-router request to. +JevRouterBackendName = Literal["openrouter"] +# OpenRouter's jev-router plugin accepts at most this many include patterns. +JEV_ROUTER_MAX_POOL = 1024 ALIAS_PREFIX = "cerberus/" VERSION_PATTERN = r"^cerberus-\d{4}-\d{2}-\d{2}\.\d+$" @@ -239,6 +243,32 @@ def outer_model(self) -> Candidate: return self.outer if self.outer is not None else self.judge +class JevRouterPolicy(BaseModel): + """Policy for a jev-router alias: Jev as one model-selection strategy. + + Cerberus decides WHICH models may serve the alias: the candidates, each a + registry-validated exact slug, filtered before every request by cost policy, + provider health, cooldowns and credential presence. Only what survives is + handed to OpenRouter's hosted Jev Router, which picks one model (and its + reasoning effort) from that pool. Jev chooses within Cerberus policy; it + never widens it. + + This is the hosted-router strategy. It can only choose among models one + OpenRouter credential reaches, so every candidate shares a provider and + credential, exactly as a fusion panel does. + """ + + model_config = ConfigDict(extra="forbid") + + backend: JevRouterBackendName = "openrouter" + # Absolute deadline for the routed request, enforced by Cerberus. + timeout_seconds: float = Field(gt=0, le=600) + # Jev picks the cheapest model strong enough for the request, which is often + # a paid one, and may add an advisor on the hardest requests. A pool with any + # paid model is therefore an explicit opt-in, as allow_paid_panel is. + allow_paid_pool: bool = False + + class Alias(BaseModel): model_config = ConfigDict(extra="forbid") @@ -247,6 +277,7 @@ class Alias(BaseModel): allow_paid_fallback: bool = False allow_experimental: bool = False fusion: FusionPolicy | None = None + jev_router: JevRouterPolicy | None = None @model_validator(mode="after") def validate_mode_coherence(self) -> "Alias": @@ -254,6 +285,14 @@ def validate_mode_coherence(self) -> "Alias": raise ValueError("fusion-mode alias requires a fusion policy block") if self.mode != "fusion" and self.fusion is not None: raise ValueError("fusion policy block is only valid on a fusion-mode alias") + if self.mode == "jev-router" and self.jev_router is None: + raise ValueError("jev-router alias requires a jev_router policy block") + if self.mode != "jev-router" and self.jev_router is not None: + raise ValueError("jev_router policy block is only valid on a jev-router alias") + if self.mode == "jev-router" and self.allow_paid_fallback: + # there is no ordered fallback for it to govern: Jev chooses from the + # pool, and paid membership is jev_router.allow_paid_pool + raise ValueError("jev-router alias must not set allow_paid_fallback; use jev_router.allow_paid_pool") if self.mode == "free" and self.allow_paid_fallback: raise ValueError("free-mode alias must not set allow_paid_fallback") return self @@ -320,6 +359,46 @@ def _validate_candidate(self, alias_name: str, alias: Alias, candidate: Candidat ) return model + def _validate_jev_router_pool(self, alias_name: str, alias: Alias, tiers: list[CostTier]) -> None: + assert alias.jev_router is not None + policy = alias.jev_router + if len(alias.candidates) > JEV_ROUTER_MAX_POOL: + raise ValueError( + f"alias {alias_name!r}: pool of {len(alias.candidates)} exceeds the jev-router limit " + f"of {JEV_ROUTER_MAX_POOL} models" + ) + anchor = alias.candidates[0] + seen: set[str] = set() + for candidate in alias.candidates: + # One routed request is one provider call under one credential, so the + # whole pool must be reachable through the same one. + if (candidate.provider, candidate.credential) != (anchor.provider, anchor.credential): + raise ValueError( + f"alias {alias_name!r}: jev-router backend {policy.backend!r} requires every candidate " + f"to use one provider/credential ({anchor.provider}/{anchor.credential}); " + f"{candidate.provider}/{candidate.model} uses {candidate.provider}/{candidate.credential}" + ) + # The plugin reads its include list as patterns: "*" is a wildcard and + # a leading "~" names a whole model family. Either could admit a model + # the registry never listed, so only exact slugs may enter the pool. + if "*" in candidate.model or candidate.model.startswith("~"): + raise ValueError( + f"alias {alias_name!r}: jev-router candidate {candidate.model!r} is a pattern; " + f"the pool admits exact registry slugs only" + ) + # Jev sets the reasoning effort for each request. A per-candidate + # budget would never be applied, so it is refused rather than ignored. + if candidate.reasoning_effort is not None or candidate.chat_template_kwargs is not None: + raise ValueError( + f"alias {alias_name!r}: jev-router candidate {candidate.model!r} sets a reasoning budget; " + f"Jev chooses reasoning effort per request, so it would never be applied" + ) + if candidate.model in seen: + raise ValueError(f"alias {alias_name!r}: jev-router candidate {candidate.model!r} is listed twice") + seen.add(candidate.model) + if not policy.allow_paid_pool and any(tier == "paid" for tier in tiers): + raise ValueError(f"alias {alias_name!r}: paid jev-router pool member requires allow_paid_pool") + @model_validator(mode="after") def validate_references(self) -> "CerberusConfig": if self.server.host not in LOOPBACK_HOSTS and not self.server.api_token_env: @@ -365,6 +444,9 @@ def validate_references(self) -> "CerberusConfig": f"alias {alias_name!r}: paid panel member, judge or outer model requires allow_paid_panel" ) + if alias.jev_router is not None: + self._validate_jev_router_pool(alias_name, alias, tiers) + for identity_name, identity in self.identities.items(): if identity.jwt_client_id is not None and self.authentik is None: raise ValueError( diff --git a/src/cerberus/router/dispatch.py b/src/cerberus/router/dispatch.py index e00f7e9..af6bcfe 100644 --- a/src/cerberus/router/dispatch.py +++ b/src/cerberus/router/dispatch.py @@ -1,7 +1,8 @@ """The failover loop — ordered candidates, scoped cooldowns, exclusion telemetry. One loop serves every alias mode (the heads share one policy engine). -Fusion-mode aliases divert to the fusion backend (cerberus.fusion.dispatch). +Fusion-mode aliases divert to the fusion backend (cerberus.fusion.dispatch); +jev-router aliases divert to the hosted Jev Router (cerberus.jev_router.dispatch). """ from __future__ import annotations diff --git a/src/cerberus/router/pool.py b/src/cerberus/router/pool.py new file mode 100644 index 0000000..77af9d8 --- /dev/null +++ b/src/cerberus/router/pool.py @@ -0,0 +1,59 @@ +"""The pool a model-selection strategy may choose from, after Cerberus's filters. + +Some aliases do not walk an ordered failover list. They hand a set of models to +a strategy that picks one — today OpenRouter's hosted Jev Router. Cerberus stays +the policy engine for those aliases: the strategy sees only what survives every +filter Cerberus applies, in this order: + + registry policy ordered_targets(): configured, validated candidates only + cost a paid model leaves the pool unless the alias allows paid + availability runtime_exclusion(): health, then cooldown, then credential + +The order matters for the same reason it does in the failover loop: a member +that fails more than one gate reports the first, so a prohibited paid model +reports the prohibition even while it is also cooled down. The availability +gates are the loop's own, read from ``router/availability.py``, so a pool and a +failover path can never disagree about whether a provider is reachable. + +Nothing here selects a model, opens a connection or applies a cooldown. Dispatch +and ``/admin/routes`` both call it, so the console reports the pool a request +would actually send. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +from cerberus.registry.schema import CerberusConfig +from cerberus.router.availability import CooldownSource, HealthSource, RuntimeExclusion, runtime_exclusion +from cerberus.router.engine import Target, ordered_targets + + +@dataclass(frozen=True, slots=True) +class PoolMember: + target: Target + # None when the strategy may choose this member right now + exclusion: RuntimeExclusion | None + + +def resolve_pool( + config: CerberusConfig, + alias_name: str, + *, + allow_paid: bool, + control_plane: HealthSource | None, + store: CooldownSource, + now: float | None = None, +) -> list[PoolMember]: + """Every configured candidate in config order, each with its first refusal.""" + + members: list[PoolMember] = [] + for target in ordered_targets(config, alias_name): + if target.cost_tier == "paid" and not allow_paid: + # schema validation already forbids this; defense in depth + members.append(PoolMember(target, RuntimeExclusion("paid_pool_prohibited"))) + continue + members.append( + PoolMember(target, runtime_exclusion(target, control_plane=control_plane, store=store, now=now)) + ) + return members diff --git a/src/cerberus/telemetry/emitter.py b/src/cerberus/telemetry/emitter.py index 8ba65c5..c3c9fa9 100644 --- a/src/cerberus/telemetry/emitter.py +++ b/src/cerberus/telemetry/emitter.py @@ -17,7 +17,7 @@ logger = logging.getLogger(__name__) -SCHEMA_VERSION = 5 +SCHEMA_VERSION = 6 AttemptOutcome = Literal[ "missing_credentials", "transport_error", @@ -25,6 +25,9 @@ "response", "invalid_response", "stream_interrupted", + # a managed router answered, but with a model outside the pool Cerberus sent + # it (or without the evidence to show otherwise); the response is withheld + "out_of_policy", ] RoutingOutcome = Literal[ "success", @@ -34,6 +37,9 @@ "unauthorized", "shadow", "fusion_unavailable", # fusion backend unreachable/unconfigured — fusion aliases fail closed + "jev_router_unavailable", # Jev Router backend unreachable/unconfigured — jev-router aliases fail closed + "out_of_policy", # a managed router served outside the Cerberus pool; response withheld + "invalid_request", # the caller sent a field the alias does not accept; nothing went upstream ] @@ -82,6 +88,9 @@ class RoutingEvent: # it vouched for in return (backend name, requested panel/analyst, returned # model, backend generation id, router metadata). None outside fusion mode. fusion: dict[str, Any] | None = None + # jev-router only: the pool Cerberus sent, what the router reported choosing, + # and why a response was withheld when it was. None outside jev-router mode. + jev_router: dict[str, Any] | None = None schema_version: int = SCHEMA_VERSION def as_payload(self) -> dict[str, Any]: @@ -91,6 +100,7 @@ def as_payload(self) -> dict[str, Any]: "alias": self.alias, "reasoning_effort": self.reasoning_effort, "fusion": self.fusion, + "jev_router": self.jev_router, "identity": self.identity, "config_version": self.config_version, "config_checksum": self.config_checksum, diff --git a/src/cerberus/ui/static/app.js b/src/cerberus/ui/static/app.js index c4c33cf..9c4bc1c 100644 --- a/src/cerberus/ui/static/app.js +++ b/src/cerberus/ui/static/app.js @@ -144,6 +144,7 @@ function aliasEntries() { // can have every credential in place and no reachable backend. function aliasRoutable(entry) { if (entry.fusion) return entry.fusion.readiness.available === true; + if (entry.jev_router) return entry.jev_router.readiness.available === true; return (entry.paths || []).some((path) => path.state === "eligible" || path.state === "standby"); } @@ -678,6 +679,7 @@ const EXCLUSION_LABEL = { cost_tier_guard: ["excluded by policy", "cost tier guard"], provider_down: ["unavailable", "provider health"], missing_credentials: ["unavailable", "credential not present"], + paid_pool_prohibited: ["excluded by policy", "paid model not allowed in this pool"], }; function describeState(path) { @@ -694,6 +696,14 @@ function describeState(path) { : { short: "excluded", note: reason.replace(/_/g, " "), tone: "t-warn" }; } +// A jev-router pool is not a ladder: Jev chooses among the eligible members, so +// none is "first" and none is standby. Exclusions read exactly as they do for a +// route path, because the same filters produced them. +function describePoolState(path) { + if (path.state === "eligible") return { short: "in pool", note: "sent to Jev, which may choose it", tone: "t-up" }; + return describeState(path); +} + function retryNote(retryAt) { if (!retryAt) return ""; const seconds = Math.max(0, Math.round(retryAt - Date.now() / 1000)); @@ -712,7 +722,9 @@ function selectedAliasEntry() { function aliasRail(aliases, current) { return el("div", { class: "alias-rail" }, ...aliases.map((entry, index) => { const on = current && entry.alias === current.alias; - const count = entry.paths ? entry.paths.length + " route paths" : entry.fusion.panel.length + " panel members"; + const count = entry.paths ? entry.paths.length + " route paths" + : entry.fusion ? entry.fusion.panel.length + " panel members" + : entry.jev_router.pool.length + " pool models"; return el("button", { class: "alias-tab" + (on ? " alias-tab-on" : ""), attrs: { type: "button", id: "alias-tab-" + index, "aria-pressed": on ? "true" : "false" }, @@ -720,20 +732,20 @@ function aliasRail(aliases, current) { }, el("code", { class: "alias-name", text: entry.alias }), el("span", { class: "alias-meta" }, - el("span", { class: "pill", text: entry.mode === "fusion" ? "Fusion" : "dedicated" }), + el("span", { class: "pill", text: entry.mode === "fusion" ? "Fusion" : entry.mode === "jev-router" ? "Jev Router" : "dedicated" }), el("code", { text: count }))); })); } -function pathRow(entry, path) { - const state = describeState(path); +function pathRow(entry, path, pool) { + const state = pool ? describePoolState(path) : describeState(path); const on = SELECTION.alias === entry.alias && SELECTION.ordinal === path.ordinal; return el("button", { class: "path-row" + (on ? " path-row-on" : ""), attrs: { type: "button", id: "path-row-" + path.ordinal, "aria-pressed": on ? "true" : "false" }, onClick: () => { SELECTION = { alias: entry.alias, ordinal: path.ordinal, decision: null, inspect: "path" }; render(); }, }, - el("code", { class: "path-ord", text: "pref " + path.ordinal }), + el("code", { class: "path-ord", text: (pool ? "#" : "pref ") + path.ordinal }), el("div", { class: "path-main" }, el("code", { class: "path-target", text: path.provider + "/" + path.model }), el("span", { class: "path-note", text: state.note })), @@ -766,6 +778,26 @@ function fusionChain(entry) { codeText(ready[k] ? "yes" : "no", ready[k] ? "t-up" : "t-bad"))))); } +function jevRouterPool(entry) { + const j = entry.jev_router; + const ready = j.readiness; + return el("div", { class: "stack" }, + el("p", { class: "chain-intro", text: "Cerberus filters this pool by cost, health, cooldown and credential before every request, then hands what remains to the Jev Router, which chooses one model. A response from outside the pool is withheld." }), + el("div", null, + el("div", { class: "path-head" }, + el("span", { text: "Member" }), el("span", { text: "Pool model" }), + el("span", { text: "Credential" }), el("span", { text: "Cost" }), el("span", { text: "State" })), + ...j.pool.map((p) => pathRow(entry, p, true))), + el("div", { class: "card" }, + el("h3", { text: "Readiness" }), + el("div", { class: "card-row" }, el("span", { text: "backend configured" }), + codeText(ready.backend_present ? "yes" : "no", ready.backend_present ? "t-up" : "t-bad")), + el("div", { class: "card-row" }, el("span", { text: "pool models sent" }), + codeText(ready.pool_available + " of " + ready.pool_size, ready.pool_available ? "t-up" : "t-bad")), + el("div", { class: "card-row" }, el("span", { text: "can route now" }), + codeText(ready.available ? "yes" : "no", ready.available ? "t-up" : "t-bad")))); +} + function inspectorRows(entry) { const projection = STATE.routes; const base = [ @@ -785,6 +817,25 @@ function inspectorRows(entry) { entry.fusion.readiness.available ? "t-up" : "t-bad"], ]); } + if (entry.jev_router) { + const j = entry.jev_router; + const rows = base.concat([ + ["router", j.router_model, "t-ink"], + ["pool", j.readiness.pool_available + " of " + j.readiness.pool_size + " sent to Jev", "t-mono"], + ["paid pool", j.allow_paid_pool ? "allowed" : "prohibited", "t-mono"], + ["routing", j.readiness.available ? "ready" : "unavailable", j.readiness.available ? "t-up" : "t-bad"], + ]); + const member = j.pool.find((p) => p.ordinal === SELECTION.ordinal); + if (!member) return rows.concat([["pool model", "select a pool model", "t-mono"]]); + const memberState = describePoolState(member); + return rows.concat([ + ["pool model", member.provider + "/" + member.model, "t-ink"], + ["credential", member.credential_ref, "t-mono"], + ["cost tier", member.cost_tier, member.cost_tier === "paid" ? "t-warn" : "t-mono"], + ["state", memberState.short, memberState.tone], + ["reason", member.exclusion ? memberState.note : "—", member.exclusion ? memberState.tone : "t-mono"], + ]); + } const path = (entry.paths || []).find((p) => p.ordinal === SELECTION.ordinal); if (!path) return base.concat([["route path", "select a route path", "t-mono"]]); const state = describeState(path); @@ -802,7 +853,7 @@ function inspectorRows(entry) { function inspector(entry) { const rows = inspectorRows(entry); return el("aside", { class: "inspector", attrs: { id: "inspector", "aria-live": "polite", tabindex: "-1" } }, - el("span", { class: "fact-label", text: entry.fusion ? "Fusion inspector" : "Route path inspector" }), + el("span", { class: "fact-label", text: entry.fusion ? "Fusion inspector" : entry.jev_router ? "Jev Router inspector" : "Route path inspector" }), el("div", { class: "inspect-rows" }, ...rows.map(([k, v, tone]) => el("div", { class: "inspect-row" }, el("span", { text: k }), codeText(v, tone)))), @@ -823,6 +874,8 @@ function viewRoutes() { const entry = selectedAliasEntry(); const body = entry.fusion ? fusionChain(entry) + : entry.jev_router + ? jevRouterPool(entry) : el("div", null, el("div", { class: "path-head" }, el("span", { text: "Order" }), el("span", { text: "Route path" }), @@ -867,6 +920,9 @@ const OUTCOME_TONE = { upstream_error: "t-bad", routing_exhausted: "t-bad", fusion_unavailable: "t-bad", + jev_router_unavailable: "t-bad", + out_of_policy: "t-bad", + invalid_request: "t-warn", }; const ATTEMPT_TONE = { @@ -876,6 +932,7 @@ const ATTEMPT_TONE = { stream_interrupted: "t-warn", missing_credentials: "t-bad", invalid_response: "t-bad", + out_of_policy: "t-bad", }; const outcomeTone = (outcome) => OUTCOME_TONE[outcome] || "t-mono"; @@ -957,16 +1014,24 @@ function exclusionList(event) { )))); } -function fusionEvidence(event) { - if (!event.fusion) return null; +// what a managed backend was asked for and vouched for, exactly as recorded +function evidenceRows(label, record) { return el("div", null, - el("span", { class: "fact-label", text: "Fusion evidence" }), + el("span", { class: "fact-label", text: label }), el("div", { class: "inspect-rows" }, - ...Object.entries(event.fusion).map(([key, value]) => el("div", { class: "inspect-row" }, + ...Object.entries(record).map(([key, value]) => el("div", { class: "inspect-row" }, el("span", { text: key }), codeText(value === null || value === undefined ? "—" : (typeof value === "object" ? JSON.stringify(value) : value)))))); } +function fusionEvidence(event) { + return event.fusion ? evidenceRows("Fusion evidence", event.fusion) : null; +} + +function jevRouterEvidence(event) { + return event.jev_router ? evidenceRows("Jev Router evidence", event.jev_router) : null; +} + function decisionInspector(event) { const rows = [ ["request id", text(event.request_id), "t-ink"], @@ -994,6 +1059,7 @@ function decisionInspector(event) { attemptLadder(event)), exclusionList(event), fusionEvidence(event), + jevRouterEvidence(event), ); } diff --git a/tests/test_admin_ui.py b/tests/test_admin_ui.py index 5321286..c3cf00f 100644 --- a/tests/test_admin_ui.py +++ b/tests/test_admin_ui.py @@ -737,7 +737,7 @@ async def test_history_is_rendered_from_the_event_never_from_current_route_state app = make_app(monkeypatch, tmp_path) script = (await fetch(app, "/admin/ui/app.js")).text - for name in ("attemptLadder", "exclusionList", "fusionEvidence", "decisionInspector"): + for name in ("attemptLadder", "exclusionList", "fusionEvidence", "jevRouterEvidence", "decisionInspector"): source = _function_source(script, name) for current in ("STATE.routes", "projectedAliases", "aliasRoutable", "STATE.status"): assert current not in source, f"{name} must not explain history with current state: {current}" diff --git a/tests/test_example_config.py b/tests/test_example_config.py index d3c30a4..65e9939 100644 --- a/tests/test_example_config.py +++ b/tests/test_example_config.py @@ -10,4 +10,9 @@ def test_example_config_validates(): doc = load_config_document(EXAMPLE, validate_credentials=False) assert doc.version == "cerberus-2026-07-16.1" - assert set(doc.config.aliases) == {"cerberus/free", "cerberus/dispatch-standard", "cerberus/fusion-code"} + assert set(doc.config.aliases) == { + "cerberus/free", + "cerberus/dispatch-standard", + "cerberus/fusion-code", + "cerberus/auto", + } diff --git a/tests/test_jev_router_dispatch.py b/tests/test_jev_router_dispatch.py new file mode 100644 index 0000000..e0d59fb --- /dev/null +++ b/tests/test_jev_router_dispatch.py @@ -0,0 +1,546 @@ +"""jev-router mode: Jev as one model-selection strategy under Cerberus policy. + +OpenRouter is mocked at the HTTP boundary (jev_router_transport); no live +traffic. These prove the Cerberus side: Cerberus filters the pool before Jev +sees it and sends it as exact slugs; Jev picks within it; a response from +outside the pool — or without the router's evidence that it stayed inside — is +withheld while its billed usage is still recorded; the caller cannot steer +routing; and a router outage fails only jev-router aliases. +""" + +import json + +import httpx +import pytest + +from cerberus.app import create_app +from cerberus.jev_router import JevRouterError, JevRouterRequest, OpenRouterJevRouterBackend +from cerberus.jev_router.backend import slug_matches +from cerberus.registry import load_config_document +from tests.test_control import ok_upstream, write_config + +POOL = ("vendor/model-a", "vendor/model-b") + + +def jev_raw(*, pool=POOL, allow_paid=False, state_path=None) -> dict: + raw = { + "metadata": {"version": "cerberus-2026-10-05.1"}, + "providers": { + "openrouter": { + "base_url": "https://openrouter.ai/api/v1", + "credentials": {"main": {"api_key_env": "OPENROUTER_API_KEY"}}, + "models": { + "vendor/model-a": {"cost_tier": "free"}, + "vendor/model-b": {"cost_tier": "free"}, + "vendor/model-paid": {"cost_tier": "paid"}, + }, + }, + }, + "identities": { + "dev": { + "credential_env": "CB_KEY_DEV", + "allowed_modes": ["dispatch", "jev-router"], + "allowed_aliases": ["cerberus/dispatch-dev", "cerberus/auto"], + }, + "dispatch-only": { + "credential_env": "CB_KEY_DISPATCH", + "allowed_modes": ["dispatch"], + "allowed_aliases": ["cerberus/dispatch-dev"], + }, + }, + "aliases": { + "cerberus/dispatch-dev": { + "mode": "dispatch", + "candidates": [{"provider": "openrouter", "credential": "main", "model": "vendor/model-a"}], + }, + "cerberus/auto": { + "mode": "jev-router", + "candidates": [{"provider": "openrouter", "credential": "main", "model": m} for m in pool], + "jev_router": {"timeout_seconds": 30, "allow_paid_pool": allow_paid}, + }, + }, + } + if state_path is not None: + raw["state"] = {"path": str(state_path)} + return raw + + +class Capture: + """Records the outbound OpenRouter request so tests can assert the wire shape.""" + + def __init__(self, respond): + self.requests: list[httpx.Request] = [] + self._respond = respond + + def __call__(self, request: httpx.Request) -> httpx.Response: + self.requests.append(request) + return self._respond(request) + + @property + def body(self) -> dict: + return json.loads(self.requests[-1].content) + + +def routed(model="vendor/model-a", *, resolved=None, stage=True, list_fallback=None, message=None, **data): + """A Jev Router completion carrying the documented jev-router pipeline stage.""" + + payload = { + "id": "gen-jev-1", + "model": model, + "provider": "SomeProvider", + "choices": [{"message": message or {"role": "assistant", "content": "routed answer"}}], + "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15, "cost": 0.0021}, + } + if model is None: + del payload["model"] + if stage: + stage_data = {"resolved_models": resolved if resolved is not None else [model], **data} + if list_fallback is not None: + stage_data["list_fallback"] = list_fallback + payload["openrouter_metadata"] = { + "pipeline": [{"name": "moderation", "data": {}}, {"name": "jev-router", "data": stage_data}] + } + + def respond(_request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json=payload) + + return respond + + +def make_app(monkeypatch, tmp_path, *, raw=None, respond=None, api_key="or-key"): + if api_key is not None: + monkeypatch.setenv("OPENROUTER_API_KEY", api_key) + else: + monkeypatch.delenv("OPENROUTER_API_KEY", raising=False) + monkeypatch.setenv("CB_KEY_DEV", "cb-dev") + monkeypatch.setenv("CB_KEY_DISPATCH", "cb-dispatch") + doc = load_config_document(write_config(tmp_path, "v.yaml", raw or jev_raw()), validate_credentials=False) + capture = Capture(respond or routed()) + app = create_app( + doc, + http_transport=httpx.MockTransport(ok_upstream), + jev_router_transport=httpx.MockTransport(capture), + ) + return app, capture + + +def client_for(app): + return httpx.AsyncClient(transport=httpx.ASGITransport(app=app, client=("127.0.0.1", 40001)), base_url="http://t") + + +AUTH = {"authorization": "Bearer cb-dev"} +JEV_REQ = {"model": "cerberus/auto", "messages": [{"role": "user", "content": "hi"}]} +DISPATCH_REQ = {"model": "cerberus/dispatch-dev", "messages": [{"role": "user", "content": "hi"}]} + + +async def call(app, body=JEV_REQ, headers=AUTH): + async with app.router.lifespan_context(app): + async with client_for(app) as client: + resp = await client.post("/v1/chat/completions", json=body, headers=headers) + events = (await client.get("/admin/events")).json()["events"] + return resp, events + + +# ---- normal request: wire shape, response contract, telemetry ---- + + +@pytest.mark.asyncio +async def test_request_becomes_one_jev_router_call_over_the_cerberus_pool(monkeypatch, tmp_path): + app, capture = make_app(monkeypatch, tmp_path) + resp, _ = await call(app, {**JEV_REQ, "stream": True, "temperature": 0.3}) + + assert resp.status_code == 200 + assert len(capture.requests) == 1 + req = capture.requests[0] + assert str(req.url) == "https://openrouter.ai/api/v1/chat/completions" + assert req.headers["authorization"] == "Bearer or-key" + # the only evidence of whether the list was honoured, so always requested + assert req.headers["x-openrouter-metadata"] == "enabled" + assert capture.body == { + "model": "typesafe/jev-router", + "plugins": [{"id": "jev-router", "models": ["vendor/model-a", "vendor/model-b"]}], + "messages": JEV_REQ["messages"], + "temperature": 0.3, + }, "exact slugs only, no stream, nothing Cerberus did not construct or allow" + + payload = resp.json() + assert payload["choices"][0]["message"]["content"] == "routed answer" + assert "openrouter_metadata" not in payload, "the caller gets the shape it would have had without it" + assert payload["cerberus"] == { + "request_id": resp.headers["x-request-id"], + "alias": "cerberus/auto", + "mode": "jev-router", + "identity": "dev", + "backend": "openrouter", + "router": "typesafe/jev-router", + "pool": ["openrouter/vendor/model-a", "openrouter/vendor/model-b"], + "model": "openrouter/vendor/model-a", + "decision": { + "resolved_models": ["vendor/model-a"], + "list_fallback": None, + "list_tier_cap": None, + "max_fallback": None, + }, + "config_version": "cerberus-2026-10-05.1", + "config_checksum": app.state.lifecycle.active.checksum, + } + + +@pytest.mark.asyncio +async def test_telemetry_records_the_pool_the_decision_and_the_billed_cost(monkeypatch, tmp_path): + respond = routed("vendor/model-b", resolved=["vendor/model-b", "vendor/model-a"], list_tier_cap="standard") + app, _ = make_app(monkeypatch, tmp_path, respond=respond) + resp, events = await call(app) + + event = events[0] + assert event["request_id"] == resp.headers["x-request-id"] + assert (event["mode"], event["outcome"], event["http_status"]) == ("jev-router", "success", 200) + assert event["model"] == "vendor/model-b", "the model Jev chose, not the router alias" + assert event["cost_tier"] == "free", "known because the served model is a registry member" + assert event["reported_cost"] == 0.0021 + assert event["token_usage"] == {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15} + assert event["candidates"] == ["openrouter/vendor/model-a", "openrouter/vendor/model-b"] + assert event["fusion"] is None + assert event["jev_router"] == { + "backend": "openrouter", + "router_model": "typesafe/jev-router", + "pool": ["vendor/model-a", "vendor/model-b"], + "returned_model": "vendor/model-b", + "generation_id": "gen-jev-1", + "decision": { + "resolved_models": ["vendor/model-b", "vendor/model-a"], + "list_fallback": None, + "list_tier_cap": "standard", + "max_fallback": None, + }, + "violation": None, + "metadata": {"provider": "SomeProvider"}, + } + assert [a["pool"] for a in event["attempts"]] == ["jev-router"] + assert event["attempts"][0]["outcome"] == "response" + + +@pytest.mark.asyncio +async def test_a_dated_revision_of_a_pool_slug_is_inside_the_pool(monkeypatch, tmp_path): + app, _ = make_app(monkeypatch, tmp_path, respond=routed("vendor/model-a-20260922")) + resp, events = await call(app) + assert resp.status_code == 200 + assert events[0]["cost_tier"] == "free" + assert resp.json()["cerberus"]["model"] == "openrouter/vendor/model-a-20260922" + + +def test_slug_matching_admits_only_dated_revisions(): + assert slug_matches("vendor/model-a", "vendor/model-a") + assert slug_matches("vendor/model-a-20260922", "vendor/model-a") + for outside in ("vendor/model-a-preview", "vendor/model-a-2026", "vendor/model-ab", "vendor/model", "x/vendor/model-a"): + assert not slug_matches(outside, "vendor/model-a"), outside + # a registered dated revision matches only itself + assert not slug_matches("vendor/model-a-20260923", "vendor/model-a-20260922") + + +# ---- out of policy: withheld, but billed usage recorded ---- + + +@pytest.mark.parametrize( + ("respond", "violation"), + [ + # OpenRouter ignored the include list and routed over its whole pool + (routed("elsewhere/expensive", list_fallback="models_ignored"), "models_ignored"), + # ... even when the model it happened to pick is one Cerberus allows + (routed("vendor/model-a", list_fallback="models_ignored"), "models_ignored"), + (routed("vendor/model-a", list_fallback="something_new"), "list_fallback"), + (routed("elsewhere/expensive", resolved=["elsewhere/expensive"]), "returned_model_outside_pool"), + (routed("vendor/model-a", resolved=["vendor/model-a", "elsewhere/expensive"]), "resolved_model_outside_pool"), + (routed("vendor/model-a-preview"), "returned_model_outside_pool"), + # no evidence either way is not evidence of staying inside + (routed("vendor/model-a", stage=False), "decision_unverifiable"), + (routed(None, resolved=["vendor/model-a"]), "returned_model_unknown"), + ], + ids=["ignored-outside", "ignored-inside", "unknown-fallback", "served-outside", "resolved-outside", + "undated-suffix", "no-stage", "no-model"], +) +@pytest.mark.asyncio +async def test_a_response_outside_the_pool_is_withheld(monkeypatch, tmp_path, respond, violation): + app, capture = make_app(monkeypatch, tmp_path, respond=respond) + resp, events = await call(app) + + assert len(capture.requests) == 1 + assert resp.status_code == 502 + body = resp.json() + assert "choices" not in body, "the answer never reaches the caller" + assert body["error"]["reason"] == violation + event = events[0] + assert (event["outcome"], event["http_status"]) == ("out_of_policy", 502) + assert event["attempts"][0]["outcome"] == "out_of_policy" + assert event["jev_router"]["violation"] == violation + # the call ran and is billed, so its cost is recorded rather than lost + assert event["token_usage"] == {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15} + assert event["reported_cost"] == 0.0021 + assert event["cost_tier"] is None + + +@pytest.mark.asyncio +async def test_models_ignored_records_what_actually_ran(monkeypatch, tmp_path): + app, _ = make_app(monkeypatch, tmp_path, respond=routed("elsewhere/expensive", list_fallback="models_ignored")) + _, events = await call(app) + record = events[0]["jev_router"] + assert record["returned_model"] == "elsewhere/expensive" + assert record["decision"]["list_fallback"] == "models_ignored" + assert events[0]["model"] == "elsewhere/expensive" + + +# ---- Cerberus filters first; Jev only sees what survives ---- + + +@pytest.mark.asyncio +async def test_a_cooled_down_model_never_reaches_jev(monkeypatch, tmp_path): + app, capture = make_app(monkeypatch, tmp_path) + app.state.cooldowns.apply( + scope="model", provider="openrouter", credential="main", model="vendor/model-b", + reason="quota_429", duration_seconds=600, + ) + resp, events = await call(app) + + assert resp.status_code == 200 + assert capture.body["plugins"] == [{"id": "jev-router", "models": ["vendor/model-a"]}] + assert events[0]["jev_router"]["pool"] == ["vendor/model-a"] + assert events[0]["exclusions"] == [ + {"provider": "openrouter", "credential": "main", "model": "vendor/model-b", + "reason": "cooldown_quota_429", "scope": "model"} + ] + + +@pytest.mark.asyncio +async def test_a_pool_model_jev_was_not_sent_is_outside_the_pool(monkeypatch, tmp_path): + """Registered is not enough: the pool is what survived the filters.""" + app, _ = make_app(monkeypatch, tmp_path, respond=routed("vendor/model-b")) + app.state.cooldowns.apply( + scope="model", provider="openrouter", credential="main", model="vendor/model-b", + reason="quota_429", duration_seconds=600, + ) + resp, events = await call(app) + assert resp.status_code == 502 + assert events[0]["jev_router"]["violation"] == "returned_model_outside_pool" + + +@pytest.mark.asyncio +async def test_an_empty_pool_fails_closed_without_calling_openrouter(monkeypatch, tmp_path): + """An empty include list matches nothing, and OpenRouter answers that by + routing over its whole pool — so an empty pool is never sent.""" + app, capture = make_app(monkeypatch, tmp_path) + app.state.cooldowns.apply( + scope="credential", provider="openrouter", credential="main", model=None, + reason="quota_429", duration_seconds=600, + ) + resp, events = await call(app) + + assert resp.status_code == 503 + assert capture.requests == [] + assert events[0]["outcome"] == "routing_exhausted" + assert [e["model"] for e in resp.json()["error"]["exclusions"]] == list(POOL) + + +@pytest.mark.asyncio +async def test_a_missing_credential_empties_the_pool(monkeypatch, tmp_path): + app, capture = make_app(monkeypatch, tmp_path, api_key=None) + resp, events = await call(app) + assert resp.status_code == 503 + assert capture.requests == [] + assert {e["reason"] for e in events[0]["exclusions"]} == {"missing_credentials"} + + +# ---- the caller cannot steer routing ---- + + +@pytest.mark.parametrize( + ("key", "value"), + [ + ("plugins", [{"id": "jev-router", "models": ["elsewhere/*"]}]), + ("models", ["elsewhere/expensive"]), + ("route", "fallback"), + ("provider", {"only": ["SomeProvider"]}), + ("preset", "@preset/anything"), + ("transforms", ["middle-out"]), # unknown fields are refused, not forwarded + ("reasoning_effort", "high"), # Jev chooses the effort + ("reasoning", {"effort": "high"}), + ("tools", [{"type": "openrouter:fusion", "parameters": {"analysis_models": ["elsewhere/expensive"]}}]), + ("tools", [{"type": "web_search"}]), + ("tool_choice", {"type": "openrouter:web_search"}), + ], +) +@pytest.mark.asyncio +async def test_caller_routing_fields_are_refused_before_any_call(monkeypatch, tmp_path, key, value): + app, capture = make_app(monkeypatch, tmp_path) + resp, events = await call(app, {**JEV_REQ, key: value}) + + assert resp.status_code == 400 + assert key in resp.json()["error"]["message"] + assert resp.json()["error"]["reason"] == "caller_field_not_allowed" + assert capture.requests == [], "nothing may reach OpenRouter" + assert (events[0]["outcome"], events[0]["http_status"]) == ("invalid_request", 400) + assert events[0]["attempts"] == [] + + +@pytest.mark.asyncio +async def test_function_tools_pass_through_and_a_tool_call_answer_is_accepted(monkeypatch, tmp_path): + tool_call = {"role": "assistant", "content": None, + "tool_calls": [{"id": "c1", "type": "function", "function": {"name": "lookup", "arguments": "{}"}}]} + app, capture = make_app(monkeypatch, tmp_path, respond=routed(message=tool_call)) + tools = [{"type": "function", "function": {"name": "lookup", "parameters": {"type": "object"}}}] + resp, _ = await call(app, {**JEV_REQ, "tools": tools, "tool_choice": "auto", "parallel_tool_calls": False}) + + assert resp.status_code == 200 + assert capture.body["tools"] == tools + assert capture.body["tool_choice"] == "auto" + assert resp.json()["choices"][0]["message"]["tool_calls"][0]["function"]["name"] == "lookup" + + +# ---- policy boundary ---- + + +@pytest.mark.asyncio +async def test_authorization_gates_jev_router_before_any_call(monkeypatch, tmp_path): + app, capture = make_app(monkeypatch, tmp_path) + async with app.router.lifespan_context(app): + async with client_for(app) as client: + anonymous = await client.post("/v1/chat/completions", json=JEV_REQ) + denied = await client.post( + "/v1/chat/completions", json=JEV_REQ, headers={"authorization": "Bearer cb-dispatch"} + ) + assert anonymous.status_code == 401 + assert denied.status_code == 403 + assert capture.requests == [] + + +@pytest.mark.asyncio +async def test_a_router_outage_fails_only_jev_router(monkeypatch, tmp_path): + def unreachable(request: httpx.Request) -> httpx.Response: + raise httpx.ConnectError("refused", request=request) + + app, _ = make_app(monkeypatch, tmp_path, respond=unreachable) + async with app.router.lifespan_context(app): + async with client_for(app) as client: + jev = await client.post("/v1/chat/completions", json=JEV_REQ, headers=AUTH) + dispatch = await client.post("/v1/chat/completions", json=DISPATCH_REQ, headers=AUTH) + events = (await client.get("/admin/events")).json()["events"] + + assert jev.status_code == 503 + assert dispatch.status_code == 200 + event = next(e for e in events if e["mode"] == "jev-router") + assert event["outcome"] == "jev_router_unavailable" + assert event["attempts"][0]["outcome"] == "transport_error" + + +@pytest.mark.parametrize( + ("respond", "status", "outcome", "attempt_outcome"), + [ + (lambda r: httpx.Response(404, json={"error": "widen allowed_models"}), 502, "upstream_error", "invalid_response"), + (lambda r: httpx.Response(429, json={"error": "rate"}), 502, "upstream_error", "retryable_status"), + (lambda r: httpx.Response(503, text="down"), 503, "jev_router_unavailable", "retryable_status"), + (lambda r: httpx.Response(200, content=b"", headers={"content-type": "text/html"}), + 502, "upstream_error", "invalid_response"), + (lambda r: httpx.Response(200, json={"id": "g", "choices": [{"message": {"role": "assistant"}}]}), + 502, "upstream_error", "invalid_response"), + ], + ids=["http-404", "http-429", "http-503", "invalid-json", "no-content"], +) +@pytest.mark.asyncio +async def test_bad_router_responses_fail_closed(monkeypatch, tmp_path, respond, status, outcome, attempt_outcome): + app, _ = make_app(monkeypatch, tmp_path, respond=respond) + resp, events = await call(app) + assert resp.status_code == status + assert "choices" not in resp.json() + assert events[0]["outcome"] == outcome + assert events[0]["attempts"][0]["outcome"] == attempt_outcome + + +# ---- admin projection and control-plane records ---- + + +@pytest.mark.asyncio +async def test_routes_project_the_pool_jev_would_be_sent(monkeypatch, tmp_path): + app, _ = make_app(monkeypatch, tmp_path) + app.state.cooldowns.apply( + scope="model", provider="openrouter", credential="main", model="vendor/model-b", + reason="quota_429", duration_seconds=600, + ) + async with app.router.lifespan_context(app): + async with client_for(app) as client: + projection = (await client.get("/admin/routes")).json() + + entry = next(a for a in projection["aliases"] if a["alias"] == "cerberus/auto") + assert entry["mode"] == "jev-router" + assert "paths" not in entry and "fusion" not in entry + jev = entry["jev_router"] + assert (jev["backend"], jev["router_model"], jev["allow_paid_pool"]) == ("openrouter", "typesafe/jev-router", False) + assert [(m["model"], m["state"]) for m in jev["pool"]] == [ + ("vendor/model-a", "eligible"), + ("vendor/model-b", "excluded"), + ], "a pool has no preference order, so never a standby" + assert jev["pool"][1]["exclusion"]["reason"] == "cooldown_quota_429" + assert jev["readiness"] == {"backend_present": True, "pool_available": 1, "pool_size": 2, "available": True} + serialized = json.dumps(projection) + assert "OPENROUTER_API_KEY" not in serialized and "or-key" not in serialized + + +@pytest.mark.asyncio +async def test_revision_materializes_the_jev_router_pool(monkeypatch, tmp_path): + import sqlite3 + + state = tmp_path / "state.sqlite3" + app, _ = make_app(monkeypatch, tmp_path, raw=jev_raw(state_path=state)) + async with app.router.lifespan_context(app): + pass + connection = sqlite3.connect(state) + try: + binding = connection.execute( + "SELECT binding_type FROM alias_bindings WHERE alias='cerberus/auto'" + ).fetchone() + members = connection.execute( + "SELECT role, ordinal, model FROM route_members WHERE alias='cerberus/auto' ORDER BY ordinal" + ).fetchall() + finally: + connection.close() + assert binding == ("jev-router",) + assert members == [("pool", 0, "vendor/model-a"), ("pool", 1, "vendor/model-b")] + + +# ---- backend unit: wire shape independent of the app ---- + + +def test_backend_builds_the_documented_body_and_owns_routing_fields(): + request = JevRouterRequest( + body={"messages": [], "model": "cerberus/auto", "stream": True, "stream_options": {"include_usage": True}, + "plugins": [{"id": "x"}], "seed": 7}, + pool_models=["vendor/model-a"], + base_url="https://openrouter.ai/api/v1/", + api_key="k", + timeout_seconds=5, + ) + assert OpenRouterJevRouterBackend.build_body(request) == { + "messages": [], + "seed": 7, + "model": "typesafe/jev-router", + "plugins": [{"id": "jev-router", "models": ["vendor/model-a"]}], + } + + +@pytest.mark.asyncio +async def test_backend_refuses_an_empty_pool_and_normalizes_errors(): + backend = OpenRouterJevRouterBackend(httpx.MockTransport(routed())) + empty = JevRouterRequest(body={"messages": []}, pool_models=[], base_url="https://openrouter.ai/api/v1", + api_key="k", timeout_seconds=5) + with pytest.raises(ValueError, match="non-empty pool"): + await backend.execute(empty) + await backend.aclose() + + def slow(request: httpx.Request) -> httpx.Response: + raise httpx.ReadTimeout("slow", request=request) + + backend = OpenRouterJevRouterBackend(httpx.MockTransport(slow)) + request = JevRouterRequest(body={"messages": []}, pool_models=["vendor/model-a"], + base_url="https://openrouter.ai/api/v1", api_key="k", timeout_seconds=5) + with pytest.raises(JevRouterError) as info: + await backend.execute(request) + assert (info.value.http_status, info.value.outcome) == (503, "jev_router_unavailable") + await backend.aclose() diff --git a/tests/test_schema.py b/tests/test_schema.py index 7da010d..f1c4d4f 100644 --- a/tests/test_schema.py +++ b/tests/test_schema.py @@ -299,3 +299,93 @@ def test_non_loopback_server_requires_api_token_env(): def test_unknown_top_level_keys_rejected(): with pytest.raises(ValidationError): CerberusConfig.model_validate(make(pools={"free": {"providers": ["openrouter"]}})) + + +# -- jev-router: Jev chooses within a pool Cerberus validates ------------------ + + +def _jev_raw(**alias_overrides) -> dict: + alias = { + "mode": "jev-router", + "candidates": [ + {"provider": "openrouter", "credential": "primary", "model": "openrouter/free"}, + ], + "jev_router": {"timeout_seconds": 60}, + } + alias.update(alias_overrides) + return make(**{"aliases.cerberus/auto": alias}) + + +def test_jev_router_alias_valid(): + config = CerberusConfig.model_validate(_jev_raw()) + policy = config.aliases["cerberus/auto"].jev_router + assert policy is not None and policy.backend == "openrouter" and policy.allow_paid_pool is False + + +def test_jev_router_mode_and_policy_block_must_agree(): + with pytest.raises(ValidationError, match="jev_router policy block"): + CerberusConfig.model_validate(_jev_raw(jev_router=None)) + raw = make() + raw["aliases"]["cerberus/free"]["jev_router"] = {"timeout_seconds": 60} + with pytest.raises(ValidationError, match="only valid on a jev-router alias"): + CerberusConfig.model_validate(raw) + + +def test_jev_router_refuses_allow_paid_fallback(): + with pytest.raises(ValidationError, match="allow_paid_pool"): + CerberusConfig.model_validate(_jev_raw(allow_paid_fallback=True)) + + +def test_jev_router_paid_pool_member_requires_opt_in(): + paid = [ + {"provider": "openrouter", "credential": "primary", "model": "openrouter/free"}, + {"provider": "openrouter", "credential": "primary", "model": "anthropic/claude-sonnet"}, + ] + with pytest.raises(ValidationError, match="allow_paid_pool"): + CerberusConfig.model_validate(_jev_raw(candidates=paid)) + assert CerberusConfig.model_validate( + _jev_raw(candidates=paid, jev_router={"timeout_seconds": 60, "allow_paid_pool": True}) + ) + + +def test_jev_router_pool_shares_one_provider_and_credential(): + mixed = [ + {"provider": "openrouter", "credential": "primary", "model": "openrouter/free"}, + {"provider": "google", "credential": "free-1", "model": "gemini-flash"}, + ] + with pytest.raises(ValidationError, match="one provider/credential"): + CerberusConfig.model_validate(_jev_raw(candidates=mixed)) + + +@pytest.mark.parametrize("pattern", ["anthropic/*", "*flash*", "~openai/gpt-luna-latest"]) +def test_jev_router_pool_admits_exact_slugs_only(pattern): + """The plugin reads its list as patterns; a registry key shaped like one + would admit models the registry never listed.""" + raw = _jev_raw(candidates=[{"provider": "openrouter", "credential": "primary", "model": pattern}]) + raw["providers"]["openrouter"]["models"][pattern] = {"cost_tier": "free"} + with pytest.raises(ValidationError, match="exact registry slugs"): + CerberusConfig.model_validate(raw) + + +@pytest.mark.parametrize( + "budget", [{"reasoning_effort": "low"}, {"chat_template_kwargs": {"enable_thinking": False}}] +) +def test_jev_router_refuses_a_per_candidate_reasoning_budget(budget): + candidate = {"provider": "openrouter", "credential": "primary", "model": "openrouter/free", **budget} + with pytest.raises(ValidationError, match="reasoning effort per request"): + CerberusConfig.model_validate(_jev_raw(candidates=[candidate])) + + +def test_jev_router_refuses_a_duplicate_pool_member(): + twice = [{"provider": "openrouter", "credential": "primary", "model": "openrouter/free"}] * 2 + with pytest.raises(ValidationError, match="listed twice"): + CerberusConfig.model_validate(_jev_raw(candidates=twice)) + + +def test_jev_router_identity_must_allow_the_mode(): + raw = _jev_raw() + raw["identities"]["recon"]["allowed_aliases"] = ["cerberus/free", "cerberus/auto"] + with pytest.raises(ValidationError, match="mode"): + CerberusConfig.model_validate(raw) + raw["identities"]["recon"]["allowed_modes"] = ["free", "jev-router"] + assert CerberusConfig.model_validate(raw) diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py index ca8943d..2488d19 100644 --- a/tests/test_telemetry.py +++ b/tests/test_telemetry.py @@ -93,7 +93,7 @@ async def telemetry(request: httpx.Request) -> httpx.Response: assert len(events) == 1 assert events[0].headers["authorization"] == "Bearer telemetry-test-value" event = json.loads(events[0].content) - assert event["schema_version"] == 5 + assert event["schema_version"] == 6 assert event["alias"] == "cerberus/controlled" assert event["attempts"][0]["used_fallback"] is False assert event["attempts"][0]["cooldown_scope"] is None From cdd06542bd7fcbcc5143287513f84374c06e721a Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 5 Oct 2026 12:54:23 +0000 Subject: [PATCH 2/2] Let jev-router own its caller-field gate after the allowlist merge The merged caller-field allowlist refuses non-function tools in app.py for every alias except fusion, ahead of mode dispatch and without a routing event. jev-router enforces a stricter allowlist of its own with an invalid_request event, so it is exempted the same way fusion is, and its allowlist is now the shared CALLER_FIELDS less the reasoning controls rather than a second copy that could drift. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01P4Nm4MVSMmG6JyZi8WumtK --- src/cerberus/app.py | 8 +++--- src/cerberus/jev_router/dispatch.py | 41 ++++++----------------------- 2 files changed, 13 insertions(+), 36 deletions(-) diff --git a/src/cerberus/app.py b/src/cerberus/app.py index 52d5edf..9153940 100644 --- a/src/cerberus/app.py +++ b/src/cerberus/app.py @@ -934,9 +934,11 @@ async def chat_completions(request: Request) -> JSONResponse | StreamingResponse status_code=403, content={"error": {"message": f"Identity not authorized for {alias_name!r}", "reason": denial}}, ) - if alias.mode != "fusion": - # fusion rejects caller tools outright (fusion_dispatch); elsewhere - # only function tools are forwarded + if alias.mode not in ("fusion", "jev-router"): + # fusion rejects caller tools outright (fusion_dispatch) and + # jev-router refuses anything outside its own allowlist + # (jev_router_dispatch), each with a routing event; elsewhere only + # function tools are forwarded server_tools = non_function_tools(body) if server_tools: return JSONResponse( diff --git a/src/cerberus/jev_router/dispatch.py b/src/cerberus/jev_router/dispatch.py index 2f9783b..2659227 100644 --- a/src/cerberus/jev_router/dispatch.py +++ b/src/cerberus/jev_router/dispatch.py @@ -19,6 +19,7 @@ from fastapi.responses import JSONResponse +from cerberus.caller_fields import CALLER_FIELDS # cerberus.router before cerberus.egress: egress.client imports router.engine, and # the router package imports egress.client in turn, so egress first is a cycle. from cerberus.router.engine import Target @@ -36,40 +37,14 @@ from cerberus.registry.schema import Alias, JevRouterPolicy from cerberus.telemetry import RoutingAttempt, RoutingEvent, TelemetryEmitter -# What a caller may set on a jev-router request. This is an allowlist, not a -# denylist: anything that decides where a request runs (model fallback lists, +# What a caller may set on a jev-router request: the standard chat fields every +# alias forwards, less the reasoning controls, because Jev chooses the effort per +# request and overriding it silently in either direction would hide the +# substitution. Unlike dispatch, which drops anything else, a jev-router alias +# refuses it: anything that decides where a request runs (model fallback lists, # route, plugins, presets, provider routing, server tools) is Cerberus's to -# construct, and anything not listed here is refused rather than forwarded, so -# a field OpenRouter adds tomorrow cannot widen the pool. reasoning_effort and -# reasoning are absent on purpose: Jev chooses the effort per request, and -# overriding it silently in either direction would hide the substitution. -_CALLER_FIELDS = frozenset( - { - "messages", - "model", # the alias; replaced by the router model - "stream", # removed; jev-router is not streamed - "stream_options", - "temperature", - "top_p", - "n", - "stop", - "max_tokens", - "max_completion_tokens", - "presence_penalty", - "frequency_penalty", - "logit_bias", - "logprobs", - "top_logprobs", - "seed", - "user", - "response_format", - "tools", # function tools only; see caller_field_violations - "tool_choice", - "parallel_tool_calls", - "functions", - "function_call", - } -) +# construct, and a caller is told rather than quietly served something else. +_CALLER_FIELDS = CALLER_FIELDS - {"reasoning_effort", "chat_template_kwargs"} _TOOL_CHOICE_MODES = frozenset({"none", "auto", "required"})