diff --git a/.github/workflows/resolve-pr-909-v2.yml b/.github/workflows/resolve-pr-909-v2.yml new file mode 100644 index 000000000..2b67bf480 --- /dev/null +++ b/.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 diff --git a/CHANGELOG.d/batch-routing-owner-model-error.md b/CHANGELOG.d/batch-routing-owner-model-error.md new file mode 100644 index 000000000..4a8636567 --- /dev/null +++ b/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. diff --git a/CHANGELOG.md b/CHANGELOG.md index 3a04c1d4f..be9214874 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/contextual_orchestrator/api_contract.py b/contextual_orchestrator/api_contract.py index 55ef57935..0b3d76f79 100644 --- a/contextual_orchestrator/api_contract.py +++ b/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": { diff --git a/contextual_orchestrator/batch_routing.py b/contextual_orchestrator/batch_routing.py index ea12aca58..82ce36600 100644 --- a/contextual_orchestrator/batch_routing.py +++ b/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 diff --git a/contextual_orchestrator/cost_router.py b/contextual_orchestrator/cost_router.py index 723dabc38..438a26fde 100644 --- a/contextual_orchestrator/cost_router.py +++ b/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 diff --git a/contextual_orchestrator/server.py b/contextual_orchestrator/server.py index 0f4bdd2de..a4db73f41 100644 --- a/contextual_orchestrator/server.py +++ b/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", diff --git a/docs/architecture.md b/docs/architecture.md index 1c710851a..df27bca82 100644 --- a/docs/architecture.md +++ b/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. diff --git a/docs/doctoring/responses-stream-usage.md b/docs/doctoring/responses-stream-usage.md index 5e5d7a7ae..46aed738f 100644 --- a/docs/doctoring/responses-stream-usage.md +++ b/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 diff --git a/docs/planning/adrs/0019-workflow-run-object-authorization.md b/docs/planning/adrs/0019-workflow-run-object-authorization.md index ac5723a56..080611b8d 100644 --- a/docs/planning/adrs/0019-workflow-run-object-authorization.md +++ b/docs/planning/adrs/0019-workflow-run-object-authorization.md @@ -20,9 +20,40 @@ confirmed across owners. The library API continues to support local single-process callers that omit an owner key. HTTP callers do not omit it. Deployments with an external bearer -verifier may use stable per-principal credentials; token rotation may revoke -access to older evidence and must be handled by the deployment's identity -policy. +verifier may inject `principal_resolver(token)` to return a stable, +tenant-scoped principal key; the gateway hashes that key and never stores the +bearer. Adapters that retain the legacy bool-only verifier contract use the +bearer digest fallback, so token rotation may revoke access to older evidence +and must be handled by the deployment's identity policy. + +Batch routing jobs follow the same boundary. The coordinator stores the +principal digest on each HTTP-created `BatchJob`; status polling and result +retrieval require an equal digest and report an owner mismatch as not found +before calling the backend. Result retrieval additionally requires both +inference and trace-purpose authorization. Legacy in-process jobs without an +owner remain available only through the library API, not through an +owner-bound HTTP request. + +```mermaid +sequenceDiagram + participant C as Client + participant H as HTTP boundary + participant S as Security principal + participant R as Batch registry/backend + C->>H: submit batch + H->>S: verify inference and derive owner digest + H->>R: persist BatchJob(owner digest) + C->>H: poll or retrieve job id + H->>S: verify required scope and derive owner digest + H->>R: lookup job by id and equal owner digest + alt owner mismatch or legacy ownerless HTTP job + R-->>H: not found + H-->>C: generic not-found response + else owner matches + R-->>H: status or results + H-->>C: permitted response + end +``` ## Consequences @@ -32,16 +63,24 @@ policy. never rendered in public payloads. - Shared static credentials represent one deployment principal; multi-principal deployments must issue distinct verified bearers. External bearer deployments - currently use the bearer credential digest as that principal key, so token - rotation can revoke access to older evidence. + should return a stable tenant/subject key through `principal_resolver`; the + legacy bool-only fallback uses the bearer credential digest, so token rotation + can revoke access to older evidence. - Old records without an owner key are not visible through the owner-bound HTTP resource routes, which is fail-closed during migration. +- A guessed batch routing job identifier cannot retrieve another principal's + provider result or trigger its cost recording. ## Acceptance evidence Owner mismatch, list filtering, evaluation ownership, digest stability, and response redaction are covered by `tests/test_workflow_run_object_authorization.py`. +Batch status/result ownership and the public trace-plus-inference security +contract are covered by `tests/test_cost_review_server.py`, +`tests/test_cost_router_boundaries.py`, and `tests/test_api_contract.py`; +stable external-principal resolution is covered by +`tests/test_security_hardening.py`. ## References diff --git a/docs/planning/adrs/0038-streamed-responses-usage-boundary.md b/docs/planning/adrs/0040-streamed-responses-usage-boundary.md similarity index 99% rename from docs/planning/adrs/0038-streamed-responses-usage-boundary.md rename to docs/planning/adrs/0040-streamed-responses-usage-boundary.md index c8572eab2..b98621e96 100644 --- a/docs/planning/adrs/0038-streamed-responses-usage-boundary.md +++ b/docs/planning/adrs/0040-streamed-responses-usage-boundary.md @@ -1,5 +1,5 @@ --- -id: "0038" +id: "0040" title: "Record streamed Responses usage at the workflow boundary" status: accepted proposed_date: "2026-08-29" diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index ff00b0643..0908e3080 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -1,5 +1,25 @@ # Product and Technical Gap Baseline +## 2026-08-29 batch-routing object-authorization slice + +Protected `main` remains +`b21645116b352967e50fc497b87eb745b9cc8c61`. The accepted workflow-object +authorization decision now has a bounded implementation branch for its listed +batch-job gap: HTTP-created batch routing jobs carry a non-secret +authenticated-principal digest, and both status and result retrieval require +the same digest. An external verifier can provide a stable tenant/subject key +through the optional principal resolver; bool-only adapters retain the +documented bearer-digest fallback. A mismatch is returned as the existing generic +`batch_job_not_found` response before the backend is called; results also keep +the separate trace-purpose gate. Local exact-branch evidence is `61 passed` +across the cost-router, HTTP, and OpenAPI contract suites. This is branch +evidence only until the implementation reaches protected `main` through the +normal review, Checks, and approval gates. + +The remaining issue #117 gaps are unchanged: tenant/resource/purpose/lifetime +claims from an external identity adapter, explicit legacy single-token +production migration, and ownership for other evidence surfaces still need +their own decisions and acceptance evidence. ## 2026-08-29 streamed Responses usage boundary Protected `main` remains diff --git a/tests/test_api_contract.py b/tests/test_api_contract.py index 397761da6..f7d4762a0 100644 --- a/tests/test_api_contract.py +++ b/tests/test_api_contract.py @@ -66,6 +66,9 @@ def test_openapi_documents_compatibility_front_door() -> None: assert OPENAPI_SPEC["components"]["securitySchemes"]["trace_bearer_auth"]["scheme"] == ( "bearer" ) + assert OPENAPI_SPEC["paths"]["/api/v1/batch_routing_jobs/{batch_routing_job_id}/results"]["post"][ + "security" + ] == [{"inference_bearer_auth": [], "trace_bearer_auth": []}] def test_openapi_documents_orchestrator_owned_embedding_model_selection() -> None: @@ -107,3 +110,17 @@ def test_openapi_capability_requests_have_endpoint_specific_contracts() -> None: test_openapi_uses_resource_oriented_operation_ids() test_openapi_documents_orchestrator_owned_embedding_model_selection() print("ok") + + + +def test_batch_job_openapi_documents_principal_hiding_404s() -> None: + """Missing and foreign batch jobs share the documented not-found surface.""" + status = OPENAPI_SPEC["paths"][ + "/api/v1/batch_routing_jobs/{batch_routing_job_id}" + ]["get"]["responses"] + results = OPENAPI_SPEC["paths"][ + "/api/v1/batch_routing_jobs/{batch_routing_job_id}/results" + ]["post"]["responses"] + assert "404" in status + assert "not owned" in status["404"]["description"] + assert results["404"] == status["404"] diff --git a/tests/test_cost_review_server.py b/tests/test_cost_review_server.py index 0770990b7..9efbb66af 100644 --- a/tests/test_cost_review_server.py +++ b/tests/test_cost_review_server.py @@ -359,6 +359,51 @@ def test_batch_routing_via_chat_completion_and_results_retrieval() -> None: server.shutdown() +def test_batch_routing_jobs_are_principal_bound_for_poll_and_results() -> None: + """A second bearer cannot confirm or retrieve another principal's job.""" + token_a = "batch-owner-a" + token_b = "batch-owner-b" + security = SecurityConfig( + bearer_verifier=lambda presented, scope: presented in {token_a, token_b} + and scope in {"admin", "inference", "trace"} + ) + server, port, _ = _serve(security) + base = f"http://127.0.0.1:{port}" + try: + status, submitted = _request( + "POST", + f"{base}/api/v1/batch_routing_jobs", + token_a, + {"requests": [{"messages": [{"role": "user", "content": "owned"}]}]}, + ) + assert status == 201, submitted + job_id = submitted["job_id"] + + status, body = _request( + "GET", f"{base}/api/v1/batch_routing_jobs/{job_id}", token_b + ) + assert status == 404 + assert body["error"]["code"] == "batch_job_not_found" + status, body = _request( + "POST", f"{base}/api/v1/batch_routing_jobs/{job_id}/results", token_b + ) + assert status == 404 + assert body["error"]["code"] == "batch_job_not_found" + + status, polled = _request( + "GET", f"{base}/api/v1/batch_routing_jobs/{job_id}", token_a + ) + assert status == 200 + assert polled["is_complete"] is True + status, retrieved = _request( + "POST", f"{base}/api/v1/batch_routing_jobs/{job_id}/results", token_a + ) + assert status == 200 + assert retrieved["result_count"] == 1 + finally: + server.shutdown() + + def test_batch_routing_jobs_endpoint_submits_multiple_requests() -> None: server, port, token = _serve() base = f"http://127.0.0.1:{port}" @@ -494,3 +539,26 @@ def test_dimension_catalog_endpoint_lists_all_dimensions() -> None: _fn() print(f"ok {_name}") print("ok") + + + +def test_batch_routing_rejects_unknown_zdr_model_as_client_error() -> None: + """An unknown explicit ZDR model is a non-retryable client error.""" + server, port, token = _serve() + try: + status, body = _request( + "POST", + f"http://127.0.0.1:{port}/api/v1/batch_routing_jobs", + token, + { + "model": "not-configured", + "zdr_only": True, + "requests": [ + {"messages": [{"role": "user", "content": "route securely"}]} + ], + }, + ) + finally: + server.shutdown() + assert status == 400 + assert body["error"]["code"] == "invalid_model" diff --git a/tests/test_cost_router_boundaries.py b/tests/test_cost_router_boundaries.py index b30938744..d09d75ab5 100644 --- a/tests/test_cost_router_boundaries.py +++ b/tests/test_cost_router_boundaries.py @@ -146,6 +146,26 @@ def test_retrieve_batch_requires_known_job_id() -> None: coordinator.retrieve_batch("nope_missing_job") +def test_batch_poll_and_retrieve_require_the_bound_owner() -> None: + """An opaque job identifier cannot cross the authenticated owner boundary.""" + coordinator = _coordinator() + submitted = coordinator.complete( + [{"role": "user", "content": "owned"}], + hints={"channel": "batch"}, + owner_id="principal-a", + ) + job_id = submitted["job_id"] + job = coordinator._batch_jobs[job_id] + + assert job.owner_id == "principal-a" + assert coordinator.poll_batch(job_id, owner_id="principal-a")["is_complete"] is True + with pytest.raises(KeyError, match="batch job"): + coordinator.poll_batch(job_id, owner_id="principal-b") + with pytest.raises(KeyError, match="batch job"): + coordinator.retrieve_batch(job_id, owner_id="principal-b") + assert coordinator.retrieve_batch(job_id, owner_id="principal-a")["result_count"] == 1 + + # --- embedding input splitting -------------------------------------------------------- @@ -423,3 +443,33 @@ def test_complete_embeddings_batch_round_trips_locally() -> None: assert document["status"] == "completed" assert document["embeddings"][0]["index"] == 0 assert document["cost_micro_usd"] >= 0 + + + +def test_batch_model_identity_error_does_not_capture_backend_value_errors() -> None: + """Only model resolution receives the client-facing invalid-model category.""" + from contextual_orchestrator.batch_routing import BatchRequest + from contextual_orchestrator.cost_router import InvalidBatchModelError + + class RejectingBackend: + name = "rejecting-backend" + + def submit(self, requests, metadata=None): # type: ignore[no-untyped-def] + del requests, metadata + raise ValueError("backend payload validation failed") + + coordinator = _coordinator(batch_backend=RejectingBackend()) + with pytest.raises(ValueError, match="backend payload validation failed") as backend_error: + coordinator.submit_batch([ + BatchRequest(messages=[{"role": "user", "content": "valid"}], model="mock-a") + ]) + assert type(backend_error.value) is ValueError + + with pytest.raises(InvalidBatchModelError, match="not configured"): + coordinator.submit_batch([ + BatchRequest( + messages=[{"role": "user", "content": "private"}], + model="not-configured", + zdr_only=True, + ) + ]) diff --git a/tests/test_orchestrated_responses_stream.py b/tests/test_orchestrated_responses_stream.py index de8561985..dd8460f75 100644 --- a/tests/test_orchestrated_responses_stream.py +++ b/tests/test_orchestrated_responses_stream.py @@ -505,3 +505,39 @@ def test_stream_failure_emits_terminal_responses_event() -> None: assert event["event_detail"]["status_code"] == 500 assert event["event_detail"]["transport_status_code"] == 200 assert event["event_detail"]["response_status"] == "failed" + + + +def test_stream_usage_failure_remains_inside_the_started_sse_protocol(monkeypatch) -> None: + """A post-header ledger failure emits Responses failure framing, never JSON HTTP.""" + token = "responses_stream_usage_failure_token" + orchestrator = TaskOrchestrator([ + ModelAgent("workflow_agent", "mock-model", base_url="mock://provider") + ]) + coordinator = CostRoutingCoordinator(orchestrator) + + def fail_usage(**_kwargs): + raise RuntimeError("ledger unavailable") + + monkeypatch.setattr(coordinator, "record_stream_usage", fail_usage) + server = build_server( + orchestrator, + port=0, + security=SecurityConfig(auth_token=token), + coordinator=coordinator, + ) + threading.Thread(target=server.serve_forever, daemon=True).start() + try: + stream = _post(server, token, "orchestrator/auto") + finally: + server.shutdown() + + events = [ + json.loads(line[6:]) + for line in stream.splitlines() + if line.startswith("data: {") + ] + assert events[-1]["type"] == "response.failed" + assert events[-1]["response"]["error"]["code"] == "usage_recording_failed" + assert all(event["type"] != "response.completed" for event in events) + assert stream.rstrip().endswith("data: [DONE]") diff --git a/tests/test_security_hardening.py b/tests/test_security_hardening.py index 2852a8e29..55987348b 100644 --- a/tests/test_security_hardening.py +++ b/tests/test_security_hardening.py @@ -40,6 +40,20 @@ def verify(token: str, scope: str) -> bool: assert security.readiness_profile()["auth_mode"] == "external_bearer_verifier" +def test_external_principal_resolver_survives_bearer_rotation() -> None: + """An adapter-provided tenant/subject key keeps resource ownership stable.""" + valid_tokens = {"old-token", "rotated-token"} + security = SecurityConfig( + bearer_verifier=lambda token, scope: token in valid_tokens and scope == "inference", + principal_resolver=lambda token: "tenant-a/user-a" if token in valid_tokens else None, + ) + old_headers = {"authorization": "Bearer old-token"} + rotated_headers = {"authorization": "Bearer rotated-token"} + security.authorize(old_headers, "inference", "127.0.0.1") + security.authorize(rotated_headers, "inference", "127.0.0.1") + assert security.principal_id(old_headers) == security.principal_id(rotated_headers) + + def post_json(url: str, payload: dict[str, object], token: str | None = None) -> tuple[int, dict[str, object]]: headers = {"content-type": "application/json", "connection": "close"} if token: