Enable opt-in realtime streaming across client protocols - #41
Conversation
Reviewer's GuideThis PR adds an opt-in, server-controlled realtime streaming path for Chat Completions, Responses, and Anthropic Messages while retaining compatible aggregation and tool repair by default. It centralizes request policy snapshots, output budgets, and terminal/tool validation, adapts incremental events with stable protocol indexes, hardens post-error cleanup and auditing, and documents configuration, client expectations, and rollback procedures. Sequence diagram for realtime streaming and terminal validationsequenceDiagram
participant Client
participant Gateway
participant Upstream
participant Accumulator as ChatSSEAccumulator
participant Adapter as ProtocolConverter
Client->>Gateway: POST stream=true
Gateway->>Gateway: _snapshot_stream_policy
Gateway->>Upstream: Open backend stream
Upstream-->>Accumulator: SSE output deltas
Accumulator->>Accumulator: merge_tool_call_delta
Accumulator-->>Adapter: Validated incremental state
Adapter->>Adapter: _flush_tool_slot
Adapter-->>Client: Text, reasoning, or tool deltas
Upstream-->>Accumulator: finish_reason
Accumulator->>Accumulator: seal_tool_identities
Accumulator->>Gateway: result
Gateway->>Adapter: set_validated_tools
Adapter->>Adapter: finish
Adapter-->>Client: Protocol terminal event
alt Invalid terminal or tool metadata
Adapter-->>Client: Protocol error event
end
File-Level Changes
Assessment against linked issues
Possibly linked issues
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 2 issues
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="converter.py" line_range="2798-2800" />
<code_context>
client_wants_stream = _client_wants_stream(payload)
body = {k: payload[k] for k in PASSTHROUGH_BODY_KEYS if k in payload}
body = await run_in_threadpool(_prepare_chat_body, body, session_payload=payload)
+ stream_policy = (_snapshot_stream_policy("chat", body) if client_wants_stream else None)
+ if stream_policy is not None:
+ observe_stream_mode(stream_policy.mode)
# Record request metadata.
</code_context>
<issue_to_address>
**issue (broader_impact):** The selected `stream_mode` is recorded only when the client requests streaming; non-streaming Chat, Responses, and Messages requests never call `observe_stream_mode`, so their audits omit the selected mode despite the request-audit requirement.
**Triggers:** When a client sends `stream: false` or omits the streaming field.
**Suggested fix:** Snapshot and record the mode for every generation request, not only inside the streaming branches.
</issue_to_address>
### Comment 2
<location path="converter.py" line_range="3767" />
<code_context>
chat_body = await run_in_threadpool(_prepare_chat_body, chat_body, session_payload=payload)
+ client_wants_stream = _client_wants_stream(payload)
+ stream_policy = (_snapshot_stream_policy("messages", chat_body) if client_wants_stream else None)
+ if stream_policy is not None:
+ observe_stream_mode(stream_policy.mode)
</code_context>
<issue_to_address>
**issue (bug_risk):** Non-streaming requests do not receive a frozen stream policy or retained-output budget; `_nonstream_adapted` invokes `_fetch_checked_chat` without `max_collect_bytes`, causing failover attempts to read the live `CONFIG` value instead of the request-start budget.
**Triggers:** When `max_collect_bytes` changes while a non-streaming request is undergoing credential failover.
**Suggested fix:** Create and pass a request-start budget snapshot through the non-streaming fetch and failover path as well.
</issue_to_address>Sourcery assessment
Needs a human reviewer. 2 findings to address first, and realtime mode changes when tool calls become externally visible and can emit partial tool arguments before terminal validation; a client could act on an incorrect or premature tool call, and reverting cannot undo already delivered output or upstream attempts consumed. The default remains compatible and the setting is reversible, but the new opt-in path spans multiple protocol adapters and its failure modes can outlive a source revert.
Blocking findings: converter.py:2800, converter.py:3767
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 6f4a0fcb36
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| state["tracker"] = tracker | ||
|
|
||
| def completed(): | ||
| merged = tracker.result() |
There was a problem hiding this comment.
When a realtime upstream response ends with content_filter, content-filter, or refusal without emitting text or tool bytes—a valid filter-only outcome—this call raises UpstreamResponseError from ChatSSEAccumulator.result() before _validate_realtime_tools(..., filtered=True) or the protocol adapter can map it. Consequently Responses emits/returns a 502 error instead of response.incomplete, despite the new realtime path explicitly treating filtered terminals as legitimate; handle these terminal reasons before applying the accumulator's empty-output rejection.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 3968facb18
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| self._close_content_blocks(events) | ||
| events.append(self._evt("content_block_start", { | ||
| "index": slot["block_idx"], | ||
| "content_block": {"type": "tool_use", "id": state["id"], | ||
| "name": state["name"], "input": {}}, |
There was a problem hiding this comment.
In realtime Messages streams, _close_content_blocks() closes only thinking/text blocks, so a second ready tool—or text/reasoning arriving after a tool—causes another content_block_start while the previous tool block remains open. For interleaved parallel calls this produces sequences such as start(tool 1), start(tool 2), …, stop(tool 1), stop(tool 2), rather than completing each Anthropic content-block lifecycle before the next block starts; strict Messages clients can reject or misassemble that stream. Buffer later blocks or close the active tool block before emitting another block start.
Useful? React with 👍 / 👎.
1 participant
Fixes #39.
Changes
stream_mode=realtimefor incremental Chat Completions, Responses and Messages output, including reasoning and tool arguments. Keepcompatibleas the default, preserving aggregate validation and bounded tool repair; non-streaming requests remain validated JSON.--stream-mode,CODEBUDDY2API_STREAM_MODEand the WebUI enum through existing configuration precedence and source locking. Freeze mode and retained-output budget before routing for every generation request, including non-streaming failovers and aggregate repair attempts; record the entry mode in request audits.response.incomplete, even without text, while ordinary empty or broken streams remain errors. Explicitly close nested streams and release capacity on errors or cancellation.Verification
6f4a0fcpass with zero-multiplierhy3throughintl-cli: all three protocols deliver text/reasoning and tool-argument deltas before their terminal event, reconstructed tool JSON matches, non-stream responses and tool-history round trips succeed, invalid input is rejected locally, and a cancelled stream releases capacity for the next request. The review follow-ups are covered by the backend regressions above.Scope and rollback
tool_callsterminal marker for successful realtime completion. Models or clients needing the previous behavior can usecompatible.max_collect_bytes=0retains the existing unlimited convention. Tool identity changes after emission are rejected. In Messages, later blocks may wait until upstream completion while the active tool remains incremental; a tool still awaiting identity does not hold unrelated text before its block starts.compatiblefor runtime fallback. Before downgrading source, stop the gateway, remove the new startup option and back up current state; ifstream_modewas saved, use the documented narrow offline removal with a revision increment and integrity check. Do not restore a stale database over newer claims, revocations or account state. The bilingual procedure was verified against a temporary database with unrelated state preserved.Summary by Sourcery
Enable opt-in realtime streaming across all client protocols while preserving compatible defaults and strengthening terminal validation, resource handling, configuration, and operational safeguards.
New Features:
Bug Fixes:
Enhancements:
Build:
Deployment:
Documentation:
Tests: