fix(deepseek): restore progressive Responses streaming - #1095
Conversation
📝 WalkthroughWalkthroughThe PR restores native Responses SSE streaming for DeepSeek V4 Flash. A five-second model-scoped repair synthesizes missing completion events only for valid output graphs. Invalid or incomplete streams end as incomplete. HTTP and WebSocket paths use the repaired stream. ChangesDeepSeek Responses terminal repair
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant DeepSeek as DeepSeek Responses endpoint
participant Repair as Terminal repair relay
participant Transport as HTTP or WebSocket transport
DeepSeek->>Repair: Send progressive SSE events
Repair->>Repair: Validate output items and terminal state
Repair->>Transport: Forward events or synthesize completion
Transport->>Transport: Apply downstream response handling
Possibly related issues
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
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. Comment |
|
✅ Deterministic PR hygiene checks passed. |
|
Reviewed as part of a sweep of the The bug is real: What makes the fix defensible rather than merely effective is that it does not simply trust the stream. It removes the forced-JSON setting for that one model, wraps only its SSE path before the inspection/client split, and synthesizes The terminal-repair tests are genuine — fragmentation, timer races, malformed calls, cancellation, budget exhaustion, plus HTTP and WebSocket integration. Each would fail without the wrapper. Your remaining blocker is mechanical, not technical. The PR is still a draft and the head is 341 commits behind Two files you are touching have moved since your base, so please rebase rather than merging Nothing else from me. Once it is rebased and out of draft this is ready for a maintainer. |
6ce487d to
71d3bfb
Compare
✅ READY
Review readiness checklist
✅ 4/4 boxes ticked. This pull request is already Ready for Review. |
|
Superseded — this update was accidentally posted from an alternate linked GitHub account. Please see the PR author's reply below. |
|
Rebased this PR onto the current Post-rebase verification:
The readiness checklist has been re-ticked for this head, and the PR is now Ready for review. |
![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: 71d3bfbe54
ℹ️ About Codex in GitHub
Codex has been enabled to automatically 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 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| appendBuffer(decoder.decode(value, { stream: true })); | ||
| const result = emitBlocks(controller); |
There was a problem hiding this comment.
When a complete output item has armed the grace timer, this read path appends bytes for the next SSE frame but does not invalidate the timer until emitBlocks() sees a full blank-line-delimited block. If the upstream starts a real response.failed/response.incomplete frame, or any malformed frame, and the delimiter is delayed past the 5-second grace window, the old timer still fires from the previous complete candidate, emits response.completed, and cancels the reader, so a delayed failure/partial frame is reported as success. Treat raw post-candidate bytes as activity, or suppress synthetic success while buffer is non-empty.
AGENTS.md reference: src/AGENTS.md:L17-L19
Useful? React with 👍 / 👎.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 15
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In
`@docs/superpowers/plans/2026-08-06-deepseek-responses-streaming-terminal-repair.md`:
- Around line 245-247: Make retained-item replacement atomic: in the retained
collector update path, reserve and commit the new serialized
metadata/completed-item charge before replacing the existing item, then release
the old charge only after the replacement succeeds. Preserve matching charges
when reservation fails, and add a test covering replacement failure to verify
the prior item and charge remain intact.
- Around line 233-243: Clarify the plan’s [DONE] ownership rule in the streaming
relay flow: define whether an upstream data: [DONE] block is forwarded or
consumed, and ensure the downstream boundary appends the sentinel only when
appropriate. Update the relay and synthetic-completion behavior so both upstream
[DONE] and generated terminal completion produce exactly one [DONE], and add
tests covering each case.
- Around line 354-358: Update the single commitTerminal(kind) gate and its
surrounding repair state to track the latest known usage from
response.completed, then copy that usage into synthetic completed and incomplete
responses. Preserve existing ordered output, status, timestamps, and sequence
handling, and add an assertion covering terminal-less repairs that verifies
usage is retained.
- Around line 279-301: Extend the Step 1 RED-test matrix with a mid-stream
source read-error fixture after a complete candidate, routed through the
failed-tail relay wrapper. Assert response.failed with the expected failed-tail
output, no synthetic terminal, source cancellation, an empty scheduler queue,
and budget.snapshot().currentBytes === 0 after teardown.
- Around line 238-243: Update the streaming wrapper’s synthetic-commit close
path to track an explicit committed-close state before cancelling the upstream
reader, and suppress any resulting cancellation error from producing
response.failed. Preserve response.completed as the sole terminal event for
intentional cancellation, and add coverage for the cancellation/read-error race.
- Around line 164-189: Update ManualScheduler so due callbacks can be queued
without executing immediately: add takeDueCallbacks() to retrieve and remove
callbacks whose deadlines have passed, and isEmpty() to report whether jobs
remain queued. Adjust advance() to only advance time and retain due jobs until
explicitly taken, enabling stale-timer tests to run callbacks after terminal
delivery and assert the scheduler is empty.
In
`@docs/superpowers/specs/2026-08-06-deepseek-responses-streaming-terminal-repair-design.md`:
- Around line 152-170: Update the synthetic-success eligibility predicate to
require exact set equality between added output indices and done-item indices,
not merely that every added index has a done item. Ensure each index has exactly
one lifecycle and reject any done-only index, preserving the existing tainting
behavior for contradictory or duplicate events.
- Around line 188-205: The design must distinguish intentional reader
cancellation after terminal commitment from genuine upstream read failures.
Update the terminal commitment and failed-tail relay flow so cancellation
initiated after the single terminal transition is ignored, while read errors
occurring before commitment still become response.failed, preserving exactly one
terminal event.
In `@src/providers/registry.ts`:
- Around line 2189-2191: Update the policy resolution logic around
modelResponsesTerminalRepair to clamp the floored positive graceMs value to a
minimum of 1 millisecond. Preserve the existing undefined behavior for missing,
non-finite, or non-positive values and continue returning the normalized integer
graceMs.
In `@src/server/responses-terminal-repair.ts`:
- Around line 250-258: Update inspectPayload’s response.output_item.done
handling to release retained state when taint is first detected, then skip
retention for that event and all subsequent events while tainted. Use
releaseRetainedState so completed, created, and related retained state are
cleared, and ensure later output_item.done branches cannot call retainCompleted
after taint.
In `@tests/deepseek-inbound-wire.test.ts`:
- Around line 47-53: Update ManualTerminalScheduler.advance to repeatedly drain
due jobs until none remain, including timers scheduled by callbacks during the
same call. Select each due job by earliest deadline rather than relying on Map
insertion order, while preserving the current-time boundary and callback
execution behavior.
- Around line 225-228: Remove the `as Parameters<typeof handleResponses>[3]`
assertion from the `options` object used in the test, passing the object
directly to `handleResponses`. Preserve `abortSignal` and
`responsesTerminalRepairScheduler` so the option name remains subject to
excess-property type checking, consistent with the sibling tests.
- Around line 266-271: Replace the bare Bun.sleep(0) before scheduler.advance in
the deep-sequence test with the existing bounded polling pattern used elsewhere
in the file, waiting until the relay has emitted the final sequence_number: 6
frame before advancing time. Preserve the existing timeout failure behavior and
subsequent terminal-close race.
- Around line 354-358: Assert after the 20-iteration poll around sent that
response.output_item.done was observed before calling scheduler.advance and
awaiting pump, matching the earlier poll’s failure behavior. Update the test’s
finally block to await or otherwise settle pump so rejected background work is
attributed to this test.
In `@tests/responses-terminal-repair.test.ts`:
- Around line 416-421: Update the chunk boundaries in the test “fragmented UTF-8
and SSE delimiters preserve lifecycle state” so one boundary falls one byte into
the multibyte UTF-8 encoding of “你” and another boundary splits an SSE “\n\n”
delimiter. Preserve the existing lifecycle assertions while ensuring the test
exercises both fragmented UTF-8 decoding and delimiter handling.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: b6dccb20-4006-42bb-a553-d626c5d9dcfb
📒 Files selected for processing (12)
docs-site/src/content/docs/guides/providers.mddocs-site/src/content/docs/zh-cn/guides/providers.mddocs/superpowers/plans/2026-08-06-deepseek-responses-streaming-terminal-repair.mddocs/superpowers/specs/2026-08-06-deepseek-responses-streaming-terminal-repair-design.mdsrc/providers/registry.tssrc/server/responses-terminal-repair.tssrc/server/responses/core.tsstructure/04_transports-and-sidecars.mdtests/deepseek-inbound-wire.test.tstests/deepseek-responses-item-id-repair.test.tstests/passthrough-abort.test.tstests/responses-terminal-repair.test.ts
| - [ ] **Step 1: Write the manual scheduler and two RED tests** | ||
|
|
||
| The test scheduler must advance callbacks synchronously without wall-clock sleep: | ||
|
|
||
| ```ts | ||
| class ManualScheduler implements ResponsesTerminalRepairScheduler { | ||
| private current = 0; | ||
| private nextId = 1; | ||
| private readonly jobs = new Map<number, { at: number; callback: () => void }>(); | ||
|
|
||
| nowMs(): number { return this.current; } | ||
| schedule(callback: () => void, delayMs: number): unknown { | ||
| const id = this.nextId++; | ||
| this.jobs.set(id, { at: this.current + delayMs, callback }); | ||
| return id; | ||
| } | ||
| cancel(handle: unknown): void { this.jobs.delete(handle as number); } | ||
| advance(ms: number): void { | ||
| this.current += ms; | ||
| const due = [...this.jobs.entries()].filter(([, job]) => job.at <= this.current); | ||
| for (const [id, job] of due) { | ||
| this.jobs.delete(id); | ||
| job.callback(); | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Make the manual scheduler controllable for stale-timer tests.
ManualScheduler.advance() runs due callbacks immediately and exposes no pending-job query. Task 3 requires a timer callback to remain queued while a real terminal arrives, then execute the stale callback. The tests must also assert an empty scheduler queue. Add takeDueCallbacks() and isEmpty(), or provide an equivalent queued scheduler.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@docs/superpowers/plans/2026-08-06-deepseek-responses-streaming-terminal-repair.md`
around lines 164 - 189, Update ManualScheduler so due callbacks can be queued
without executing immediately: add takeDueCallbacks() to retrieve and remove
callbacks whose deadlines have passed, and isEmpty() to report whether jobs
remain queued. Adjust advance() to only advance time and retain due jobs until
explicitly taken, enabling stale-timer tests to run callbacks after terminal
delivery and assert the scheduler is empty.
| - frame blocks with `nextSseBlock()` and parse payloads with `sseDataPayload()`; | ||
| - relay normal blocks with their original delimiter; | ||
| - record a valid `response.created.response` snapshot; | ||
| - track added and completed items by integer `output_index`; | ||
| - arm the grace timer only after the candidate predicate succeeds; | ||
| - on timer expiry, enqueue | ||
| `event: response.completed\ndata: <payload>\n\n`, then close and cancel the | ||
| reader; | ||
| - rely on the downstream terminal boundary to append `[DONE]` in production; | ||
| the unit harness may compose `relaySseWithFailedTail` to assert the final sentinel; | ||
| - cancel timers and release all retained budget in one idempotent disposer. |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Define one owner for [DONE].
The plan requires normal blocks to keep their delimiters, but also says the downstream boundary appends [DONE]. It does not define whether an upstream data: [DONE] block is forwarded or consumed. Forwarding and appending both produces duplicate sentinels. Consuming the block breaks pass-through behavior. Define the ownership rule and test exactly one [DONE] for upstream [DONE] and synthetic completion.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@docs/superpowers/plans/2026-08-06-deepseek-responses-streaming-terminal-repair.md`
around lines 233 - 243, Clarify the plan’s [DONE] ownership rule in the
streaming relay flow: define whether an upstream data: [DONE] block is forwarded
or consumed, and ensure the downstream boundary appends the sentinel only when
appropriate. Update the relay and synthetic-completion behavior so both upstream
[DONE] and generated terminal completion produce exactly one [DONE], and add
tests covering each case.
| - on timer expiry, enqueue | ||
| `event: response.completed\ndata: <payload>\n\n`, then close and cancel the | ||
| reader; | ||
| - rely on the downstream terminal boundary to append `[DONE]` in production; | ||
| the unit harness may compose `relaySseWithFailedTail` to assert the final sentinel; | ||
| - cancel timers and release all retained budget in one idempotent disposer. |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Suppress the intentional reader cancellation after synthetic commit.
The wrapper cancels the upstream reader after it emits synthetic response.completed. The failed-tail path converts read errors into response.failed. If the intentional cancellation is observed as a read error, one stream can emit response.completed followed by response.failed. Carry an explicit committed-close state and test this race.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@docs/superpowers/plans/2026-08-06-deepseek-responses-streaming-terminal-repair.md`
around lines 238 - 243, Update the streaming wrapper’s synthetic-commit close
path to track an explicit committed-close state before cancelling the upstream
reader, and suppress any resulting cancellation error from producing
response.failed. Preserve response.completed as the sole terminal event for
intentional cancellation, and add coverage for the cancellation/read-error race.
| Charge serialized retained response metadata and completed items under | ||
| `{ kind: "retained_collectors" }`; release the previous charge before replacing | ||
| an item and release every remaining charge during disposal. |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Make budget replacement atomic.
If the previous charge is released before the new item reservation succeeds, a reservation failure can leave retained state without a matching budget charge. Reserve and commit the new charge first. Then replace the item and release the old charge. Add a replacement-failure test.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@docs/superpowers/plans/2026-08-06-deepseek-responses-streaming-terminal-repair.md`
around lines 245 - 247, Make retained-item replacement atomic: in the retained
collector update path, reserve and commit the new serialized
metadata/completed-item charge before replacing the existing item, then release
the old charge only after the replacement succeeds. Preserve matching charges
when reservation fails, and add a test covering replacement failure to verify
the prior item and charge remain intact.
| - [ ] **Step 1: Add RED tests for every fail-closed boundary** | ||
|
|
||
| Use the Task 2 `ManualScheduler`, controlled source, SSE block builder, stream | ||
| drain, and terminal-type helpers. Add one complete test per row: | ||
|
|
||
| | Test | Exact fixture/action | Required assertions | | ||
| |---|---|---| | ||
| | New activity resets grace | Complete reasoning item; advance 4,999 ms; add and complete a message item; advance 4,999 then 1 ms | No early terminal; final output includes both ordered items; one completed terminal | | ||
| | Complete EOF | Created plus one completed message, then source close | Completed appears immediately before close; source has one terminal | | ||
| | Complete `[DONE]` | Created plus one completed message, then `data: [DONE]` | Completed precedes exactly one `[DONE]` | | ||
| | Open item EOF | Created plus `output_item.added`, then close | One incomplete terminal; no completed terminal | | ||
| | Invalid function arguments | Done function call whose `arguments` is `{broken`, then close | One incomplete terminal; no completed terminal | | ||
| | Unknown item | Done item with `type:"computer_call"`, then close | One incomplete terminal; no completed terminal | | ||
| | Contradictory index | Two different added items reuse index 0, followed by one done item | State stays tainted; incomplete on close | | ||
| | Real terminal precedence | Run completed, failed, and incomplete subcases before grace expiry | Upstream terminal byte-preserved; no synthetic terminal | | ||
| | Timer/terminal race | Queue the timer callback, deliver real completed, then execute queued callback | Exactly one real completed terminal | | ||
| | Fragmentation | Split a multibyte reasoning delta and `\r\n\r\n` delimiters across chunks | Same terminal sequence and completed output as one-chunk control | | ||
| | Cancel/abort | Cancel client before grace; separately abort upstream before grace | No synthetic terminal; timer queue empty; source reader cancelled | | ||
| | Budget overflow | Use `createTestTranslatorBudget({ maxTurnBytes: 128 })` and a done item larger than 128 bytes | `translation_buffer_limit`; no completed terminal; retained bytes return to zero | | ||
|
|
||
| Every test must assert the complete terminal type sequence, expected source | ||
| cancellation, an empty scheduler queue, and | ||
| `budget.snapshot().currentBytes === 0` after teardown. |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Cover read errors through the new wrapper.
The design requires a mid-stream read error to remain owned by the failed-tail relay and become response.failed. This test matrix covers EOF, [DONE], cancellation, and abort, but not a read error after a complete candidate. Add that fixture and assert no synthetic terminal, correct failed-tail output, timer cleanup, source cancellation, and zero retained bytes.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@docs/superpowers/plans/2026-08-06-deepseek-responses-streaming-terminal-repair.md`
around lines 279 - 301, Extend the Step 1 RED-test matrix with a mid-stream
source read-error fixture after a complete candidate, routed through the
failed-tail relay wrapper. Assert response.failed with the expected failed-tail
output, no synthetic terminal, source cancellation, an empty scheduler queue,
and budget.snapshot().currentBytes === 0 after teardown.
| advance(ms: number): void { | ||
| this.current += ms; | ||
| for (const [id, job] of [...this.jobs.entries()]) { | ||
| if (job.at > this.current || !this.jobs.delete(id)) continue; | ||
| job.callback(); | ||
| } | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
Make ManualTerminalScheduler.advance drain timers that are scheduled during a callback.
advance snapshots this.jobs at line 49 before running any callback. A job that schedule()s a follow-up timer inside its own callback is therefore never executed by that same advance call, even when the new timer's deadline is already at or before this.current.
Today this is latent, not a live bug: commitSynthetic in src/server/responses-terminal-repair.ts never re-arms a timer, so the single-timer flow the current tests exercise is correct. The risk is a future repair path that re-arms. That test would silently observe "no terminal" instead of failing on a real defect.
Jobs also run in Map insertion order rather than deadline order. A drain loop fixes both.
♻️ Proposed drain loop with deadline ordering
advance(ms: number): void {
this.current += ms;
- for (const [id, job] of [...this.jobs.entries()]) {
- if (job.at > this.current || !this.jobs.delete(id)) continue;
- job.callback();
+ for (;;) {
+ const due = [...this.jobs.entries()]
+ .filter(([, job]) => job.at <= this.current)
+ .sort(([, left], [, right]) => left.at - right.at);
+ if (due.length === 0) return;
+ for (const [id, job] of due) {
+ if (!this.jobs.delete(id)) continue;
+ job.callback();
+ }
}
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| advance(ms: number): void { | |
| this.current += ms; | |
| for (const [id, job] of [...this.jobs.entries()]) { | |
| if (job.at > this.current || !this.jobs.delete(id)) continue; | |
| job.callback(); | |
| } | |
| } | |
| advance(ms: number): void { | |
| this.current += ms; | |
| for (;;) { | |
| const due = [...this.jobs.entries()] | |
| .filter(([, job]) => job.at <= this.current) | |
| .sort(([, left], [, right]) => left.at - right.at); | |
| if (due.length === 0) return; | |
| for (const [id, job] of due) { | |
| if (!this.jobs.delete(id)) continue; | |
| job.callback(); | |
| } | |
| } | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/deepseek-inbound-wire.test.ts` around lines 47 - 53, Update
ManualTerminalScheduler.advance to repeatedly drain due jobs until none remain,
including timers scheduled by callbacks during the same call. Select each due
job by earliest deadline rather than relying on Map insertion order, while
preserving the current-time boundary and callback execution behavior.
| const options = { | ||
| abortSignal: testAbort.signal, | ||
| responsesTerminalRepairScheduler: scheduler, | ||
| } as Parameters<typeof handleResponses>[3]; |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Drop the as Parameters<typeof handleResponses>[3] assertion so the option name stays typechecked.
HandleResponsesOptions now declares responsesTerminalRepairScheduler at src/server/responses/core.ts lines 589-590, so this object literal is already structurally assignable. The assertion adds nothing and it defeats excess-property checking: if responsesTerminalRepairScheduler is renamed in core.ts, this test keeps compiling and silently stops injecting the deterministic scheduler. The test would then depend on a real 5-second wall-clock timer and become flaky rather than failing loudly.
The two sibling tests in this same file already pass the option without a cast (lines 313-317 and line 427), so removing it here also makes the file self-consistent.
♻️ Proposed fix
- const options = {
- abortSignal: testAbort.signal,
- responsesTerminalRepairScheduler: scheduler,
- } as Parameters<typeof handleResponses>[3];
+ const options = {
+ abortSignal: testAbort.signal,
+ responsesTerminalRepairScheduler: scheduler,
+ } satisfies Parameters<typeof handleResponses>[3];📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| const options = { | |
| abortSignal: testAbort.signal, | |
| responsesTerminalRepairScheduler: scheduler, | |
| } as Parameters<typeof handleResponses>[3]; | |
| const options = { | |
| abortSignal: testAbort.signal, | |
| responsesTerminalRepairScheduler: scheduler, | |
| } satisfies Parameters<typeof handleResponses>[3]; |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/deepseek-inbound-wire.test.ts` around lines 225 - 228, Remove the `as
Parameters<typeof handleResponses>[3]` assertion from the `options` object used
in the test, passing the object directly to `handleResponses`. Preserve
`abortSignal` and `responsesTerminalRepairScheduler` so the option name remains
subject to excess-property type checking, consistent with the sibling tests.
| await Bun.sleep(0); | ||
| scheduler.advance(5_000); | ||
| const remainder = await Promise.race([ | ||
| drainReader(reader), | ||
| new Promise<never>((_, reject) => setTimeout(() => reject(new Error("terminal repair did not close")), 200)), | ||
| ]); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Replace the bare Bun.sleep(0) with a bounded wait for the last pushed frame before advancing the scheduler.
Line 266 assumes one macrotask yield is enough for the repair relay to read the chunk pushed at lines 250-265, record both output_item.done events, and call maybeArmTimer(). That chunk crosses a ReadableStream boundary, the repair relay's async pump, and then the eager-or-tee relay before the timer is armed.
Failure mode: if the yield is too short, no timer is armed when scheduler.advance(5_000) runs at line 267. commitSynthetic never fires, no synthetic terminal is emitted, and the 200 ms race at lines 268-271 rejects with "terminal repair did not close". The test fails intermittently for a scheduling reason rather than a real regression.
This fails loudly rather than passing silently, so it is not a correctness hole. It is a flake source. The same file already solves this correctly with a bounded poll at lines 332-334 and 354-356; reuse that shape here by waiting until the relay has emitted the final sequence_number: 6 frame.
🔧 Proposed deterministic wait
- await Bun.sleep(0);
- scheduler.advance(5_000);
- const remainder = await Promise.race([
- drainReader(reader),
- new Promise<never>((_, reject) => setTimeout(() => reject(new Error("terminal repair did not close")), 200)),
- ]);
+ const beforeTerminal = await readUntil(reader, '"sequence_number":6');
+ scheduler.advance(5_000);
+ const remainder = beforeTerminal + await Promise.race([
+ drainReader(reader),
+ new Promise<never>((_, reject) => setTimeout(() => reject(new Error("terminal repair did not close")), 200)),
+ ]);This mirrors the pattern the item-id test already uses at lines 445-447.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| await Bun.sleep(0); | |
| scheduler.advance(5_000); | |
| const remainder = await Promise.race([ | |
| drainReader(reader), | |
| new Promise<never>((_, reject) => setTimeout(() => reject(new Error("terminal repair did not close")), 200)), | |
| ]); | |
| const beforeTerminal = await readUntil(reader, '"sequence_number":6'); | |
| scheduler.advance(5_000); | |
| const remainder = beforeTerminal + await Promise.race([ | |
| drainReader(reader), | |
| new Promise<never>((_, reject) => setTimeout(() => reject(new Error("terminal repair did not close")), 200)), | |
| ]); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/deepseek-inbound-wire.test.ts` around lines 266 - 271, Replace the bare
Bun.sleep(0) before scheduler.advance in the deep-sequence test with the
existing bounded polling pattern used elsewhere in the file, waiting until the
relay has emitted the final sequence_number: 6 frame before advancing time.
Preserve the existing timeout failure behavior and subsequent terminal-close
race.
| for (let i = 0; i < 20 && !sent.some(frame => JSON.parse(frame).type === "response.output_item.done"); i += 1) { | ||
| await Bun.sleep(0); | ||
| } | ||
| scheduler.advance(5_000); | ||
| await pump; |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Assert the poll at lines 354-356 succeeded before advancing the scheduler.
The poll at lines 332-334 is followed by a hard assertion at line 335, so an exhausted poll fails with a clear message. The poll at lines 354-356 has no such assertion. If it exhausts all 20 iterations because response.output_item.done never reached the WebSocket, the loop exits silently, scheduler.advance(5_000) runs at line 357 against an unarmed timer, and await pump at line 358 blocks until the whole test times out.
The result is an opaque timeout instead of a precise failure that names the missing frame. Adding the mirror assertion restores symmetry with lines 332-335 and keeps the diagnosis local.
Separately, pump is created at line 326 but first awaited at line 358 and is not awaited in the finally block at lines 386-389. If any assertion between lines 335 and 357 throws, pump becomes a floating promise; a rejection then surfaces as an unhandled rejection rather than as part of this test's failure.
🔧 Proposed fix
for (let i = 0; i < 20 && !sent.some(frame => JSON.parse(frame).type === "response.output_item.done"); i += 1) {
await Bun.sleep(0);
}
+ expect(sent.some(frame => JSON.parse(frame).type === "response.output_item.done")).toBe(true);
scheduler.advance(5_000);
await pump;And in the finally block, settle the pump so a late rejection is attributed to this test:
} finally {
abort.abort("test cleanup");
source.cancel();
+ await pump.catch(() => { /* teardown: failure already reported by the assertion above */ });
}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/deepseek-inbound-wire.test.ts` around lines 354 - 358, Assert after the
20-iteration poll around sent that response.output_item.done was observed before
calling scheduler.advance and awaiting pump, matching the earlier poll’s failure
behavior. Update the test’s finally block to await or otherwise settle pump so
rejected background work is attributed to this test.
| test("fragmented UTF-8 and SSE delimiters preserve lifecycle state", async () => { | ||
| const input = completedMessageLifecycle("你好") | ||
| + sse({ type: "response.completed", response: { id: "resp_message", status: "completed" }, sequence_number: 3 }) | ||
| + "data: [DONE]\n\n"; | ||
| const bytes = encoder.encode(input); | ||
| const chunks = [bytes.subarray(0, 37), bytes.subarray(37, 91), bytes.subarray(91, 173), bytes.subarray(173)]; |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Split the UTF-8 character across chunks.
Line 421 places all bytes for "你好" in the final chunk. The test validates fragmented SSE frames, but it does not validate fragmented UTF-8 decoding. A decoder regression that corrupts a multibyte character across chunk boundaries will pass.
Calculate a boundary one byte into "你". Also split one SSE \n\n delimiter.
Proposed test change
const bytes = encoder.encode(input);
- const chunks = [bytes.subarray(0, 37), bytes.subarray(37, 91), bytes.subarray(91, 173), bytes.subarray(173)];
+ const delimiterSplit = input.indexOf("\n\n") + 1;
+ const utf8Split = encoder.encode(input.slice(0, input.indexOf("你"))).byteLength + 1;
+ const chunks = [
+ bytes.subarray(0, delimiterSplit),
+ bytes.subarray(delimiterSplit, utf8Split),
+ bytes.subarray(utf8Split),
+ ];As per path instructions, tests/** changes must provide focused regression coverage for the affected behavior.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| test("fragmented UTF-8 and SSE delimiters preserve lifecycle state", async () => { | |
| const input = completedMessageLifecycle("你好") | |
| + sse({ type: "response.completed", response: { id: "resp_message", status: "completed" }, sequence_number: 3 }) | |
| + "data: [DONE]\n\n"; | |
| const bytes = encoder.encode(input); | |
| const chunks = [bytes.subarray(0, 37), bytes.subarray(37, 91), bytes.subarray(91, 173), bytes.subarray(173)]; | |
| test("fragmented UTF-8 and SSE delimiters preserve lifecycle state", async () => { | |
| const input = completedMessageLifecycle("你好") | |
| sse({ type: "response.completed", response: { id: "resp_message", status: "completed" }, sequence_number: 3 }) | |
| "data: [DONE]\n\n"; | |
| const bytes = encoder.encode(input); | |
| const delimiterSplit = input.indexOf("\n\n") + 1; | |
| const utf8Split = encoder.encode(input.slice(0, input.indexOf("你"))).byteLength + 1; | |
| const chunks = [ | |
| bytes.subarray(0, delimiterSplit), | |
| bytes.subarray(delimiterSplit, utf8Split), | |
| bytes.subarray(utf8Split), | |
| ]; |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/responses-terminal-repair.test.ts` around lines 416 - 421, Update the
chunk boundaries in the test “fragmented UTF-8 and SSE delimiters preserve
lifecycle state” so one boundary falls one byte into the multibyte UTF-8
encoding of “你” and another boundary splits an SSE “\n\n” delimiter. Preserve
the existing lifecycle assertions while ensuring the test exercises both
fragmented UTF-8 decoding and delimiter handling.
Source: Path instructions
Wibias
left a comment
Wibias
left a comment
There was a problem hiding this comment.
Maintainer review — request changes / superseded
Do not merge this implementation.
The functional goal of this PR (restore progressive DeepSeek Responses streaming) has already been superseded on dev by 0b8e608c06a4a81ba676019ee99b10b6e201dcd1, which removes the DeepSeek bounded-JSON opt-in and relies on the existing Responses terminal-event boundary. That is materially smaller and is backed by the current live-wire tests. Re-introducing this PR's timer-based terminal-repair state machine would add a second lifecycle authority for a problem that no longer requires it.
I also completed a maintainer pass over the current head for correctness, resource retention, edge cases, and security. The open bot/reviewer findings are not merely cosmetic; several are real blockers in this implementation:
Confirmed blockers in this PR
-
Stale success timer can race a partial next SSE frame. The grace timer is invalidated only after a complete SSE block is parsed. If bytes for the next frame arrive without the delimiter, the old timer can still fire and synthesize
response.completedbefore a delayedresponse.failed/response.incompleteframe becomes parseable. -
Tainted reconstruction continues retaining item state. After a mismatch taints the candidate, later
output_item.doneevents still retain objects/source bytes. A state that can no longer synthesize success should release reconstruction state immediately and stop accumulating it. -
Invalid lifecycle ordering is accepted. Duplicate
response.createdand output-item lifecycle events beforeresponse.createdare not rejected. A malformed stream can therefore become eligible for synthetic success after an invalid lifecycle. -
Sparse
output_indexvalues are lossy. A matched item at only index5, for example, can pass the parity check and later be synthesized as array position0, silently changing the protocol meaning. Synthetic reconstruction needs a contiguous-index invariant or must fail closed. -
Synthetic completion does not preserve all authoritative terminal metadata (including usage). The reconstructed terminal is assembled from the created snapshot plus retained output, so later response metadata can be lost.
-
The framing/timer contract is more complicated than the underlying provider contract. The timer is based on locally processed complete frames rather than an authoritative upstream terminal and creates new backpressure/timing states that the smaller
devimplementation avoids entirely.
Open review findings
The remaining CodeRabbit/Codex/maintainer threads around scheduler semantics, [DONE] ownership, intentional reader cancellation, atomic budget replacement, mid-stream read-error coverage, usage preservation, exact item parity, grace clamping, tainted-state retention, deterministic tests, polling cleanup, and split UTF-8/SSE delimiter coverage are all part of this disposition. Some documentation comments describe code that has since changed, but the substantive correctness/resource findings above remain enough to reject this approach.
Security / leak pass
I did not find a new auth bypass, credential leak, SSRF primitive, command injection, or path traversal in this PR. The meaningful security risk here is availability/resource correctness: the repair state machine retains protocol reconstruction state and introduces timer/race surfaces on attacker-controlled upstream SSE.
During the takeover I found a separate issue in the current replacement path: the client-facing HTTP terminal boundary and WebSocket SSE pump could retain an arbitrarily large unterminated SSE frame. That is being fixed separately in maintainer PR #1241 with a hard byte cap and regression coverage; it is not a reason to merge this larger repair state machine.
Disposition: requested changes; close as superseded rather than iterating this branch.
|
Maintainer takeover disposition: closing this PR as superseded. The progressive DeepSeek Responses streaming fix is already on The full audit also found one independent hardening gap in the replacement path (unbounded retention of an unterminated client-facing SSE frame). That follow-up is now isolated in maintainer PR #1241 with bounded raw-byte framing for HTTP + WebSocket and dedicated regression tests. So the split is intentional:
|
4 participants
Summary
deepseek-v4-flashon the Responses passthrough path so Codex receives reasoning, text, and tool-call progress incrementally instead of waiting for one bounded JSON body.response.completedafter every opened output item has a structurally completeoutput_item.done; partial, malformed, duplicate, contradictory, or unknown lifecycles fail closed withresponse.incomplete.This branch is rebased directly onto
devcommit1fc24f03999a309737a9f01b214b7b8c1fa9ca85.Verification
bun run typecheck— passed in a clean worktree.codex-auth5-second timeout reproduces unchanged on a clean checkout of the exactdevbase.71d3bfbe547f9356571b1af13d7f2140428056f4.Checklist
Review readiness checklist
This PR stays in draft until every box below is ticked. Tick all four boxes once the requirements are met:
All CI tests are green on my local testing.
I pushed my PR to the latest dev commit.
I fixed all correct Codex and CodeRabbit findings.
My PR is ready for review.
Summary by CodeRabbit
New Features
Bug Fixes
Documentation