From f11b66ebb8a4ee3355e3b0441a7ccf0accba0b2f Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 29 Aug 2026 04:19:52 +0900 Subject: [PATCH 01/28] fix(security): bind batch jobs to authenticated owners --- CHANGELOG.md | 3 ++ contextual_orchestrator/api_contract.py | 8 ++-- contextual_orchestrator/batch_routing.py | 3 ++ contextual_orchestrator/cost_router.py | 24 +++++----- contextual_orchestrator/server.py | 22 +++++++-- docs/architecture.md | 4 ++ .../0019-workflow-run-object-authorization.md | 34 ++++++++++++++ docs/product-technical-gap-baseline.md | 19 ++++++++ tests/test_api_contract.py | 3 ++ tests/test_cost_review_server.py | 45 +++++++++++++++++++ tests/test_cost_router_boundaries.py | 20 +++++++++ 11 files changed, 168 insertions(+), 17 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index bbe29be85..31d03890d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -90,6 +90,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 ee15186e5..e0a673b71 100644 --- a/contextual_orchestrator/api_contract.py +++ b/contextual_orchestrator/api_contract.py @@ -842,7 +842,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, @@ -866,7 +866,7 @@ "/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"}} @@ -877,8 +877,8 @@ "/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"}} ], diff --git a/contextual_orchestrator/batch_routing.py b/contextual_orchestrator/batch_routing.py index 5a40f5dc6..ada8cf982 100644 --- a/contextual_orchestrator/batch_routing.py +++ b/contextual_orchestrator/batch_routing.py @@ -182,6 +182,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 1f81007ef..eb709af20 100644 --- a/contextual_orchestrator/cost_router.py +++ b/contextual_orchestrator/cost_router.py @@ -259,7 +259,9 @@ def complete( attribution=dict(attribution or {}), mode=mode, ) - 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, @@ -578,20 +580,22 @@ 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, optionally binding it to an authenticated owner.""" job = self.batch_backend.submit(requests, metadata=metadata) + job.owner_id = owner_id self._batch_jobs[job.job_id] = job return job - 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: @@ -633,9 +637,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 997d350cf..a1a937e40 100644 --- a/contextual_orchestrator/server.py +++ b/contextual_orchestrator/server.py @@ -5382,7 +5382,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 @@ -6183,6 +6187,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), )) # Batch-channel Completions return a job handle (202), not a # text_completion body — match chat Completions honesty so @@ -6519,6 +6524,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), )) # Latency-tolerant requests get dispatched to the batch backend. if result.get("channel") == "batch": @@ -6743,7 +6749,13 @@ def register_video_job(agent: ModelAgent, provider_result: dict[str, Any]) -> di _reject_unknown_keys(body, ALLOWED_BATCH_KEYS) batch_requests = _validate_batch_requests(body, security.expose_trace_by_default) metadata = {"actor_scope": "inference"} - job = self._run(lambda: coordinator.submit_batch(batch_requests, metadata=metadata)) + job = self._run( + lambda: coordinator.submit_batch( + batch_requests, + metadata=metadata, + owner_id=security.principal_id(self.headers), + ) + ) orchestrator.record_analytics_event( "batch_routing_job_created", { @@ -6766,7 +6778,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 diff --git a/docs/architecture.md b/docs/architecture.md index 64de0e26d..320a18312 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -68,6 +68,10 @@ 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. - `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/planning/adrs/0019-workflow-run-object-authorization.md b/docs/planning/adrs/0019-workflow-run-object-authorization.md index ac5723a56..1c9230b07 100644 --- a/docs/planning/adrs/0019-workflow-run-object-authorization.md +++ b/docs/planning/adrs/0019-workflow-run-object-authorization.md @@ -24,6 +24,35 @@ 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. +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 - A bearer cannot use a guessed workflow identifier to read another owner's @@ -36,12 +65,17 @@ policy. 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`. ## References diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index acd5b6470..ee07b8f19 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -1,5 +1,24 @@ # 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 chat/completions batch jobs carry a non-secret +authenticated-principal digest, and both status and result retrieval require +the same digest. 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 `50 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-27 20:10 KST main trace-rpds regression slice Protected `main` briefly carried a merge-order regression from PR #891 merged diff --git a/tests/test_api_contract.py b/tests/test_api_contract.py index d4ae0cb9a..73ea2a237 100644 --- a/tests/test_api_contract.py +++ b/tests/test_api_contract.py @@ -62,6 +62,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: diff --git a/tests/test_cost_review_server.py b/tests/test_cost_review_server.py index 0085c6f98..06a63b371 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}" diff --git a/tests/test_cost_router_boundaries.py b/tests/test_cost_router_boundaries.py index ec4339050..477cf555c 100644 --- a/tests/test_cost_router_boundaries.py +++ b/tests/test_cost_router_boundaries.py @@ -117,6 +117,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 -------------------------------------------------------- From e849a38f4e34d7e61fa08afd71ee33d9fd200b45 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 29 Aug 2026 04:27:06 +0900 Subject: [PATCH 02/28] docs: align batch ownership evidence --- docs/product-technical-gap-baseline.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index ee07b8f19..7547bf44c 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -5,11 +5,11 @@ Protected `main` remains `b21645116b352967e50fc497b87eb745b9cc8c61`. The accepted workflow-object authorization decision now has a bounded implementation branch for its listed -batch-job gap: HTTP-created chat/completions batch jobs carry a non-secret +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. 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 `50 passed` +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. From d3d2e31df62a5b773ae5077dd538472fa2a6ec18 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 29 Aug 2026 04:38:56 +0900 Subject: [PATCH 03/28] fix(security): support stable external principal keys --- contextual_orchestrator/server.py | 28 +++++++++++++------ .../0019-workflow-run-object-authorization.md | 17 +++++++---- docs/product-technical-gap-baseline.md | 4 ++- tests/test_security_hardening.py | 14 ++++++++++ 4 files changed, 47 insertions(+), 16 deletions(-) diff --git a/contextual_orchestrator/server.py b/contextual_orchestrator/server.py index a1a937e40..478a44e88 100644 --- a/contextual_orchestrator/server.py +++ b/contextual_orchestrator/server.py @@ -344,6 +344,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) @@ -451,13 +454,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 @@ -494,14 +511,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 diff --git a/docs/planning/adrs/0019-workflow-run-object-authorization.md b/docs/planning/adrs/0019-workflow-run-object-authorization.md index 1c9230b07..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,11 @@ 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 @@ -61,8 +63,9 @@ sequenceDiagram 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 @@ -75,7 +78,9 @@ 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`. +`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/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index 7547bf44c..100957f1c 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -7,7 +7,9 @@ Protected `main` remains 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. A mismatch is returned as the existing generic +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 diff --git a/tests/test_security_hardening.py b/tests/test_security_hardening.py index cfb219004..645f15e75 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: From e570534878b91da89105bb54d2a2813d8aff783d Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 29 Aug 2026 01:35:12 -0700 Subject: [PATCH 04/28] chore: run one-shot conflict resolver for PR #909 --- .github/workflows/resolve-pr-909.yml | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) create mode 100644 .github/workflows/resolve-pr-909.yml diff --git a/.github/workflows/resolve-pr-909.yml b/.github/workflows/resolve-pr-909.yml new file mode 100644 index 000000000..5f4d21227 --- /dev/null +++ b/.github/workflows/resolve-pr-909.yml @@ -0,0 +1,17 @@ +name: Resolve PR 909 against protected main + +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.yml From 069d5dcded0ec9a6fc00420bf8a9417ed7fa6ebe Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 29 Aug 2026 01:37:11 -0700 Subject: [PATCH 05/28] chore: rerun conflict diagnostics for PR #909 --- .github/workflows/resolve-pr-909.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/resolve-pr-909.yml b/.github/workflows/resolve-pr-909.yml index 5f4d21227..2dcc12270 100644 --- a/.github/workflows/resolve-pr-909.yml +++ b/.github/workflows/resolve-pr-909.yml @@ -1,5 +1,6 @@ name: Resolve PR 909 against protected main +# Diagnostic retry: report bounded diff3 hunks for executable conflicts. on: push: branches: [fix/batch-routing-owner-20260829] From 3dd220ca8b3b5a8fdc8e42d65143b77e7b893edb Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 29 Aug 2026 01:42:53 -0700 Subject: [PATCH 06/28] chore: semantically resolve PR #909 conflicts --- .github/workflows/resolve-pr-909.yml | 192 ++++++++++++++++++++++++++- 1 file changed, 185 insertions(+), 7 deletions(-) diff --git a/.github/workflows/resolve-pr-909.yml b/.github/workflows/resolve-pr-909.yml index 2dcc12270..5888a7488 100644 --- a/.github/workflows/resolve-pr-909.yml +++ b/.github/workflows/resolve-pr-909.yml @@ -1,6 +1,5 @@ -name: Resolve PR 909 against protected main +name: Resolve PR 909 semantically against protected main -# Diagnostic retry: report bounded diff3 hunks for executable conflicts. on: push: branches: [fix/batch-routing-owner-20260829] @@ -9,10 +8,189 @@ permissions: contents: write pull-requests: write +concurrency: + group: resolve-pr-909-semantic + cancel-in-progress: false + 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.yml + runs-on: ubuntu-latest + timeout-minutes: 20 + steps: + - uses: actions/checkout@v4 + with: + fetch-depth: 0 + + - name: Merge main and compose owner isolation with ZDR routing + env: + EXPECTED_BRANCH: fix/batch-routing-owner-20260829 + CALLER_FILE: .github/workflows/resolve-pr-909.yml + shell: bash + run: | + set -euo pipefail + test "$(git branch --show-current)" = "$EXPECTED_BRANCH" + git config user.name "github-actions[bot]" + git config user.email "41898282+github-actions[bot]@users.noreply.github.com" + git fetch --no-tags origin main + + set +e + git merge --no-ff --no-commit origin/main + merge_rc=$? + set -e + if [[ $merge_rc -ne 0 ]]; then + mapfile -t conflicts < <(git diff --name-only --diff-filter=U) + unsafe=() + for path in "${conflicts[@]}"; do + case "$path" in + *.md|*.mdx) + tmpdir="$(mktemp -d)" + git show ":2:$path" > "$tmpdir/ours" + git show ":1:$path" > "$tmpdir/base" + git show ":3:$path" > "$tmpdir/theirs" + git merge-file --union "$tmpdir/ours" "$tmpdir/base" "$tmpdir/theirs" + cp "$tmpdir/ours" "$path" + rm -rf "$tmpdir" + git add "$path" + ;; + contextual_orchestrator/cost_router.py|contextual_orchestrator/server.py) + git checkout --conflict=diff3 -- "$path" + ;; + *) unsafe+=("$path") ;; + esac + done + if (( ${#unsafe[@]} )); then + printf 'Unexpected conflicts:\n%s\n' "${unsafe[*]}" + git merge --abort + exit 1 + fi + + python - <<'PY' + from __future__ import annotations + + from pathlib import Path + from typing import Callable + + + def resolve_diff3(path: str, resolver: Callable[[int, str, str, str], str], expected: int) -> None: + source = Path(path) + lines = source.read_text(encoding="utf-8").splitlines(keepends=True) + output: list[str] = [] + index = 0 + cursor = 0 + while cursor < len(lines): + if not lines[cursor].startswith("<<<<<<<"): + output.append(lines[cursor]) + cursor += 1 + continue + ours_start = cursor + 1 + base_marker = next( + i for i in range(ours_start, len(lines)) if lines[i].startswith("|||||||") + ) + split_marker = next( + i for i in range(base_marker + 1, len(lines)) if lines[i].startswith("=======") + ) + end_marker = next( + i for i in range(split_marker + 1, len(lines)) if lines[i].startswith(">>>>>>>") + ) + ours = "".join(lines[ours_start:base_marker]) + base = "".join(lines[base_marker + 1:split_marker]) + theirs = "".join(lines[split_marker + 1:end_marker]) + output.append(resolver(index, ours, base, theirs)) + index += 1 + cursor = end_marker + 1 + if index != expected: + raise RuntimeError(f"{path}: expected {expected} conflicts, found {index}") + merged = "".join(output) + if any(marker in merged for marker in ("<<<<<<<", "|||||||", "=======", ">>>>>>>")): + raise RuntimeError(f"{path}: conflict marker remains") + source.write_text(merged, encoding="utf-8") + + + def cost_router_resolution(index: int, ours: str, base: str, theirs: str) -> str: + del ours, base + merged = theirs + if index == 0: + old_doc = ' """Submit a batch of requests to the configured batch backend."""\n' + new_doc = ' """Submit a batch, resolve its targets, and bind its authenticated owner."""\n' + if old_doc not in merged: + raise RuntimeError("submit_batch docstring anchor changed") + merged = merged.replace(old_doc, new_doc, 1) + anchor = " job = self.batch_backend.submit(prepared_requests, metadata=metadata)\n" + if merged.count(anchor) != 1: + raise RuntimeError("submit_batch prepared-request anchor changed") + merged = merged.replace(anchor, anchor + " job.owner_id = owner_id\n", 1) + return merged + if index == 1: + replacements = ( + ( + " def poll_batch(self, job_id: str) -> Dict[str, Any]:\n", + " def poll_batch(self, job_id: str, *, owner_id: Optional[str] = None) -> Dict[str, Any]:\n", + ), + ( + ' """Poll a previously submitted batch job by id."""\n', + ' """Poll a previously submitted batch job owned by ``owner_id``."""\n', + ), + ( + " def retrieve_batch(self, job_id: str) -> Dict[str, Any]:\n", + " def retrieve_batch(self, job_id: str, *, owner_id: Optional[str] = None) -> Dict[str, Any]:\n", + ), + ( + ' """Retrieve batch results and record usage + cost for each completion."""\n', + ' """Retrieve results for a batch owned by ``owner_id`` and record usage."""\n', + ), + ( + " def _require_job(self, job_id: str) -> BatchJob:\n", + " def _require_job(self, job_id: str, *, owner_id: Optional[str] = None) -> BatchJob:\n", + ), + ( + " if job is None:\n", + " if job is None or job.owner_id != owner_id:\n", + ), + ) + for old, new in replacements: + if merged.count(old) != 1: + raise RuntimeError(f"cost_router anchor changed: {old!r}") + merged = merged.replace(old, new, 1) + require_anchor = " job = self._require_job(job_id)\n" + if merged.count(require_anchor) != 2: + raise RuntimeError("expected poll/retrieve job lookup anchors") + merged = merged.replace( + require_anchor, + " job = self._require_job(job_id, owner_id=owner_id)\n", + ) + return merged + raise RuntimeError(f"unexpected cost_router conflict {index}") + + + def server_resolution(index: int, ours: str, base: str, theirs: str) -> str: + del base + if index not in (0, 1): + raise RuntimeError(f"unexpected server conflict {index}") + if "owner_id=security.principal_id(self.headers)," not in ours: + raise RuntimeError("owner-id side changed") + if "zdr_only=zdr_only," not in theirs: + raise RuntimeError("ZDR side changed") + return ours + theirs + + + resolve_diff3( + "contextual_orchestrator/cost_router.py", + cost_router_resolution, + expected=2, + ) + resolve_diff3( + "contextual_orchestrator/server.py", + server_resolution, + expected=2, + ) + PY + git add contextual_orchestrator/cost_router.py contextual_orchestrator/server.py + fi + + test -z "$(git diff --name-only --diff-filter=U)" + rm -f "$CALLER_FILE" + git add -A + git diff --cached --check + python -m compileall -q contextual_orchestrator + git commit -m "chore: merge protected main into PR #909" + git push origin "HEAD:$EXPECTED_BRANCH" From ef12921a23175d7c91364ed832855a56239a2e9e Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 29 Aug 2026 01:47:37 -0700 Subject: [PATCH 07/28] chore: generalize semantic conflict blocks for PR #909 --- .github/workflows/resolve-pr-909.yml | 142 ++++++++++++++++----------- 1 file changed, 84 insertions(+), 58 deletions(-) diff --git a/.github/workflows/resolve-pr-909.yml b/.github/workflows/resolve-pr-909.yml index 5888a7488..45906f9cd 100644 --- a/.github/workflows/resolve-pr-909.yml +++ b/.github/workflows/resolve-pr-909.yml @@ -71,12 +71,12 @@ jobs: from typing import Callable - def resolve_diff3(path: str, resolver: Callable[[int, str, str, str], str], expected: int) -> None: + def resolve_diff3(path: str, resolver: Callable[[str, str, str], str]) -> int: source = Path(path) lines = source.read_text(encoding="utf-8").splitlines(keepends=True) output: list[str] = [] - index = 0 cursor = 0 + conflicts = 0 while cursor < len(lines): if not lines[cursor].startswith("<<<<<<<"): output.append(lines[cursor]) @@ -95,77 +95,101 @@ jobs: ours = "".join(lines[ours_start:base_marker]) base = "".join(lines[base_marker + 1:split_marker]) theirs = "".join(lines[split_marker + 1:end_marker]) - output.append(resolver(index, ours, base, theirs)) - index += 1 + output.append(resolver(ours, base, theirs)) + conflicts += 1 cursor = end_marker + 1 - if index != expected: - raise RuntimeError(f"{path}: expected {expected} conflicts, found {index}") + if conflicts == 0: + raise RuntimeError(f"{path}: expected at least one conflict") merged = "".join(output) if any(marker in merged for marker in ("<<<<<<<", "|||||||", "=======", ">>>>>>>")): raise RuntimeError(f"{path}: conflict marker remains") source.write_text(merged, encoding="utf-8") + return conflicts - def cost_router_resolution(index: int, ours: str, base: str, theirs: str) -> str: - del ours, base - merged = theirs - if index == 0: - old_doc = ' """Submit a batch of requests to the configured batch backend."""\n' - new_doc = ' """Submit a batch, resolve its targets, and bind its authenticated owner."""\n' - if old_doc not in merged: - raise RuntimeError("submit_batch docstring anchor changed") - merged = merged.replace(old_doc, new_doc, 1) + def replace_once_if_present(text: str, old: str, new: str) -> tuple[str, bool]: + count = text.count(old) + if count > 1: + raise RuntimeError(f"ambiguous merge anchor: {old!r}") + return (text.replace(old, new, 1), True) if count == 1 else (text, False) + + + def cost_router_resolution(ours: str, base: str, theirs: str) -> str: + del base + combined = ours + theirs + if "prepared_requests" in theirs and "job.owner_id = owner_id" in ours: + merged = theirs + merged, changed_doc = replace_once_if_present( + merged, + ' """Submit a batch of requests to the configured batch backend."""\n', + ' """Submit a batch, resolve its targets, and bind its authenticated owner."""\n', + ) anchor = " job = self.batch_backend.submit(prepared_requests, metadata=metadata)\n" if merged.count(anchor) != 1: raise RuntimeError("submit_batch prepared-request anchor changed") merged = merged.replace(anchor, anchor + " job.owner_id = owner_id\n", 1) + if not changed_doc: + raise RuntimeError("submit_batch docstring anchor changed") return merged - if index == 1: - replacements = ( - ( - " def poll_batch(self, job_id: str) -> Dict[str, Any]:\n", - " def poll_batch(self, job_id: str, *, owner_id: Optional[str] = None) -> Dict[str, Any]:\n", - ), - ( - ' """Poll a previously submitted batch job by id."""\n', - ' """Poll a previously submitted batch job owned by ``owner_id``."""\n', - ), - ( - " def retrieve_batch(self, job_id: str) -> Dict[str, Any]:\n", - " def retrieve_batch(self, job_id: str, *, owner_id: Optional[str] = None) -> Dict[str, Any]:\n", - ), - ( - ' """Retrieve batch results and record usage + cost for each completion."""\n', - ' """Retrieve results for a batch owned by ``owner_id`` and record usage."""\n', - ), - ( - " def _require_job(self, job_id: str) -> BatchJob:\n", - " def _require_job(self, job_id: str, *, owner_id: Optional[str] = None) -> BatchJob:\n", - ), - ( - " if job is None:\n", - " if job is None or job.owner_id != owner_id:\n", - ), + + if not any( + token in combined + for token in ( + "def poll_batch", + "def retrieve_batch", + "def _require_job", + "def _resolve_batch_request", ) - for old, new in replacements: - if merged.count(old) != 1: - raise RuntimeError(f"cost_router anchor changed: {old!r}") - merged = merged.replace(old, new, 1) - require_anchor = " job = self._require_job(job_id)\n" - if merged.count(require_anchor) != 2: - raise RuntimeError("expected poll/retrieve job lookup anchors") + ): + raise RuntimeError("unrecognized cost_router conflict block") + + merged = theirs + changed = False + transformations = ( + ( + " def poll_batch(self, job_id: str) -> Dict[str, Any]:\n", + " def poll_batch(self, job_id: str, *, owner_id: Optional[str] = None) -> Dict[str, Any]:\n", + ), + ( + ' """Poll a previously submitted batch job by id."""\n', + ' """Poll a previously submitted batch job owned by ``owner_id``."""\n', + ), + ( + " def retrieve_batch(self, job_id: str) -> Dict[str, Any]:\n", + " def retrieve_batch(self, job_id: str, *, owner_id: Optional[str] = None) -> Dict[str, Any]:\n", + ), + ( + ' """Retrieve batch results and record usage + cost for each completion."""\n', + ' """Retrieve results for a batch owned by ``owner_id`` and record usage."""\n', + ), + ( + " def _require_job(self, job_id: str) -> BatchJob:\n", + " def _require_job(self, job_id: str, *, owner_id: Optional[str] = None) -> BatchJob:\n", + ), + ( + " if job is None:\n", + " if job is None or job.owner_id != owner_id:\n", + ), + ) + for old, new in transformations: + merged, did_change = replace_once_if_present(merged, old, new) + changed = changed or did_change + + lookup = " job = self._require_job(job_id)\n" + lookup_count = merged.count(lookup) + if lookup_count: merged = merged.replace( - require_anchor, + lookup, " job = self._require_job(job_id, owner_id=owner_id)\n", ) - return merged - raise RuntimeError(f"unexpected cost_router conflict {index}") + changed = True + if not changed: + raise RuntimeError("cost_router owner-isolation anchors were absent") + return merged - def server_resolution(index: int, ours: str, base: str, theirs: str) -> str: + def server_resolution(ours: str, base: str, theirs: str) -> str: del base - if index not in (0, 1): - raise RuntimeError(f"unexpected server conflict {index}") if "owner_id=security.principal_id(self.headers)," not in ours: raise RuntimeError("owner-id side changed") if "zdr_only=zdr_only," not in theirs: @@ -173,16 +197,18 @@ jobs: return ours + theirs - resolve_diff3( + cost_count = resolve_diff3( "contextual_orchestrator/cost_router.py", cost_router_resolution, - expected=2, ) - resolve_diff3( + server_count = resolve_diff3( "contextual_orchestrator/server.py", server_resolution, - expected=2, ) + if cost_count < 2 or server_count != 2: + raise RuntimeError( + f"unexpected conflict topology: cost_router={cost_count}, server={server_count}" + ) PY git add contextual_orchestrator/cost_router.py contextual_orchestrator/server.py fi From 68a421fa1d837f00b1868c59d7f443324ac441bb Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 29 Aug 2026 02:06:06 -0700 Subject: [PATCH 08/28] chore: request protected checks for PR #909 --- docs/.pr-909-recheck | 1 + 1 file changed, 1 insertion(+) create mode 100644 docs/.pr-909-recheck diff --git a/docs/.pr-909-recheck b/docs/.pr-909-recheck new file mode 100644 index 000000000..178cf4ae7 --- /dev/null +++ b/docs/.pr-909-recheck @@ -0,0 +1 @@ +temporary exact-head check trigger From 6ce2a82927f8516e1bc4a5a24f64f5fbe72dd9a8 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 29 Aug 2026 02:06:15 -0700 Subject: [PATCH 09/28] chore: finalize protected check trigger for PR #909 --- docs/.pr-909-recheck | 1 - 1 file changed, 1 deletion(-) delete mode 100644 docs/.pr-909-recheck diff --git a/docs/.pr-909-recheck b/docs/.pr-909-recheck deleted file mode 100644 index 178cf4ae7..000000000 --- a/docs/.pr-909-recheck +++ /dev/null @@ -1 +0,0 @@ -temporary exact-head check trigger From 35e6a73a5da9cf06f6f9415c766698620f8990ae Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 29 Aug 2026 02:33:53 -0700 Subject: [PATCH 10/28] chore: refresh PR #909 against current protected main --- .github/workflows/resolve-pr-909-refresh.yml | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) create mode 100644 .github/workflows/resolve-pr-909-refresh.yml diff --git a/.github/workflows/resolve-pr-909-refresh.yml b/.github/workflows/resolve-pr-909-refresh.yml new file mode 100644 index 000000000..5f98adee1 --- /dev/null +++ b/.github/workflows/resolve-pr-909-refresh.yml @@ -0,0 +1,17 @@ +name: Refresh PR 909 against protected main + +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-refresh.yml From 34c5b80ede093ef364942696d8816acda70780a9 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 29 Aug 2026 03:23:48 -0700 Subject: [PATCH 11/28] chore: add one-shot ADR number repair for PR #909 --- .github/workflows/fix-adr-number-pr-909.yml | 73 +++++++++++++++++++++ 1 file changed, 73 insertions(+) create mode 100644 .github/workflows/fix-adr-number-pr-909.yml diff --git a/.github/workflows/fix-adr-number-pr-909.yml b/.github/workflows/fix-adr-number-pr-909.yml new file mode 100644 index 000000000..739286727 --- /dev/null +++ b/.github/workflows/fix-adr-number-pr-909.yml @@ -0,0 +1,73 @@ +name: Repair duplicate ADR number on PR 909 + +on: + push: + branches: [fix/batch-routing-owner-20260829] + +permissions: + contents: write + +concurrency: + group: repair-pr-909-adr-number + cancel-in-progress: false + +jobs: + repair: + runs-on: ubuntu-latest + timeout-minutes: 20 + steps: + - name: Check out exact PR head + uses: actions/checkout@v4 + with: + fetch-depth: 0 + + - name: Renumber streamed Responses usage ADR and validate + env: + EXPECTED_PARENT: 96c27059e49e50d4e508ce4612463e922fbff447 + BRANCH: fix/batch-routing-owner-20260829 + SELF: .github/workflows/fix-adr-number-pr-909.yml + shell: bash + run: | + set -euo pipefail + test "$(git branch --show-current)" = "$BRANCH" + test "$(git rev-parse HEAD^)" = "$EXPECTED_PARENT" + old='docs/planning/adrs/0038-streamed-responses-usage-boundary.md' + new='docs/planning/adrs/0040-streamed-responses-usage-boundary.md' + test -f "$old" + test ! -e "$new" + git mv "$old" "$new" + python - <<'PY' + from pathlib import Path + + adr = Path('docs/planning/adrs/0040-streamed-responses-usage-boundary.md') + text = adr.read_text(encoding='utf-8') + old = 'id: "0038"' + if text.count(old) != 1: + raise SystemExit(f'expected exactly one {old!r} in {adr}') + adr.write_text(text.replace(old, 'id: "0040"', 1), encoding='utf-8') + + architecture = Path('docs/architecture.md') + text = architecture.read_text(encoding='utf-8') + old = 'See [ADR 0038](planning/adrs/0038-streamed-responses-usage-boundary.md).' + new = 'See [ADR 0040](planning/adrs/0040-streamed-responses-usage-boundary.md).' + if text.count(old) != 1: + raise SystemExit(f'expected exactly one streamed ADR link in {architecture}') + architecture.write_text(text.replace(old, new, 1), encoding='utf-8') + + changelog = Path('CHANGELOG.md') + text = changelog.read_text(encoding='utf-8') + marker = 'answer, and nested gateway upstreams remain compatible (ADR 0038).' + replacement = 'answer, and nested gateway upstreams remain compatible (ADR 0040).' + if text.count(marker) != 1: + raise SystemExit(f'expected exactly one streamed ADR reference in {changelog}') + changelog.write_text(text.replace(marker, replacement, 1), encoding='utf-8') + PY + rm -f "$SELF" + git add -A + git diff --cached --check + uv run --locked --extra api --extra db --extra queue --group dev \ + python -m pytest -q tests/test_planning_adr_identifiers.py + git config user.name 'github-actions[bot]' + git config user.email '41898282+github-actions[bot]@users.noreply.github.com' + git commit -m 'docs: assign unique ADR number to streamed Responses usage' + git push origin "HEAD:$BRANCH" From 97eb8d5a0a0ed1ba1f4d5a6a34367652c18288fe Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 29 Aug 2026 03:27:34 -0700 Subject: [PATCH 12/28] chore: complete one-shot PR #909 repair validation --- .github/workflows/fix-adr-number-pr-909.yml | 143 ++++++++++++++++---- 1 file changed, 116 insertions(+), 27 deletions(-) diff --git a/.github/workflows/fix-adr-number-pr-909.yml b/.github/workflows/fix-adr-number-pr-909.yml index 739286727..927764649 100644 --- a/.github/workflows/fix-adr-number-pr-909.yml +++ b/.github/workflows/fix-adr-number-pr-909.yml @@ -1,4 +1,4 @@ -name: Repair duplicate ADR number on PR 909 +name: Repair PR 909 merge-result regressions on: push: @@ -8,7 +8,7 @@ permissions: contents: write concurrency: - group: repair-pr-909-adr-number + group: repair-pr-909-merge-result cancel-in-progress: false jobs: @@ -17,13 +17,23 @@ jobs: timeout-minutes: 20 steps: - name: Check out exact PR head - uses: actions/checkout@v4 + uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 with: fetch-depth: 0 - - name: Renumber streamed Responses usage ADR and validate + - name: Set up Python + uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 + with: + python-version: "3.12" + + - name: Set up uv + uses: astral-sh/setup-uv@d0cc045d04ccac9d8b7881df0226f9e82c39688e + with: + version: "0.12.5" + + - name: Repair ADR numbering and invalid-model semantics env: - EXPECTED_PARENT: 96c27059e49e50d4e508ce4612463e922fbff447 + EXPECTED_PARENT: 34c5b80ede093ef364942696d8816acda70780a9 BRANCH: fix/batch-routing-owner-20260829 SELF: .github/workflows/fix-adr-number-pr-909.yml shell: bash @@ -39,35 +49,114 @@ jobs: python - <<'PY' from pathlib import Path - adr = Path('docs/planning/adrs/0040-streamed-responses-usage-boundary.md') - text = adr.read_text(encoding='utf-8') - old = 'id: "0038"' - if text.count(old) != 1: - raise SystemExit(f'expected exactly one {old!r} in {adr}') - adr.write_text(text.replace(old, 'id: "0040"', 1), encoding='utf-8') + def replace_exact(path: str, old: str, new: str) -> None: + target = Path(path) + text = target.read_text(encoding="utf-8") + if text.count(old) != 1: + raise SystemExit( + f"expected exactly one replacement target in {path}; found {text.count(old)}" + ) + target.write_text(text.replace(old, new, 1), encoding="utf-8") + + replace_exact( + "docs/planning/adrs/0040-streamed-responses-usage-boundary.md", + 'id: "0038"', + 'id: "0040"', + ) + replace_exact( + "docs/architecture.md", + "See [ADR 0038](planning/adrs/0038-streamed-responses-usage-boundary.md).", + "See [ADR 0040](planning/adrs/0040-streamed-responses-usage-boundary.md).", + ) + replace_exact( + "CHANGELOG.md", + "answer, and nested gateway upstreams remain compatible (ADR 0038).", + "answer, and nested gateway upstreams remain compatible (ADR 0040).", + ) + replace_exact( + "contextual_orchestrator/cost_router.py", + ''' try: + prepared_requests = [self._resolve_batch_request(request) for request in requests] + except (RuntimeError, ValueError) as exc: + raise BatchModelSelectionError( + "no eligible model-group member is available for this batch request" + ) from exc + ''', + ''' try: + prepared_requests = [self._resolve_batch_request(request) for request in requests] + except ValueError: + raise + except RuntimeError as exc: + raise BatchModelSelectionError( + "no eligible model-group member is available for this batch request" + ) from exc + ''', + ) + replace_exact( + "contextual_orchestrator/server.py", + ''' job = self._run( + lambda: coordinator.submit_batch( + batch_requests, + metadata=metadata, + owner_id=security.principal_id(self.headers), + ) + ) + ''', + ''' try: + job = self._run( + lambda: coordinator.submit_batch( + batch_requests, + metadata=metadata, + owner_id=security.principal_id(self.headers), + ) + ) + except ValueError as exc: + raise RequestError(400, "invalid_model", str(exc)) from exc + ''', + ) - architecture = Path('docs/architecture.md') - text = architecture.read_text(encoding='utf-8') - old = 'See [ADR 0038](planning/adrs/0038-streamed-responses-usage-boundary.md).' - new = 'See [ADR 0040](planning/adrs/0040-streamed-responses-usage-boundary.md).' - if text.count(old) != 1: - raise SystemExit(f'expected exactly one streamed ADR link in {architecture}') - architecture.write_text(text.replace(old, new, 1), encoding='utf-8') + tests = Path("tests/test_cost_review_server.py") + test_text = tests.read_text(encoding="utf-8") + test_name = "test_batch_routing_rejects_unknown_zdr_model_as_client_error" + if test_name in test_text: + raise SystemExit(f"{test_name} already exists") + tests.write_text( + test_text + + ''' - changelog = Path('CHANGELOG.md') - text = changelog.read_text(encoding='utf-8') - marker = 'answer, and nested gateway upstreams remain compatible (ADR 0038).' - replacement = 'answer, and nested gateway upstreams remain compatible (ADR 0040).' - if text.count(marker) != 1: - raise SystemExit(f'expected exactly one streamed ADR reference in {changelog}') - changelog.write_text(text.replace(marker, replacement, 1), encoding='utf-8') + 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" + ''', + encoding="utf-8", + ) PY rm -f "$SELF" git add -A git diff --cached --check uv run --locked --extra api --extra db --extra queue --group dev \ - python -m pytest -q tests/test_planning_adr_identifiers.py + python -m pytest -q \ + tests/test_planning_adr_identifiers.py \ + tests/test_cost_review_server.py \ + tests/test_cost_router_boundaries.py git config user.name 'github-actions[bot]' git config user.email '41898282+github-actions[bot]@users.noreply.github.com' - git commit -m 'docs: assign unique ADR number to streamed Responses usage' + git commit -m 'fix: repair merge-result ADR and batch model contracts' git push origin "HEAD:$BRANCH" From 76d1254918d1756a7226d509c8ce9500c7a9963d Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 29 Aug 2026 03:30:00 -0700 Subject: [PATCH 13/28] chore: use structural replacements in PR #909 repair --- .github/workflows/fix-adr-number-pr-909.yml | 118 ++++++++++---------- 1 file changed, 59 insertions(+), 59 deletions(-) diff --git a/.github/workflows/fix-adr-number-pr-909.yml b/.github/workflows/fix-adr-number-pr-909.yml index 927764649..d9da6e6fd 100644 --- a/.github/workflows/fix-adr-number-pr-909.yml +++ b/.github/workflows/fix-adr-number-pr-909.yml @@ -33,7 +33,7 @@ jobs: - name: Repair ADR numbering and invalid-model semantics env: - EXPECTED_PARENT: 34c5b80ede093ef364942696d8816acda70780a9 + EXPECTED_PARENT: 97eb8d5a0a0ed1ba1f4d5a6a34367652c18288fe BRANCH: fix/batch-routing-owner-20260829 SELF: .github/workflows/fix-adr-number-pr-909.yml shell: bash @@ -48,16 +48,28 @@ jobs: git mv "$old" "$new" python - <<'PY' from pathlib import Path + import re def replace_exact(path: str, old: str, new: str) -> None: target = Path(path) text = target.read_text(encoding="utf-8") - if text.count(old) != 1: + count = text.count(old) + if count != 1: raise SystemExit( - f"expected exactly one replacement target in {path}; found {text.count(old)}" + f"expected exactly one replacement target in {path}; found {count}" ) target.write_text(text.replace(old, new, 1), encoding="utf-8") + def replace_regex(path: str, pattern: str, replacement: str) -> None: + target = Path(path) + text = target.read_text(encoding="utf-8") + updated, count = re.subn(pattern, replacement, text, flags=re.MULTILINE) + if count != 1: + raise SystemExit( + f"expected exactly one structural replacement in {path}; found {count}" + ) + target.write_text(updated, encoding="utf-8") + replace_exact( "docs/planning/adrs/0040-streamed-responses-usage-boundary.md", 'id: "0038"', @@ -73,46 +85,33 @@ jobs: "answer, and nested gateway upstreams remain compatible (ADR 0038).", "answer, and nested gateway upstreams remain compatible (ADR 0040).", ) - replace_exact( + replace_regex( "contextual_orchestrator/cost_router.py", + r'''^ try:\n prepared_requests = \[self\._resolve_batch_request\(request\) for request in requests\]\n except \(RuntimeError, ValueError\) as exc:\n raise BatchModelSelectionError\(\n "no eligible model-group member is available for this batch request"\n \) from exc\n''', ''' try: - prepared_requests = [self._resolve_batch_request(request) for request in requests] - except (RuntimeError, ValueError) as exc: - raise BatchModelSelectionError( - "no eligible model-group member is available for this batch request" - ) from exc - ''', - ''' try: - prepared_requests = [self._resolve_batch_request(request) for request in requests] - except ValueError: - raise - except RuntimeError as exc: - raise BatchModelSelectionError( - "no eligible model-group member is available for this batch request" - ) from exc - ''', + prepared_requests = [self._resolve_batch_request(request) for request in requests] + except ValueError: + raise + except RuntimeError as exc: + raise BatchModelSelectionError( + "no eligible model-group member is available for this batch request" + ) from exc +''', ) - replace_exact( + replace_regex( "contextual_orchestrator/server.py", - ''' job = self._run( - lambda: coordinator.submit_batch( - batch_requests, - metadata=metadata, - owner_id=security.principal_id(self.headers), - ) - ) - ''', + r'''^ job = self\._run\(\n lambda: coordinator\.submit_batch\(\n batch_requests,\n metadata=metadata,\n owner_id=security\.principal_id\(self\.headers\),\n \)\n \)\n''', ''' try: - job = self._run( - lambda: coordinator.submit_batch( - batch_requests, - metadata=metadata, - owner_id=security.principal_id(self.headers), - ) - ) - except ValueError as exc: - raise RequestError(400, "invalid_model", str(exc)) from exc - ''', + job = self._run( + lambda: coordinator.submit_batch( + batch_requests, + metadata=metadata, + owner_id=security.principal_id(self.headers), + ) + ) + except ValueError as exc: + raise RequestError(400, "invalid_model", str(exc)) from exc +''', ) tests = Path("tests/test_cost_review_server.py") @@ -124,27 +123,28 @@ jobs: test_text + ''' - 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" - ''', + +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" +''', encoding="utf-8", ) PY From 0130f94c24c108e8a1f9d5d1b56c4dd1d3ff772f Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 29 Aug 2026 03:32:24 -0700 Subject: [PATCH 14/28] chore: add one-shot PR #909 repair script --- scripts/ci/repair_pr_909.py | 121 ++++++++++++++++++++++++++++++++++++ 1 file changed, 121 insertions(+) create mode 100644 scripts/ci/repair_pr_909.py diff --git a/scripts/ci/repair_pr_909.py b/scripts/ci/repair_pr_909.py new file mode 100644 index 000000000..043d2acad --- /dev/null +++ b/scripts/ci/repair_pr_909.py @@ -0,0 +1,121 @@ +"""One-shot exact-head repair for PR #909; deleted by its caller after validation.""" + +from __future__ import annotations + +from pathlib import Path + + +def replace_exact(path: str, old: str, new: str) -> None: + """Replace exactly one literal block or fail closed.""" + target = Path(path) + text = target.read_text(encoding="utf-8") + count = text.count(old) + if count != 1: + raise SystemExit( + f"expected exactly one replacement target in {path}; found {count}" + ) + target.write_text(text.replace(old, new, 1), encoding="utf-8") + + +def main() -> None: + """Apply the reviewed ADR and batch-model contract repairs.""" + old_adr = Path("docs/planning/adrs/0038-streamed-responses-usage-boundary.md") + new_adr = Path("docs/planning/adrs/0040-streamed-responses-usage-boundary.md") + if not old_adr.is_file() or new_adr.exists(): + raise SystemExit("streamed Responses ADR rename preconditions are not met") + old_adr.rename(new_adr) + + replace_exact( + str(new_adr), + 'id: "0038"', + 'id: "0040"', + ) + replace_exact( + "docs/architecture.md", + "See [ADR 0038](planning/adrs/0038-streamed-responses-usage-boundary.md).", + "See [ADR 0040](planning/adrs/0040-streamed-responses-usage-boundary.md).", + ) + replace_exact( + "CHANGELOG.md", + "answer, and nested gateway upstreams remain compatible (ADR 0038).", + "answer, and nested gateway upstreams remain compatible (ADR 0040).", + ) + replace_exact( + "contextual_orchestrator/cost_router.py", + ''' try: + prepared_requests = [self._resolve_batch_request(request) for request in requests] + except (RuntimeError, ValueError) as exc: + raise BatchModelSelectionError( + "no eligible model-group member is available for this batch request" + ) from exc +''', + ''' try: + prepared_requests = [self._resolve_batch_request(request) for request in requests] + except ValueError: + raise + except RuntimeError as exc: + raise BatchModelSelectionError( + "no eligible model-group member is available for this batch request" + ) from exc +''', + ) + replace_exact( + "contextual_orchestrator/server.py", + ''' job = self._run( + lambda: coordinator.submit_batch( + batch_requests, + metadata=metadata, + owner_id=security.principal_id(self.headers), + ) + ) +''', + ''' try: + job = self._run( + lambda: coordinator.submit_batch( + batch_requests, + metadata=metadata, + owner_id=security.principal_id(self.headers), + ) + ) + except ValueError as exc: + raise RequestError(400, "invalid_model", str(exc)) from exc +''', + ) + + tests = Path("tests/test_cost_review_server.py") + test_text = tests.read_text(encoding="utf-8") + test_name = "test_batch_routing_rejects_unknown_zdr_model_as_client_error" + if test_name in test_text: + raise SystemExit(f"{test_name} already exists") + tests.write_text( + test_text + + ''' + + +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" +''', + encoding="utf-8", + ) + + +if __name__ == "__main__": + main() From 54d409114f50c05400567f12aefe479f312fc175 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 29 Aug 2026 03:32:36 -0700 Subject: [PATCH 15/28] chore: invoke bounded PR #909 repair script --- .github/workflows/fix-adr-number-pr-909.yml | 120 ++------------------ 1 file changed, 8 insertions(+), 112 deletions(-) diff --git a/.github/workflows/fix-adr-number-pr-909.yml b/.github/workflows/fix-adr-number-pr-909.yml index d9da6e6fd..95547b8fc 100644 --- a/.github/workflows/fix-adr-number-pr-909.yml +++ b/.github/workflows/fix-adr-number-pr-909.yml @@ -2,7 +2,8 @@ name: Repair PR 909 merge-result regressions on: push: - branches: [fix/batch-routing-owner-20260829] + branches: + - fix/batch-routing-owner-20260829 permissions: contents: write @@ -31,124 +32,19 @@ jobs: with: version: "0.12.5" - - name: Repair ADR numbering and invalid-model semantics + - name: Apply, validate, and commit the bounded repair env: - EXPECTED_PARENT: 97eb8d5a0a0ed1ba1f4d5a6a34367652c18288fe + EXPECTED_PARENT: 0130f94c24c108e8a1f9d5d1b56c4dd1d3ff772f BRANCH: fix/batch-routing-owner-20260829 - SELF: .github/workflows/fix-adr-number-pr-909.yml shell: bash run: | set -euo pipefail test "$(git branch --show-current)" = "$BRANCH" test "$(git rev-parse HEAD^)" = "$EXPECTED_PARENT" - old='docs/planning/adrs/0038-streamed-responses-usage-boundary.md' - new='docs/planning/adrs/0040-streamed-responses-usage-boundary.md' - test -f "$old" - test ! -e "$new" - git mv "$old" "$new" - python - <<'PY' - from pathlib import Path - import re - - def replace_exact(path: str, old: str, new: str) -> None: - target = Path(path) - text = target.read_text(encoding="utf-8") - count = text.count(old) - if count != 1: - raise SystemExit( - f"expected exactly one replacement target in {path}; found {count}" - ) - target.write_text(text.replace(old, new, 1), encoding="utf-8") - - def replace_regex(path: str, pattern: str, replacement: str) -> None: - target = Path(path) - text = target.read_text(encoding="utf-8") - updated, count = re.subn(pattern, replacement, text, flags=re.MULTILINE) - if count != 1: - raise SystemExit( - f"expected exactly one structural replacement in {path}; found {count}" - ) - target.write_text(updated, encoding="utf-8") - - replace_exact( - "docs/planning/adrs/0040-streamed-responses-usage-boundary.md", - 'id: "0038"', - 'id: "0040"', - ) - replace_exact( - "docs/architecture.md", - "See [ADR 0038](planning/adrs/0038-streamed-responses-usage-boundary.md).", - "See [ADR 0040](planning/adrs/0040-streamed-responses-usage-boundary.md).", - ) - replace_exact( - "CHANGELOG.md", - "answer, and nested gateway upstreams remain compatible (ADR 0038).", - "answer, and nested gateway upstreams remain compatible (ADR 0040).", - ) - replace_regex( - "contextual_orchestrator/cost_router.py", - r'''^ try:\n prepared_requests = \[self\._resolve_batch_request\(request\) for request in requests\]\n except \(RuntimeError, ValueError\) as exc:\n raise BatchModelSelectionError\(\n "no eligible model-group member is available for this batch request"\n \) from exc\n''', - ''' try: - prepared_requests = [self._resolve_batch_request(request) for request in requests] - except ValueError: - raise - except RuntimeError as exc: - raise BatchModelSelectionError( - "no eligible model-group member is available for this batch request" - ) from exc -''', - ) - replace_regex( - "contextual_orchestrator/server.py", - r'''^ job = self\._run\(\n lambda: coordinator\.submit_batch\(\n batch_requests,\n metadata=metadata,\n owner_id=security\.principal_id\(self\.headers\),\n \)\n \)\n''', - ''' try: - job = self._run( - lambda: coordinator.submit_batch( - batch_requests, - metadata=metadata, - owner_id=security.principal_id(self.headers), - ) - ) - except ValueError as exc: - raise RequestError(400, "invalid_model", str(exc)) from exc -''', - ) - - tests = Path("tests/test_cost_review_server.py") - test_text = tests.read_text(encoding="utf-8") - test_name = "test_batch_routing_rejects_unknown_zdr_model_as_client_error" - if test_name in test_text: - raise SystemExit(f"{test_name} already exists") - tests.write_text( - test_text - + ''' - - -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" -''', - encoding="utf-8", - ) - PY - rm -f "$SELF" + python scripts/ci/repair_pr_909.py + rm -f \ + scripts/ci/repair_pr_909.py \ + .github/workflows/fix-adr-number-pr-909.yml git add -A git diff --cached --check uv run --locked --extra api --extra db --extra queue --group dev \ From b6c63b46f4f1e9f382529a74d7bc93e149a00742 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Sat, 29 Aug 2026 10:33:11 +0000 Subject: [PATCH 16/28] fix: repair merge-result ADR and batch model contracts --- .github/workflows/fix-adr-number-pr-909.yml | 58 --------- CHANGELOG.md | 2 +- contextual_orchestrator/cost_router.py | 4 +- contextual_orchestrator/server.py | 15 ++- docs/architecture.md | 2 +- ...0040-streamed-responses-usage-boundary.md} | 2 +- scripts/ci/repair_pr_909.py | 121 ------------------ tests/test_cost_review_server.py | 23 ++++ 8 files changed, 38 insertions(+), 189 deletions(-) delete mode 100644 .github/workflows/fix-adr-number-pr-909.yml rename docs/planning/adrs/{0038-streamed-responses-usage-boundary.md => 0040-streamed-responses-usage-boundary.md} (99%) delete mode 100644 scripts/ci/repair_pr_909.py diff --git a/.github/workflows/fix-adr-number-pr-909.yml b/.github/workflows/fix-adr-number-pr-909.yml deleted file mode 100644 index 95547b8fc..000000000 --- a/.github/workflows/fix-adr-number-pr-909.yml +++ /dev/null @@ -1,58 +0,0 @@ -name: Repair PR 909 merge-result regressions - -on: - push: - branches: - - fix/batch-routing-owner-20260829 - -permissions: - contents: write - -concurrency: - group: repair-pr-909-merge-result - cancel-in-progress: false - -jobs: - repair: - runs-on: ubuntu-latest - timeout-minutes: 20 - steps: - - name: Check out exact PR head - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 - with: - fetch-depth: 0 - - - name: Set up Python - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 - with: - python-version: "3.12" - - - name: Set up uv - uses: astral-sh/setup-uv@d0cc045d04ccac9d8b7881df0226f9e82c39688e - with: - version: "0.12.5" - - - name: Apply, validate, and commit the bounded repair - env: - EXPECTED_PARENT: 0130f94c24c108e8a1f9d5d1b56c4dd1d3ff772f - BRANCH: fix/batch-routing-owner-20260829 - shell: bash - run: | - set -euo pipefail - test "$(git branch --show-current)" = "$BRANCH" - test "$(git rev-parse HEAD^)" = "$EXPECTED_PARENT" - python scripts/ci/repair_pr_909.py - rm -f \ - scripts/ci/repair_pr_909.py \ - .github/workflows/fix-adr-number-pr-909.yml - git add -A - git diff --cached --check - uv run --locked --extra api --extra db --extra queue --group dev \ - python -m pytest -q \ - tests/test_planning_adr_identifiers.py \ - tests/test_cost_review_server.py \ - tests/test_cost_router_boundaries.py - git config user.name 'github-actions[bot]' - git config user.email '41898282+github-actions[bot]@users.noreply.github.com' - git commit -m 'fix: repair merge-result ADR and batch model contracts' - git push origin "HEAD:$BRANCH" diff --git a/CHANGELOG.md b/CHANGELOG.md index 07d45aaff..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 diff --git a/contextual_orchestrator/cost_router.py b/contextual_orchestrator/cost_router.py index c6225ff82..778b771a6 100644 --- a/contextual_orchestrator/cost_router.py +++ b/contextual_orchestrator/cost_router.py @@ -659,7 +659,9 @@ def submit_batch( """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: + raise + except RuntimeError as exc: raise BatchModelSelectionError( "no eligible model-group member is available for this batch request" ) from exc diff --git a/contextual_orchestrator/server.py b/contextual_orchestrator/server.py index 14cd8c3e3..96fa66a2c 100644 --- a/contextual_orchestrator/server.py +++ b/contextual_orchestrator/server.py @@ -6976,13 +6976,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, - owner_id=security.principal_id(self.headers), + try: + job = self._run( + lambda: coordinator.submit_batch( + batch_requests, + metadata=metadata, + owner_id=security.principal_id(self.headers), + ) ) - ) + except ValueError as exc: + raise RequestError(400, "invalid_model", str(exc)) from exc orchestrator.record_analytics_event( "batch_routing_job_created", { diff --git a/docs/architecture.md b/docs/architecture.md index 2174054ab..df27bca82 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -77,7 +77,7 @@ bounded, authenticated recursion protocol; it is not administratively disabled. 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/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/scripts/ci/repair_pr_909.py b/scripts/ci/repair_pr_909.py deleted file mode 100644 index 043d2acad..000000000 --- a/scripts/ci/repair_pr_909.py +++ /dev/null @@ -1,121 +0,0 @@ -"""One-shot exact-head repair for PR #909; deleted by its caller after validation.""" - -from __future__ import annotations - -from pathlib import Path - - -def replace_exact(path: str, old: str, new: str) -> None: - """Replace exactly one literal block or fail closed.""" - target = Path(path) - text = target.read_text(encoding="utf-8") - count = text.count(old) - if count != 1: - raise SystemExit( - f"expected exactly one replacement target in {path}; found {count}" - ) - target.write_text(text.replace(old, new, 1), encoding="utf-8") - - -def main() -> None: - """Apply the reviewed ADR and batch-model contract repairs.""" - old_adr = Path("docs/planning/adrs/0038-streamed-responses-usage-boundary.md") - new_adr = Path("docs/planning/adrs/0040-streamed-responses-usage-boundary.md") - if not old_adr.is_file() or new_adr.exists(): - raise SystemExit("streamed Responses ADR rename preconditions are not met") - old_adr.rename(new_adr) - - replace_exact( - str(new_adr), - 'id: "0038"', - 'id: "0040"', - ) - replace_exact( - "docs/architecture.md", - "See [ADR 0038](planning/adrs/0038-streamed-responses-usage-boundary.md).", - "See [ADR 0040](planning/adrs/0040-streamed-responses-usage-boundary.md).", - ) - replace_exact( - "CHANGELOG.md", - "answer, and nested gateway upstreams remain compatible (ADR 0038).", - "answer, and nested gateway upstreams remain compatible (ADR 0040).", - ) - replace_exact( - "contextual_orchestrator/cost_router.py", - ''' try: - prepared_requests = [self._resolve_batch_request(request) for request in requests] - except (RuntimeError, ValueError) as exc: - raise BatchModelSelectionError( - "no eligible model-group member is available for this batch request" - ) from exc -''', - ''' try: - prepared_requests = [self._resolve_batch_request(request) for request in requests] - except ValueError: - raise - except RuntimeError as exc: - raise BatchModelSelectionError( - "no eligible model-group member is available for this batch request" - ) from exc -''', - ) - replace_exact( - "contextual_orchestrator/server.py", - ''' job = self._run( - lambda: coordinator.submit_batch( - batch_requests, - metadata=metadata, - owner_id=security.principal_id(self.headers), - ) - ) -''', - ''' try: - job = self._run( - lambda: coordinator.submit_batch( - batch_requests, - metadata=metadata, - owner_id=security.principal_id(self.headers), - ) - ) - except ValueError as exc: - raise RequestError(400, "invalid_model", str(exc)) from exc -''', - ) - - tests = Path("tests/test_cost_review_server.py") - test_text = tests.read_text(encoding="utf-8") - test_name = "test_batch_routing_rejects_unknown_zdr_model_as_client_error" - if test_name in test_text: - raise SystemExit(f"{test_name} already exists") - tests.write_text( - test_text - + ''' - - -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" -''', - encoding="utf-8", - ) - - -if __name__ == "__main__": - main() diff --git a/tests/test_cost_review_server.py b/tests/test_cost_review_server.py index 404769707..9efbb66af 100644 --- a/tests/test_cost_review_server.py +++ b/tests/test_cost_review_server.py @@ -539,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" From eaaa9e20f1da7afe02061cc0935ea151e10a69be Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 29 Aug 2026 03:33:42 -0700 Subject: [PATCH 17/28] fix: document batch owner and invalid-model contracts --- CHANGELOG.d/batch-routing-owner-model-error.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 CHANGELOG.d/batch-routing-owner-model-error.md 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..d8d6a4cd1 --- /dev/null +++ b/CHANGELOG.d/batch-routing-owner-model-error.md @@ -0,0 +1 @@ +Bound HTTP-created batch routing jobs to the authenticated principal, preserved cross-owner not-found behavior, and report an unknown explicit ZDR model as the non-retryable `400 invalid_model` client error instead of a retryable backend outage. From 5f7ae2e8f6314a8924b42cb065a436ddd69d0c07 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 29 Aug 2026 03:35:08 -0700 Subject: [PATCH 18/28] docs: point streamed Responses doctoring to ADR 0040 --- docs/doctoring/responses-stream-usage.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 From 661360d2ae8c350ddd77b7f789b64997185f3245 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 29 Aug 2026 03:49:36 -0700 Subject: [PATCH 19/28] chore: add bounded PR #909 ZDR classification repair --- scripts/ci/repair_pr_909_zdr_selection.py | 52 +++++++++++++++++++++++ 1 file changed, 52 insertions(+) create mode 100644 scripts/ci/repair_pr_909_zdr_selection.py diff --git a/scripts/ci/repair_pr_909_zdr_selection.py b/scripts/ci/repair_pr_909_zdr_selection.py new file mode 100644 index 000000000..a8ec30ecb --- /dev/null +++ b/scripts/ci/repair_pr_909_zdr_selection.py @@ -0,0 +1,52 @@ +"""One-shot exact-head repair for PR #909 ZDR model error classification.""" + +from __future__ import annotations + +from pathlib import Path + + +def main() -> None: + """Distinguish an unknown model from a configured but ZDR-ineligible model.""" + path = Path("contextual_orchestrator/cost_router.py") + text = path.read_text(encoding="utf-8") + old = ''' with self.orchestrator.request_policy(request.zdr_only): + agent = self.orchestrator._requested_agent(request.model) + if agent is None: + text = self.orchestrator._latest_user_text(request.messages) + agent = self.orchestrator._select_agent( + text, + "worker", + free_only=request.model + == getattr(self.orchestrator, "FREE_MODEL", object()), + ) +''' + new = ''' with self.orchestrator.request_policy(request.zdr_only): + 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( + text, + "worker", + free_only=request.model + == getattr(self.orchestrator, "FREE_MODEL", object()), + ) +''' + count = text.count(old) + if count != 1: + raise SystemExit(f"expected one ZDR selection block; found {count}") + path.write_text(text.replace(old, new, 1), encoding="utf-8") + + +if __name__ == "__main__": + main() From 95dff0ec466a53b81342ab1c63cd9826fcb655df Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 29 Aug 2026 03:49:51 -0700 Subject: [PATCH 20/28] chore: run bounded PR #909 ZDR classification repair --- .../workflows/repair-pr-909-zdr-selection.yml | 58 +++++++++++++++++++ 1 file changed, 58 insertions(+) create mode 100644 .github/workflows/repair-pr-909-zdr-selection.yml diff --git a/.github/workflows/repair-pr-909-zdr-selection.yml b/.github/workflows/repair-pr-909-zdr-selection.yml new file mode 100644 index 000000000..07dc8b860 --- /dev/null +++ b/.github/workflows/repair-pr-909-zdr-selection.yml @@ -0,0 +1,58 @@ +name: Repair PR 909 ZDR model classification + +on: + push: + branches: + - fix/batch-routing-owner-20260829 + +permissions: + contents: write + +concurrency: + group: repair-pr-909-zdr-selection + cancel-in-progress: false + +jobs: + repair: + runs-on: ubuntu-latest + timeout-minutes: 15 + steps: + - name: Check out exact PR head + uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 + with: + fetch-depth: 0 + + - name: Set up Python + uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 + with: + python-version: "3.12" + + - name: Set up uv + uses: astral-sh/setup-uv@d0cc045d04ccac9d8b7881df0226f9e82c39688e + with: + version: "0.12.5" + + - name: Apply, validate, and commit the classification repair + env: + EXPECTED_PARENT: 661360d2ae8c350ddd77b7f789b64997185f3245 + BRANCH: fix/batch-routing-owner-20260829 + shell: bash + run: | + set -euo pipefail + test "$(git branch --show-current)" = "$BRANCH" + test "$(git rev-parse HEAD^)" = "$EXPECTED_PARENT" + python scripts/ci/repair_pr_909_zdr_selection.py + rm -f \ + scripts/ci/repair_pr_909_zdr_selection.py \ + .github/workflows/repair-pr-909-zdr-selection.yml + git add -A + git diff --cached --check + uv run --locked --extra api --extra db --extra queue --group dev \ + python -m pytest -q \ + tests/test_cost_router.py::test_zdr_batch_rejects_an_explicit_non_zdr_configured_model \ + tests/test_cost_review_server.py::test_batch_routing_rejects_unknown_zdr_model_as_client_error \ + tests/test_cost_router_boundaries.py + git config user.name 'github-actions[bot]' + git config user.email '41898282+github-actions[bot]@users.noreply.github.com' + git commit -m 'fix: preserve ZDR batch model error taxonomy' + git push origin "HEAD:$BRANCH" From 2323095acb1bead6e88bbdfe4afb057993752fb3 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Sat, 29 Aug 2026 10:50:10 +0000 Subject: [PATCH 21/28] fix: preserve ZDR batch model error taxonomy --- .../workflows/repair-pr-909-zdr-selection.yml | 58 ------------------- contextual_orchestrator/cost_router.py | 13 ++++- scripts/ci/repair_pr_909_zdr_selection.py | 52 ----------------- 3 files changed, 12 insertions(+), 111 deletions(-) delete mode 100644 .github/workflows/repair-pr-909-zdr-selection.yml delete mode 100644 scripts/ci/repair_pr_909_zdr_selection.py diff --git a/.github/workflows/repair-pr-909-zdr-selection.yml b/.github/workflows/repair-pr-909-zdr-selection.yml deleted file mode 100644 index 07dc8b860..000000000 --- a/.github/workflows/repair-pr-909-zdr-selection.yml +++ /dev/null @@ -1,58 +0,0 @@ -name: Repair PR 909 ZDR model classification - -on: - push: - branches: - - fix/batch-routing-owner-20260829 - -permissions: - contents: write - -concurrency: - group: repair-pr-909-zdr-selection - cancel-in-progress: false - -jobs: - repair: - runs-on: ubuntu-latest - timeout-minutes: 15 - steps: - - name: Check out exact PR head - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 - with: - fetch-depth: 0 - - - name: Set up Python - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 - with: - python-version: "3.12" - - - name: Set up uv - uses: astral-sh/setup-uv@d0cc045d04ccac9d8b7881df0226f9e82c39688e - with: - version: "0.12.5" - - - name: Apply, validate, and commit the classification repair - env: - EXPECTED_PARENT: 661360d2ae8c350ddd77b7f789b64997185f3245 - BRANCH: fix/batch-routing-owner-20260829 - shell: bash - run: | - set -euo pipefail - test "$(git branch --show-current)" = "$BRANCH" - test "$(git rev-parse HEAD^)" = "$EXPECTED_PARENT" - python scripts/ci/repair_pr_909_zdr_selection.py - rm -f \ - scripts/ci/repair_pr_909_zdr_selection.py \ - .github/workflows/repair-pr-909-zdr-selection.yml - git add -A - git diff --cached --check - uv run --locked --extra api --extra db --extra queue --group dev \ - python -m pytest -q \ - tests/test_cost_router.py::test_zdr_batch_rejects_an_explicit_non_zdr_configured_model \ - tests/test_cost_review_server.py::test_batch_routing_rejects_unknown_zdr_model_as_client_error \ - tests/test_cost_router_boundaries.py - git config user.name 'github-actions[bot]' - git config user.email '41898282+github-actions[bot]@users.noreply.github.com' - git commit -m 'fix: preserve ZDR batch model error taxonomy' - git push origin "HEAD:$BRANCH" diff --git a/contextual_orchestrator/cost_router.py b/contextual_orchestrator/cost_router.py index 778b771a6..e387f7de0 100644 --- a/contextual_orchestrator/cost_router.py +++ b/contextual_orchestrator/cost_router.py @@ -675,7 +675,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( diff --git a/scripts/ci/repair_pr_909_zdr_selection.py b/scripts/ci/repair_pr_909_zdr_selection.py deleted file mode 100644 index a8ec30ecb..000000000 --- a/scripts/ci/repair_pr_909_zdr_selection.py +++ /dev/null @@ -1,52 +0,0 @@ -"""One-shot exact-head repair for PR #909 ZDR model error classification.""" - -from __future__ import annotations - -from pathlib import Path - - -def main() -> None: - """Distinguish an unknown model from a configured but ZDR-ineligible model.""" - path = Path("contextual_orchestrator/cost_router.py") - text = path.read_text(encoding="utf-8") - old = ''' with self.orchestrator.request_policy(request.zdr_only): - agent = self.orchestrator._requested_agent(request.model) - if agent is None: - text = self.orchestrator._latest_user_text(request.messages) - agent = self.orchestrator._select_agent( - text, - "worker", - free_only=request.model - == getattr(self.orchestrator, "FREE_MODEL", object()), - ) -''' - new = ''' with self.orchestrator.request_policy(request.zdr_only): - 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( - text, - "worker", - free_only=request.model - == getattr(self.orchestrator, "FREE_MODEL", object()), - ) -''' - count = text.count(old) - if count != 1: - raise SystemExit(f"expected one ZDR selection block; found {count}") - path.write_text(text.replace(old, new, 1), encoding="utf-8") - - -if __name__ == "__main__": - main() From 0cba6660f4d95c4ac4a3f1e69ecc4f36ec1bc477 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 29 Aug 2026 03:50:52 -0700 Subject: [PATCH 22/28] docs: clarify ZDR batch model error taxonomy --- CHANGELOG.d/batch-routing-owner-model-error.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.d/batch-routing-owner-model-error.md b/CHANGELOG.d/batch-routing-owner-model-error.md index d8d6a4cd1..4a8636567 100644 --- a/CHANGELOG.d/batch-routing-owner-model-error.md +++ b/CHANGELOG.d/batch-routing-owner-model-error.md @@ -1 +1 @@ -Bound HTTP-created batch routing jobs to the authenticated principal, preserved cross-owner not-found behavior, and report an unknown explicit ZDR model as the non-retryable `400 invalid_model` client error instead of a retryable backend outage. +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. From b87a75ce807593aa501e643a46906cc8a06b2411 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 29 Aug 2026 04:06:56 -0700 Subject: [PATCH 23/28] chore: add bounded PR #909 final contract repair --- scripts/ci/repair_pr_909_final_contracts.py | 240 ++++++++++++++++++++ 1 file changed, 240 insertions(+) create mode 100644 scripts/ci/repair_pr_909_final_contracts.py diff --git a/scripts/ci/repair_pr_909_final_contracts.py b/scripts/ci/repair_pr_909_final_contracts.py new file mode 100644 index 000000000..7290a5d15 --- /dev/null +++ b/scripts/ci/repair_pr_909_final_contracts.py @@ -0,0 +1,240 @@ +"""One-shot exact-head repair for PR #909's final review contracts.""" + +from __future__ import annotations + +from pathlib import Path + + +def replace_once(path: str, old: str, new: str) -> None: + """Replace one reviewed block or fail closed.""" + target = Path(path) + text = target.read_text(encoding="utf-8") + count = text.count(old) + if count != 1: + raise SystemExit(f"expected one replacement in {path}; found {count}") + target.write_text(text.replace(old, new, 1), encoding="utf-8") + + +def append_once(path: str, marker: str, addition: str) -> None: + """Append one regression test only when its marker is absent.""" + target = Path(path) + text = target.read_text(encoding="utf-8") + if marker in text: + raise SystemExit(f"{marker} already exists in {path}") + target.write_text(text + addition, encoding="utf-8") + + +def main() -> None: + """Apply exact model, OpenAPI, and already-started SSE error boundaries.""" + replace_once( + "contextual_orchestrator/cost_router.py", + '''class BatchModelSelectionError(RuntimeError): + """Raised when a batch request has no eligible model-group member.""" + + +class CostRoutingCoordinator: +''', + '''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: +''', + ) + replace_once( + "contextual_orchestrator/cost_router.py", + ''' try: + prepared_requests = [self._resolve_batch_request(request) for request in requests] + except ValueError: + raise + except RuntimeError as exc: +''', + ''' try: + prepared_requests = [self._resolve_batch_request(request) for request in requests] + except ValueError as exc: + raise InvalidBatchModelError(str(exc)) from exc + except RuntimeError as exc: +''', + ) + replace_once( + "contextual_orchestrator/server.py", + '''from .cost_router import BatchModelSelectionError, CostRoutingCoordinator +''', + '''from .cost_router import ( + BatchModelSelectionError, + CostRoutingCoordinator, + InvalidBatchModelError, +) +''', + ) + replace_once( + "contextual_orchestrator/server.py", + ''' except ValueError as exc: + raise RequestError(400, "invalid_model", str(exc)) from exc +''', + ''' except InvalidBatchModelError as exc: + raise RequestError(400, "invalid_model", str(exc)) from exc +''', + ) + replace_once( + "contextual_orchestrator/api_contract.py", + ''' "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" + }, + }, +''', + ) + replace_once( + "contextual_orchestrator/api_contract.py", + ''' "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" + }, + }, +''', + ) + replace_once( + "contextual_orchestrator/server.py", + ''' if coordinator is not None: + result = { + **result, + **coordinator.record_stream_usage( + result=result, + attribution=attribution, + model_name=model_name, + ), + } +''', + ''' if coordinator is not None: + 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} +''', + ) + + append_once( + "tests/test_cost_router_boundaries.py", + "test_batch_model_identity_error_does_not_capture_backend_value_errors", + ''' + + +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, + ) + ]) +''', + ) + append_once( + "tests/test_api_contract.py", + "test_batch_job_openapi_documents_principal_hiding_404s", + ''' + + +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"] +''', + ) + append_once( + "tests/test_orchestrated_responses_stream.py", + "test_stream_usage_failure_remains_inside_the_started_sse_protocol", + ''' + + +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]") +''', + ) + + +if __name__ == "__main__": + main() From 42fd380c168aaf95af18d23c6e23eb735ce3f890 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 29 Aug 2026 04:07:16 -0700 Subject: [PATCH 24/28] chore: run bounded PR #909 final contract repair --- .../repair-pr-909-final-contracts.yml | 60 +++++++++++++++++++ 1 file changed, 60 insertions(+) create mode 100644 .github/workflows/repair-pr-909-final-contracts.yml diff --git a/.github/workflows/repair-pr-909-final-contracts.yml b/.github/workflows/repair-pr-909-final-contracts.yml new file mode 100644 index 000000000..f08fc57f1 --- /dev/null +++ b/.github/workflows/repair-pr-909-final-contracts.yml @@ -0,0 +1,60 @@ +name: Repair PR 909 final contracts + +on: + push: + branches: + - fix/batch-routing-owner-20260829 + +permissions: + contents: write + +concurrency: + group: repair-pr-909-final-contracts + cancel-in-progress: false + +jobs: + repair: + runs-on: ubuntu-latest + timeout-minutes: 20 + steps: + - name: Check out exact PR head + uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 + with: + fetch-depth: 0 + + - name: Set up Python + uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 + with: + python-version: "3.12" + + - name: Set up uv + uses: astral-sh/setup-uv@d0cc045d04ccac9d8b7881df0226f9e82c39688e + with: + version: "0.12.5" + + - name: Apply, validate, and commit the final review contracts + env: + EXPECTED_PARENT: b87a75ce807593aa501e643a46906cc8a06b2411 + BRANCH: fix/batch-routing-owner-20260829 + shell: bash + run: | + set -euo pipefail + test "$(git branch --show-current)" = "$BRANCH" + test "$(git rev-parse HEAD^)" = "$EXPECTED_PARENT" + python scripts/ci/repair_pr_909_final_contracts.py + rm -f \ + scripts/ci/repair_pr_909_final_contracts.py \ + .github/workflows/repair-pr-909-final-contracts.yml + git add -A + git diff --cached --check + uv run --locked --extra api --extra db --extra queue --group dev \ + python -m pytest -q \ + tests/test_cost_router_boundaries.py::test_batch_model_identity_error_does_not_capture_backend_value_errors \ + tests/test_cost_review_server.py::test_batch_routing_rejects_unknown_zdr_model_as_client_error \ + tests/test_cost_router.py::test_zdr_batch_rejects_an_explicit_non_zdr_configured_model \ + tests/test_api_contract.py::test_batch_job_openapi_documents_principal_hiding_404s \ + tests/test_orchestrated_responses_stream.py::test_stream_usage_failure_remains_inside_the_started_sse_protocol + git config user.name 'github-actions[bot]' + git config user.email '41898282+github-actions[bot]@users.noreply.github.com' + git commit -m 'fix: preserve final batch and SSE protocol boundaries' + git push origin "HEAD:$BRANCH" From 2283f4c92dbed1eb6ff1570ef1fb30c10e5faefa Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 29 Aug 2026 04:10:58 -0700 Subject: [PATCH 25/28] chore: scope PR #909 server replacement to batch submission --- scripts/ci/repair_pr_909_final_contracts.py | 22 +++++++++++++++++++-- 1 file changed, 20 insertions(+), 2 deletions(-) diff --git a/scripts/ci/repair_pr_909_final_contracts.py b/scripts/ci/repair_pr_909_final_contracts.py index 7290a5d15..b996edd9c 100644 --- a/scripts/ci/repair_pr_909_final_contracts.py +++ b/scripts/ci/repair_pr_909_final_contracts.py @@ -73,11 +73,29 @@ class CostRoutingCoordinator: ) replace_once( "contextual_orchestrator/server.py", - ''' except ValueError as exc: + ''' try: + job = self._run( + lambda: coordinator.submit_batch( + batch_requests, + metadata=metadata, + owner_id=security.principal_id(self.headers), + ) + ) + except ValueError as exc: raise RequestError(400, "invalid_model", str(exc)) from exc + orchestrator.record_analytics_event( ''', - ''' except InvalidBatchModelError as exc: + ''' 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( ''', ) replace_once( From 6d0ecba8e5ea63186df67eb5d17f96ec49ed9596 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 29 Aug 2026 04:11:19 -0700 Subject: [PATCH 26/28] chore: rerun scoped PR #909 final contract repair --- .github/workflows/repair-pr-909-final-contracts.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/repair-pr-909-final-contracts.yml b/.github/workflows/repair-pr-909-final-contracts.yml index f08fc57f1..97a750473 100644 --- a/.github/workflows/repair-pr-909-final-contracts.yml +++ b/.github/workflows/repair-pr-909-final-contracts.yml @@ -34,7 +34,7 @@ jobs: - name: Apply, validate, and commit the final review contracts env: - EXPECTED_PARENT: b87a75ce807593aa501e643a46906cc8a06b2411 + EXPECTED_PARENT: 2283f4c92dbed1eb6ff1570ef1fb30c10e5faefa BRANCH: fix/batch-routing-owner-20260829 shell: bash run: | From 42959f8b54cef1ba85ed131622398cf6bb299844 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Sat, 29 Aug 2026 11:20:22 +0000 Subject: [PATCH 27/28] fix: preserve final batch and SSE protocol boundaries --- .../repair-pr-909-final-contracts.yml | 60 ---- contextual_orchestrator/api_contract.py | 14 +- contextual_orchestrator/cost_router.py | 8 +- contextual_orchestrator/server.py | 29 +- scripts/ci/repair_pr_909_final_contracts.py | 258 ------------------ tests/test_api_contract.py | 14 + tests/test_cost_router_boundaries.py | 30 ++ tests/test_orchestrated_responses_stream.py | 36 +++ 8 files changed, 120 insertions(+), 329 deletions(-) delete mode 100644 .github/workflows/repair-pr-909-final-contracts.yml delete mode 100644 scripts/ci/repair_pr_909_final_contracts.py diff --git a/.github/workflows/repair-pr-909-final-contracts.yml b/.github/workflows/repair-pr-909-final-contracts.yml deleted file mode 100644 index 97a750473..000000000 --- a/.github/workflows/repair-pr-909-final-contracts.yml +++ /dev/null @@ -1,60 +0,0 @@ -name: Repair PR 909 final contracts - -on: - push: - branches: - - fix/batch-routing-owner-20260829 - -permissions: - contents: write - -concurrency: - group: repair-pr-909-final-contracts - cancel-in-progress: false - -jobs: - repair: - runs-on: ubuntu-latest - timeout-minutes: 20 - steps: - - name: Check out exact PR head - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 - with: - fetch-depth: 0 - - - name: Set up Python - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 - with: - python-version: "3.12" - - - name: Set up uv - uses: astral-sh/setup-uv@d0cc045d04ccac9d8b7881df0226f9e82c39688e - with: - version: "0.12.5" - - - name: Apply, validate, and commit the final review contracts - env: - EXPECTED_PARENT: 2283f4c92dbed1eb6ff1570ef1fb30c10e5faefa - BRANCH: fix/batch-routing-owner-20260829 - shell: bash - run: | - set -euo pipefail - test "$(git branch --show-current)" = "$BRANCH" - test "$(git rev-parse HEAD^)" = "$EXPECTED_PARENT" - python scripts/ci/repair_pr_909_final_contracts.py - rm -f \ - scripts/ci/repair_pr_909_final_contracts.py \ - .github/workflows/repair-pr-909-final-contracts.yml - git add -A - git diff --cached --check - uv run --locked --extra api --extra db --extra queue --group dev \ - python -m pytest -q \ - tests/test_cost_router_boundaries.py::test_batch_model_identity_error_does_not_capture_backend_value_errors \ - tests/test_cost_review_server.py::test_batch_routing_rejects_unknown_zdr_model_as_client_error \ - tests/test_cost_router.py::test_zdr_batch_rejects_an_explicit_non_zdr_configured_model \ - tests/test_api_contract.py::test_batch_job_openapi_documents_principal_hiding_404s \ - tests/test_orchestrated_responses_stream.py::test_stream_usage_failure_remains_inside_the_started_sse_protocol - git config user.name 'github-actions[bot]' - git config user.email '41898282+github-actions[bot]@users.noreply.github.com' - git commit -m 'fix: preserve final batch and SSE protocol boundaries' - git push origin "HEAD:$BRANCH" diff --git a/contextual_orchestrator/api_contract.py b/contextual_orchestrator/api_contract.py index 52eb2de7e..0b3d76f79 100644 --- a/contextual_orchestrator/api_contract.py +++ b/contextual_orchestrator/api_contract.py @@ -892,7 +892,12 @@ "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": { @@ -903,7 +908,12 @@ "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/cost_router.py b/contextual_orchestrator/cost_router.py index e387f7de0..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``.""" @@ -659,8 +663,8 @@ def submit_batch( """Submit a batch, resolve its targets, and bind its authenticated owner.""" try: prepared_requests = [self._resolve_batch_request(request) for request in requests] - except ValueError: - raise + 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" diff --git a/contextual_orchestrator/server.py b/contextual_orchestrator/server.py index 96fa66a2c..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, @@ -6984,7 +6988,7 @@ def register_video_job(agent: ModelAgent, provider_result: dict[str, Any]) -> di owner_id=security.principal_id(self.headers), ) ) - except ValueError as exc: + except InvalidBatchModelError as exc: raise RequestError(400, "invalid_model", str(exc)) from exc orchestrator.record_analytics_event( "batch_routing_job_created", @@ -7926,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/scripts/ci/repair_pr_909_final_contracts.py b/scripts/ci/repair_pr_909_final_contracts.py deleted file mode 100644 index b996edd9c..000000000 --- a/scripts/ci/repair_pr_909_final_contracts.py +++ /dev/null @@ -1,258 +0,0 @@ -"""One-shot exact-head repair for PR #909's final review contracts.""" - -from __future__ import annotations - -from pathlib import Path - - -def replace_once(path: str, old: str, new: str) -> None: - """Replace one reviewed block or fail closed.""" - target = Path(path) - text = target.read_text(encoding="utf-8") - count = text.count(old) - if count != 1: - raise SystemExit(f"expected one replacement in {path}; found {count}") - target.write_text(text.replace(old, new, 1), encoding="utf-8") - - -def append_once(path: str, marker: str, addition: str) -> None: - """Append one regression test only when its marker is absent.""" - target = Path(path) - text = target.read_text(encoding="utf-8") - if marker in text: - raise SystemExit(f"{marker} already exists in {path}") - target.write_text(text + addition, encoding="utf-8") - - -def main() -> None: - """Apply exact model, OpenAPI, and already-started SSE error boundaries.""" - replace_once( - "contextual_orchestrator/cost_router.py", - '''class BatchModelSelectionError(RuntimeError): - """Raised when a batch request has no eligible model-group member.""" - - -class CostRoutingCoordinator: -''', - '''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: -''', - ) - replace_once( - "contextual_orchestrator/cost_router.py", - ''' try: - prepared_requests = [self._resolve_batch_request(request) for request in requests] - except ValueError: - raise - except RuntimeError as exc: -''', - ''' try: - prepared_requests = [self._resolve_batch_request(request) for request in requests] - except ValueError as exc: - raise InvalidBatchModelError(str(exc)) from exc - except RuntimeError as exc: -''', - ) - replace_once( - "contextual_orchestrator/server.py", - '''from .cost_router import BatchModelSelectionError, CostRoutingCoordinator -''', - '''from .cost_router import ( - BatchModelSelectionError, - CostRoutingCoordinator, - InvalidBatchModelError, -) -''', - ) - replace_once( - "contextual_orchestrator/server.py", - ''' try: - job = self._run( - lambda: coordinator.submit_batch( - batch_requests, - metadata=metadata, - owner_id=security.principal_id(self.headers), - ) - ) - except ValueError as exc: - raise RequestError(400, "invalid_model", str(exc)) from exc - orchestrator.record_analytics_event( -''', - ''' 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( -''', - ) - replace_once( - "contextual_orchestrator/api_contract.py", - ''' "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" - }, - }, -''', - ) - replace_once( - "contextual_orchestrator/api_contract.py", - ''' "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" - }, - }, -''', - ) - replace_once( - "contextual_orchestrator/server.py", - ''' if coordinator is not None: - result = { - **result, - **coordinator.record_stream_usage( - result=result, - attribution=attribution, - model_name=model_name, - ), - } -''', - ''' if coordinator is not None: - 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} -''', - ) - - append_once( - "tests/test_cost_router_boundaries.py", - "test_batch_model_identity_error_does_not_capture_backend_value_errors", - ''' - - -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, - ) - ]) -''', - ) - append_once( - "tests/test_api_contract.py", - "test_batch_job_openapi_documents_principal_hiding_404s", - ''' - - -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"] -''', - ) - append_once( - "tests/test_orchestrated_responses_stream.py", - "test_stream_usage_failure_remains_inside_the_started_sse_protocol", - ''' - - -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]") -''', - ) - - -if __name__ == "__main__": - main() diff --git a/tests/test_api_contract.py b/tests/test_api_contract.py index 09b60a5f4..f7d4762a0 100644 --- a/tests/test_api_contract.py +++ b/tests/test_api_contract.py @@ -110,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_router_boundaries.py b/tests/test_cost_router_boundaries.py index 5b43950e9..d09d75ab5 100644 --- a/tests/test_cost_router_boundaries.py +++ b/tests/test_cost_router_boundaries.py @@ -443,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]") From 35d9f54ff84653bebf9e3f8baac312b41462c105 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 29 Aug 2026 04:22:53 -0700 Subject: [PATCH 28/28] chore: run validated semantic resolver for PR #909 --- .github/workflows/resolve-pr-909-v2.yml | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) create mode 100644 .github/workflows/resolve-pr-909-v2.yml 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