fix(security): bind batch routing jobs to authenticated owners

.github/workflows/resolve-pr-909-v2.yml

@@ -0,0 +1,17 @@
+name: Resolve PR 909 with validated semantic merge
+
+on:
+  push:
+    branches: [fix/batch-routing-owner-20260829]
+
+permissions:
+  contents: write
+  pull-requests: write
+
+jobs:
+  resolve:
+    uses: ContextualWisdomLab/contextual-orchestrator/.github/workflows/reusable-pr-conflict-resolver.yml@automation/one-shot-pr-conflict-resolver
+    with:
+      pr_number: 909
+      branch: fix/batch-routing-owner-20260829
+      caller_file: .github/workflows/resolve-pr-909-v2.yml

CHANGELOG.d/batch-routing-owner-model-error.md

@@ -0,0 +1 @@
+Bound HTTP-created batch routing jobs to the authenticated principal and preserved cross-owner not-found behavior. An unknown explicit ZDR model is now the non-retryable `400 invalid_model` client error, while a configured model that is ineligible for the requested ZDR policy retains the retryable batch-selection failure contract.

CHANGELOG.md

@@ -77,7 +77,7 @@ and this project uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html)
   SSE usage, record per-step `stream` cost-ledger rows, and expose cost status
   plus usage-record identities. Missing provider usage is explicitly
   unavailable; the gateway does not estimate billing tokens from the final
-  answer, and nested gateway upstreams remain compatible (ADR 0038).
+  answer, and nested gateway upstreams remain compatible (ADR 0040).
 
 ### Fixed
 
@@ -129,6 +129,9 @@ and this project uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html)
   `fast-mlsirm` and its `numpy` dependency are installed in CI and locally.
 - Validate orchestration-trace requests before every chat execution branch and
   require trace-purpose authorization before access-report lookup.
+- Bind HTTP-created batch routing jobs to the authenticated principal and
+  require the same owner for status polling and trace-bearing result retrieval;
+  owner mismatches fail closed as not found.
 - Mixed structured workflows now retain a cost-ledger row for calls whose
   provider omitted usage, using the existing token-counting fallback while
   preserving reported counts for the other calls in the same workflow.

contextual_orchestrator/api_contract.py

