fix: orchestrate structured provider features

contextual_orchestrator/cost_ledger.py

@@ -330,6 +330,7 @@ class NoopUsageTelemetrySink:
     """Default sink for callers that do not wire telemetry yet."""
 
     def emit_usage(self, event: UsageTelemetryEvent) -> None:
+        """Discard one prompt-safe usage event."""
         return None
 
 
@@ -342,12 +343,14 @@ def __init__(self, max_events: int = 512) -> None:
         self._lock = threading.Lock()
 
     def emit_usage(self, event: UsageTelemetryEvent) -> None:
+        """Append one usage event while retaining only the configured limit."""
         with self._lock:
             self._events.append(event)
             if len(self._events) > self._max_events:
                 del self._events[: len(self._events) - self._max_events]
 
     def events(self) -> List[UsageTelemetryEvent]:
+        """Return a thread-safe snapshot of retained usage events."""
         with self._lock:
             return list(self._events)
 
@@ -363,6 +366,7 @@ class UsageTelemetryHealth:
     last_error_type: Optional[str] = None
 
     def as_dict(self) -> Dict[str, Any]:
+        """Return operator-safe counters as a serializable mapping."""
         return {
             "records_accepted": self.records_accepted,
             "records_stored": self.records_stored,
@@ -442,6 +446,7 @@ def flush(self, timeout: Optional[float] = None) -> bool:
         return True
 
     def telemetry_health(self) -> Dict[str, Any]:
+        """Return a thread-safe snapshot of persistence health counters."""
         with self._lock:
             return self._health.as_dict()
 
@@ -669,7 +674,7 @@ def query(self, start: Optional[int] = None, end: Optional[int] = None) -> List[
             _USAGE_QUERY_SQL[(self._paramstyle, start is not None, end is not None)],
             tuple(params),
         )
-        return [dict(zip(_USAGE_COLUMNS, values)) for values in cur.fetchall()]
+        return [dict(zip(_USAGE_COLUMNS, values, strict=True)) for values in cur.fetchall()]
 
 
 def _within_window(created_at: int, start: Optional[int], end: Optional[int]) -> bool:

contextual_orchestrator/server.py

@@ -35,7 +35,7 @@
     sse_stream_body,
 )
 
-# OpenAI request params forwarded verbatim to the provider on passthrough.
+# OpenAI request params forwarded to the final provider after orchestration.
 OPENAI_PASSTHROUGH_PARAM_KEYS = {
     "temperature", "top_p", "max_tokens", "max_completion_tokens", "n", "stop",
     "seed", "presence_penalty", "frequency_penalty", "logit_bias", "logprobs",
@@ -70,7 +70,7 @@
     "max_output_tokens",
     # Tool-loop budget — accepted only for explicit unsupported error (no multi-step tool loop).
     "max_tool_calls",
-    # Gateway cost/routing control plane (stripped before provider passthrough).
+    # Gateway cost/routing control plane (stripped before final provider transport).
     "attribution", "routing",
     # previous_response_id / conversation / truncation / include fail closed
     # with named unsupported errors. Official text.format is validated
@@ -3254,7 +3254,7 @@ def _validate_responses_store(body: dict[str, Any]) -> bool | None:
     """Responses API ``store`` — strict boolean; ``true`` is not supported.
 
     OpenAI may persist Responses when ``store=true``. This gateway's Responses
-    path is a single-agent passthrough without a persistence plane, so
+    path has no provider-response persistence plane, so
     ``store=true`` fails closed rather than silently dropping a buyer-visible
     storage control. ``store=false`` and omit remain valid.
     """
@@ -5123,24 +5123,13 @@ def do_POST(self) -> None:  # noqa: N802
                         attribution["service"] = "completions_api"
                     routing = _validate_routing(body.get("routing"))
                     started_at = time.perf_counter()
-                    # Apply request sampling knobs to the provider client for this call.
-                    model_client = orchestrator.client
-                    previous_max_tokens = model_client.max_output_tokens
-                    previous_temperature = model_client.default_temperature
-                    previous_top_p = model_client.default_top_p
-                    previous_presence = model_client.default_presence_penalty
-                    previous_frequency = model_client.default_frequency_penalty
-                    if max_tokens is not None:
-                        model_client.max_output_tokens = max_tokens
-                    if temperature is not None:
-                        model_client.default_temperature = temperature
-                    if top_p is not None:
-                        model_client.default_top_p = top_p
-                    if presence_penalty is not None:
-                        model_client.default_presence_penalty = presence_penalty
-                    if frequency_penalty is not None:
-                        model_client.default_frequency_penalty = frequency_penalty
-                    try:
+                    with orchestrator.client.request_settings(
+                        max_output_tokens=max_tokens,
+                        temperature=temperature,
+                        top_p=top_p,
+                        presence_penalty=presence_penalty,
+                        frequency_penalty=frequency_penalty,
+                    ):
                         result = self._run(lambda: coordinator.complete(
                             messages,
                             mode="route",
@@ -5149,12 +5138,6 @@ def do_POST(self) -> None:  # noqa: N802
                             model_name=model_name,
                             workflow_run_id=f"run_{uuid.uuid4().hex}",
                         ))
-                    finally:
-                        model_client.max_output_tokens = previous_max_tokens
-                        model_client.default_temperature = previous_temperature
-                        model_client.default_top_p = previous_top_p
-                        model_client.default_presence_penalty = previous_presence
-                        model_client.default_frequency_penalty = previous_frequency
                     # Batch-channel Completions return a job handle (202), not a
                     # text_completion body — match chat Completions honesty so
                     # clients never receive a 500 on a valid batch routing hint.
@@ -5325,7 +5308,9 @@ def do_POST(self) -> None:  # noqa: N802
                                 "tool-loop passthrough requires stream=false",
                             )
                         started_at = time.perf_counter()
