Refuse caller-supplied model selection on dispatch-path aliases

docs/architecture.md

@@ -21,8 +21,11 @@ constitute a general billing cap or a global distributed rate limiter.
 The backend sees only what Cerberus sends it: a request Cerberus has already
 authorized against the caller's identity and alias policy, under the credential
 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.
+surface of a fusion request: a body carrying `tools`, `tool_choice`, or any of
+the model-selection keys every alias refuses (see Dispatch control plane) is
+answered 400 before the backend is called. A caller cannot reach the backend
+except through a fusion alias it is allowed to use. Fusion-mode deliberation is
+not streamed.
 
 ## Dispatch control plane
 
@@ -43,6 +46,23 @@ per-request failures are excluded at their own scope by the cooldown store.
 Normal label lookup reads the in-memory snapshot and performs no SQL.
 Routing and usage records are written asynchronously after a decision.
 
+The request body cannot add a path either. Cerberus overwrites `model` with the
+target it selected, and refuses with a 400, before any upstream call, a body
+carrying a key an upstream reads as model selection: `models` and `route`
+(OpenRouter's fallback list), `plugins`, `preset`, or a server tool, meaning an
+entry in `tools` or a `tool_choice` whose `type` starts with `openrouter:`
+(`openrouter:fusion` names a panel and analyst of its own). Any of them would let
+a caller run, on the operator's credential, a model the pinned revision never
+named and the free/paid cost gate never weighed. The rule is
+`engine.caller_model_selection`. It applies to every alias mode and every
+provider, OpenRouter or not, because the failover loop may reach any candidate.
+Refusing rather than stripping keeps the substitution from ever being silent, as
+with `reasoning_effort_override`. The refusal is recorded as a routing event,
+outcome `upstream_error` with status 400 and no attempts, the same shape fusion
+records. Function tools, a `tool_choice` naming one, and keys that leave the
+model alone pass through unchanged. That includes `provider` preferences, which
+only choose among hosts for the model Cerberus already pinned.
+
 `/health` is the only ungated endpoint and answers liveness only —
 `{"status": "ok", "service": "cerberus"}` — which is what a container or
 orchestrator probe needs. Operational diagnostics (config version and checksum,

src/cerberus/fusion/dispatch.py

@@ -22,12 +22,15 @@
 from cerberus.fusion.backend import FusionBackend, FusionError, FusionRequest
 from cerberus.registry.loader import ConfigDocument
 from cerberus.registry.schema import Alias, CerberusConfig
+from cerberus.router.engine import CALLER_ROUTING_KEYS
 from cerberus.telemetry import RoutingAttempt, RoutingEvent, TelemetryEmitter
 
 # Caller-supplied keys a fusion alias cannot honor: the backend owns the tool
 # surface during deliberation, so accepting them would silently drop or reroute
-# the caller's intent. Rejected up front rather than forwarded.
-_REJECTED_CALLER_KEYS = ("tools", "tool_choice", "plugins")
+# the caller's intent. Rejected up front rather than forwarded. The keys every
+# alias refuses because they choose models upstream (models, route, ...) would
+# otherwise reach OpenRouter beside the deliberation, so they are refused too.
+_REJECTED_CALLER_KEYS = ("tools", "tool_choice", *CALLER_ROUTING_KEYS)
 
 
 def fusion_aliases(config: CerberusConfig) -> list[str]:

src/cerberus/router/dispatch.py

@@ -26,7 +26,7 @@
 )
 from cerberus.registry.loader import ConfigDocument
 from cerberus.router.availability import runtime_exclusion
-from cerberus.router.engine import Target, cost_eligible, ordered_targets
+from cerberus.router.engine import Target, caller_model_selection, cost_eligible, ordered_targets
 from cerberus.state.cooldowns import InMemoryCooldownStore
 from cerberus.telemetry import RoutingAttempt, RoutingEvent, RoutingOutcome, TelemetryEmitter
 
@@ -199,6 +199,41 @@ async def dispatch(
 
     targets = ordered_targets(config, alias_name)
     ordered = [f"{t.provider_id}/{t.credential_id}/{t.model}" for t in targets]
+
+    # The pinned revision alone chooses the model. A body that would choose or
+    # add one upstream is refused before any provider sees it, whichever
+    # candidate the loop would have reached, so the refusal is never silent.
+    rejected = caller_model_selection(body)
+    if rejected:
+        telemetry.emit(
+            _event(
+                request_id=request_id,
+                alias=alias_name,
+                mode=alias.mode,
+                target=None,
+                attempts=[],
+                http_status=400,
+                outcome="upstream_error",
+                started_at=started_at,
+                usage=None,
+                streaming=streaming,
+                fallback=False,
+                identity=identity_name,
+                config_version=document.version,
+                config_checksum=document.checksum,
+                candidates=ordered,
+            )
+        )
+        return JSONResponse(
+            status_code=400,
+            content={
+                "error": {
+                    "message": f"alias does not accept caller-supplied model selection: {', '.join(rejected)}",
+                    "request_id": request_id,
+                }
+            },
+        )
+
     eligible, exclusions = cost_eligible(alias, targets)
     attempts: list[RoutingAttempt] = []
     attempted = 0

src/cerberus/router/engine.py

@@ -1,8 +1,9 @@
-"""Deterministic target resolution and cost-tier gating.
+"""Deterministic target resolution, cost-tier gating and the caller routing gate.
 
 Policy dispatch is deterministic: same config version + same alias -> same
 ordered target list. Runtime failover over that order is state-dependent and
-belongs to the dispatch loop, not here.
+belongs to the dispatch loop, not here. Neither may a request body widen that
+list: caller_model_selection names the keys that would choose models upstream.
 """
 
 from __future__ import annotations
@@ -13,6 +14,17 @@
 
 from cerberus.registry.schema import Alias, CerberusConfig, CostTier
 
+# Caller-supplied body keys an upstream reads as model selection. OpenRouter
+# honours every one: models/route are its fallback list, and plugins and preset
+# can carry routing of their own. Forwarding any of them would let a caller run,
+# on the operator's credential, a model the pinned revision never named and
+# cost_eligible never gated.
+CALLER_ROUTING_KEYS = ("models", "route", "plugins", "preset")
+# OpenRouter server tools execute upstream under the operator's credential, and
+# some run models of their own (openrouter:fusion names a whole panel and
+# analyst). Refused as a class by type prefix; function tools are untouched.
+SERVER_TOOL_PREFIX = "openrouter:"
+
 
 @dataclass(frozen=True, slots=True)
 class Target:
@@ -96,3 +108,25 @@ def cost_eligible(alias: Alias, targets: list[Target]) -> tuple[list[Target], li
                 continue
         eligible.append(target)
     return eligible, exclusions
+
+
+def caller_model_selection(body: Mapping[str, Any]) -> list[str]:
+    """The request-body keys that would choose or add models upstream.
+
+    Empty means the body leaves model selection to the pinned revision. A
+    non-empty answer is refused before any upstream call rather than stripped,
+    so the caller learns its routing was not honoured instead of being served
+    by a model it did not ask for.
+    """
+    found = [key for key in CALLER_ROUTING_KEYS if key in body]
+    tools = body.get("tools")
+    if any(_is_server_tool(tool) for tool in (tools if isinstance(tools, list) else [tools])):
+        found.append("tools")
+    if _is_server_tool(body.get("tool_choice")):
+        found.append("tool_choice")
+    return found
+
+
+def _is_server_tool(value: Any) -> bool:
+    kind = value.get("type") if isinstance(value, Mapping) else value
+    return isinstance(kind, str) and kind.strip().lower().startswith(SERVER_TOOL_PREFIX)

tests/test_caller_model_selection.py

@@ -0,0 +1,132 @@
+"""A request body cannot choose models upstream on a dispatch-path alias.
+
+``build_upstream_request`` overwrites only ``model``, so any other key reaches
+the provider as sent. OpenRouter reads several of them as model selection
+(``models``/``route`` fallback lists, ``plugins``, ``preset``, ``openrouter:*``
+server tools), which would let a caller run models absent from the pinned
+revision on the operator's credential. The upstream is mocked at the HTTP
+boundary; these prove such a body is refused before it, while tool calling and
+keys that leave the model alone are forwarded untouched.
+"""
+
+import httpx
+import pytest
+
+from cerberus.app import create_app
+from cerberus.registry import load_config_document
+from cerberus.router.engine import caller_model_selection
+from tests.test_control import ok_upstream, write_config
+from tests.test_fusion_dispatch import AUTH, Capture, client_for, fusion_raw
+
+FUSION_TOOL = {
+    "type": "openrouter:fusion",
+    "parameters": {"analysis_models": ["openai/gpt-5", "anthropic/claude-opus-4"], "model": "openai/gpt-5"},
+}
+FUNCTION_TOOL = {
+    "type": "function",
+    "function": {"name": "lookup", "parameters": {"type": "object", "properties": {"q": {"type": "string"}}}},
+}
+ALIASES = ("cerberus/dispatch-dev", "cerberus/free-dev")
+
+
+def raw_config() -> dict:
+    """The fusion-dev slice (an OpenRouter dispatch candidate) plus a free alias."""
+    raw = fusion_raw()
+    raw["aliases"]["cerberus/free-dev"] = {
+        "mode": "free",
+        "candidates": [{"provider": "openrouter", "credential": "main", "model": "free-b"}],
+    }
+    raw["identities"]["dev"]["allowed_modes"].append("free")
+    raw["identities"]["dev"]["allowed_aliases"].append("cerberus/free-dev")
+    return raw
+
+
+def make_app(monkeypatch, tmp_path):
+    monkeypatch.setenv("OPENROUTER_API_KEY", "or-key")
+    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_config()), validate_credentials=False)
+    upstream = Capture(ok_upstream)
+    return create_app(doc, http_transport=httpx.MockTransport(upstream)), upstream
+
+
+async def call(app, body):
+    async with app.router.lifespan_context(app):
+        async with client_for(app) as client:
+            resp = await client.post("/v1/chat/completions", json=body, headers=AUTH)
+            events = (await client.get("/admin/events")).json()["events"]
+    return resp, events
+
+
+def request(alias: str, **extra) -> dict:
+    return {"model": alias, "messages": [{"role": "user", "content": "hi"}], **extra}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("alias", ALIASES)
+@pytest.mark.parametrize(
+    ("key", "extra"),
+    [
+        ("models", {"models": ["openai/gpt-5", "anthropic/claude-opus-4"]}),
+        ("route", {"route": "fallback"}),
+        ("plugins", {"plugins": [{"id": "jev-router"}]}),
+        ("preset", {"preset": "operator-preset"}),
+        ("tools", {"tools": [FUNCTION_TOOL, FUSION_TOOL]}),
+        ("tool_choice", {"tool_choice": {"type": "openrouter:fusion"}}),
+        ("models", {"models": ["openai/gpt-5"], "stream": True}),
+    ],
+    ids=["models", "route", "plugins", "preset", "openrouter-fusion-tool", "openrouter-tool-choice", "streaming"],
+)
+async def test_model_selection_keys_are_rejected_before_any_upstream_call(monkeypatch, tmp_path, alias, key, extra):
+    app, upstream = make_app(monkeypatch, tmp_path)
+    resp, events = await call(app, request(alias, **extra))
+
+    assert upstream.requests == [], "a body that selects models must never reach the provider"
+    assert resp.status_code == 400
+    error = resp.json()["error"]
+    assert key in error["message"] and error["request_id"]
+    assert len(events) == 1
+    event = events[0]
+    assert event["outcome"] == "upstream_error" and event["http_status"] == 400
+    assert event["alias"] == alias and event["identity"] == "dev"
+    assert event["request_id"] == error["request_id"]
+    assert event["attempt_count"] == 0 and event["provider"] is None and event["model"] is None
+    assert event["streaming"] is bool(extra.get("stream"))
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("alias", ALIASES)
+async def test_function_tools_and_non_routing_keys_are_forwarded_unchanged(monkeypatch, tmp_path, alias):
+    app, upstream = make_app(monkeypatch, tmp_path)
+    tool_choice = {"type": "function", "function": {"name": "lookup"}}
+    # provider preferences narrow which host serves the pinned model; they never pick another one
+    provider = {"data_collection": "deny", "allow_fallbacks": False}
+    resp, events = await call(
+        app, request(alias, tools=[FUNCTION_TOOL], tool_choice=tool_choice, provider=provider, temperature=0.2)
+    )
+
+    assert resp.status_code == 200
+    assert len(upstream.requests) == 1
+    sent = upstream.body
+    assert sent["model"] == ("free-a" if alias == "cerberus/dispatch-dev" else "free-b")
+    assert sent["tools"] == [FUNCTION_TOOL]
+    assert sent["tool_choice"] == tool_choice
+    assert sent["provider"] == provider and sent["temperature"] == 0.2
+    assert events[0]["outcome"] == "success"
+
+
+@pytest.mark.parametrize(
+    ("body", "expected"),
+    [
+        ({"messages": []}, []),
+        ({"tools": [FUNCTION_TOOL], "tool_choice": "auto"}, []),
+        ({"tool_choice": "required", "provider": {"order": ["x"]}}, []),
+        ({"tools": None, "tool_choice": None}, []),
+        ({"tools": [{"type": "OpenRouter:Web_Search"}]}, ["tools"]),
+        ({"tools": FUSION_TOOL}, ["tools"]),
+        ({"tool_choice": "openrouter:fusion"}, ["tool_choice"]),
+        ({"models": [], "route": None, "plugins": [], "preset": ""}, ["models", "route", "plugins", "preset"]),
+    ],
+)
+def test_caller_model_selection_classifies_bodies(body, expected):
+    assert caller_model_selection(body) == expected

tests/test_fusion_dispatch.py

@@ -216,9 +216,11 @@ async def test_caller_cannot_override_panel_or_tool_surface(monkeypatch, tmp_pat
     resp, _ = await call(app, {**FUSION_REQ, "stream": True})
     assert resp.status_code == 200
     assert capture.body["model"] == "free-a" and "stream" not in capture.body
-    # ... but tools/tool_choice/plugins cannot be honored through a fusion alias: reject, never forward
+    # ... but tools/tool_choice/plugins cannot be honored through a fusion alias, and models/route/preset
+    # would choose models beside the deliberation: reject, never forward
     for key, value in (("plugins", [{"id": "fusion", "analysis_models": ["free-c"]}]), ("tool_choice", "none"),
-                       ("tools", [{"type": "function", "function": {"name": "x"}}])):
+                       ("tools", [{"type": "function", "function": {"name": "x"}}]),
+                       ("models", ["paid-x"]), ("route", "fallback"), ("preset", "operator-preset")):
         app, capture = make_app(monkeypatch, tmp_path)
         resp, events = await call(app, {**FUSION_REQ, key: value})
         assert resp.status_code == 400, key