From fea207fc459be6c11b09ee399e4102452b23d23f Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sun, 20 Sep 2026 15:47:41 +0900 Subject: [PATCH] fix(routing): restore request receipts and consume proxy responses --- contextual_orchestrator/orchestrator.py | 193 +++++++++++++++++- .../doctoring/http_test_resource_lifecycle.md | 35 ++++ tests/test_api_contract.py | 4 + tests/test_http_resource_lifecycle.py | 39 ++++ tests/test_psychometric_routing.py | 2 +- tests/test_rate_limit_aware_admission.py | 11 +- 6 files changed, 273 insertions(+), 11 deletions(-) diff --git a/contextual_orchestrator/orchestrator.py b/contextual_orchestrator/orchestrator.py index 08285ac9f..8bb78f25b 100644 --- a/contextual_orchestrator/orchestrator.py +++ b/contextual_orchestrator/orchestrator.py @@ -104,6 +104,46 @@ ) +_REQUEST_EXECUTION_SNAPSHOT: ContextVar[ + tuple[object, object, OrchestrationPolicy] | None +] = ContextVar( + "contextual_orchestrator_request_execution_snapshot", default=None +) +_REQUEST_SELECTION_ATTEMPTS: ContextVar[list | None] = ContextVar( + "contextual_orchestrator_request_selection_attempts", default=None +) + + +def _request_execution_scoped(method: Callable) -> Callable: + """Share policy and effort across nested calls without leaking suspended streams.""" + if inspect.isgeneratorfunction(method): + @wraps(method) + def scoped_stream(self, *args, **kwargs): + """Advance and close a stream in its own captured request context.""" + with self._request_execution_scope(): + context = copy_context() + stream = method(self, *args, **kwargs) + exhausted = object() + try: + while True: + item = context.run(next, stream, exhausted) + if item is exhausted: + return + yield item + finally: + context.run(stream.close) + + return scoped_stream + + @wraps(method) + def scoped_call(self, *args, **kwargs): + """Restore the caller's context after a synchronous result or exception.""" + with self._request_execution_scope(): + return method(self, *args, **kwargs) + + return scoped_call + + _REQUEST_ENDPOINT_AGENT_IDS: ContextVar[frozenset[str] | None] = ContextVar( "contextual_orchestrator_request_endpoint_agent_ids", default=None ) @@ -6091,6 +6131,7 @@ def _reload_state(self) -> None: } ) + @_request_execution_scoped def proxy_completion( self, body: dict[str, Any], @@ -6423,6 +6464,7 @@ def proxy_completion( effort_profile, api_surface=api_surface, ) + consumed_response = None try: send_once = getattr(self.client, "proxy_send_once", None) if not callable(send_once): @@ -6430,6 +6472,8 @@ def proxy_completion( record_initial_selection([candidate.id], "automatic_proxy") result = send_once(candidate, endpoint, candidate_payload) except Exception as exc: # noqa: BLE001 - provider trust boundary + if isinstance(exc, urllib.error.HTTPError): + consumed_response = exc classified = classify_provider_failure( exc, agent_id=candidate.id, @@ -6563,6 +6607,12 @@ def proxy_completion( if candidate.group_name and not skip_breaker: self._group_router.observe_failure(candidate.id) continue + finally: + if consumed_response is not None: + try: + consumed_response.close() + except Exception: + pass # Preserve classification and failover on cleanup failure. self._record_success(candidate.id) if candidate.group_name: self._group_router.observe_success( @@ -7762,6 +7812,7 @@ def _requested_agent(self, requested_model: Any) -> ModelAgent | None: raise ValueError(f"requested model {requested_model!r} is not configured") return next((candidate for candidate in matches if not candidate.disabled), matches[0]) + @_request_execution_scoped def complete( self, messages: list[ChatMessage], @@ -7891,6 +7942,7 @@ def would_route( text = self._latest_user_text(messages) return not self._needs_workflow(text) + @_request_execution_scoped def stream_route( self, messages: list[ChatMessage], @@ -8083,6 +8135,10 @@ def stream_route( trace_step["usage"] = usage if isinstance(output_budget, dict): trace_step.update(output_budget) + attempted = list(_REQUEST_SELECTION_ATTEMPTS.get() or [agent]) + trace_step["selection_design"] = self._selection_design_receipt( + [agent], attempted, agent + ) record = self._with_effort_snapshot( { "workflow_run_id": workflow_run_id or f"run_{uuid.uuid4().hex}", @@ -8134,6 +8190,13 @@ def _cache_key( "max_output_tokens": getattr(self.client, "max_output_tokens", None), } parameters = {**parameters, "zdr_only": _REQUEST_ZDR_ONLY.get()} + effort_snapshot = self._effort_snapshot() + parameters["effort_snapshot_hash"] = ( + None if effort_snapshot is None else effort_snapshot.snapshot_hash + ) + parameters["policy_snapshot_hash"] = hashlib.sha256( + json.dumps(self.policy.as_dict(), sort_keys=True, separators=(",", ":")).encode("utf-8") + ).hexdigest() if resolved_mode is not None: parameters["resolved_mode"] = resolved_mode endpoint_partition = _request_endpoint_partition() @@ -8150,6 +8213,7 @@ def _cache_key( partition=cache_partition, ) + @_request_execution_scoped def run( self, messages: list[ChatMessage], @@ -8299,6 +8363,7 @@ def _trace_budget_spend( ) return output_tokens, round(output_cost, 6) + @_request_execution_scoped def batch_route(self, prompts: list[str]) -> list[dict[str, Any]]: """Route many prompts through the provider's Batch API and persist each run. @@ -8534,6 +8599,8 @@ def _persist_pending_batch_row( "latency_ms": batch_latency_ms, "subtask": "Direct route (batched)", "access": [], "output": result["content"], } + attempted = list(_REQUEST_SELECTION_ATTEMPTS.get() or [agent]) + row["selection_design"] = self._selection_design_receipt([agent], attempted, agent) if result.get("usage") is not None: row["usage"] = result["usage"] run_id = f"run_{uuid.uuid4().hex}" @@ -9271,6 +9338,7 @@ def _last_route_evidence(self, value: dict[str, Any] | None) -> None: """Store this thread's pending route attempt evidence.""" self._route_evidence_local.value = value + @_request_execution_scoped def route_once( self, messages: list[ChatMessage], @@ -9350,6 +9418,7 @@ def route_once( break tried_ids.add(candidate.id) start = time.perf_counter() + selection_start = len(_REQUEST_SELECTION_ATTEMPTS.get() or ()) attempt_answer, attempt_served_id, attempt_served_model, attempt_usage = ( self._invoke_with_rate_limit_recovery( candidate, @@ -9361,6 +9430,7 @@ def route_once( prompt_token_lower_bound=prompt_bound, ) ) + attempted = list((_REQUEST_SELECTION_ATTEMPTS.get() or [candidate])[selection_start:]) current_route_evidence = self._last_route_evidence self._last_route_evidence = None if isinstance(current_route_evidence, dict): @@ -9453,6 +9523,13 @@ def route_once( "accepted": verification["accepted"], "reason": verification["reason"], } + served = next( + (item for item in ranked_pool if item.id == attempt_served_id), + candidate, + ) + row["selection_design"] = self._selection_design_receipt( + ranked_pool, attempted, served + ) trace_rows.append(row) if verification["accepted"]: break @@ -9578,6 +9655,7 @@ def _usage_total_tokens(usage: dict[str, Any] | None) -> int | None: return None return tokens + @_request_execution_scoped def conduct( self, messages: list[ChatMessage], @@ -9768,6 +9846,8 @@ def conduct( if served_id != agent.id: # pragma: no cover row["served_agent_id"] = served_id row["failover_from"] = agent.id + attempted = list(_REQUEST_SELECTION_ATTEMPTS.get() or [agent]) + row["selection_design"] = self._selection_design_receipt([agent], attempted, agent) trace.append(row) if progress is not None: _notify_progress(progress, step.role, "completed", redact_value(output)) @@ -9905,25 +9985,120 @@ def _require_role_effort_pool( ) def _role_effort_profile(self, role: str) -> ReasoningEffortProfile | None: - """Return the opt-in profile bound to one workflow role.""" - if self.role_effort_catalog is None: + """Use the active request revision; preserve standalone single-role adapters.""" + active = _REQUEST_EXECUTION_SNAPSHOT.get() + if active is None or active[0] is not self: + catalog = self.role_effort_catalog + return catalog.get(role) if catalog is not None else None + snapshot = active[1] + if snapshot is None: return None - return self.role_effort_catalog.get(role) + profile = snapshot.role_profiles.get(role) + return ReasoningEffortProfile(**profile) if profile is not None else None + + @property + def policy(self) -> OrchestrationPolicy: + """Read the active request's policy, or the configured policy between requests.""" + active = _REQUEST_EXECUTION_SNAPSHOT.get() + return active[2] if active is not None and active[0] is self else self._configured_policy + + @policy.setter + def policy(self, policy: OrchestrationPolicy) -> None: + """Publish policy for later requests without changing work already in progress.""" + self._configured_policy = policy + + @contextmanager + def _request_execution_scope(self): + """Capture policy and validate effort once; preserve nested calls and other instances.""" + active = _REQUEST_EXECUTION_SNAPSHOT.get() + if active is not None and active[0] is self: + yield + return + policy = self._configured_policy + attempts: list = [] + snapshot_token = _REQUEST_EXECUTION_SNAPSHOT.set((self, self._effort_snapshot(), policy)) + attempt_token = _REQUEST_SELECTION_ATTEMPTS.set(attempts) + try: + yield + finally: + _REQUEST_SELECTION_ATTEMPTS.reset(attempt_token) + _REQUEST_EXECUTION_SNAPSHOT.reset(snapshot_token) + + def _effort_snapshot(self): + """Read this request's catalog, or validate a fresh standalone-operation copy.""" + active = _REQUEST_EXECUTION_SNAPSHOT.get() + if active is not None and active[0] is self: + return active[1] + catalog = self.role_effort_catalog + return snapshot_role_effort_catalog(dict(catalog)) if catalog is not None else None + + def _note_selection_attempt(self, agent: ModelAgent) -> None: + """Record one real provider call for this request's selection receipt.""" + attempts = _REQUEST_SELECTION_ATTEMPTS.get() + if attempts is not None: + attempts.append(agent) + + def _psychometric_candidate_id(self, agent: ModelAgent) -> str: + """Bind routing evidence to the declared deployment and decode policy.""" + return self._psychometric_candidate_ids((agent,))[0] + + def _psychometric_candidate_ids(self, agents: Iterable[ModelAgent]) -> list[str]: + """Bind an ordered batch to the request's validated decode-policy snapshot.""" + agents = list(agents) + if not agents: + return [] + effort_snapshot = self._effort_snapshot() + effort_catalog = effort_snapshot.snapshot_hash if effort_snapshot is not None else None + candidate_ids = [] + for agent in agents: + configuration = json.dumps( + {"agent": agent.to_config(), "role_effort_catalog": effort_catalog}, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + ) + revision = hashlib.sha256(configuration.encode("utf-8")).hexdigest() + candidate_ids.append(f"{agent.id}:{revision}") + return candidate_ids + + def _selection_design_receipt( + self, + candidates: Iterable[ModelAgent], + attempted: Iterable[ModelAgent], + selected: ModelAgent, + ) -> dict[str, Any]: + """Describe the observed deterministic assignment without inventing propensity.""" + policy = json.dumps( + self.policy.as_dict(), sort_keys=True, separators=(",", ":") + ).encode("utf-8") + candidates = list(candidates) + attempted = list(attempted) + deployment_ids = self._psychometric_candidate_ids([*candidates, *attempted, selected]) + return { + "assignment_mechanism": "deterministic_ranked", + "propensity_status": "not_identified", + "selected_probability": None, + "policy_snapshot_hash": hashlib.sha256(policy).hexdigest(), + "candidate_deployment_ids": deployment_ids[:len(candidates)], + "attempted_deployment_ids": deployment_ids[len(candidates):-1], + "selected_deployment_id": deployment_ids[-1], + } def _with_effort_snapshot(self, result: dict[str, Any]) -> dict[str, Any]: - """Attach a replayable role-effort snapshot when the operator opted in. + """Attach effective policy and an opt-in replayable role-effort snapshot. Buyer next action: compare ``reasoning_effort_snapshot.snapshot_hash`` on ``complete``, ``run``, ``stream_route``, and ``batch_route``. Omit - the constructor catalog to keep today's payload. + the constructor catalog to omit role-effort metadata. """ - if self.role_effort_catalog is None: + result["policy_snapshot"] = self.policy.as_dict() + snapshot = self._effort_snapshot() + if snapshot is None: return result - snapshot = snapshot_role_effort_catalog(self.role_effort_catalog) result["reasoning_effort_snapshot"] = { "profile_version": snapshot.profile_version, "snapshot_hash": snapshot.snapshot_hash, - "role_profiles": snapshot.role_profiles, + "role_profiles": copy.deepcopy(snapshot.role_profiles), } return result @@ -11077,6 +11252,7 @@ def call( with self.client.request_settings(**request_settings), tool_scope: record_initial_selection([member.id for member in race_members], "text_race", attempt_id=decision_attempt_id) + self._note_selection_attempt(agent) output = ( self.client.chat(agent, messages, effort_profile=effort_profile) if effort_profile is not None @@ -11183,6 +11359,7 @@ def attempt(agent: ModelAgent) -> EndpointAttempt[Any]: ) with transport_scope, tool_scope: record_initial_selection([agent.id], "invocation_" + role) + self._note_selection_attempt(agent) output = ( self.client.chat(agent, messages, effort_profile=effort_profile) if effort_profile is not None diff --git a/docs/doctoring/http_test_resource_lifecycle.md b/docs/doctoring/http_test_resource_lifecycle.md index 11c881a7c..06122b9cc 100644 --- a/docs/doctoring/http_test_resource_lifecycle.md +++ b/docs/doctoring/http_test_resource_lifecycle.md @@ -48,6 +48,41 @@ removed because concrete Responses streaming is rejected as invalid_stream and Chat streaming uses SSE rather than the assumed non-streaming JSON contract; no streaming fix or acceptance is claimed. +## Request receipt and virtual-response follow-up, 2026-09-20 + +At #1212 `898a7cb97fac7b10d35e7ca6e5e0d2484d3a29cf`, the remaining +selection_design KeyError comes from dropped implementation: the tests remain, +but request snapshot scoping, selection attempt collection, and receipt +construction are absent. Reuse the request/receipt hunks of existing #1088 +commit `e569cefa70079a64734ac0f5e47965f9efec3f3d` and its deployment-hash +fixture update. Do not copy its dependency, ranking/prior, or persistence changes. +Worker receipts capture attempts before the nested judge and exclude earlier +worker rounds. A strengthened API regression checks that the judge is absent. + +An expanded strict run passed 175 test bodies but exited 1 during final cleanup: +five raw 429 responses leaked from the virtual proxy caller of proxy_send_once. +This is production ownership, distinct from #1211's test-owned classifiers. +The virtual loop converts those raw responses to typed errors; it now closes +in finally after classification/cooldown diagnostics, before failover or return. +Cleanup Exception does not replace the original outcome; BaseException remains +unmasked. Two direct closure assertions fail on the unchanged #1212 source and +pass after repair, including an injected cleanup OSError. Together with the +original KeyError, this bounded RED is 3 failed, exit 1. + +A wait-round test used a 10ms real cooldown with a fake sleep. Under host pressure +the cooldown expired during execution and the test missed its intended branch. +It now advances a controlled monotonic clock with its sleep hook; production +cooldown policy is unchanged. + +Final source validation: 179 passed, 21.91s, process exit 0, under -W error across +rate-limit admission, provider reliability, HTTP resource lifecycle, API contract, +request policy/effort snapshots, and the existing stream receipt and race-failover +receipt cases. Four selection-receipt identity tests separately pass, exit 0. +The isolated environment needed binary-only numpy 2.5.3 and fast-mlsirm 0.11.3 +wheels to collect the latter module; no native build or numerical experiment ran. +These dependency versions do not establish locked-install, complete-suite, +hosted-gate, independent-review or protected-merge acceptance. + ## Trace HTTP fixture successor, 2026-09-13 Base: #1140 at `38c0603af2fd8fcb204f65be47081ada9d6bd35c`. diff --git a/tests/test_api_contract.py b/tests/test_api_contract.py index b00110a13..381f495cc 100644 --- a/tests/test_api_contract.py +++ b/tests/test_api_contract.py @@ -482,6 +482,10 @@ def nested_judge(**kwargs): # noqa: ANN003 - mirrors the owned judge boundary "primary_worker", "primary_worker", "fallback_worker", "primary_worker" ] assert [attempt["agent_id"] for attempt in result["route"]["attempted"]] == client.calls[:3] + assert [ + value.split(":", 1)[0] + for value in result["trace"][0]["selection_design"]["attempted_deployment_ids"] + ] == client.calls[:3] assert [attempt["outcome"] for attempt in result["route"]["attempted"]] == [ "retryable_transport", "retryable_transport", diff --git a/tests/test_http_resource_lifecycle.py b/tests/test_http_resource_lifecycle.py index 68a2e8e13..738926eb6 100644 --- a/tests/test_http_resource_lifecycle.py +++ b/tests/test_http_resource_lifecycle.py @@ -253,3 +253,42 @@ def raise_response_error(*args, **kwargs): assert "private cleanup diagnostic" not in str(captured.value) finally: original_close() + + +@pytest.mark.parametrize("cleanup_fails", [False, True]) +def test_virtual_proxy_closes_consumed_error_before_failover(monkeypatch, cleanup_fails): + """The virtual caller owns raw responses transferred by proxy_send_once.""" + from test_rate_limit_aware_admission import SequencedRateLimitClient, _two_agent_pool + + response = urllib.error.HTTPError( + "https://provider.example/v1", 429, "rate limited", {"Retry-After": "30"}, io.BytesIO(b"{}") + ) + original_close = response.close + if cleanup_fails: + def failing_close(): + original_close() + raise OSError("cleanup failed") + monkeypatch.setattr(response, "close", failing_close) + client = SequencedRateLimitClient({ + "primary_agent": [response], "fallback_agent": [{"id": "served", "choices": []}], + }) + send_once = client.proxy_send_once + + def checked_send(agent, endpoint, payload): + if agent.id == "fallback_agent": + assert response.closed + return send_once(agent, endpoint, payload) + + client.proxy_send_once = checked_send + orchestrator = _two_agent_pool(client) + try: + result = orchestrator.proxy_completion({ + "model": "orchestrator/auto", "messages": [{"role": "user", "content": "hello"}], + }) + assert result["id"] == "served" + assert response.closed + assert client.calls == ["primary_agent", "fallback_agent"] + assert orchestrator._circuit == {} + finally: + original_close() + orchestrator.close() diff --git a/tests/test_psychometric_routing.py b/tests/test_psychometric_routing.py index 36c95a2bb..e2f7e9604 100644 --- a/tests/test_psychometric_routing.py +++ b/tests/test_psychometric_routing.py @@ -1301,7 +1301,7 @@ def attempted_agents(): selected_id = receipt["selected_deployment_id"] assert selected_id == ( "audit_candidate:" - "f30639fe9ae8729e57eb659445325038a92c6c6e8d6810b8f84b36f3bce46bb9" + "91dad5c97d95337b359e020fd9d2636cfaeebb0257622cc5490620ed7cf5b13e" ) assert receipt["candidate_deployment_ids"] == [selected_id] assert receipt["attempted_deployment_ids"] == [selected_id] diff --git a/tests/test_rate_limit_aware_admission.py b/tests/test_rate_limit_aware_admission.py index 0081c47ba..5813f2389 100644 --- a/tests/test_rate_limit_aware_admission.py +++ b/tests/test_rate_limit_aware_admission.py @@ -699,13 +699,20 @@ def test_http_fail_closed_after_503_keeps_route_and_stops() -> None: assert "must-not-leak" not in json.dumps(body) -def test_route_once_fail_closed_keeps_attempts_from_the_wait_round() -> None: +def test_route_once_fail_closed_keeps_attempts_from_the_wait_round(monkeypatch) -> None: """A tool stop after a 503 storm still carries the waited round's attempts.""" orchestrator = TaskOrchestrator( _free_route_agents(), tool_retry_attempts=0, rate_limit_wait_seconds=5.0 ) slept: list[float] = [] - orchestrator._rate_limit_sleep = slept.append + now = [100.0] + monkeypatch.setattr(time, "monotonic", lambda: now[0]) + + def advance_clock(seconds: float) -> None: + slept.append(seconds) + now[0] += seconds + + orchestrator._rate_limit_sleep = advance_clock chat_outcomes = QueuedChatOutcomes( { "primary_free_agent": [