-                        raw_response = self._run(lambda: orchestrator.proxy_completion(body))
+                        raw_response = self._run(
+                            lambda: orchestrator.proxy_completion(body, single_agent=True)
+                        )
                         orchestrator.record_analytics_event(
                             "chat_completion_tool_passthrough",
                             {
@@ -5391,23 +5376,13 @@ def do_POST(self) -> None:  # noqa: N802
                     if "metadata" in body:
                         _validate_openai_metadata(body)
                     started_at = time.perf_counter()
-                    model_client = orchestrator.client
-                    previous_max_tokens = model_client.max_output_tokens
-                    previous_temperature = model_client.default_temperature
-                    previous_top_p = model_client.default_top_p
-                    previous_presence = model_client.default_presence_penalty
-                    previous_frequency = model_client.default_frequency_penalty
-                    if max_tokens is not None:
-                        model_client.max_output_tokens = max_tokens
-                    if temperature is not None:
-                        model_client.default_temperature = temperature
-                    if top_p is not None:
-                        model_client.default_top_p = top_p
-                    if presence_penalty is not None:
-                        model_client.default_presence_penalty = presence_penalty
-                    if frequency_penalty is not None:
-                        model_client.default_frequency_penalty = frequency_penalty
-                    try:
+                    with orchestrator.client.request_settings(
+                        max_output_tokens=max_tokens,
+                        temperature=temperature,
+                        top_p=top_p,
+                        presence_penalty=presence_penalty,
+                        frequency_penalty=frequency_penalty,
+                    ):
                         if stream and orchestrator.would_route(messages, mode):
                             self._stream_route_completion(orchestrator, security, messages, model_name)
                             orchestrator.record_analytics_event(
@@ -5431,12 +5406,6 @@ def do_POST(self) -> None:  # noqa: N802
                             workflow_run_id=f"run_{uuid.uuid4().hex}",
                             response_format=structured_response_format,
                         ))
-                    finally:
-                        model_client.max_output_tokens = previous_max_tokens
-                        model_client.default_temperature = previous_temperature
-                        model_client.default_top_p = previous_top_p
-                        model_client.default_presence_penalty = previous_presence
-                        model_client.default_frequency_penalty = previous_frequency
                     if structured_response_format is not None and result.get("channel") != "batch":
                         _validate_structured_completion_answer(result.get("answer"), structured_response_format)
                     # Latency-tolerant requests get dispatched to the batch backend.
@@ -5659,14 +5628,10 @@ def do_POST(self) -> None:  # noqa: N802
                     if "stream_options" in body:
                         _validate_responses_stream_options(body)
                     # Sampling knobs: type/range fail-closed before provider passthrough.
-                    if "temperature" in body:
-                        _validate_completions_temperature(body)
-                    if "top_p" in body:
-                        _validate_completions_top_p(body)
-                    if "presence_penalty" in body:
-                        _validate_completions_presence_penalty(body)
-                    if "frequency_penalty" in body:
-                        _validate_completions_frequency_penalty(body)
+                    _validate_completions_temperature(body)
+                    _validate_completions_top_p(body)
+                    _validate_completions_presence_penalty(body)
+                    _validate_completions_frequency_penalty(body)
                     if "n" in body:
                         _validate_responses_n(body)
                     if "seed" in body:
@@ -5677,12 +5642,9 @@ def do_POST(self) -> None:  # noqa: N802
                         _validate_responses_logit_bias(body)
                     if "logprobs" in body or "top_logprobs" in body:
                         _validate_responses_logprobs(body)
-                    if "max_tokens" in body:
-                        _validate_completions_max_tokens(body)
-                    if "max_completion_tokens" in body:
-                        _validate_chat_max_completion_tokens(body)
-                    if "max_output_tokens" in body:
-                        _validate_responses_max_output_tokens(body)
+                    responses_max_tokens = _validate_completions_max_tokens(body)
+                    responses_max_completion_tokens = _validate_chat_max_completion_tokens(body)
+                    responses_max_output_tokens = _validate_responses_max_output_tokens(body)
                     if "max_tool_calls" in body:
                         _validate_responses_max_tool_calls(body)
                     _validate_openai_sdk_control_fields(body, endpoint_path="/v1/responses")
@@ -5818,7 +5780,11 @@ def do_POST(self) -> None:  # noqa: N802
                         # requested stream or accept a missing input.
                         started_at = time.perf_counter()
                         raw_response = self._run(
-                            lambda: orchestrator.proxy_completion(body, endpoint="responses")
+                            lambda: orchestrator.proxy_completion(
+                                body,
+                                endpoint="responses",
+                                single_agent=True,
+                            )
                         )
                         orchestrator.record_analytics_event(
                             "responses_tool_passthrough",
@@ -5854,12 +5820,17 @@ def do_POST(self) -> None:  # noqa: N802
                             }
                     _reject_responses_orchestration_controls(body)
                     chat_payload = _responses_to_chat_payload(body)
-                    model_client = orchestrator.client
-                    previous_max_tokens = model_client.max_output_tokens
-                    if isinstance(body.get("max_output_tokens"), int):
-                        model_client.max_output_tokens = body["max_output_tokens"]
+                    response_max_tokens = (
+                        responses_max_output_tokens
+                        if responses_max_output_tokens is not None
+                        else responses_max_completion_tokens
+                        if responses_max_completion_tokens is not None
+                        else responses_max_tokens
+                    )
                     started_at = time.perf_counter()
-                    try:
+                    with orchestrator.client.request_settings(
+                        max_output_tokens=response_max_tokens,
+                    ):
                         result = self._run(lambda: coordinator.complete(
                             chat_payload["messages"],
                             mode="conduct",
@@ -5869,8 +5840,6 @@ def do_POST(self) -> None:  # noqa: N802
                             workflow_run_id=f"run_{uuid.uuid4().hex}",
                             response_format=response_contract,
                         ))
-                    finally:
-                        model_client.max_output_tokens = previous_max_tokens
                     if response_contract is not None:
                         _validate_structured_completion_answer(result.get("answer"), response_contract)
                     chat_response = chat_completion_response(

docs/architecture.md

@@ -28,6 +28,17 @@ The useful split is quality-latency, not separate products:
 - Low-latency routing: select one worker for the current query or turn.
 - Deep orchestration: create a multi-step workflow when the task needs decomposition, independent attempts, verification, or synthesis.
 
+Structured provider features do not create a third single-agent path. A
+non-null `response_format` or Responses request enters the conducted workflow
+and reaches one final provider only after the
+Thinker/Worker/Verifier/Synthesizer evidence has been assembled. The final
+provider call preserves the validated wire feature; it is a transport boundary,
+not a bypass of orchestration. JSON object and JSON schema are both covered by
+[ADR 0011](planning/adrs/0011-structured-provider-features-stay-orchestrated.md).
+The explicit client-owned tool-loop exception remains the single-worker
+contract defined by [ADR 0014](planning/adrs/0014-gateway-owned-model-selection.md);
+ordinary tool declarations without that opt-in fail closed.
+
 TRINITY contributes the compact coordinator idea: a small model representation plus a lightweight head can choose agent and role over multiple turns. Its Thinker, Worker, and Verifier contracts are practical enough to implement directly.
 
 Conductor contributes the workflow representation: each step is a natural-language subtask, an assigned worker, and an access list of prior step outputs. This is the key piece for preventing every worker from being dragged into the same transcript while still allowing deliberate collaboration.