feat(routing): add stateless candidate controls

CHANGELOG.md

@@ -10,6 +10,14 @@ and this project uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html)
 
 ## [0.2.0] - Unreleased
 
+### Added
+
+- Add request-local `routing.candidate_id` and `exclude_candidate_ids` controls
+  for trusted virtual-model chat and Responses calls. Pins and exclusions are
+  strictly validated, never persisted, honored by route, structured, and
+  streaming paths, and disclosed only as per-response routing evidence; the
+  public model catalog remains unchanged.
+
 ### Deprecated
 
 - Internal callers now use
@@ -20,6 +28,51 @@ and this project uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html)
 
 ### Fixed
 
+- Auto-mode triage now runs a live decision call even when `zdr_only` is
+  active and `routing.candidate_id` pins a paid ZDR-eligible candidate.
+  `_compute_triage_verdict`'s free-only ranking is always empty in that
+  shape (the pin restricts every candidate list to that one agent, and a
+  paid agent never satisfies `free_only`), and the empty-pool fallback was
+  gated behind `not zdr_only` even though the fallback's own per-agent
+  filter (`_zdr_agent_allowed` plus the active pin/exclusion) already makes
+  it safe to consult under an active ZDR policy. The gate discarded the one
+  legitimate evidence source instead of protecting against contacting a
+  non-ZDR provider, so the request silently took the direct route with zero
+  triage call and zero routing evidence regardless of task complexity
+  (Devin Review, PR #983: "ZDR pins skip workflow triage").
+- The published OpenAPI schema for `CandidateRoutingControls.exclude_candidate_ids`
+  no longer declares `maxItems: 32`. The runtime validator's own repository-authored
+  32-ID cardinality cutoff was already removed as unsupported; the schema still
+  publishing that limit meant generated/OpenAPI clients rejected exclusion lists the
+  runtime intentionally accepts, making the schema false at source. `uniqueItems`,
+  lexical ID constraints, and normal authenticated request-size bounds are
+  unchanged; no replacement cardinality heuristic was introduced.
+- The real-time model judge no longer selects a verifier-excluded agent as
+  the judge when it is the sole candidate. `_ranked_agents` deliberately
+  still returns role-ineligible members (appended after every eligible one),
+  so a caller that wants only role-eligible candidates must re-apply
+  `role not in agent.provider_exclusions` itself — `_plan_generated` and
+  `_parse_workflow_plan` already do; `_model_judge_verification`'s judge
+  selection did not. With a single-candidate pool excluded from `verifier`
+  (e.g. a worker-only pinned agent under PR #983's `orchestrator/free`
+  provable-route carve-out), that judge selection picked the ineligible
+  agent anyway, producing an extra, unrequested live call once fast-mlsirm
+  is actually importable (observed as a duplicate served-candidate call in
+  hosted CI, which every earlier sandboxed verification round of this PR
+  could not reproduce locally because the sandbox's blocked fast-mlsirm
+  archive download always short-circuits the judge to its fail-closed path
+  first). `_invoke`'s own failover already enforced this exclusion for a
+  *backup* judge; this closes the same gap for the *primary* selection.
+- `route_once` and `stream_route` no longer stamp `served_agent_id` on every
+  trace row unconditionally. An earlier no-heuristics repair for candidate-
+  routing evidence made that stamp unconditional to give
+  `_candidate_routing_evidence` an explicit serving fact even when unchanged,
+  but this broke the pre-existing contract (regression-guarded by
+  `test_provider_reliability.py` and `test_tool_execution_fallback.py`) that
+  the ordinary, no-candidate-policy path never carries failover metadata for
+  an unchanged serving agent. The stamp is now unconditional only while
+  request-local candidate-attempt tracking is actually active (i.e. inside a
+  `candidate_routing_policy` scope); the ordinary path is unaffected.
 - Workflow workers now preserve the caller message array exactly once, while
   the added envelope carries only the subtask and Conductor-style prior-step
   access list instead of duplicating the task or source attachments.
@@ -1256,3 +1309,4 @@ This is the current development baseline, not a published release. It
 provides the OpenAI-compatible gateway, route/conduct orchestration, workflow
 and access evidence, provider credential boundaries, cost and readiness
 reporting, and security-focused contract tests.
+- Candidate routing controls no longer impose an unsupported 32-ID exclusion cutoff or infer serving identity from output text/trace order; serving identity now requires explicit provenance and otherwise fails closed.

README.md

@@ -264,6 +264,29 @@ is read from a **KV config store**, never `os.getenv`.
   (`{"routing": {"latency_tolerant": true}}` on `/v1/chat/completions`) plus
   KV thresholds. Interactive requests stay on the fast sync path; latency-tolerant
   or bulk requests are dispatched to a batch backend.
+- **Stateless candidate control.** Trusted callers may add
+  `routing.candidate_id` to pin one private agent ID and
+  `routing.exclude_candidate_ids` (unique exact IDs) to omit evidence-ineligible
+  candidates for one virtual-model request. The gateway validates the full set
+  before any provider call, forces synchronous execution, and returns
+  requested, excluded, attempted, and served IDs under
+  `orchestration.routing`. Omitting both keys -- or supplying only an empty
+  `exclude_candidate_ids` array with no `candidate_id` -- preserves the
+  existing request and response contract: neither constrains selection, so
+  there is no routing decision to attest to. Concrete provider model names
+  cannot be combined with these controls; candidate IDs remain absent from
+  `/v1/models`.
+
+  ```json
+  {
+    "model": "orchestrator/auto",
+    "messages": [{"role": "user", "content": "Review this change"}],
+    "routing": {
+      "candidate_id": "candidate_b",
+      "exclude_candidate_ids": ["candidate_a"]
+    }
+  }
+  ```
 - **Batch routing to pg-llm-batch.** The production batch backend is an injected
   [`pg-llm-batch`](https://github.com/ContextualWisdomLab/pg-llm-batch)
   OpenAI-compatible Batch API client (submit JSONL -> poll -> retrieve). A local

contextual_orchestrator/api_contract.py

@@ -22,6 +22,79 @@
             },
         },
         "schemas": {
+            "CandidateRoutingControls": {
+                "type": "object",
+                "properties": {
+                    "channel": {"type": "string", "enum": ["sync", "batch"]},
+                    "latency_tolerant": {"type": "boolean"},
+                    "priority": {
+                        "type": "string",
+                        "enum": ["interactive", "normal", "bulk"],
+                    },
+                    "candidate_id": {
+                        "type": "string",
+                        "minLength": 1,
+                        "pattern": r"^\S(?:[^\r\n]*\S)?(?![\s\S])",
+                        "description": "Exact private agent ID to use for this request.",
+                    },
+                    "exclude_candidate_ids": {
+                        "type": "array",
+                        "uniqueItems": True,
+                        "items": {
+                            "type": "string",
+                            "minLength": 1,
+                            "pattern": r"^\S(?:[^\r\n]*\S)?(?![\s\S])",
+                        },
+                    },
+                    "endpoint": {
+                        "type": "string",
+                        "minLength": 1,
+                        "pattern": r"^\S(?:[^\r\n]*\S)?(?![\s\S])",
+                        "description": (
+                            "Pin the request to one configured endpoint selector. "
+                            "Forces synchronous routing (channel=sync); cannot be "
+                            "combined with channel=batch or latency_tolerant=true."
+                        ),
+                    },
+                },
+                "additionalProperties": False,
+            },
+            "CandidateRoutingEvidence": {
+                "type": "object",
+                "description": (
+                    "Per-response disclosure of how active candidate controls "
+                    "were applied. Present only when the request supplied "
+                    "routing.candidate_id and/or routing.exclude_candidate_ids."
+                ),
+                "properties": {
+                    "candidate_id": {
+                        "type": "string",
+                        "description": "Echoes the request's routing.candidate_id, when pinned.",
+                    },
+                    "exclude_candidate_ids": {
+                        "type": "array",
+                        "items": {"type": "string"},
+                        "description": "The request's routing.exclude_candidate_ids, sorted.",
+                    },
+                    "attempted_candidate_ids": {
+                        "type": "array",
+                        "items": {"type": "string"},
+                        "description": (
+                            "Every private agent ID a provider call was attempted "
+                            "against while serving this request, in first-attempt order."
+                        ),
+                    },
+                    "served_candidate_id": {
+                        "type": "string",
+                        "description": (
+                            "The private agent ID whose output was actually returned "
+                            "to the caller, when determinable."
+                        ),
+                    },
+                },
+                "required": ["exclude_candidate_ids", "attempted_candidate_ids"],
+                "additionalProperties": False,
+            },
             "AuthoritativeUsage": {
                 "type": ["object", "null"],
                 "required": ["prompt_tokens", "completion_tokens"],
@@ -74,7 +147,12 @@
                         "type": "string",
                         "enum": ["measured", "unavailable"],
                     },
-                    "orchestration": {"type": "object"},
+                    "orchestration": {
+                        "type": "object",
+                        "properties": {
+                            "routing": {"$ref": "#/components/schemas/CandidateRoutingEvidence"},
+                        },
+                    },
                 },
             },
             "ModelGroupWrite": {
@@ -214,6 +292,9 @@
                                         "description": "When true, select only model-group members with ZDR evidence.",
                                     },
                                     "response_format": {"type": "object"},