@@ -859,7 +859,7 @@
         "/api/v1/batch_routing_jobs": {
             "post": {
                 "operationId": "create_batch_routing_job",
-                "summary": "Submit a batch of latency-tolerant requests to the batch backend (pg-llm-batch)",
+                "summary": "Submit a principal-owned batch of latency-tolerant requests to the batch backend (pg-llm-batch)",
                 "security": [{"inference_bearer_auth": []}],
                 "requestBody": {
                     "required": True,
@@ -887,23 +887,33 @@
         "/api/v1/batch_routing_jobs/{batch_routing_job_id}": {
             "get": {
                 "operationId": "get_batch_routing_job",
-                "summary": "Poll a submitted batch routing job",
+                "summary": "Poll a submitted batch routing job owned by the authenticated principal",
                 "security": [{"admin_bearer_auth": []}],
                 "parameters": [
                     {"name": "batch_routing_job_id", "in": "path", "required": True, "schema": {"type": "string"}}
                 ],
-                "responses": {"200": {"description": "Batch routing job status"}},
+                "responses": {
+                    "200": {"description": "Batch routing job status"},
+                    "404": {
+                        "description": "Batch job is missing or is not owned by the authenticated principal"
+                    },
+                },
             }
         },
         "/api/v1/batch_routing_jobs/{batch_routing_job_id}/results": {
             "post": {
                 "operationId": "create_batch_routing_job_results",
-                "summary": "Retrieve batch results and record their usage + cost",
-                "security": [{"inference_bearer_auth": []}],
+                "summary": "Retrieve principal-owned batch results and record their usage + cost",
+                "security": [{"inference_bearer_auth": [], "trace_bearer_auth": []}],
                 "parameters": [
                     {"name": "batch_routing_job_id", "in": "path", "required": True, "schema": {"type": "string"}}
                 ],
-                "responses": {"200": {"description": "Batch results with recorded usage"}},
+                "responses": {
+                    "200": {"description": "Batch results with recorded usage"},
+                    "404": {
+                        "description": "Batch job is missing or is not owned by the authenticated principal"
+                    },
+                },
             }
         },
         "/v1/batch/embeddings": {

contextual_orchestrator/batch_routing.py

@@ -187,6 +187,9 @@ class BatchJob:
     status: str = "submitted"
     submitted_at: int = field(default_factory=lambda: int(time.time()))
     request_count: int = 0
+    # HTTP callers bind this opaque digest to the authenticated principal;
+    # library-only jobs may remain unowned for standalone use.
+    owner_id: Optional[str] = None
 
 
 @dataclass

contextual_orchestrator/cost_router.py

@@ -57,6 +57,10 @@ class BatchModelSelectionError(RuntimeError):
     """Raised when a batch request has no eligible model-group member."""
 
 
+class InvalidBatchModelError(ValueError):
+    """Raised only for an unknown client-supplied batch model identity."""
+
+
 class CostRoutingCoordinator:
     """Wire routing + cost accounting around a ``TaskOrchestrator``."""
 
@@ -269,7 +273,9 @@ def complete(
                 mode=mode,
                 zdr_only=zdr_only,
             )
-            job = self.submit_batch([request], metadata={"routing_reason": decision.reason})
+            job = self.submit_batch(
+                [request], metadata={"routing_reason": decision.reason}, owner_id=owner_id
+            )
             return {
                 "channel": "batch",
                 "routing_reason": decision.reason,
@@ -652,15 +658,19 @@ def submit_batch(
         self,
         requests: List[BatchRequest],
         metadata: Optional[Dict[str, Any]] = None,
+        owner_id: Optional[str] = None,
     ) -> BatchJob:
-        """Submit a batch of requests to the configured batch backend."""
+        """Submit a batch, resolve its targets, and bind its authenticated owner."""
         try:
             prepared_requests = [self._resolve_batch_request(request) for request in requests]
-        except (RuntimeError, ValueError) as exc:
+        except ValueError as exc:
+            raise InvalidBatchModelError(str(exc)) from exc
+        except RuntimeError as exc:
             raise BatchModelSelectionError(
                 "no eligible model-group member is available for this batch request"
             ) from exc
         job = self.batch_backend.submit(prepared_requests, metadata=metadata)
+        job.owner_id = owner_id
         self._batch_jobs[job.job_id] = job
         return job
 
@@ -669,7 +679,18 @@ def _resolve_batch_request(self, request: BatchRequest) -> BatchRequest:
         if not request.zdr_only:
             return request
         with self.orchestrator.request_policy(request.zdr_only):
-            agent = self.orchestrator._requested_agent(request.model)
+            try:
+                agent = self.orchestrator._requested_agent(request.model)
+            except ValueError as exc:
+                configured_exact = any(
+                    candidate.model == request.model
+                    for candidate in self.orchestrator.candidates
+                )
+                if configured_exact:
+                    raise RuntimeError(
+                        "requested model is configured but not eligible for ZDR batch routing"
+                    ) from exc
+                raise
             if agent is None:
                 text = self.orchestrator._latest_user_text(request.messages)
                 agent = self.orchestrator._select_agent(
@@ -680,14 +701,14 @@ def _resolve_batch_request(self, request: BatchRequest) -> BatchRequest:
                 )
         return replace(request, model=agent.model)
 
-    def poll_batch(self, job_id: str) -> Dict[str, Any]:
-        """Poll a previously submitted batch job by id."""
-        job = self._require_job(job_id)
+    def poll_batch(self, job_id: str, *, owner_id: Optional[str] = None) -> Dict[str, Any]:
+        """Poll a previously submitted batch job owned by ``owner_id``."""
+        job = self._require_job(job_id, owner_id=owner_id)
         return self.batch_backend.poll(job)
 
-    def retrieve_batch(self, job_id: str) -> Dict[str, Any]:
-        """Retrieve batch results and record usage + cost for each completion."""
-        job = self._require_job(job_id)
+    def retrieve_batch(self, job_id: str, *, owner_id: Optional[str] = None) -> Dict[str, Any]:
+        """Retrieve results for a batch owned by ``owner_id`` and record usage."""
+        job = self._require_job(job_id, owner_id=owner_id)
         items: List[BatchResultItem] = self.batch_backend.retrieve(job)
         recorded: List[Dict[str, Any]] = []
         for item in items:
@@ -729,9 +750,9 @@ def _resolve_batch_provider_model(self, item: BatchResultItem) -> tuple[str, str
             provider = "unknown"
         return provider, item.model
 
-    def _require_job(self, job_id: str) -> BatchJob:
+    def _require_job(self, job_id: str, *, owner_id: Optional[str] = None) -> BatchJob:
         job = self._batch_jobs.get(job_id)
-        if job is None:
+        if job is None or job.owner_id != owner_id:
             raise KeyError(f"batch job {job_id!r} not found")
         return job
 

contextual_orchestrator/server.py

@@ -26,7 +26,11 @@
 from .admin import ADMIN_HTML, ADMIN_TRANSLATIONS
 from .api_contract import OPENAPI_SPEC
 from .cost_ledger import ATTRIBUTION_DIMENSIONS, dimension_catalog
-from .cost_router import BatchModelSelectionError, CostRoutingCoordinator
+from .cost_router import (
+    BatchModelSelectionError,
+    CostRoutingCoordinator,
+    InvalidBatchModelError,
+)
 from .batch_routing import BatchRequest
 from .orchestrator import (
     BudgetExceededError,
@@ -346,6 +350,9 @@ class SecurityConfig:
     # relying-party adapter). The core deliberately does not decode JWTs with
     # an unsafe hand-rolled parser or own Keycloak admin credentials.
     bearer_verifier: Callable[[str, str], bool] | None = None
+    # Optional companion seam for external verifiers that can expose a stable,
+    # tenant-scoped principal key without exposing the bearer itself.
+    principal_resolver: Callable[[str], str | None] | None = None
     _rate_buckets: dict[str, tuple[int, float]] = field(default_factory=dict, init=False, repr=False)
     _rate_lock: threading.Lock = field(default_factory=threading.Lock, init=False, repr=False)
     _run_semaphore: threading.BoundedSemaphore = field(init=False, repr=False)
@@ -471,13 +478,27 @@ def principal_id(self, headers: Any) -> str:
                 if principal:
                     return principal
             raise RequestError(401, "unauthorized", "authenticated principal is required")
+        return self._principal_digest(token)
+
+    def _principal_digest(self, token: str) -> str:
+        """Hash a stable deployment principal without retaining bearer material."""
         if self.bearer_verifier is None:
             if self.admin_token and self.inference_token:
                 principal_material = f"split:{self.admin_token}\x00{self.inference_token}"
             else:
                 principal_material = f"single:{self.auth_token}"
-        else:
+        elif self.principal_resolver is None:
+            # Back-compatible fallback for adapters that only return bool;
+            # token rotation can intentionally revoke old resource access.
             principal_material = f"bearer:{token}"
+        else:
+            try:
+                resolved = self.principal_resolver(token)
+            except Exception as exc:  # noqa: BLE001 - identity adapter failure denies access
+                raise RequestError(401, "unauthorized", "authenticated principal is unavailable") from exc
+            if not isinstance(resolved, str) or not resolved.strip():
+                raise RequestError(401, "unauthorized", "authenticated principal is unavailable")
+            principal_material = f"principal:{resolved}"
         return hashlib.sha256(principal_material.encode("utf-8")).hexdigest()
 
     @staticmethod
@@ -514,14 +535,7 @@ def establish_admin_session(self, presented_token: str) -> str:
             raise RequestError(401, "unauthorized", "bearer token is invalid for this scope")
         session_id = secrets.token_urlsafe(32)
         expires_at = time.monotonic() + float(self.admin_session_ttl_seconds)
-        if self.bearer_verifier is None:
-            if self.admin_token and self.inference_token:
-                principal_material = f"split:{self.admin_token}\x00{self.inference_token}"
-            else:
-                principal_material = f"single:{self.auth_token}"
-        else:
-            principal_material = f"bearer:{presented_token}"
-        principal = hashlib.sha256(principal_material.encode("utf-8")).hexdigest()
+        principal = self._principal_digest(presented_token)
         with self._session_lock:
             self._purge_expired_admin_sessions_locked(time.monotonic())
             overflow = len(self._admin_sessions) - self.max_admin_sessions + 1
@@ -5551,7 +5565,11 @@ def do_GET(self) -> None:  # noqa: N802
                 if path.startswith("/api/v1/batch_routing_jobs/"):
                     job_id = path.rsplit("/", 1)[-1]
                     try:
-                        self._send(coordinator.poll_batch(job_id))
+                        self._send(
+                            coordinator.poll_batch(
+                                job_id, owner_id=security.principal_id(self.headers)
+                            )
+                        )
                     except KeyError:
                         self._send_error(404, "batch_job_not_found", f"batch job {job_id} not found")
                     return
@@ -6356,6 +6374,7 @@ def register_video_job(agent: ModelAgent, provider_result: dict[str, Any]) -> di
                             workflow_run_id=f"run_{uuid.uuid4().hex}",
                             cache_bypass=cache_bypass,
                             cache_partition=cache_partition,
+                            owner_id=security.principal_id(self.headers),
                             zdr_only=zdr_only,
                         ))
                     # Batch-channel Completions return a job handle (202), not a
@@ -6722,6 +6741,7 @@ def register_video_job(agent: ModelAgent, provider_result: dict[str, Any]) -> di
                             workflow_run_id=f"run_{uuid.uuid4().hex}",
                             cache_bypass=cache_bypass,
                             cache_partition=cache_partition,
+                            owner_id=security.principal_id(self.headers),
                             zdr_only=zdr_only,
                         ))
                     # Latency-tolerant requests get dispatched to the batch backend.
@@ -6960,7 +6980,16 @@ def register_video_job(agent: ModelAgent, provider_result: dict[str, Any]) -> di
                         zdr_only=zdr_only,
                     )
                     metadata = {"actor_scope": "inference"}
-                    job = self._run(lambda: coordinator.submit_batch(batch_requests, metadata=metadata))
+                    try:
+                        job = self._run(
+                            lambda: coordinator.submit_batch(
+                                batch_requests,
+                                metadata=metadata,
+                                owner_id=security.principal_id(self.headers),
+                            )
+                        )
+                    except InvalidBatchModelError as exc:
+                        raise RequestError(400, "invalid_model", str(exc)) from exc
                     orchestrator.record_analytics_event(
                         "batch_routing_job_created",
                         {
@@ -6983,7 +7012,11 @@ def register_video_job(agent: ModelAgent, provider_result: dict[str, Any]) -> di
                     job_id = path[len("/api/v1/batch_routing_jobs/"):-len("/results")]
                     self._authorize_trace_access()
                     try:
-                        retrieved = self._run(lambda: coordinator.retrieve_batch(job_id))
+                        retrieved = self._run(
+                            lambda: coordinator.retrieve_batch(
+                                job_id, owner_id=security.principal_id(self.headers)
+                            )
+                        )
                     except KeyError:
                         self._send_error(404, "batch_job_not_found", f"batch job {job_id} not found")
                         return
@@ -7897,14 +7930,25 @@ def progress(role: str, status: str) -> None:
                     self._write_sse("data: [DONE]\n\n")
                     return False
                 if coordinator is not None:
-                    result = {
-                        **result,
-                        **coordinator.record_stream_usage(
+                    try:
+                        stream_usage = coordinator.record_stream_usage(
                             result=result,
                             attribution=attribution,
                             model_name=model_name,
-                        ),
-                    }
+                        )
+                    except Exception:  # noqa: BLE001 - headers sent; remain inside SSE
+                        failed = {
+                            **created_response,
+                            "status": "failed",
+                            "error": {
+                                "code": "usage_recording_failed",
+                                "message": "Usage evidence could not be recorded for this response.",
+                            },
+                        }
+                        emit("response.failed", response=failed)
+                        self._write_sse("data: [DONE]\n\n")
+                        return False
+                    result = {**result, **stream_usage}
                 reasoning_done = {
                     **reasoning_item,
                     "status": "completed",

docs/architecture.md

@@ -68,12 +68,16 @@ bounded, authenticated recursion protocol; it is not administratively disabled.
 - `WorkflowStep.access`: Conductor-style visibility control.
 - `ModelClient`: OpenAI-compatible HTTP client, with `mock://` for local checks.
 - `contextual_orchestrator.server`: small `/v1/chat/completions` HTTP server.
+- Batch routing jobs carry a non-secret authenticated-principal digest from
+  submission through status and result retrieval; mismatched owners receive
+  the same not-found response before backend access. Results require the
+  separate trace purpose in addition to inference authorization.
 - Streamed `/v1/responses` workflow runs preserve optional provider usage on
   each trace step and record one `stream` cost-ledger row per completed step.
   Missing provider counts remain `unavailable`; the gateway never derives
   billing tokens from the final answer. The final Responses event uses the
   standard `input_tokens`/`output_tokens`/`total_tokens` usage shape only when
-  all workflow steps are measured. See [ADR 0038](planning/adrs/0038-streamed-responses-usage-boundary.md).
+  all workflow steps are measured. See [ADR 0040](planning/adrs/0040-streamed-responses-usage-boundary.md).
 - `ResponsiveThreadingHTTPServer`: I/O-bound provider waits run in independent
   daemon request threads, the accept queue uses the operating system's native
   `SOMAXCONN`, and fixed-length responses use HTTP/1.1 persistent connections.

docs/doctoring/responses-stream-usage.md

@@ -2,7 +2,7 @@
 title: "Streamed Responses usage and cost evidence"
 status: "implemented on feature branch"
 date: "2026-08-29"
-scope: "ADR 0038"
+scope: "ADR 0040"
 ---
 
 # Streamed Responses usage and cost evidence