fix: accept Chat Completions streaming usage options

[seonghobae]

Summary

• Accept stream_options.include_usage=true on /v1/chat/completions.
• Emit provider-reported usage in a usage-only SSE chunk when available.
• Keep include_obfuscation=true and unsupported Responses stream flags fail-closed.

This fixes the exact upstream contract failure observed by ContextualWisdomLab/accounting-information-platform#39: Strix sent stream_options.include_usage=true and the gateway returned invalid_stream_options 400. No provider fallback or gate bypass is changed.

Validation

• 2526 passed
• ruff, compileall, and git diff --check passed
• Changed streaming usage branches covered; repository baseline coverage remains 96% because of pre-existing uncovered modules.

---

[coderabbitai]

Warning

Review limit reached

Next included review available in 34 seconds.

View limit details

Limit details: You’ve used the included review currently available.

You've used all free OSS reviews for now. Wait for the free limit to reset to keep reviewing this public repository.

Learn how review limits work.

Review configuration:

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 08fa3a4f-853f-48c3-a3fb-a4ed606ed400

📥 Commits

Reviewing files that changed from the base of the PR and between 8dfa7d8 and d57563a.

📒 Files selected for processing (9)

• CHANGELOG.md
• README.md
• contextual_orchestrator/orchestrator.py
• contextual_orchestrator/server.py
• tests/test_chat_tools_passthrough_controls_http_honesty.py
• tests/test_http_response_write_disconnect_safety.py
• tests/test_stream_options_null_flags_noop_http_honesty.py
• tests/test_streaming.py
• tests/test_true_streaming.py

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[devin-ai-integration]

Note

This report is out of date. Scroll down for Devin Review's latest report on this PR.

Devin Review found 0 new potential issues.

[devin-ai-integration]

Note

This report is out of date. Scroll down for Devin Review's latest report on this PR.

Devin Review found 0 new potential issues.

[seonghobae]

@opencode-agent Review exact current head 3db6b77ca7f5b25f47488e61371b0007a34f0dbb only. Use the configured NVIDIA_NIM_API_KEY review path; do not use COPILOT_GITHUB_TOKEN. Submit an authenticated current-head formal verdict through the Reviews API. Treat predecessor-head, deterministic fallback, synthetic/model-unavailable, or status-only evidence as non-authoritative. Review the Chat Completions stream_options.include_usage compatibility change, provider usage capture, usage-only SSE ordering, fail-closed unsupported flags, disconnect/thread-local isolation, and exact regression coverage. Do not mutate the branch.

[github-actions]