+                                    "routing": {
+                                        "$ref": "#/components/schemas/CandidateRoutingControls"
+                                    },
                                     "include_orchestration_trace": {
                                         "type": "boolean",
                                         "description": "Requires the same caller to have the trace purpose",
@@ -448,6 +529,9 @@
                                     "model": {"type": "string"},
                                     "input": {"oneOf": [{"type": "string"}, {"type": "array"}]},
                                     "stream": {"type": "boolean"},
+                                    "routing": {
+                                        "$ref": "#/components/schemas/CandidateRoutingControls"
+                                    },
                                     "zdr_only": {
                                         "type": "boolean",
                                         "description": "When true, select only model-group members with ZDR evidence.",

contextual_orchestrator/cost_router.py

@@ -18,6 +18,7 @@
 
 from __future__ import annotations
 
+import contextlib
 import hashlib
 import re
 from contextvars import ContextVar
@@ -543,6 +544,7 @@ def complete(
         provider_request: Optional[Dict[str, Any]] = None,
         provider_endpoint: str = "chat/completions",
         zdr_only: bool = False,
+        candidate_scope_open: bool = False,
     ) -> Dict[str, Any]:
         """Route a request (sync or batch) and record its usage + cost.
 
@@ -554,17 +556,47 @@ def complete(
         Each trace step backed by valid provider token counts is ``measured``.
         A missing count is recorded with an ``unavailable`` status and numeric
         storage sentinels; API usage and cost remain null.
+
+        ``candidate_scope_open=True`` tells the ``messages``-based sync path
+        (``provider_request is None``) that the caller already has a
+        ``candidate_routing_policy`` scope open -- e.g. because it ran
+        :meth:`TaskOrchestrator.would_route`'s triage call under that scope
+        before deciding to conduct rather than route -- so this call must not
+        open a second, independent scope that would discard the triage
+        attempt from ``attempted_candidate_ids`` (#983). Ignored on the
+        ``provider_request`` path, which always scopes itself around its own
+        ``proxy_completion`` call.
         """
         if not isinstance(cache_bypass, bool):
             raise TypeError("cache_bypass must be a boolean")
         if type(zdr_only) is not bool:
             raise TypeError("zdr_only must be a boolean")
+        routing_controls = hints if isinstance(hints, dict) else {}
+        # TaskOrchestrator._has_active_candidate_controls is the single
+        # source of truth: it detects an active control by key *presence*,
+        # not truthiness, so an explicitly malformed value (candidate_id=
+        # None, exclude_candidate_ids=None or a non-list/tuple) still
+        # forces the sync path below, giving candidate_routing_policy's
+        # real validation a chance to reject it instead of silently falling
+        # through the batch branch's early return and dropping the
+        # malformed control entirely. An explicit empty exclude_candidate_ids
+        # list/tuple is the one genuine no-op (#983 Devin/CodeRabbit
+        # finding: direct Python API callers can lose or bypass routing
+        # validation). The same predicate also gates whether this method's
+        # own provider-response evidence handling below may trust a
+        # `_candidate_routing` field (#983 Devin finding: "Provider fields
+        # forge routing evidence").
+        has_candidate_controls = self.orchestrator._has_active_candidate_controls(
+            routing_controls
+        )
         routing_hints = hints if isinstance(hints, RoutingHints) else RoutingHints.from_mapping(hints)
         try:
             prompt_tokens = self.token_counter.count_messages(messages, model_name)
         except TokenCountUnavailable:
             prompt_tokens = None
         decision = self.policy.decide(routing_hints, prompt_tokens)
+        if has_candidate_controls and decision.channel == "batch":
+            decision = replace(decision, channel="sync", reason="candidate controls require sync routing")
 
         if decision.channel == "batch" and provider_request is None:
             request = BatchRequest(
@@ -600,7 +632,17 @@ def complete(
             }
             race_token = self._race_usage_context.set(race_context)
             try:
-                with self.orchestrator.request_policy(zdr_only):
+                with self.orchestrator.request_policy(zdr_only), self.orchestrator.candidate_routing_policy(
+                    routing_controls,
+                    model_name=model_name,
+                    required_roles=("thinker", "worker", "verifier", "synthesizer")
+                    if self.orchestrator.proxy_completion_requires_conduct(
+                        provider_request,
+                        endpoint=provider_endpoint,
+                        single_agent=False,
+                    )
+                    else ("worker",),
+                ):
                     provider_response = self.orchestrator.proxy_completion(
                         provider_request,
                         endpoint=provider_endpoint,
@@ -621,6 +663,17 @@ def complete(
             ):
                 raise RuntimeError("provider completion omitted orchestration lineage")
             result = dict(self.orchestrator.get_workflow_run(lineage["workflow_run_id"]))
+            routing_evidence = provider_response.pop("_candidate_routing", None)
+            # Only republish gateway-computed evidence: the raw provider
+            # response is untrusted (#983 Devin finding: "Provider fields
+            # forge routing evidence"). Without an active candidate control
+            # on this request, proxy_completion() never sets
+            # `_candidate_routing` itself, so a `_candidate_routing` field
+            # observed here with no active control can only have arrived
+            # already-present on the provider's own response body -- never
+            # trust it as gateway evidence in that case.
+            if has_candidate_controls and routing_evidence is not None:
+                lineage["routing"] = routing_evidence
             race_records = list(race_context["records"])
             records = list(race_records)
             # The caller's request prompt is attributed at most once per
@@ -734,8 +787,28 @@ def complete(
         }
         race_token = self._race_usage_context.set(race_context)
         try:
-            with self.orchestrator.request_policy(zdr_only):
+            # candidate_scope_open=True means the caller already has a
+            # candidate_routing_policy scope open (see the docstring above);
+            # entering a second, independent one here would reset the
+            # attempted-candidate ContextVar and discard whatever the caller
+            # already recorded under it, so reuse a no-op context instead.
+            candidate_scope = (
+                contextlib.nullcontext()
+                if candidate_scope_open
+                else self.orchestrator.candidate_routing_policy(
+                    routing_controls,
+                    model_name=model_name,
+                    required_roles=self.orchestrator.candidate_pin_required_roles(
+                        mode, model_name
+                    ),
+                )
+            )
+            with self.orchestrator.request_policy(zdr_only), candidate_scope:
                 result = self.orchestrator.run(messages, **run_kwargs)
+                routing_evidence = self.orchestrator._candidate_routing_evidence(result)
+                if routing_evidence is not None:
+                    result = dict(result)
+                    result["candidate_routing"] = routing_evidence
             if isinstance(result.get("workflow_run_id"), str):
                 race_context["workflow_run_id"] = result["workflow_run_id"]
             race_context["workflow_ready"] = True