fix: accept Chat Completions streaming usage options

CHANGELOG.md

@@ -75,6 +75,12 @@ and this project uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html)
 
 ### Fixed
 
+- Accept the standard Chat Completions `stream_options.include_usage=true`
+  request and emit provider-reported usage in a usage-only SSE chunk when
+  available after the terminal stop chunk; pass the option through live provider
+  streams, and reject structured `tools`/`response_format` passthrough before
+  execution because it cannot emit that SSE contract; keep unsupported
+  obfuscation flags fail-closed.
 - Billing usage-export failures now appear in the operator-safe telemetry health
   counters instead of only in emitted error events.
 - Billing usage export now follows accepted ledger writes and skips duplicate,

README.md

@@ -152,7 +152,7 @@ Non-mock providers must use `https://` URLs and a **resolvable KV credential** 
 One public interface:
 
 - `contextual-orchestrator` is the model-like control-plane candidate exposed to callers. `/v1/models` lists it first, followed by every configured worker candidate, including disabled candidates with their status.
-- `/v1/chat/completions` accepts normal chat messages, and `"stream": true` returns an OpenAI-compatible `text/event-stream` of `chat.completion.chunk` deltas terminated by `data: [DONE]`. In **route** mode the worker's tokens are streamed live as they arrive from the provider (real token streaming); in **conduct** mode the multi-step answer is produced then framed as deltas (a workflow can't honestly token-stream a synthesizer that hasn't run yet).
+- `/v1/chat/completions` accepts normal chat messages, and `"stream": true` returns an OpenAI-compatible `text/event-stream` of `chat.completion.chunk` deltas terminated by `data: [DONE]`. `stream_options.include_usage=true` is accepted for ordinary chat streams and emits a provider-reported usage-only chunk after the terminal stop chunk when usage is available; structured `tools`/`response_format` passthrough rejects that combination before provider execution. In **route** mode the worker's tokens are streamed live as they arrive from the provider (real token streaming); in **conduct** mode the multi-step answer is produced then framed as deltas (a workflow can't honestly token-stream a synthesizer that hasn't run yet).
 - `TaskOrchestrator.complete()` decides whether to route to one worker or run a short workflow.
 - `TaskOrchestrator.compare_to_baseline(prompts, mode)` (CLI `--eval PROMPT...`) measures the orchestration engine against a single-worker baseline — per-prompt and aggregate latency plus a structural coverage delta (contributing steps + verifier-pass presence). It is a measured tradeoff report, not a human-quality claim.
 - Responses include orchestration mode metadata, and trusted callers can request the full trace for audit.

contextual_orchestrator/orchestrator.py

@@ -1686,11 +1686,11 @@ def stream_chat(
         """
         if type(include_usage) is not bool:
             raise TypeError("include_usage must be a boolean")
+        self._local.usage = None
         if not is_chat_compatible_model_id(agent.model):
             raise ValueError(
                 f"model {agent.model!r} is not chat-compatible and cannot serve {agent.id!r}"
             )
-        self._local.usage = None
         if agent.base_url.startswith("mock://"):
             answer = self._mock(agent, messages)
             for start in range(0, len(answer), 24):
@@ -1754,10 +1754,14 @@ def _stream_send(
                         chunk = json.loads(data)
                     except json.JSONDecodeError:
                         continue
+                    choices = chunk.get("choices")
                     usage = chunk.get("usage")
                     if isinstance(usage, dict):
                         self._local.usage = usage
-                    choices = chunk.get("choices") or [{}]
+                    if choices == []:
+                        continue
+                    if not isinstance(choices, list) or not choices:
+                        continue
                     delta = (choices[0] or {}).get("delta", {}).get("content")
                     if delta:
                         yield delta
@@ -13838,23 +13842,41 @@ def chat_completion_chunks(
     include_trace: bool = False,
     include_usage: bool = False,
 ) -> list[dict[str, Any]]:
-    """Frame an orchestration result as OpenAI-compatible ``chat.completion.chunk`` deltas.
+    """Frame an orchestration result as OpenAI-compatible chat completion chunks.
 
-    The engine produces the full answer before framing, so this yields a correct-shape
-    SSE stream (role delta, content deltas, terminal stop delta) rather than true
-    token-by-token streaming — real token streaming requires a streaming ModelClient.
+    Only provider-reported usage may be emitted. Gateway estimates remain internal
+    because presenting estimates as provider usage would violate the wire contract.
     """
     answer = result.get("answer", "")
     completion_id = _new_chat_completion_id()
     created = int(time.time())
-    base = {"id": completion_id, "object": "chat.completion.chunk", "created": created, "model": model}
+    base = {
+        "id": completion_id,
+        "object": "chat.completion.chunk",
+        "created": created,
+        "model": model,
+    }
+    if include_usage:
+        base["usage"] = None
 
     chunks: list[dict[str, Any]] = [
-        {**base, "choices": [{"index": 0, "delta": {"role": "assistant"}, "finish_reason": None}]}
+        {
+            **base,
+            "choices": [
+                {"index": 0, "delta": {"role": "assistant"}, "finish_reason": None}
+            ],
+        }
     ]
-    for start in range(0, len(answer), _STREAM_CHUNK_SIZE):
-        piece = answer[start : start + _STREAM_CHUNK_SIZE]
-        chunks.append({**base, "choices": [{"index": 0, "delta": {"content": piece}, "finish_reason": None}]})
+    for offset in range(0, len(answer), _STREAM_CHUNK_SIZE):
+        piece = answer[offset : offset + _STREAM_CHUNK_SIZE]
+        chunks.append(
+            {
+                **base,
+                "choices": [
+                    {"index": 0, "delta": {"content": piece}, "finish_reason": None}
+                ],
+            }
+        )
 
     orchestration = {
         "workflow_run_id": result.get("workflow_run_id"),
@@ -13863,30 +13885,25 @@ def chat_completion_chunks(
     }
     if include_trace and "trace" in result:
         orchestration["trace"] = redact_value(result["trace"])
-    final = {**base, "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]}
-    final["orchestration"] = {key: value for key, value in orchestration.items() if value is not None}
+
+    final = {
+        **base,
+        "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
+        "orchestration": {
+            key: value for key, value in orchestration.items() if value is not None
+        },
+    }
     chunks.append(final)
-    if include_usage:
-        reported_usage = result.get("usage")
-        if isinstance(reported_usage, dict):
-            usage = {**reported_usage, "usage_source": "reported"}
-        else:
-            prompt_text = result.get("prompt_text", "")
-            estimated_prompt_tokens = estimate_tokens(
-                prompt_text if isinstance(prompt_text, str) else str(prompt_text)
-            )
-            estimated_completion_tokens = estimate_tokens(answer)
-            usage = {
-                "prompt_tokens": estimated_prompt_tokens,
-                "completion_tokens": estimated_completion_tokens,
-                "total_tokens": estimated_prompt_tokens + estimated_completion_tokens,
-                "usage_source": "estimated",
-            }
-        chunks.append({
-            **base,
-            "choices": [],
-            "usage": usage,
-        })
+
+    usage = result.get("usage")
+    cost = result.get("cost")
+    if (
+        include_usage
+        and isinstance(cost, dict)
+        and cost.get("measurement_status") == "measured"
+        and isinstance(usage, dict)
+    ):
+        chunks.append({**base, "choices": [], "usage": usage})
     return chunks
 
 

contextual_orchestrator/server.py

@@ -6492,6 +6492,12 @@ def register_video_job(agent: ModelAgent, provider_result: dict[str, Any]) -> di
                     # Explicit JSON null on trigger keys is omit-equivalent (SDK optional
                     # defaults) — do not force single-agent passthrough for null-only keys.
                     if body.get("response_format") or tools_list:
+                        if stream and include_usage:
+                            raise RequestError(
+                                400,
+                                "invalid_stream_options",
+                                "stream_options.include_usage=true is not supported with tools or response_format",
+                            )
                         if explicit_trace:
                             raise RequestError(
                                 400,
@@ -7923,11 +7929,10 @@ def _stream_route_completion(
             *,
             include_usage: bool = False,
         ) -> None:
-            """Pipe a worker's live deltas out as OpenAI chat.completion.chunk SSE frames."""
+            """Pipe live provider deltas as OpenAI chat-completion SSE frames."""
             run_id = f"run_{uuid.uuid4().hex}"
             completion_id = _new_chat_completion_id()
             created = int(time.time())
-            streamed_parts: list[str] = []
             stream_usage: dict[str, Any] | None = None
 
             def capture_usage(usage: dict[str, Any] | None) -> None:
@@ -7940,13 +7945,30 @@ def frame(delta: dict[str, Any], finish: str | None = None) -> str:
                     "object": "chat.completion.chunk",
                     "created": created,
                     "model": model_name,
-                    "choices": [{"index": 0, "delta": delta, "finish_reason": finish}],
+                    "choices": [
+                        {"index": 0, "delta": delta, "finish_reason": finish}
+                    ],
+                }
+                if include_usage:
+                    payload["usage"] = None
+                return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
+
+            def usage_frame(usage: dict[str, Any]) -> str:
+                payload = {
+                    "id": completion_id,
+                    "object": "chat.completion.chunk",
+                    "created": created,
+                    "model": model_name,
+                    "choices": [],
+                    "usage": usage,
                 }
                 return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
 
             security.acquire_run_slot()
             try:
-                if not self._begin_sse() or not self._write_sse(frame({"role": "assistant"})):
+                if not self._begin_sse() or not self._write_sse(
+                    frame({"role": "assistant"})
+                ):
                     return
                 try:
                     stream_kwargs: dict[str, Any] = {
@@ -7958,39 +7980,16 @@ def frame(delta: dict[str, Any], finish: str | None = None) -> str:
                             {"include_usage": True, "usage_callback": capture_usage}
                         )
                     for delta in orchestrator.stream_route(messages, **stream_kwargs):
-                        streamed_parts.append(delta)
                         if not self._write_sse(frame({"content": delta})):
                             return
                     if not self._write_sse(frame({}, finish="stop")):
                         return
-                    if include_usage:
-                        if isinstance(stream_usage, dict):
-                            usage = {**stream_usage, "usage_source": "reported"}
-                        else:
-                            estimated_prompt_tokens = estimate_tokens(
-                                json.dumps(messages, ensure_ascii=False)
-                            )
-                            estimated_completion_tokens = estimate_tokens(
-                                "".join(streamed_parts)
-                            )
-                            usage = {
-                                "prompt_tokens": estimated_prompt_tokens,
-                                "completion_tokens": estimated_completion_tokens,
-                                "total_tokens": estimated_prompt_tokens + estimated_completion_tokens,
-                                "usage_source": "estimated",
-                            }
-                        usage_payload = {
-                            "id": completion_id,
-                            "object": "chat.completion.chunk",
-                            "created": created,
-                            "model": model_name,
-                            "choices": [],
-                            "usage": usage,
-                        }
-                        if not self._write_sse(
-                            f"data: {json.dumps(usage_payload, ensure_ascii=False)}\n\n"
-                        ):
-                            return
+                    if (
+                        include_usage
+                        and isinstance(stream_usage, dict)
+                        and not self._write_sse(usage_frame(stream_usage))
+                    ):
+                        return
                 except ToolFallbackStoppedError as exc:
                     detail = {
                         "request_id": uuid.uuid4().hex,
@@ -8007,7 +8006,7 @@ def frame(delta: dict[str, Any], finish: str | None = None) -> str:
                         return
                     if not self._write_sse(frame({}, finish="error")):
                         return
-                except Exception:  # noqa: BLE001 - headers already sent; surface as a terminal error frame
+                except Exception:  # noqa: BLE001 - headers already sent
                     if not self._write_sse(frame({}, finish="error")):
                         return
                 self._write_sse("data: [DONE]\n\n")

tests/test_chat_tools_passthrough_controls_http_honesty.py

@@ -146,6 +146,28 @@ def test_http_tools_passthrough_rejects_invalid_user_and_stream_options() -> Non
         thread.join(timeout=5)
 
 
+def test_http_structured_stream_usage_fails_closed_before_execution() -> None:
+    """Structured passthrough cannot emit usage SSE, so reject it before provider work."""
+    server, thread, port = _server()
+    try:
+        for payload in (
+            _base(stream=True, stream_options={"include_usage": True}),
+            {
+                "model": "mock-planner",
+                "messages": [{"role": "user", "content": "structured"}],
+                "response_format": {"type": "json_object"},
+                "stream": True,
+                "stream_options": {"include_usage": True},
+            },
+        ):
+            status, body = _post(port, payload)
+            assert status == 400, (payload, body)
+            assert "invalid_stream_options" in json.dumps(body)
+    finally:
+        server.shutdown()
+        thread.join(timeout=5)
+
+
 def test_http_tools_passthrough_accepts_coerced_sampling() -> None:
     server, thread, port = _server()
     try:
@@ -185,6 +207,7 @@ def test_http_response_format_passthrough_rejects_seed() -> None:
     test_http_tools_passthrough_rejects_invalid_temperature()
     test_http_tools_passthrough_rejects_unsupported_seed_store_stop_n()
     test_http_tools_passthrough_rejects_invalid_user_and_stream_options()
+    test_http_structured_stream_usage_fails_closed_before_execution()
     test_http_tools_passthrough_accepts_coerced_sampling()
     test_http_response_format_passthrough_rejects_seed()
     print("ok")

tests/test_http_response_write_disconnect_safety.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import json
 from pathlib import Path
 import sys
 
@@ -92,6 +93,81 @@ def _write_sse(self, _frame):
         server.server_close()
 
 
+def test_stream_route_emits_provider_usage_when_requested() -> None:
+    """A successful live route includes provider usage after its stop frame."""
+    server = build_server(build(), port=0, security=SecurityConfig(auth_token=_TEST_AUTH_TOKEN))
+    frames: list[str] = []
+
+    class Client:
+        def take_usage(self):
+            return {"prompt_tokens": 2, "completion_tokens": 4, "total_tokens": 6}
+
+    class Orchestrator:
+        client = Client()
+
+        def stream_route(
+            self,
+            messages,
+            workflow_run_id,
+            *,
+            model_name,
+            include_usage=False,
+            usage_callback=None,
+        ):
+            del messages, workflow_run_id, model_name
+            assert include_usage is True
+            yield "answer"
+            assert usage_callback is not None
+            usage_callback(
+                {"prompt_tokens": 2, "completion_tokens": 4, "total_tokens": 6}
+            )
+
+    class Security:
+        def acquire_run_slot(self):
+            return None
+
+        def release_run_slot(self):
+            return None
+
+    class Handler:
+        def _begin_sse(self):
+            return True
+
+        def _write_sse(self, frame):
+            frames.append(frame)
+            return True
+
+    try:
+        server.RequestHandlerClass._stream_route_completion(
+            Handler(),
+            Orchestrator(),
+            Security(),
+            [],
+            "model-group",
+            include_usage=True,
+        )
+    finally:
+        server.server_close()
+
+    payloads = [
+        json.loads(frame[6:])
+        for frame in frames
+        if frame.startswith("data: ") and frame != "data: [DONE]\n\n"
+    ]
+    usage_frames = [payload for payload in payloads if payload.get("choices") == []]
+    assert len(usage_frames) == 1
+    assert payloads[-2]["choices"][0]["finish_reason"] == "stop"
+    usage_frame = usage_frames[0]
+    assert usage_frame["object"] == "chat.completion.chunk"
+    assert usage_frame["model"] == "model-group"
+    assert usage_frame["choices"] == []
+    assert usage_frame["usage"] == {
+        "prompt_tokens": 2,
+        "completion_tokens": 4,
+        "total_tokens": 6,
+    }
+
+
 def test_responses_stream_does_not_start_orchestration_after_header_disconnect() -> None:
     """A dead Responses peer must not trigger any paid provider work."""
     server = build_server(build(), port=0, security=SecurityConfig(auth_token=_TEST_AUTH_TOKEN))