Reusable conflict resolver stopped fail-closed. Executable or structured-data conflicts require semantic resolution at exact head ce26cd9b59d0f89eb2f48eb606dee8492bca3bd4 against protected main 9b0a356daa4f6bfcb5f83a314f11a7b273cd2623:\n\n```\ncontextual_orchestrator/orchestrator.py
contextual_orchestrator/server.py
tests/test_chat_stream_options_http_honesty.py
tests/test_stream_options_null_flags_noop_http_honesty.py
tests/test_streaming.py

[devin-ai-integration]

Note

This report is out of date. Scroll down for Devin Review's latest report on this PR.

Devin Review found 0 new potential issues.

[github-actions]

Reusable conflict resolver stopped fail-closed.

Exact PR head: 68617470d95b784e1bb59b2abb530aa5a9226efc
Protected main: 9b0a356daa4f6bfcb5f83a314f11a7b273cd2623

Executable or structured-data conflicts require semantic resolution:

contextual_orchestrator/orchestrator.py

  1672         """Yield content deltas from a mock or OpenAI-compatible streaming endpoint.
  1673 
  1674         Real token streaming: the provider is called with stream=true and its SSE deltas
  1675         are yielded as they arrive (not computed-then-framed). The mock path yields its
  1676         answer in fixed chunks so behavior shape stays testable and unchanged.
  1677         """
  1678 <<<<<<< ours
  1679         if type(include_usage) is not bool:
  1680             raise TypeError("include_usage must be a boolean")
  1681         self._local.usage = None
  1682 ||||||| base
  1683 =======
  1684         if type(include_usage) is not bool:
  1685             raise TypeError("include_usage must be a boolean")
  1686 >>>>>>> theirs
  1687         if not is_chat_compatible_model_id(agent.model):
  1688             raise ValueError(
  1689                 f"model {agent.model!r} is not chat-compatible and cannot serve {agent.id!r}"
  1690             )
  1691         self._local.usage = None
  1692         if agent.base_url.startswith("mock://"):
  1751                     if data == "[DONE]":
  1752                         break
  1753                     try:
  1754                         chunk = json.loads(data)
  1755                     except json.JSONDecodeError:
  1756                         continue
  1757 <<<<<<< ours
  1758                     choices = chunk.get("choices")
  1759                     usage = chunk.get("usage")
  1760                     if choices == [] and isinstance(usage, dict):
  1761                         self._local.usage = usage
  1762                         continue
  1763                     if not isinstance(choices, list) or not choices:
  1764                         continue
  1765 ||||||| base
  1766                     choices = chunk.get("choices") or [{}]
  1767 =======
  1768                     usage = chunk.get("usage")
  1769                     if isinstance(usage, dict):
  1770                         self._local.usage = usage
  1771                     choices = chunk.get("choices") or [{}]
  1772 >>>>>>> theirs
  1773                     delta = (choices[0] or {}).get("delta", {}).get("content")
  1774                     if delta:
  1775                         yield delta
  1776         except Exception as exc:  # noqa: BLE001 - provider error boundary (CWE-209)
  1777             # The gateway's own terminal tool-stop contract must survive the
  1778             # boundary: convert the provider HTTP shape into the package-owned
  4198         self,
  4199         messages: list[ChatMessage],
  4200         workflow_run_id: str | None = None,
  4201         *,
  4202         model_name: str = "contextual-orchestrator",
  4203         owner_id: str | None = None,
  4204 <<<<<<< ours
  4205         include_usage: bool = False,
  4206 ||||||| base
  4207 =======
  4208         include_usage: bool = False,
  4209         usage_callback: Callable[[dict[str, Any] | None], None] | None = None,
  4210 >>>>>>> theirs
  4211     ):
  4212         """Stream a single worker's content deltas as they arrive, then persist the run.
  4213 
  4214         True streaming for the route path. ponytail: no cross-agent failover here — bytes
  4215         already sent can't be recalled, so a mid-stream provider failure surfaces to the caller.
  4216         """
 13882         and isinstance(usage, dict)
 13883     ):
 13884         usage_chunk = {**base, "choices": [], "usage": usage}
 13885     final = {**base, "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]}
 13886     final["orchestration"] = {key: value for key, value in orchestration.items() if value is not None}
 13887     chunks.append(final)
 13888 <<<<<<< ours
 13889     if usage_chunk is not None:
 13890         chunks.append(usage_chunk)
 13891 ||||||| base
 13892 =======
 13893     if include_usage:
 13894         reported_usage = result.get("usage")
 13895         if isinstance(reported_usage, dict):
 13896             usage = {**reported_usage, "usage_source": "reported"}
 13897         else:
 13898             prompt_text = result.get("prompt_text", "")
 13899             estimated_prompt_tokens = estimate_tokens(
 13900                 prompt_text if isinstance(prompt_text, str) else str(prompt_text)
 13901             )
 13902             estimated_completion_tokens = estimate_tokens(answer)
 13903             usage = {
 13904                 "prompt_tokens": estimated_prompt_tokens,
 13905                 "completion_tokens": estimated_completion_tokens,
 13906                 "total_tokens": estimated_prompt_tokens + estimated_completion_tokens,
 13907                 "usage_source": "estimated",
 13908             }
 13909         chunks.append({
 13910             **base,
 13911             "choices": [],
 13912             "usage": usage,
 13913         })
 13914 >>>>>>> theirs
 13915     return chunks
 13916 
 13917 
 13918 def _new_chat_completion_id() -> str:
 13919     """Create a collision-resistant OpenAI-compatible completion identifier."""
 13920     return f"chatcmpl-{uuid.uuid4().hex}"

contextual_orchestrator/server.py

  1796     return opts
  1797 
  1798 
  1799 
  1800 
  1801 def _validate_chat_stream_options(body: dict[str, Any], stream: bool) -> dict[str, Any] | None:
  1802 <<<<<<< ours
  1803     """Validate Chat Completions ``stream_options`` for the gateway SSE contract.
  1804 ||||||| base
  1805     """Chat Completions ``stream_options`` — requires stream=true; include_usage unsupported.
  1806 =======
  1807     """Chat Completions ``stream_options`` — validate supported streaming flags.
  1808 >>>>>>> theirs
  1809 
  1810     Shape matches OpenAI (include_usage / include_obfuscation booleans). This
  1811 <<<<<<< ours
  1812     gateway's SSE route emits a final usage-only chunk when provider usage is
  1813     available. It does not apply stream obfuscation, so
  1814     include_obfuscation=true fails closed.
  3414     if "reasoning_effort" in body:
  3415         _validate_chat_reasoning_effort(body)
  3416     if "service_tier" in body:
  3417         _validate_service_tier(body, endpoint_path="/v1/chat/completions")
  3418     if "user" in body:
  3419         _validate_completions_user(body)
  3420 <<<<<<< ours
  3421     if "stream_options" in body:
  3422         stream_options = _validate_chat_stream_options(body, stream)
  3423         sampling["include_usage"] = bool(
  3424             stream_options and stream_options.get("include_usage") is True
  3425         )
  3426 ||||||| base
  3427     if "stream_options" in body:
  3428         _validate_chat_stream_options(body, stream)
  3429 =======
  3430     stream_options = _validate_chat_stream_options(body, stream) if "stream_options" in body else None
  3431     sampling["include_usage"] = bool(
  3432         stream_options and stream_options.get("include_usage") is True
  3433     )
  3434 >>>>>>> theirs
  3435     return sampling
  3436 
  3437 
  3438 def _validate_completions_tools_surface(body: dict[str, Any]) -> None:
  3439     """Reject chat-era tool fields on legacy Completions with a migration path.
  3440 
  7992 
  7993             security.acquire_run_slot()
  7994             try:
  7995                 if not self._begin_sse() or not self._write_sse(frame({"role": "assistant"})):
  7996                     return
  7997                 try:
  7998 <<<<<<< ours
  7999                     stream_kwargs: dict[str, Any] = {
  8000                         "workflow_run_id": run_id,
  8001                         "model_name": model_name,
  8002                     }
  8003                     if include_usage:
  8004                         stream_kwargs["include_usage"] = True
  8005                     for delta in orchestrator.stream_route(messages, **stream_kwargs):
  8006 ||||||| base
  8007                     for delta in orchestrator.stream_route(
  8008                         messages, workflow_run_id=run_id, model_name=model_name
  8009                     ):
  8010 =======
  8011                     stream_kwargs: dict[str, Any] = {
  8012                         "workflow_run_id": run_id,
  8013                         "model_name": model_name,
  8014                     }
  8015                     if include_usage:
  8016                         stream_kwargs.update(
  8017                             {"include_usage": True, "usage_callback": capture_usage}
  8018                         )
  8019                     for delta in orchestrator.stream_route(messages, **stream_kwargs):
  8020                         streamed_parts.append(delta)
  8021 >>>>>>> theirs
  8022                         if not self._write_sse(frame({"content": delta})):
  8023                             return
  8024                     if not self._write_sse(frame({}, finish="stop")):
  8025                         return
  8026 <<<<<<< ours
  8027                     if include_usage:

tests/test_chat_stream_options_http_honesty.py

     1 <<<<<<< ours
     2 """Chat stream_options honesty: requires stream=true and preserves usage requests."""
     3 ||||||| base
     4 """Chat stream_options honesty: requires stream=true; include_usage true fail-closed."""
     5 =======
     6 """Chat stream_options honesty: requires stream=true; include_usage is supported."""
     7 >>>>>>> theirs
     8 
     9 from __future__ import annotations
    10 
    11 import json
    12 import threading
    13 import urllib.error
    95         assert "invalid_stream_options" in json.dumps(body)
    96     finally:
    97         server.shutdown()
    98         thread.join(timeout=5)
    99 
   100 
   101 <<<<<<< ours
   102 def test_http_chat_stream_options_include_usage_true_streams() -> None:
   103 ||||||| base
   104 def test_http_chat_stream_options_include_usage_true_fail_closed() -> None:
   105 =======
   106 def test_http_chat_stream_options_include_usage_true_is_accepted() -> None:
   107 >>>>>>> theirs
   108     server, thread, port = _server()
   109     try:
   110         status, body = _post(
   111             port,
   112             {
   113                 "model": "mock-generalist",
   113                 "model": "mock-generalist",
   114                 "messages": [{"role": "user", "content": "hi"}],
   115                 "stream": True,
   116                 "stream_options": {"include_usage": True},
   117             },
   118         )
   119 <<<<<<< ours
   120         assert status == 200, body
   121         assert isinstance(body, str) and body.endswith("data: [DONE]\n\n")
   122 ||||||| base
   123         assert status == 400, body
   124         assert "invalid_stream_options" in json.dumps(body)
   125 =======
   126         assert status == 200, body
   127         assert isinstance(body, str)
   128         assert '"usage"' in body
   129 >>>>>>> theirs
   130     finally:
   131         server.shutdown()
   132         thread.join(timeout=5)
   133 
   134 
   135 def test_http_chat_stream_options_include_usage_false_with_stream_ok() -> None:
   188         thread.join(timeout=5)
   189 
   190 
   191 if __name__ == "__main__":
   192     test_http_chat_stream_options_all_false_without_stream_as_omit()
   193     test_http_chat_stream_options_true_without_stream_fail_closed()
   194 <<<<<<< ours
   195     test_http_chat_stream_options_include_usage_true_streams()
   196 ||||||| base
   197     test_http_chat_stream_options_include_usage_true_fail_closed()
   198 =======
   199     test_http_chat_stream_options_include_usage_true_is_accepted()
   200 >>>>>>> theirs
   201     test_http_chat_stream_options_include_usage_false_with_stream_ok()
   202     test_http_chat_stream_options_non_object_fail_closed()
   203     test_http_chat_omits_stream_options_ok()
   204     print("ok")

tests/test_stream_options_null_flags_noop_http_honesty.py

    38         with urllib.request.urlopen(request, timeout=15) as response:
    39             return response.status, json.loads(response.read().decode("utf-8"))
    40     except urllib.error.HTTPError as exc:
    41         return exc.code, json.loads(exc.read().decode("utf-8"))
    42 
    43 
    44 <<<<<<< ours
    45 def _post_text(port: int, path: str, payload: dict) -> tuple[int, str]:
    46     request = urllib.request.Request(
    47         f"http://127.0.0.1:{port}{path}",
    48         data=json.dumps(payload).encode("utf-8"),
    49         headers={
    50             "content-type": "application/json",
    51             "authorization": f"Bearer {_TEST_AUTH_TOKEN}",
    52             "connection": "close",
    53         },
    54         method="POST",
    55     )
    56     try:
    57         with urllib.request.urlopen(request, timeout=15) as response:
    58             return response.status, response.read().decode("utf-8")
    59     except urllib.error.HTTPError as exc:
    60         return exc.code, exc.read().decode("utf-8")
    61 
    62 
    63 ||||||| base
    64 =======
    65 def _post_raw(port: int, path: str, payload: dict) -> tuple[int, str, str]:
    66     request = urllib.request.Request(
    67         f"http://127.0.0.1:{port}{path}",
    68         data=json.dumps(payload).encode("utf-8"),
    69         headers={
    70             "content-type": "application/json",
    71             "authorization": f"Bearer {_TEST_AUTH_TOKEN}",
    72             "connection": "close",
    73         },
    74         method="POST",
    75     )
    76     try:
    77         with urllib.request.urlopen(request, timeout=15) as response:
    78             return (
    79                 response.status,
    80                 response.headers.get("content-type", ""),
    81                 response.read().decode("utf-8"),
    82             )
    83     except urllib.error.HTTPError as exc:
    84         return exc.code, exc.headers.get("content-type", ""), exc.read().decode("utf-8")
    85 
    86 
    87 >>>>>>> theirs
    88 def _server():
    89     server = build_server(
    90         build(),
    91         port=0,
    92         security=SecurityConfig(auth_token=_TEST_AUTH_TOKEN, rate_limit_requests=10_000),
    93     )
   163         assert status == 200, body
   164     finally:
   165         server.shutdown()
   166         thread.join(timeout=5)
   167 
   168 
   169 <<<<<<< ours
   170 def test_http_chat_accepts_include_usage_true_from_openai_sdk() -> None:
   171 ||||||| base
   172 def test_http_chat_still_rejects_include_usage_true() -> None:
   173 =======
   174 def test_http_chat_accepts_include_usage_true() -> None:
   175 >>>>>>> theirs
   176     server, thread, port = _server()
   177     try:
   178 <<<<<<< ours
   179         status, body = _post_text(
   180 ||||||| base
   181         status, body = _post(
   188                 "model": "mock-planner",
   189                 "messages": [{"role": "user", "content": "usage compatibility"}],
   190                 "stream": True,
   191                 "stream_options": {"include_usage": True},
   192             },
   193         )
   194 <<<<<<< ours
   195         assert status == 200, body
   196         assert body.startswith("data: ")
   197         assert body.endswith("data: [DONE]\n\n")
   198 ||||||| base
   199         assert status == 400, body
   200         blob = json.dumps(body)
   201         assert "invalid_stream_options" in blob
   202         assert "unknown_fields" not in blob
   203 =======
   204         assert status == 200, sse
   205         assert content_type.startswith("text/event-stream")
   206         frames = [
   207             json.loads(frame[len("data: "):])
   208             for frame in sse.split("\n\n")
   209             if frame.startswith("data: ") and frame != "data: [DONE]"
   210         ]
   211         usage = next(frame for frame in frames if frame.get("choices") == [])
   212         assert usage["usage"]["usage_source"] == "estimated"
   213         assert usage["usage"]["completion_tokens"] > 0
   214     finally:
   215         server.shutdown()
   216         thread.join(timeout=5)
   217 
   218 
   219 def test_http_chat_structured_streams_include_usage() -> None:
   220     server, thread, port = _server()
   221     try:
   222         for structured in (
   223             {
   224                 "tools": [
   225                     {
   226                         "type": "function",
   227                         "function": {
   228                             "name": "lookup",
   229                             "parameters": {"type": "object", "properties": {}},
   230                         },
   231                     }
   232                 ]
   233             },
   234             {"response_format": {"type": "json_object"}},
   235         ):
   236             status, content_type, sse = _post_raw(
   237                 port,
   238                 "/v1/chat/completions",
   239                 {
   240                     "model": "mock-planner",
   241                     "messages": [{"role": "user", "content": "structured usage"}],
   242                     "stream": True,
   243                     "stream_options": {"include_usage": True},
   244                     **structured,
   245                 },
   246             )
   247             assert status == 200, (structured, sse)
   248             assert content_type.startswith("text/event-stream")
   249             frames = [
   250                 json.loads(frame[len("data: "):])
   251                 for frame in sse.split("\n\n")
   252                 if frame.startswith("data: ") and frame != "data: [DONE]"
   253             ]
   254             usage = next(frame for frame in frames if frame.get("choices") == [])
   255             assert usage["usage"]["usage_source"] in {"reported", "estimated"}
   256             assert usage["usage"]["total_tokens"] >= 0
   257 >>>>>>> theirs
   258     finally:
   259         server.shutdown()
   260         thread.join(timeout=5)
   261 
   262 
   263 def test_http_chat_rejects_non_boolean_non_null_flag() -> None:
   281 
   282 
   283 if __name__ == "__main__":
   284     test_http_chat_accepts_stream_options_null_flags_without_stream()
   285     test_http_completions_accepts_stream_options_null_flags_without_stream()
   286     test_http_responses_accepts_stream_options_null_flags()
   287 <<<<<<< ours
   288     test_http_chat_accepts_include_usage_true_from_openai_sdk()
   289 ||||||| base
   290     test_http_chat_still_rejects_include_usage_true()
   291 =======
   292     test_http_chat_accepts_include_usage_true()
   293     test_http_chat_structured_streams_include_usage()
   294 >>>>>>> theirs
   295     test_http_chat_rejects_non_boolean_non_null_flag()
   296     print("ok")

tests/test_streaming.py

    74     chunks = chat_completion_chunks({"answer": "", "mode": "route"})
    75     assert len(chunks) == 2  # role delta + stop delta, no content frames
    76     assert chunks[0]["choices"][0]["delta"] == {"role": "assistant"}
    77     assert chunks[1]["choices"][0]["finish_reason"] == "stop"
    78 
    79 
    80 <<<<<<< ours
    81 def test_chunks_include_requested_provider_usage_after_stop() -> None:
    82     usage = {"prompt_tokens": 3, "completion_tokens": 5, "total_tokens": 8}
    83     chunks = chat_completion_chunks(
    84         {
    85             "answer": "OK",
    86             "mode": "route",
    87             "usage": usage,
    88             "cost": {"measurement_status": "measured"},
    89         },
    90         include_usage=True,
    91     )
    92 
    93     assert chunks[-2]["choices"][0]["finish_reason"] == "stop"
    94     assert chunks[-1]["choices"] == []
    95     assert chunks[-1]["usage"] == usage
    96     assert all("usage" in chunk and chunk["usage"] is None for chunk in chunks[:-1])
    97 
    98 
    99 def test_chunks_omit_estimated_usage_even_when_requested() -> None:
   100     chunks = chat_completion_chunks(
   101         {
   102             "answer": "OK",
   103             "mode": "conduct",
   104             "usage": {"prompt_tokens": 3, "completion_tokens": 5},
   105             "cost": {"measurement_status": "estimated"},
   106         },
   107         include_usage=True,
   108     )
   109 
   110     assert all(chunk.get("usage") is None for chunk in chunks)
   111 
   112 
   113 ||||||| base
   114 =======
   115 def test_include_usage_adds_openai_usage_chunk() -> None:
   116     chunks = chat_completion_chunks(
   117         {
   118             "answer": "abc",
   119             "mode": "route",
   120             "usage": {"prompt_tokens": 2, "completion_tokens": 3, "total_tokens": 5},
   121         },
   122         include_usage=True,
   123     )
   124 
   125     assert chunks[-1]["choices"] == []
   126     assert chunks[-1]["usage"] == {
   127         "prompt_tokens": 2,
   128         "completion_tokens": 3,
   129         "total_tokens": 5,
   130         "usage_source": "reported",
   131     }
   132 
   133 
   134 >>>>>>> theirs
   135 def test_completion_ids_remain_unique_when_created_in_one_millisecond() -> None:
   136     result = {"answer": "OK", "mode": "route"}
   137     with patch("contextual_orchestrator.orchestrator.time.time", return_value=1_786_698_100.0):
   138         response_ids = {chat_completion_response(result)["id"] for _ in range(128)}
   139         chunk_ids = {chat_completion_chunks(result)[0]["id"] for _ in range(128)}
   140 

[devin-ai-integration]

Note

This report is out of date. Scroll down for Devin Review's latest report on this PR.

Devin Review found 0 new potential issues.

[devin-ai-integration]

Note

This report is out of date. Scroll down for Devin Review's latest report on this PR.

Devin Review found 0 new potential issues.

[devin-ai-integration]

Devin Review found 3 new potential issues.

[devin-ai-integration]

🟡 Cached streams fabricate provider usage

On a cached conducted stream, chat_completion_chunks emits measured zero-token usage although no provider ran. Clients mistake cache metadata for provider-reported usage.

Prompt for agents

Prevent chat_completion_chunks from treating cache ledger usage as provider-reported usage. CostRoutingCoordinator.complete creates a measured zero-token cache record for cache hits, so cost.measurement_status alone cannot prove that result.usage came from a provider. Preserve enough provenance in the completion result, or explicitly gate out cache_status == "hit", and add an HTTP or unit regression test for stream_options.include_usage=true on a cached conducted response.

---

Was this helpful? React with 👍 or 👎 to provide feedback.

[devin-ai-integration]

📝 Info: Live usage ordering remains intact

_stream_send captures usage before skipping empty choices. stream_route delivers it after provider iteration, preserving stop-then-usage ordering without stale values.

---

Was this helpful? React with 👍 or 👎 to provide feedback.

[devin-ai-integration]

📝 Info: Structured requests fail before execution

The streaming-usage guard runs before either structured execution branch. Rejected tools and response-format requests cannot start provider work or return misleading usage.

---

Was this helpful? React with 👍 or 👎 to provide feedback.