fix(routing): restore request receipts and close consumed proxy responses

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,13 +6464,16 @@ 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):
                         send_once = self.client.proxy_send
                     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

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`.