diff --git a/docs-site/src/content/docs/guides/providers.md b/docs-site/src/content/docs/guides/providers.md index af1d0ef4ab..9bc61576f4 100644 --- a/docs-site/src/content/docs/guides/providers.md +++ b/docs-site/src/content/docs/guides/providers.md @@ -279,6 +279,10 @@ free-experimentation model. Most use the `openai-chat` adapter with a bearer key; a few that expose only an Anthropic-compatible endpoint (e.g. **Xiaomi MiMo**) use the `anthropic` adapter (`x-api-key`). Volcengine Agent Plan uses its native Responses endpoint through `openai-responses`. +The built-in DeepSeek preset also routes `deepseek-v4-flash` over its native Responses endpoint and +keeps upstream SSE streaming enabled. If that model finishes every output item but omits the final +Responses event, opencodex applies a five-second model-scoped grace repair; malformed or partial +streams close as incomplete rather than being reported as successful. > **Three Volcengine billing routes:** `volcengine` is the pay-as-you-go Ark API, > `volcengine-coding-plan` consumes Coding Plan quota, and `volcengine-agent-plan` consumes Agent diff --git a/docs-site/src/content/docs/zh-cn/guides/providers.md b/docs-site/src/content/docs/zh-cn/guides/providers.md index 490fd696a4..34a31bef70 100644 --- a/docs-site/src/content/docs/zh-cn/guides/providers.md +++ b/docs-site/src/content/docs/zh-cn/guides/providers.md @@ -193,6 +193,9 @@ Cline IDE/CLI 中提供，不能通过 API 使用；`minimax/minimax-m2.5` 是 大多数使用带 bearer 密钥的 `openai-chat` adapter；少数仅暴露 Anthropic 兼容端点的提供商（例如 **Xiaomi MiMo**）使用 `anthropic` adapter（`x-api-key`）。 火山方舟 Agent Plan 通过 `openai-responses` adapter 使用原生 Responses 端点。 +内置 DeepSeek preset 同样会让 `deepseek-v4-flash` 使用原生 Responses 端点，并保留上游 SSE +流式输出。如果该模型已经完成全部输出项却缺少最终 Responses 事件，opencodex 会应用模型级 +5 秒宽限修复；不完整或格式异常的流会以 incomplete 结束，不会被误报为成功。 > **三条火山方舟计费线路：**`volcengine` 是按量付费方舟 API，`volcengine-coding-plan` > 消耗 Coding Plan 额度，`volcengine-agent-plan` 消耗 Agent Plan 额度。密钥与端点需要属于 diff --git a/docs/superpowers/plans/2026-08-06-deepseek-responses-streaming-terminal-repair.md b/docs/superpowers/plans/2026-08-06-deepseek-responses-streaming-terminal-repair.md new file mode 100644 index 0000000000..de1983ef81 --- /dev/null +++ b/docs/superpowers/plans/2026-08-06-deepseek-responses-streaming-terminal-repair.md @@ -0,0 +1,650 @@ +# DeepSeek Responses Streaming Terminal Repair Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Restore progressive `deepseek-v4-flash` Responses streaming and safely synthesize a single completed terminal only for structurally complete terminal-less output. + +**Architecture:** Replace the built-in DeepSeek forced bounded-JSON hint with a registry-only terminal-repair policy. A bounded SSE wrapper runs before the existing inspection/client split, relays healthy streams unchanged, and uses a five-second post-completion grace timer to synthesize one `response.completed` event when every output item is safely complete. + +**Tech Stack:** Bun-native TypeScript, Web `ReadableStream`, Responses SSE, `bun:test`, existing translator-budget and SSE framing helpers. + +## Global Constraints + +- Work only on `agent/fix-deepseek-responses-streaming`, based on `origin/dev`; do not modify PR #1047's branch. +- Keep the repair registry-only and limited to the official built-in `deepseek-v4-flash` Responses route. +- Do not change DeepSeek Chat Completions, Anthropic replay, global JSON timeouts, or other providers. +- A real upstream terminal is authoritative and must never be duplicated or replaced. +- Never synthesize success for partial, malformed, tainted, oversized, unknown-type, cancelled, or aborted output. +- Use TDD for every behavior change: observe the focused test fail for the expected reason before production edits. +- Preserve existing item-id repair, reasoning replay, continuation-state, WebSocket, cancellation, and failed-tail contracts. +- Do not log prompts, API keys, raw credentials, or private account identifiers. + +--- + +## File responsibility map + +- `src/providers/registry.ts` — declares and resolves the built-in per-model terminal-repair policy. +- `src/server/responses-terminal-repair.ts` — owns SSE lifecycle tracking, bounded retained state, grace scheduling, and synthetic terminal creation. +- `src/server/responses/core.ts` — activates the repair before existing transport-specific relay branches. +- `tests/responses-terminal-repair.test.ts` — unit state-machine and stream-race coverage. +- `tests/deepseek-inbound-wire.test.ts` — end-to-end official DeepSeek wire and HTTP activation. +- `tests/ws-endpoint.test.ts` — WebSocket event parity for real and repaired terminals. +- `structure/04_transports-and-sidecars.md` — architectural contract for the provider-scoped streaming repair. +- `docs/superpowers/specs/2026-08-06-deepseek-responses-streaming-terminal-repair-design.md` — approved design authority; implementation must remain consistent with it. + +--- + +### Task 1: Replace the DeepSeek bounded-JSON hint with a terminal-repair policy + +**Files:** +- Modify: `src/providers/registry.ts:150-170` +- Modify: `src/providers/registry.ts:1143-1162` +- Modify: `src/providers/registry.ts:1834-1842` +- Modify: `tests/deepseek-inbound-wire.test.ts:1-175` +- Modify: `tests/deepseek-responses-item-id-repair.test.ts` + +**Interfaces:** +- Produces: `ResponsesTerminalRepairPolicy` +- Produces: `providerModelResponsesTerminalRepair(id, provider, modelId): ResponsesTerminalRepairPolicy | undefined` +- Preserves: `providerModelResponsesUpstreamStreaming(...)` for providers that still need bounded JSON. + +- [ ] **Step 1: Write failing registry and outbound-wire tests** + +Import the new resolver and change the existing DeepSeek transport expectations: + +```ts +import { + getProviderRegistryEntry, + providerModelResponsesTerminalRepair, +} from "../src/providers/registry"; + +test("the official DeepSeek Responses route opts into terminal repair", () => { + const provider = deepseekProvider(); + expect(providerModelResponsesTerminalRepair("deepseek", provider, MODEL)).toEqual({ graceMs: 5_000 }); + expect(providerModelResponsesTerminalRepair("deepseek", provider, "deepseek-chat")).toBeUndefined(); + expect(providerModelResponsesTerminalRepair("custom-deepseek", provider, MODEL)).toBeUndefined(); +}); + +test("Codex HTTP and WebSocket turns keep DeepSeek streaming upstream", async () => { + expect((await drive("responses")).body.stream).toBe(true); + expect((await drive("responses", "websocket")).body.stream).toBe(true); +}); +``` + +Delete or rewrite the tests whose asserted contract is specifically +`stream:false`/bounded JSON for built-in DeepSeek. Keep the bounded-JSON helper +coverage that is provider-neutral. In +`tests/deepseek-responses-item-id-repair.test.ts`, retain the pure +`repairResponsesJsonItemIds()` unit test but remove the built-in-DeepSeek HTTP +activation assertion; Task 4 replaces it with streaming repair composition. + +- [ ] **Step 2: Run the test and verify RED** + +Run: + +```bash +bun test tests/deepseek-inbound-wire.test.ts +``` + +Expected: compile/test failure because `providerModelResponsesTerminalRepair` +does not exist and the current outbound body still contains `stream:false`. + +- [ ] **Step 3: Add the registry policy and resolver** + +Add the exact registry-only type and field: + +```ts +export interface ResponsesTerminalRepairPolicy { + graceMs: number; +} +``` + +Add this exact field inside the existing `ProviderRegistryEntry` interface: + +```ts +modelResponsesTerminalRepair?: Record; +``` + +In the DeepSeek entry, remove the `false` streaming override and declare: + +```ts +modelResponsesTerminalRepair: { + "deepseek-v4-flash": { graceMs: 5_000 }, +}, +``` + +Add the resolver beside `providerModelResponsesUpstreamStreaming`: + +```ts +export function providerModelResponsesTerminalRepair( + id: string, + provider: Pick & Partial>, + modelId: string, +): ResponsesTerminalRepairPolicy | undefined { + const entry = getProviderRegistryEntry(id); + if (!entry?.modelResponsesTerminalRepair || !providerMatchesRegistryTransport(id, provider)) return undefined; + const policy = entry.modelResponsesTerminalRepair[modelId.trim().toLowerCase()]; + if (!policy || !Number.isFinite(policy.graceMs) || policy.graceMs <= 0) return undefined; + return { graceMs: Math.floor(policy.graceMs) }; +} +``` + +- [ ] **Step 4: Run focused tests and verify GREEN** + +Run: + +```bash +bun test tests/deepseek-inbound-wire.test.ts tests/deepseek-responses-item-id-repair.test.ts tests/provider-registry-parity.test.ts +``` + +Expected: all selected tests pass; captured HTTP and WebSocket request bodies +carry `stream:true`. + +- [ ] **Step 5: Commit** + +```bash +git add src/providers/registry.ts tests/deepseek-inbound-wire.test.ts tests/deepseek-responses-item-id-repair.test.ts +git commit -m "fix(deepseek): restore Responses upstream streaming" +``` + +--- + +### Task 2: Build the healthy-stream and grace-completion core + +**Files:** +- Create: `src/server/responses-terminal-repair.ts` +- Create: `tests/responses-terminal-repair.test.ts` + +**Interfaces:** +- Consumes: `ResponsesTerminalRepairPolicy` +- Consumes: `TranslatorBudget` +- Produces: `ResponsesTerminalRepairScheduler` +- Produces: `relayResponsesSseWithTerminalRepair(body, upstream, policy, budget, scheduler?)` + +- [ ] **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 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(); + } + } +} +``` + +Add one test with the live-captured lifecycle shape and a real +`response.completed`; assert output is byte-identical and contains one terminal. +Add one terminal-less fixture containing `response.created`, reasoning added/done, +function-call added/arguments.done/output_item.done; advance 4,999 ms (no +terminal), then one more millisecond and assert exactly one synthetic completed +terminal followed by one `[DONE]`. + +- [ ] **Step 2: Run tests and verify RED** + +Run: + +```bash +bun test tests/responses-terminal-repair.test.ts +``` + +Expected: module-not-found failure for +`src/server/responses-terminal-repair.ts`. + +- [ ] **Step 3: Implement the public API and minimal healthy/grace path** + +Create these exact public interfaces: + +```ts +export interface ResponsesTerminalRepairScheduler { + nowMs(): number; + schedule(callback: () => void, delayMs: number): unknown; + cancel(handle: unknown): void; +} + +export function relayResponsesSseWithTerminalRepair( + body: ReadableStream, + upstream: AbortController, + policy: ResponsesTerminalRepairPolicy, + budget: TranslatorBudget, + scheduler: ResponsesTerminalRepairScheduler = systemScheduler, +): ReadableStream; +``` + +The default scheduler wraps `Date.now()`, `setTimeout`, and `clearTimeout`. +The relay must: + +- 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: \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. + +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. + +- [ ] **Step 4: Run the two tests and verify GREEN** + +Run: + +```bash +bun test tests/responses-terminal-repair.test.ts +``` + +Expected: healthy pass-through and five-second grace completion both pass; +translator-budget current bytes return to zero after drain. + +- [ ] **Step 5: Commit** + +```bash +git add src/server/responses-terminal-repair.ts tests/responses-terminal-repair.test.ts +git commit -m "feat(responses): repair complete terminal-less streams" +``` + +--- + +### Task 3: Harden the terminal-repair state machine + +**Files:** +- Modify: `src/server/responses-terminal-repair.ts` +- Modify: `tests/responses-terminal-repair.test.ts` + +**Interfaces:** +- Preserves Task 2 public signatures. +- Adds no user-facing configuration. + +- [ ] **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. + +- [ ] **Step 2: Run tests and verify RED** + +Run: + +```bash +bun test tests/responses-terminal-repair.test.ts +``` + +Expected: the newly added boundary tests fail because Task 2 implements only +the healthy and basic grace paths. + +- [ ] **Step 3: Implement strict item validation and singular terminal commitment** + +Implement these private rules: + +```ts +function isPlainRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function isCompleteItem(item: Record): boolean { + if (item.status !== "completed") return false; + if (item.type === "reasoning") { + return typeof item.id === "string" && item.id.length > 0 + && Array.isArray(item.content) + && item.content.every(part => isPlainRecord(part) + && part.type === "reasoning_text" && typeof part.text === "string"); + } + if (item.type === "message") { + return typeof item.id === "string" && item.id.length > 0 + && item.role === "assistant" && Array.isArray(item.content) + && item.content.every(part => isPlainRecord(part) + && part.type === "output_text" && typeof part.text === "string"); + } + if (item.type === "function_call") { + if (typeof item.id !== "string" || item.id.length === 0) return false; + if (typeof item.call_id !== "string" || item.call_id.length === 0) return false; + if (typeof item.name !== "string" || item.name.length === 0) return false; + if (typeof item.arguments !== "string") return false; + try { + const parsed = JSON.parse(item.arguments) as unknown; + return isPlainRecord(parsed); + } catch { return false; } + } + return false; +} +``` + +Require a valid created snapshot, at least one done item, exact added/done index +parity, no taint, and all done items passing `isCompleteItem()`. + +Implement one `commitTerminal(kind)` gate. For `completed`, build the response +from created metadata with ordered output, `status:"completed"`, injected +`completed_at`, and next sequence number. For incomplete EOF/DONE, emit a +canonical `response.incomplete` with `incomplete_details.reason` set to +`"missing_terminal_event"`. + +Every new non-terminal event increments a generation counter and cancels the +old timer. Timer callbacks capture the generation and re-check terminal, +abort/cancel, taint, and candidate completeness immediately before enqueueing. + +- [ ] **Step 4: Run the hardening tests and verify GREEN** + +Run: + +```bash +bun test tests/responses-terminal-repair.test.ts tests/sse-failed-tail.test.ts tests/relay-eager.test.ts +``` + +Expected: all tests pass with no duplicate terminal, timer leak, or retained +budget after teardown. + +- [ ] **Step 5: Commit** + +```bash +git add src/server/responses-terminal-repair.ts tests/responses-terminal-repair.test.ts +git commit -m "fix(responses): fail closed on unsafe terminal repair" +``` + +--- + +### Task 4: Integrate repair before HTTP/WebSocket transport branching + +**Files:** +- Modify: `src/server/responses/core.ts:95-115` +- Modify: `src/server/responses/core.ts:2030-2230` +- Modify: `tests/deepseek-inbound-wire.test.ts:115-330` +- Modify: `tests/deepseek-responses-item-id-repair.test.ts` +- Modify: `tests/ws-endpoint.test.ts` + +**Interfaces:** +- Consumes: `providerModelResponsesTerminalRepair(...)` +- Consumes: `relayResponsesSseWithTerminalRepair(...)` +- Adds: `HandleResponsesOptions.responsesTerminalRepairScheduler?` as a narrow + clock/timer dependency injection seam used by deterministic integration tests. +- Preserves: existing payload/block rewrite composition and terminal inspection. + +- [ ] **Step 1: Write failing HTTP progressive-delivery and repair-composition tests** + +Replace the old bounded-JSON DeepSeek fixture with an SSE source that exposes +manual `push()` and `close()` controls. Assert: + +```ts +expect(capturedRequest.body.stream).toBe(true); + +source.push(reasoningDeltaBlock); +const firstRead = await reader.read(); +expect(new TextDecoder().decode(firstRead.value)).toContain("response.reasoning_text.delta"); + +source.push(functionCallDoneBlock); +scheduler.advance(5_000); +const remainder = await drainReader(reader); +expect(remainder).toContain("response.completed"); +expect(remainder).toContain("data: [DONE]"); +``` + +The fixture must use UUID reasoning/message ids and assert that existing item-id +repair rewrites added/delta/done/synthetic-terminal payloads consistently while +leaving `function_call.id` and `call_id` unchanged. + +- [ ] **Step 2: Add a failing WebSocket parity test** + +Drive the same terminal-less complete function call through `/v1/responses` +WebSocket handling. Assert the client receives progressive delta frames, one +`response.output_item.done`, and one `response.completed`, and that a second +`response.create` carrying `function_call_output` remains accepted. + +- [ ] **Step 3: Run tests and verify RED** + +Run: + +```bash +bun test tests/deepseek-inbound-wire.test.ts tests/ws-endpoint.test.ts +``` + +Expected: requests now carry `stream:true` from Task 1 but no provider-scoped +repair is activated, so the terminal-less fixture does not close at the injected +grace boundary. + +- [ ] **Step 4: Wrap the upstream SSE body before existing branches** + +Import both new resolvers and build one body before the eager/tee split: + +```ts +const terminalRepairPolicy = providerModelResponsesTerminalRepair( + route.providerName, + route.provider, + route.modelId, +); +const passthroughSseBody = terminalRepairPolicy + ? relayResponsesSseWithTerminalRepair( + upstreamResponse.body, + upstream, + terminalRepairPolicy, + translatorBudget, + options.responsesTerminalRepairScheduler, + ) + : upstreamResponse.body; +``` + +Add the optional scheduler to `HandleResponsesOptions`: + +```ts +/** Internal deterministic clock/timer seam for provider terminal repair. */ +responsesTerminalRepairScheduler?: ResponsesTerminalRepairScheduler; +``` + +Use `passthroughSseBody` in both: + +- the eager single-reader call to `relaySseEagerBounded()`; +- the default `passthroughSseBody.tee()` path. + +Do not place the wrapper only on the client branch: background inspection and +continuation persistence must see the synthetic terminal too. Leave the +existing JSON branch and bounded-JSON synthesis intact for providers whose +streaming resolver still returns `false`. + +- [ ] **Step 5: Run HTTP/WebSocket and continuation tests and verify GREEN** + +Run: + +```bash +bun test tests/deepseek-inbound-wire.test.ts tests/ws-endpoint.test.ts tests/responses-state.test.ts tests/deepseek-responses-item-id-repair.test.ts tests/deepseek-reasoning-replay.test.ts +``` + +Expected: progressive output precedes terminal, both transports close once, +item ids are stable, and continuation state retains reasoning/function output. + +- [ ] **Step 6: Commit** + +```bash +git add src/server/responses/core.ts tests/deepseek-inbound-wire.test.ts tests/deepseek-responses-item-id-repair.test.ts tests/ws-endpoint.test.ts +git commit -m "fix(deepseek): repair terminal-less Responses streams" +``` + +--- + +### Task 5: Synchronize architecture documentation and remove stale assertions + +**Files:** +- Modify: `structure/04_transports-and-sidecars.md:205-225` +- Modify: `src/providers/registry.ts:1150-1165` + +**Interfaces:** +- Documents the registry-only `modelResponsesTerminalRepair` policy and rollback path. + +- [ ] **Step 1: Find stale policy wording** + +Run: + +```bash +rg -n "DeepSeek.*bounded|bounded.*DeepSeek|modelResponsesUpstreamStreaming|stream:false" structure src tests docs-site +``` + +Expected: identify every statement that specifically claims the built-in +DeepSeek route is forced to bounded JSON. + +- [ ] **Step 2: Update documentation and comments** + +Document that official DeepSeek uses native Responses streaming and a +provider/model-scoped five-second repair only after complete output items. +State that the bounded-JSON mechanism remains available for other providers and +as a rollback capability. + +Do not describe synthetic completion as a generic Responses behavior. + +- [ ] **Step 3: Run repository hygiene checks** + +Run: + +```bash +git diff --check +bun run privacy:scan +``` + +Expected: both exit 0; no local path, credential, raw probe id, or private +prompt appears in tracked files. + +- [ ] **Step 4: Commit** + +```bash +git add structure/04_transports-and-sidecars.md src/providers/registry.ts +git commit -m "docs(deepseek): describe streaming terminal repair" +``` + +--- + +### Task 6: Complete verification and live smoke test + +**Files:** +- No planned source changes; fix only defects exposed by verification, each with a RED test first. + +**Interfaces:** +- Verifies every acceptance criterion in the approved design. + +- [ ] **Step 1: Run the focused regression matrix** + +Run: + +```bash +bun test tests/responses-terminal-repair.test.ts tests/deepseek-inbound-wire.test.ts tests/ws-endpoint.test.ts tests/sse-failed-tail.test.ts tests/relay-eager.test.ts tests/responses-item-id-repair.test.ts tests/deepseek-responses-item-id-repair.test.ts tests/deepseek-reasoning-replay.test.ts tests/responses-state.test.ts +``` + +Expected: 0 failures. + +- [ ] **Step 2: Run typecheck and full repository gates** + +Run: + +```bash +bun run typecheck +bun run test +bun run privacy:scan +bun run prepush +``` + +Expected: every command exits 0. Record exact pass/skip/fail counts from the +fresh `prepush` output. + +- [ ] **Step 3: Run one minimal official-DeepSeek smoke through the new code** + +Start an isolated one-off opencodex server from this worktree on an unused +loopback port, using the existing local config without printing its API key. +Send a prompt containing no private data and a no-op function tool. Verify: + +- the upstream request remains `stream:true`; +- at least one reasoning/function delta reaches the client before terminal; +- exactly one real `response.completed` and one `[DONE]` arrive; +- the request log status is 200 and `firstOutputMs` is populated; +- no `upstream JSON response stalled before completing` error occurs. + +Stop only the one-off process; do not restart or replace the installed service +until the user separately authorizes deployment. + +- [ ] **Step 4: Review final diff and branch state** + +Run: + +```bash +git status -sb +git diff origin/dev...HEAD --check +git diff origin/dev...HEAD --stat +git log --oneline origin/dev..HEAD +``` + +Expected: only the design, plan, targeted source, tests, and architecture doc +are changed; working tree is clean. + +--- + +### Task 7: Publish the independent pull request + +**Files:** +- No local code changes expected. + +**Interfaces:** +- Produces an independent PR targeting `lidge-jun/opencodex:dev`. + +- [ ] **Step 1: Rebase or merge the latest `origin/dev` only if required** + +Fetch current `origin/dev`, check ancestry, and update the branch without +touching PR #1047. If baseline movement creates conflicts, resolve only within +this branch and rerun Task 6 gates. + +- [ ] **Step 2: Push the branch to the user's fork** + +```bash +git push -u fork agent/fix-deepseek-responses-streaming +``` + +- [ ] **Step 3: Open a draft PR using the repository template** + +Target `dev`. The PR summary must state: + +- the observed 30-second bounded-JSON failure mode; +- the 2026-08-06 official-stream capture showing a valid terminal; +- the provider-scoped five-second completion repair; +- fail-closed conditions and transport parity; +- exact local verification counts. + +Do not include API keys, local paths, private prompts, account identifiers, or +the user's production conversation data. + +- [ ] **Step 4: Verify PR state** + +Confirm target branch, head SHA, template completeness, CI/check state, and +that the PR is independent of #1047. Leave it draft until the repository's +review-readiness checklist is satisfied against the exact final SHA. diff --git a/docs/superpowers/specs/2026-08-06-deepseek-responses-streaming-terminal-repair-design.md b/docs/superpowers/specs/2026-08-06-deepseek-responses-streaming-terminal-repair-design.md new file mode 100644 index 0000000000..9acf1aab7e --- /dev/null +++ b/docs/superpowers/specs/2026-08-06-deepseek-responses-streaming-terminal-repair-design.md @@ -0,0 +1,317 @@ +# DeepSeek Responses Streaming Terminal Repair Design + +## Goal + +Restore progressive Responses streaming for the built-in `deepseek-v4-flash` +route while keeping Codex turns bounded and terminal-complete when an upstream +stream finishes its output items but omits or indefinitely delays the protocol +terminal event. + +The fix must remove the current slow-JSON failure mode without weakening +failure honesty: only a fully closed, structurally valid output graph may be +promoted to a synthetic `response.completed` event. + +## Context and evidence + +The built-in DeepSeek registry currently declares +`modelResponsesUpstreamStreaming["deepseek-v4-flash"] = false`. Final route +normalization therefore changes `stream:true` to `stream:false` before sending +the request to `POST https://api.deepseek.com/responses`. For an HTTP client +that requested streaming, opencodex waits for the complete JSON response and +only then reframes it as SSE. + +That compatibility policy was introduced for a historical DeepSeek stream that +could deliver output without closing on a Responses terminal event. It now has +two observable costs: + +1. Codex receives no progressive deltas. +2. Non-streaming JSON is guarded by a 30-second body inactivity deadline. Long + context or high-reasoning turns can cross that boundary and fail with + `upstream JSON response stalled before completing` even while the upstream + is still generating a legitimate response. + +A live, minimal capture against the official DeepSeek endpoint on 2026-08-06 +showed that the current `deepseek-v4-flash` stream emits the complete native +Responses lifecycle: + +1. `response.created` +2. reasoning item and reasoning deltas +3. `response.output_item.done` for reasoning +4. function-call argument deltas and `response.function_call_arguments.done` +5. `response.output_item.done` for the function call +6. `response.completed` + +The captured function-call turn completed in roughly eight seconds. This +supports restoring upstream streaming while retaining a provider-scoped repair +for regressions in terminal delivery. + +## Scope + +### In scope + +- The official built-in DeepSeek provider when the resolved model is + `deepseek-v4-flash` and the resolved wire is `openai-responses`. +- HTTP/SSE and Codex WebSocket clients. +- Native reasoning, message, and function-call output items. +- Existing client-facing item-id repair, continuation-state recording, usage + inspection, cancellation, and failed-tail behavior. +- A five-second post-completion grace window for a missing terminal event. + +### Out of scope + +- Changing Chat Completions behavior for DeepSeek, Claude Code, or other + OpenAI-compatible clients. +- Changing the global 30-second bounded-JSON inactivity limit. +- Removing the transport-neutral bounded-JSON capability; other providers may + still need it. +- Treating silence, partial output, malformed tool calls, or unknown output item + types as success. +- Retrying a committed upstream generation. +- Adding a user-facing configuration option in this change. + +## Options considered + +### Remove the non-streaming override only + +This restores progressive output with the smallest diff, but a future terminal +regression would again leave Codex waiting after otherwise complete output. + +### Repair terminal events in the shared relay for every provider + +This covers more gateways but changes global Responses semantics. A heuristic +safe for DeepSeek may be incorrect for another provider, so the blast radius is +not justified. + +### Provider-scoped streaming terminal repair + +This is the selected design. DeepSeek returns to native streaming, while a +registry-only model capability opts its stream into a narrowly defined repair +state machine. Other providers and custom DeepSeek-compatible endpoints remain +unchanged unless they match the built-in registry transport and model. + +## Architecture + +### Registry policy + +Replace DeepSeek's forced non-streaming entry with a registry-only terminal +repair policy for `deepseek-v4-flash`. The policy carries the five-second grace +duration and is resolved only when `providerMatchesRegistryTransport()` accepts +the built-in provider transport. + +The policy is intentionally not persisted into user configuration. It is a +compatibility fact about the official endpoint, analogous to the existing +registry-only wire and streaming hints. + +### Terminal repair stream + +Add a focused module under `src/server/` that wraps a native Responses SSE body. +It has one responsibility: relay complete SSE blocks while tracking whether a +safe synthetic terminal can be emitted. + +The wrapper runs before the body is split for client delivery and background +inspection. Consequently, both branches observe the same real or synthetic +terminal. Existing item-id repair, lifecycle snapshot repair, request logging, +continuation recording, HTTP/SSE delivery, and WebSocket reframing remain +downstream and keep their current ownership. + +The wrapper must use existing SSE framing helpers and the per-turn translator +budget. It may retain only the response-created metadata and completed output +items required to construct a terminal response. Retained state is released on +every terminal, EOF, cancellation, error, and disposal path. + +### Data flow + +```text +DeepSeek Responses SSE + -> provider-scoped terminal repair + -> existing payload/block rewrites (item ids, image calls, snapshots) + -> existing failed-tail and terminal-boundary relay + -> Codex HTTP/SSE or WebSocket client + +The repaired stream is also inspected for: + -> request outcome and usage metadata + -> completed-response continuation state +``` + +## Completion state machine + +### Tracked state + +- The most recent valid `response.created.response` object. +- The highest valid numeric `sequence_number` seen. +- Every valid `response.output_item.added`, keyed by `output_index`. +- Every valid `response.output_item.done`, keyed by `output_index`. +- Whether a real `response.completed`, `response.failed`, or + `response.incomplete` event has arrived. +- Whether a real `data: [DONE]` event has arrived. +- One generation token for the active grace timer, preventing a stale timer + from committing after later activity. + +### Candidate-complete predicate + +A stream is eligible for synthetic success only when all conditions hold: + +1. No real Responses terminal has been observed. +2. A valid `response.created` event with an object-valued response snapshot + has been observed. +3. At least one `response.output_item.done` has been observed. +4. Every added output index has exactly one corresponding done item. +5. No done item exists for an index whose lifecycle is contradictory or + tainted by malformed duplicate events. +6. Every retained item has `status: "completed"`. +7. Item types are limited to `reasoning`, `message`, and `function_call`. +8. A function call has a non-empty `name`, non-empty `call_id`, string + `arguments`, and arguments that parse as JSON. +9. A message contains only completed output content carried by its done item. +10. The retained state remains inside the existing per-turn translator budget. + +Any malformed, contradictory, oversized, or unsupported item permanently +taints synthetic success for that stream. A later real upstream terminal stays +authoritative and is still relayed. + +### Grace behavior + +When the candidate-complete predicate first becomes true, arm a five-second +timer. Any subsequent non-terminal SSE event invalidates that timer generation, +updates the state, and re-evaluates the predicate. If the stream remains a +complete candidate for the entire grace window, emit one synthetic +`response.completed` event and close the repaired source. + +The synthetic response is based on the created response metadata, with: + +- `status: "completed"` +- `completed_at` set from the injected clock +- `output` set to completed items ordered by `output_index` +- `usage` left unchanged when known and otherwise absent or null +- `sequence_number` set to the next valid sequence number + +After emitting the terminal, cancel the upstream reader. The existing +terminal-boundary relay appends exactly one `[DONE]` sentinel when necessary. + +### EOF and `[DONE]` + +- If EOF or `[DONE]` arrives with a complete candidate and no real terminal, + emit the synthetic completed terminal immediately before closing. +- If EOF or `[DONE]` arrives without a complete candidate, emit + `response.incomplete`, never `response.completed`. +- A mid-stream read error remains owned by the existing failed-tail relay and + becomes `response.failed`. + +### Terminal races + +A real terminal event always wins over the grace timer. Terminal commitment is +guarded by a single boolean transition, and both the timer callback and stream +reader re-check it immediately before enqueueing. Late events after terminal +commitment are dropped, and the upstream reader is cancelled. + +## Error and cancellation behavior + +- A client cancellation follows the existing client-gone and bounded drain + behavior. It never triggers synthetic success. +- A server shutdown abort suppresses synthetic terminal generation. +- A translator-budget overflow fails through the existing typed failure path; + retained repair state is released. +- A malformed SSE block is relayed according to existing passthrough behavior + but taints synthetic success when it affects lifecycle state. +- A real upstream `response.failed` or `response.incomplete` is byte-preserved + apart from already configured downstream rewrites. +- The repair never resends a request after output has been committed. + +## Integration details + +The implementation is expected to touch these responsibility boundaries: + +- `src/providers/registry.ts`: registry-only DeepSeek terminal-repair policy; + remove the official model's forced bounded-JSON streaming override. +- `src/server/responses-terminal-repair.ts`: bounded state machine and stream + wrapper. +- `src/server/responses/core.ts`: resolve the provider policy and wrap the SSE + body before transport-specific relay branches. +- `tests/responses-terminal-repair.test.ts`: unit state-machine coverage. +- `tests/deepseek-inbound-wire.test.ts`: end-to-end wire, progressive delivery, + repair composition, and WebSocket/HTTP activation. +- Existing relay and item-id tests only where an explicit integration contract + needs to be pinned. +- `structure/04_transports-and-sidecars.md`: replace the bounded-JSON-only + DeepSeek description with the streaming plus provider-scoped repair policy. + +No unrelated refactor of `core.ts`, the shared relays, or provider configuration +is part of this change. + +## Test design + +### Unit activation + +1. A healthy captured-shape stream containing reasoning, a function call, and a + real `response.completed` is relayed without a synthetic terminal. +2. A complete reasoning plus function-call sequence that goes silent produces + one synthetic `response.completed` after the injected five-second deadline. +3. A new item during the grace window invalidates the old timer and restarts the + deadline only after the new item completes. +4. A clean EOF and a `[DONE]` event synthesize success immediately only for a + complete candidate. +5. Open items, invalid function arguments, unknown item types, contradictory + indices, malformed lifecycle frames, and budget overflow never synthesize + success. +6. Real completed, failed, and incomplete terminals beat the timer and remain + singular. +7. Client cancellation, upstream reset, and shutdown abort preserve their + existing accounting and terminal behavior. +8. Fragmented UTF-8 and SSE block boundaries produce the same state as a + single-chunk stream. + +Tests use an injected clock/timer seam rather than wall-clock sleeps. + +### Integration activation + +1. A Codex Responses request sends `stream:true` to the official DeepSeek + `/responses` endpoint. +2. The first reasoning or output delta is observable before the upstream emits + its terminal event. +3. A terminal-less, complete function-call fixture closes within the injected + grace deadline and preserves `call_id`, `name`, and `arguments`. +4. HTTP/SSE and WebSocket clients receive equivalent item and terminal + lifecycles. +5. UUID message and reasoning ids remain normalized consistently in added, + delta, done, and synthetic terminal payloads. +6. Chat and Anthropic inbound requests continue to use + `/chat/completions` with no terminal repair activation. + +### Verification gates + +- Focused terminal-repair, DeepSeek wire, relay, WebSocket, item-id, reasoning + replay, and continuation-state tests. +- `bun run typecheck` +- `bun run test` +- `bun run privacy:scan` +- `bun run prepush` +- One minimal live official-DeepSeek streaming smoke test, with no private + prompt content and no credential output. + +## Rollout and compatibility + +The change is provider- and model-scoped. Official DeepSeek Responses clients +gain progressive streaming; Chat and Anthropic clients are unchanged. Custom +providers that happen to use the name `deepseek` but do not match the registry +transport do not inherit the repair. + +The bounded-JSON machinery remains available as a rollback path. If live or CI +evidence reveals an unsafe terminal synthesis condition, the registry can +restore `modelResponsesUpstreamStreaming: false` without changing shared relay +behavior. + +## Acceptance criteria + +- Official `deepseek-v4-flash` Responses requests remain `stream:true` + upstream. +- Codex receives progressive reasoning, text, and function-call deltas. +- Normal live streams preserve the upstream terminal without duplication. +- A fully complete terminal-less output graph closes after five seconds with + exactly one synthetic `response.completed` and one `[DONE]`. +- Partial, malformed, tainted, or unsupported output never becomes synthetic + success. +- Function calls remain executable and continuation state retains completed + reasoning and output items. +- HTTP/SSE and WebSocket behavior agree. +- Existing providers and DeepSeek Chat Completions behavior remain unchanged. +- Focused and full repository verification gates pass. diff --git a/src/providers/registry.ts b/src/providers/registry.ts index 20518adb46..8c3e6d21ae 100644 --- a/src/providers/registry.ts +++ b/src/providers/registry.ts @@ -32,6 +32,11 @@ export type InboundWire = "responses" | "chat" | "anthropic"; */ export type ModelWireDefault = string | { wire: string; inbound: readonly InboundWire[] }; +export interface ResponsesTerminalRepairPolicy { + /** Quiet time after a structurally complete output graph before synthesizing completion. */ + graceMs: number; +} + export type ProviderModelDiscoveryScalar = string | number | boolean; export type ProviderModelDiscoveryPredicate = @@ -162,6 +167,8 @@ export interface ProviderRegistryEntry { * can omit or indefinitely delay the terminal event. */ modelResponsesUpstreamStreaming?: Record; + /** Registry-only repair for a model whose native Responses stream may omit its terminal. */ + modelResponsesTerminalRepair?: Record; /** * Registry-only client-facing item-id repair policy (#938), filled onto the * runtime provider only when the user has no explicit policy (derive.ts); @@ -1307,10 +1314,9 @@ export const PROVIDER_REGISTRY: readonly ProviderRegistryEntry[] = [ // for no gain. "deepseek-v4-flash": { wire: "openai-responses", inbound: ["responses"] }, }, - // DeepSeek's Codex Responses stream can deliver output without closing on the - // terminal event. Keep Codex on WebSocket, but use the provider's bounded JSON - // response upstream so the bridge can synthesize a complete WS event sequence. - modelResponsesUpstreamStreaming: { "deepseek-v4-flash": false }, + // Current official streams deliver progressive reasoning/tool deltas and a real + // terminal. Retain a narrow grace repair for the historical missing-terminal shape. + modelResponsesTerminalRepair: { "deepseek-v4-flash": { graceMs: 5_000 } }, // DeepSeek's Responses route emits bare UUID item ids, which leave Codex // clients stuck on an uncommitted turn (#938). Client-facing only — raw // continuation snapshots keep the upstream ids. @@ -2172,6 +2178,19 @@ export function providerModelResponsesUpstreamStreaming( return entry.modelResponsesUpstreamStreaming[modelId.trim().toLowerCase()]; } +/** Resolve a registry-only terminal-repair policy for native Responses streams. */ +export function providerModelResponsesTerminalRepair( + id: string, + provider: Pick & Partial>, + modelId: string, +): ResponsesTerminalRepairPolicy | undefined { + const entry = getProviderRegistryEntry(id); + if (!entry?.modelResponsesTerminalRepair || !providerMatchesRegistryTransport(id, provider)) return undefined; + const policy = entry.modelResponsesTerminalRepair[modelId.trim().toLowerCase()]; + if (!policy || !Number.isFinite(policy.graceMs) || policy.graceMs <= 0) return undefined; + return { graceMs: Math.floor(policy.graceMs) }; +} + /** * Effective Codex account mode for a provider. For canonical `openai`, a valid persisted * `codexAccountMode` on the provider config wins and a missing/invalid value defaults to diff --git a/src/server/responses-terminal-repair.ts b/src/server/responses-terminal-repair.ts new file mode 100644 index 0000000000..4d6749a913 --- /dev/null +++ b/src/server/responses-terminal-repair.ts @@ -0,0 +1,342 @@ +import type { TranslatorBudget } from "../lib/translator-budget"; +import type { ResponsesTerminalRepairPolicy } from "../providers/registry"; +import { nextSseBlock, sseDataPayload } from "./sse-payload-rewrite"; + +export interface ResponsesTerminalRepairScheduler { + nowMs(): number; + schedule(callback: () => void, delayMs: number): unknown; + cancel(handle: unknown): void; +} + +const systemScheduler: ResponsesTerminalRepairScheduler = { + nowMs: () => Date.now(), + schedule(callback, delayMs) { + const handle = setTimeout(callback, delayMs); + (handle as { unref?: () => void }).unref?.(); + return handle; + }, + cancel(handle) { clearTimeout(handle as ReturnType); }, +}; + +function isPlainRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function outputIndex(value: unknown): number | null { + return Number.isInteger(value) && (value as number) >= 0 ? value as number : null; +} + +function isCompleteItem(item: Record): boolean { + if (item.status !== "completed") return false; + if (item.type === "reasoning") { + return typeof item.id === "string" && item.id.length > 0 + && Array.isArray(item.content) + && item.content.every(part => isPlainRecord(part) + && part.type === "reasoning_text" && typeof part.text === "string"); + } + if (item.type === "message") { + return typeof item.id === "string" && item.id.length > 0 + && item.role === "assistant" && Array.isArray(item.content) + && item.content.every(part => isPlainRecord(part) + && part.type === "output_text" && typeof part.text === "string"); + } + if (item.type === "function_call") { + if (typeof item.id !== "string" || item.id.length === 0) return false; + if (typeof item.call_id !== "string" || item.call_id.length === 0) return false; + if (typeof item.name !== "string" || item.name.length === 0) return false; + if (typeof item.arguments !== "string") return false; + try { + return isPlainRecord(JSON.parse(item.arguments)); + } catch { + return false; + } + } + return false; +} + +/** + * Relay a native Responses SSE body while repairing the narrow DeepSeek shape where every + * output item is complete but the protocol terminal is missing or indefinitely delayed. + */ +export function relayResponsesSseWithTerminalRepair( + body: ReadableStream, + upstream: AbortController, + policy: ResponsesTerminalRepairPolicy, + budget: TranslatorBudget, + scheduler: ResponsesTerminalRepairScheduler = systemScheduler, +): ReadableStream { + const reader = body.getReader(); + const decoder = new TextDecoder(); + const encoder = new TextEncoder(); + const added = new Map(); + const completed = new Map; bytes: number }>(); + let created: Record | null = null; + let createdBytes = 0; + let maxSequence = -1; + let buffer = ""; + let bufferBytes = 0; + let timer: unknown; + let timerGeneration = 0; + let realTerminalSeen = false; + let tainted = false; + let disposed = false; + let controllerRef: ReadableStreamDefaultController | null = null; + let activeRead: Promise | null = null; + + const onUpstreamAbort = (): void => { + if (disposed) return; + dispose(); + reader.cancel(upstream.signal.reason).catch(() => {}); + try { controllerRef?.close(); } catch { /* already closed */ } + }; + + const releaseRetainedState = (): void => { + if (createdBytes > 0) budget.releaseRetained(createdBytes, { kind: "retained_collectors" }); + createdBytes = 0; + for (const retained of completed.values()) { + budget.releaseRetained(retained.bytes, { kind: "retained_collectors" }); + } + completed.clear(); + created = null; + }; + + const releaseBuffer = (): void => { + if (bufferBytes > 0) budget.releaseRetained(bufferBytes, { kind: "live_transient" }); + buffer = ""; + bufferBytes = 0; + }; + + const cancelTimer = (): void => { + timerGeneration += 1; + if (timer !== undefined) scheduler.cancel(timer); + timer = undefined; + }; + + const dispose = (): void => { + if (disposed) return; + disposed = true; + upstream.signal.removeEventListener("abort", onUpstreamAbort); + cancelTimer(); + releaseRetainedState(); + releaseBuffer(); + }; + + const replaceBuffer = (next: string): void => { + const nextBytes = encoder.encode(next).byteLength; + const reservation = budget.reserveTransient(nextBytes, { kind: "live_transient" }); + reservation.commitRetained(); + if (bufferBytes > 0) budget.releaseRetained(bufferBytes, { kind: "live_transient" }); + buffer = next; + bufferBytes = nextBytes; + }; + + const appendBuffer = (fragment: string): void => { + if (!fragment) return; + replaceBuffer(buffer + fragment); + }; + + const retainCreated = (response: Record): void => { + const bytes = encoder.encode(JSON.stringify(response)).byteLength; + budget.chargeRetained(bytes, { kind: "retained_collectors" }); + if (createdBytes > 0) budget.releaseRetained(createdBytes, { kind: "retained_collectors" }); + created = response; + createdBytes = bytes; + }; + + const retainCompleted = (index: number, item: Record): void => { + const bytes = encoder.encode(JSON.stringify(item)).byteLength; + budget.chargeRetained(bytes, { kind: "retained_collectors" }); + const previous = completed.get(index); + if (previous) budget.releaseRetained(previous.bytes, { kind: "retained_collectors" }); + completed.set(index, { item, bytes }); + }; + + const completeCandidate = (): boolean => { + if (realTerminalSeen || tainted || !created || completed.size === 0 || added.size !== completed.size) return false; + for (const index of added.keys()) { + const retained = completed.get(index); + if (!retained || !isCompleteItem(retained.item)) return false; + } + for (const index of completed.keys()) if (!added.has(index)) return false; + return true; + }; + + const syntheticTerminal = (kind: "completed" | "incomplete"): Uint8Array => { + const output = [...completed.entries()] + .sort(([left], [right]) => left - right) + .map(([, retained]) => retained.item); + const response = { + ...(created ?? {}), + status: kind, + completed_at: Math.floor(scheduler.nowMs() / 1_000), + output, + ...(kind === "incomplete" + ? { incomplete_details: { reason: "missing_terminal_event" } } + : {}), + }; + const type = `response.${kind}`; + return encoder.encode(`event: ${type}\ndata: ${JSON.stringify({ + type, + response, + sequence_number: maxSequence + 1, + })}\n\n`); + }; + + const emitSynthetic = ( + kind: "completed" | "incomplete", + controller: ReadableStreamDefaultController, + ): boolean => { + if (disposed || realTerminalSeen) return false; + realTerminalSeen = true; + cancelTimer(); + controller.enqueue(syntheticTerminal(kind)); + releaseRetainedState(); + return true; + }; + + const commitSynthetic = (generation: number): void => { + if (disposed || realTerminalSeen || generation !== timerGeneration || !completeCandidate()) return; + timer = undefined; + try { + if (controllerRef && emitSynthetic("completed", controllerRef)) controllerRef.close(); + } catch { + /* downstream already closed */ + } + reader.cancel("Responses terminal repaired after complete output").catch(() => {}); + dispose(); + }; + + const maybeArmTimer = (): void => { + if (!completeCandidate()) return; + const generation = timerGeneration; + timer = scheduler.schedule(() => commitSynthetic(generation), policy.graceMs); + }; + + const inspectPayload = (payload: string | null): "done" | "ordinary" => { + if (payload === "[DONE]") return "done"; + if (!payload || realTerminalSeen) return "ordinary"; + let parsed: unknown; + try { + parsed = JSON.parse(payload); + } catch { + tainted = true; + return "ordinary"; + } + if (!isPlainRecord(parsed)) { + tainted = true; + return "ordinary"; + } + if (Number.isInteger(parsed.sequence_number)) { + maxSequence = Math.max(maxSequence, parsed.sequence_number as number); + } + const type = parsed.type; + if (type === "response.completed" || type === "response.failed" || type === "response.incomplete") { + realTerminalSeen = true; + cancelTimer(); + releaseRetainedState(); + return "ordinary"; + } + + cancelTimer(); + if (type === "response.created" && isPlainRecord(parsed.response)) { + retainCreated(parsed.response); + } else if (type === "response.output_item.added") { + const index = outputIndex(parsed.output_index); + if (index === null || !isPlainRecord(parsed.item) || added.has(index) || completed.has(index)) { + tainted = true; + } else { + added.set(index, { type: parsed.item.type, id: parsed.item.id }); + } + } else if (type === "response.output_item.done") { + const index = outputIndex(parsed.output_index); + if (index === null || !isPlainRecord(parsed.item) || !added.has(index) || completed.has(index)) { + tainted = true; + } else { + const opened = added.get(index)!; + if (opened.type !== parsed.item.type || opened.id !== parsed.item.id) tainted = true; + retainCompleted(index, parsed.item); + } + } + maybeArmTimer(); + return "ordinary"; + }; + + const emitBlocks = ( + controller: ReadableStreamDefaultController, + ): { closed: boolean; emitted: boolean } => { + let next: ReturnType; + let emitted = false; + while ((next = nextSseBlock(buffer))) { + replaceBuffer(next.rest); + const kind = inspectPayload(sseDataPayload(next.block)); + if (kind === "done" && !realTerminalSeen) { + emitSynthetic(completeCandidate() ? "completed" : "incomplete", controller); + } + controller.enqueue(encoder.encode(next.block + next.delimiter)); + emitted = true; + if (kind === "done") { + reader.cancel("Responses stream ended with DONE").catch(() => {}); + dispose(); + controller.close(); + return { closed: true, emitted }; + } + } + return { closed: false, emitted }; + }; + + const readOnce = async (controller: ReadableStreamDefaultController): Promise => { + try { + for (;;) { + const { done, value } = await reader.read(); + if (disposed) return; + if (done) { + appendBuffer(decoder.decode()); + if (buffer.length > 0) { + const kind = inspectPayload(sseDataPayload(buffer)); + if (kind === "done" && !realTerminalSeen) { + emitSynthetic(completeCandidate() ? "completed" : "incomplete", controller); + } + controller.enqueue(encoder.encode(buffer)); + } + if (!realTerminalSeen) { + emitSynthetic(completeCandidate() ? "completed" : "incomplete", controller); + } + releaseBuffer(); + dispose(); + controller.close(); + return; + } + appendBuffer(decoder.decode(value, { stream: true })); + const result = emitBlocks(controller); + if (result.closed || result.emitted) return; + } + } catch (error) { + if (disposed) return; + dispose(); + controller.error(error); + } + }; + + return new ReadableStream({ + start(controller) { + controllerRef = controller; + if (upstream.signal.aborted) { + onUpstreamAbort(); + return; + } + upstream.signal.addEventListener("abort", onUpstreamAbort, { once: true }); + }, + pull(controller) { + if (disposed) return; + if (!activeRead) { + activeRead = readOnce(controller).finally(() => { activeRead = null; }); + } + return activeRead; + }, + cancel(reason) { + dispose(); + upstream.abort(reason); + return reader.cancel(reason); + }, + }); +} diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 5396172082..aef08ecb85 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -103,7 +103,11 @@ import { applyOpenAiVirtualModel, resolveOpenAiCompactModel } from "../../provid import { isUsageDebugEnabled } from "../../usage/debug"; import { readJsonRequestBody, DecompressedBodyTooLargeError, UnsupportedContentEncodingError } from "../request-decompress"; import { resolveAdapter, resolveWireProtocolOverride } from "../adapter-resolve"; -import { providerModelResponsesUpstreamStreaming, type InboundWire } from "../../providers/registry"; +import { + providerModelResponsesTerminalRepair, + providerModelResponsesUpstreamStreaming, + type InboundWire, +} from "../../providers/registry"; import type { AdapterRequest } from "../../adapters/base"; import { hasKeyPoolFailover, @@ -159,6 +163,10 @@ import { sanitizePassthroughHeaders, } from "../relay"; import { relaySseEagerBounded } from "../relay-eager"; +import { + relayResponsesSseWithTerminalRepair, + type ResponsesTerminalRepairScheduler, +} from "../responses-terminal-repair"; import { isWin32EagerRewrite, selectEagerPath } from "../../lib/bun-stream-caps"; import { cancelBodyOnAbort } from "../../lib/abort"; import { @@ -578,6 +586,8 @@ export interface HandleResponsesOptions { setTerminalOutcomeRecorder?: (recorder: ((status: ResponsesTerminalStatus, httpStatusOverride?: number) => void) | undefined) => void; onNativePassthroughTerminal?: (status: ResponsesTerminalStatus) => void; onNativePassthroughCancel?: () => void; + /** Internal deterministic clock/timer seam for provider terminal repair. */ + responsesTerminalRepairScheduler?: ResponsesTerminalRepairScheduler; /** * When true, body `prompt_cache_key` is a Claude Desktop shared cache cohort * (system/tools hash), not a per-session id — do not use it for Anthropic pool affinity. @@ -2085,6 +2095,20 @@ async function handleResponsesInner( // devlog/_fin/260731_macos_rss_retention/100_darwin_eager_optin.md). // The bundled known-bad runtime remains on tee by default on both platforms. if (isEventStream && upstreamResponse.body) { + const terminalRepairPolicy = providerModelResponsesTerminalRepair( + route.providerName, + route.provider, + route.modelId, + ); + const passthroughSseBody = terminalRepairPolicy + ? relayResponsesSseWithTerminalRepair( + upstreamResponse.body, + upstream, + terminalRepairPolicy, + translatorBudget, + options.responsesTerminalRepairScheduler, + ) + : upstreamResponse.body; const repairConfig = route.provider.responsesItemIdRepair; const snapshotRepairEnabled = hasResponsesSnapshotRepair(route.provider.responsesSnapshotRepair); const githubCopilotRepairEnabled = route.providerName === "github-copilot"; @@ -2170,7 +2194,7 @@ async function handleResponsesInner( onFirstOutput: options.onFirstOutput, pinCompletedResponseIdToFirstSeen: githubCopilotRepairEnabled, }); - const eagerBody = relaySseEagerBounded(upstreamResponse.body, turnAc, { + const eagerBody = relaySseEagerBounded(passthroughSseBody, turnAc, { inspectChunk: chunk => inspector.feed(chunk), finishInspection: () => inspector.finish(), disposeInspection: () => inspector.dispose(), @@ -2206,7 +2230,7 @@ async function handleResponsesInner( })), ); } - const [nativeBody, inspectBody] = upstreamResponse.body.tee(); + const [nativeBody, inspectBody] = passthroughSseBody.tee(); const turnAc = new AbortController(); const clientGone = new AbortController(); linkAbortSignal(upstream, turnAc.signal); diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index 24535af60f..f82a175dd5 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -216,8 +216,19 @@ upstream Responses endpoint for bounded JSON on ANY client transport — WebSock HTTP/SSE. The bridge reframes that JSON into the same Responses event sequence (`src/server/responses-json-events.ts`): WS turns send the frames as WebSocket messages, while HTTP clients that requested streaming receive a synthesized terminal SSE body (created → -output_item.done → terminal → `[DONE]`). DeepSeek V4 Flash uses this path because its Codex -streaming response can deliver output without closing on a terminal event. +output_item.done → terminal → `[DONE]`). This remains available for providers that explicitly +declare the compatibility hint. + +DeepSeek V4 Flash instead keeps native Responses streaming for progressive reasoning, text, and +tool-call delivery. Its registry entry enables a model-scoped terminal repair before the existing +inspection/client split. A real `response.completed`, `response.failed`, or `response.incomplete` +event always passes through unchanged. If every opened output item has a structurally complete +`output_item.done` and no real terminal arrives for five seconds, the repair emits exactly one +`response.completed` snapshot and closes the upstream reader. EOF or `[DONE]` uses the same strict +completion check; open, malformed, duplicate, contradictory, or unknown output graphs fail closed +as `response.incomplete`, never synthetic success. The repair shares the per-turn translator byte +budget, preserves backpressure, and composes ahead of item-id/snapshot rewrites so HTTP/SSE and +WebSocket clients observe the same canonical lifecycle. `ws-bridge.ts` preserves upstream `failed` and `incomplete` status values in the final WebSocket frame rather than always emitting `response.completed`. If the response status is `failed`, a diff --git a/tests/deepseek-inbound-wire.test.ts b/tests/deepseek-inbound-wire.test.ts index b560c5c2b3..89bfe4d7ba 100644 --- a/tests/deepseek-inbound-wire.test.ts +++ b/tests/deepseek-inbound-wire.test.ts @@ -13,10 +13,15 @@ */ import { afterEach, describe, expect, test } from "bun:test"; import { providerConfigSeed } from "../src/providers/derive"; -import { getProviderRegistryEntry } from "../src/providers/registry"; +import { + getProviderRegistryEntry, + providerModelResponsesTerminalRepair, +} from "../src/providers/registry"; import { createResponsesPassthroughAdapter as createResponsesPassthroughAdapterProduction } from "../src/adapters/openai-responses"; import { resolveWireProtocolOverride } from "../src/server/adapter-resolve"; import { handleResponses } from "../src/server/responses/core"; +import type { ResponsesTerminalRepairScheduler } from "../src/server/responses-terminal-repair"; +import { sendResponseToWebSocket } from "../src/server/ws-bridge"; import type { OcxConfig, OcxProviderConfig } from "../src/types"; import { withTestTranslatorBudget } from "./helpers/translator-budget"; @@ -24,6 +29,68 @@ const createResponsesPassthroughAdapter = (...args: Parameters 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; + for (const [id, job] of [...this.jobs.entries()]) { + if (job.at > this.current || !this.jobs.delete(id)) continue; + job.callback(); + } + } +} + +function sse(event: Record): string { + return `event: ${String(event.type)}\ndata: ${JSON.stringify(event)}\n\n`; +} + +function controlledSse(): { + stream: ReadableStream; + push(text: string): void; + cancel(): void; +} { + let controller: ReadableStreamDefaultController | null = null; + return { + stream: new ReadableStream({ start(next) { controller = next; } }), + push(text) { controller?.enqueue(encoder.encode(text)); }, + cancel() { try { controller?.close(); } catch { /* already closed */ } }, + }; +} + +async function readUntil( + reader: ReadableStreamDefaultReader, + pattern: string, +): Promise { + let out = ""; + while (!out.includes(pattern)) { + const { done, value } = await reader.read(); + if (done) throw new Error(`stream closed before ${pattern}`); + out += decoder.decode(value, { stream: true }); + } + return out; +} + +async function drainReader(reader: ReadableStreamDefaultReader): Promise { + let out = ""; + for (;;) { + const { done, value } = await reader.read(); + if (done) return out + decoder.decode(); + out += decoder.decode(value, { stream: true }); + } +} function deepseekProvider(): OcxProviderConfig { return { ...providerConfigSeed(getProviderRegistryEntry("deepseek")!), apiKey: "sk-test" }; @@ -67,6 +134,13 @@ describe("DeepSeek wire selection is scoped to the inbound protocol", () => { .toBe("openai-chat"); } }); + + test("the official DeepSeek Responses route opts into terminal repair", () => { + const provider = deepseekProvider(); + expect(providerModelResponsesTerminalRepair("deepseek", provider, MODEL)).toEqual({ graceMs: 5_000 }); + expect(providerModelResponsesTerminalRepair("deepseek", provider, "deepseek-chat")).toBeUndefined(); + expect(providerModelResponsesTerminalRepair("custom-deepseek", provider, MODEL)).toBeUndefined(); + }); }); describe("the inbound scope survives the handleResponses replay", () => { @@ -126,20 +200,32 @@ describe("the inbound scope survives the handleResponses replay", () => { expect((await drive("chat")).url).toBe("https://api.deepseek.com/chat/completions"); }); - test("a Codex WebSocket turn asks DeepSeek for bounded JSON upstream", async () => { - const request = await drive("responses", "websocket"); - expect(request.url).toBe("https://api.deepseek.com/responses"); - expect(request.body.stream).toBe(false); + test("Codex HTTP and WebSocket turns keep DeepSeek streaming upstream", async () => { + const http = await drive("responses"); + const websocket = await drive("responses", "websocket"); + expect(http.url).toBe("https://api.deepseek.com/responses"); + expect(websocket.url).toBe("https://api.deepseek.com/responses"); + expect(http.body.stream).toBe(true); + expect(websocket.body.stream).toBe(true); }); - test("a Codex WebSocket turn keeps plain JSON downstream (no SSE synthesis)", async () => { - globalThis.fetch = (async () => Response.json({ - id: "resp_deepseek", - object: "response", - status: "completed", - output: [], - })) as typeof fetch; + test("HTTP streams a DeepSeek delta before safely repairing a missing terminal", async () => { + const source = controlledSse(); + const scheduler = new ManualTerminalScheduler(); + const requestBodies: Record[] = []; + const testAbort = new AbortController(); + globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + requestBodies.push(JSON.parse(String(init?.body ?? "{}")) as Record); + return new Response(source.stream, { + status: 200, + headers: { "content-type": "text/event-stream" }, + }); + }) as typeof fetch; const config = { providers: { deepseek: deepseekProvider() } } as unknown as OcxConfig; + const options = { + abortSignal: testAbort.signal, + responsesTerminalRepairScheduler: scheduler, + } as Parameters[3]; const response = await handleResponses( new Request("http://localhost/v1/responses", { method: "POST", @@ -148,47 +234,74 @@ describe("the inbound scope survives the handleResponses replay", () => { }), config, { model: "", provider: "" }, - { inboundTransport: "websocket" }, + options, ); - expect(response.headers.get("content-type")).not.toContain("text/event-stream"); - }); - - test("ordinary HTTP Responses requests also use bounded JSON upstream (#875)", async () => { - // The reliability policy is transport-neutral: DeepSeek's Responses stream can - // deliver output without a terminal, so HTTP turns get the same bounded JSON - // upstream as WS turns — and a synthesized terminal SSE back. - const request = await drive("responses"); - expect(request.body.stream).toBe(false); + const reader = response.body!.getReader(); + try { + source.push([ + sse({ type: "response.created", response: { id: "resp_http", status: "in_progress", output: [] }, sequence_number: 0 }), + sse({ type: "response.output_item.added", item: { type: "reasoning", id: "rs_http", status: "in_progress", content: [] }, output_index: 0, sequence_number: 1 }), + sse({ type: "response.reasoning_text.delta", item_id: "rs_http", output_index: 0, delta: "thinking", sequence_number: 2 }), + ].join("")); + const first = await readUntil(reader, "response.reasoning_text.delta"); + expect(first).toContain("thinking"); + expect(requestBodies[0]?.stream).toBe(true); + + source.push([ + sse({ + type: "response.output_item.done", + item: { type: "reasoning", id: "rs_http", status: "completed", content: [{ type: "reasoning_text", text: "thinking" }], summary: [] }, + output_index: 0, + sequence_number: 3, + }), + sse({ type: "response.output_item.added", item: { type: "function_call", id: "fc_http", status: "in_progress", arguments: "", call_id: "call_http", name: "probe" }, output_index: 1, sequence_number: 4 }), + sse({ type: "response.function_call_arguments.done", item_id: "fc_http", output_index: 1, arguments: "{\"text\":\"OK\"}", sequence_number: 5 }), + sse({ + type: "response.output_item.done", + item: { type: "function_call", id: "fc_http", status: "completed", arguments: "{\"text\":\"OK\"}", call_id: "call_http", name: "probe" }, + output_index: 1, + sequence_number: 6, + }), + ].join("")); + await Bun.sleep(0); + scheduler.advance(5_000); + const remainder = await Promise.race([ + drainReader(reader), + new Promise((_, reject) => setTimeout(() => reject(new Error("terminal repair did not close")), 200)), + ]); + expect(remainder).toContain("response.completed"); + expect(remainder).toContain("data: [DONE]"); + expect(remainder).toContain('"call_id":"call_http"'); + } finally { + testAbort.abort("test cleanup"); + source.cancel(); + try { await reader.cancel(); } catch { /* already closed */ } + } }); - test("an HTTP streaming client receives a synthesized terminal SSE instead of a stall (#875)", async () => { - globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { - const body = JSON.parse(String(init?.body ?? "{}")) as { stream?: boolean }; - if (body.stream === true) { - // Old world: a terminal-less SSE that never closes — the stall the issue - // reported. The policy must never send stream:true, so fail loudly here. - return new Response(new ReadableStream({ start() {} }), { + test("WebSocket delivery preserves progressive DeepSeek frames and accepts the repaired tool result", async () => { + const source = controlledSse(); + const scheduler = new ManualTerminalScheduler(); + const requestBodies: Record[] = []; + let requestNumber = 0; + globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + requestBodies.push(JSON.parse(String(init?.body ?? "{}")) as Record); + requestNumber += 1; + if (requestNumber === 1) { + return new Response(source.stream, { status: 200, headers: { "content-type": "text/event-stream" }, }); } return Response.json({ - id: "resp_deepseek", + id: "resp_ws_followup", object: "response", status: "completed", - output: [{ - type: "function_call", - id: "fc_1", - call_id: "call_1", - name: "search", - arguments: "{\"q\":\"docs\"}", - status: "completed", - }], + output: [], }); }) as typeof fetch; - const config = { providers: { deepseek: deepseekProvider() } } as unknown as OcxConfig; - const deadline = AbortSignal.timeout(5_000); + const abort = new AbortController(); const response = await handleResponses( new Request("http://localhost/v1/responses", { method: "POST", @@ -197,38 +310,85 @@ describe("the inbound scope survives the handleResponses replay", () => { }), config, { model: "", provider: "" }, - { abortSignal: deadline }, + { + abortSignal: abort.signal, + inboundTransport: "websocket", + responsesTerminalRepairScheduler: scheduler, + }, ); - expect(response.status).toBe(200); - expect(response.headers.get("content-type")).toContain("text/event-stream"); - const text = await response.text(); - const sequence = [...text.matchAll(/"type":"(response\.[^"]+)"/g)].map(match => match[1]); - expect(sequence).toEqual([ - "response.created", - "response.output_item.done", - "response.completed", - ]); - expect(text).toContain("data: [DONE]"); - // The function-call item survives with id/call_id byte-identical. - expect(text).toContain('"fc_1"'); - expect(text).toContain('"call_1"'); + const sent: string[] = []; + const ws = { + readyState: 1, + data: {}, + send(message: string) { sent.push(message); return 1; }, + } as Parameters[0]; + try { + const pump = sendResponseToWebSocket(ws, response, () => true); + source.push([ + sse({ type: "response.created", response: { id: "resp_ws", status: "in_progress", output: [] }, sequence_number: 0 }), + sse({ type: "response.output_item.added", item: { type: "reasoning", id: "rs_ws", status: "in_progress", content: [] }, output_index: 0, sequence_number: 1 }), + sse({ type: "response.reasoning_text.delta", item_id: "rs_ws", output_index: 0, delta: "thinking", sequence_number: 2 }), + ].join("")); + for (let i = 0; i < 20 && !sent.some(frame => JSON.parse(frame).type === "response.reasoning_text.delta"); i += 1) { + await Bun.sleep(0); + } + expect(sent.some(frame => JSON.parse(frame).type === "response.reasoning_text.delta")).toBe(true); + expect(sent.some(frame => JSON.parse(frame).type === "response.completed")).toBe(false); + + source.push([ + sse({ + type: "response.output_item.done", + item: { type: "reasoning", id: "rs_ws", status: "completed", content: [{ type: "reasoning_text", text: "thinking" }], summary: [] }, + output_index: 0, + sequence_number: 3, + }), + sse({ type: "response.output_item.added", item: { type: "function_call", id: "fc_ws", status: "in_progress", arguments: "", call_id: "call_ws", name: "probe" }, output_index: 1, sequence_number: 4 }), + sse({ type: "response.function_call_arguments.done", item_id: "fc_ws", output_index: 1, arguments: "{\"text\":\"OK\"}", sequence_number: 5 }), + sse({ + type: "response.output_item.done", + item: { type: "function_call", id: "fc_ws", status: "completed", arguments: "{\"text\":\"OK\"}", call_id: "call_ws", name: "probe" }, + output_index: 1, + sequence_number: 6, + }), + ].join("")); + 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; + + const eventTypes = sent.map(frame => JSON.parse(frame).type as string); + expect(eventTypes.filter(type => type === "response.completed")).toHaveLength(1); + expect(eventTypes).toContain("response.output_item.done"); + + const followup = await handleResponses( + new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: MODEL, + input: [ + { type: "function_call", id: "fc_ws", call_id: "call_ws", name: "probe", arguments: "{\"text\":\"OK\"}" }, + { type: "function_call_output", call_id: "call_ws", output: "OK" }, + ], + stream: true, + }), + }), + config, + { model: "", provider: "" }, + { inboundTransport: "websocket" }, + ); + await followup.text(); + expect(requestBodies[1]?.input).toEqual([ + { type: "function_call", call_id: "call_ws", name: "probe", arguments: "{\"text\":\"OK\"}" }, + { type: "function_call_output", call_id: "call_ws", output: "OK" }, + ]); + } finally { + abort.abort("test cleanup"); + source.cancel(); + } }); - /** - * Review finding on this layer: the bounded-JSON answer never touches the SSE - * relay, so it never picks up the relay's item-id rewrite. Without the - * normalization added here, enabling this reliability policy would silently - * DISABLE id repair for a provider that has it configured — the client would - * get canonical ids while streaming and placeholder ids the moment the policy - * switched the upstream to bounded JSON. - */ - function repairingProvider(): OcxProviderConfig { - return { - ...deepseekProvider(), - responsesItemIdRepair: { message: ["msg_placeholder"], reasoning: ["rs_placeholder"] }, - } as OcxProviderConfig; - } - function completedWithPlaceholderIds(): Response { return Response.json({ id: "resp_deepseek", @@ -247,30 +407,15 @@ describe("the inbound scope survives the handleResponses replay", () => { }); } - test("the synthesized terminal SSE carries repaired item ids, not the upstream placeholders", async () => { - globalThis.fetch = (async () => completedWithPlaceholderIds()) as typeof fetch; - const config = { providers: { deepseek: repairingProvider() } } as unknown as OcxConfig; - const response = await handleResponses( - new Request("http://localhost/v1/responses", { - method: "POST", - headers: { "content-type": "application/json" }, - body: JSON.stringify({ model: MODEL, input: "ping", stream: true }), - }), - config, - { model: "", provider: "" }, - { abortSignal: AbortSignal.timeout(5_000) }, - ); - expect(response.headers.get("content-type")).toContain("text/event-stream"); - const text = await response.text(); - expect(text).not.toContain("msg_placeholder"); - expect(text).not.toContain("rs_placeholder"); - expect(text).toMatch(/"id":"msg_ocx_[0-9a-f]{8}/); - expect(text).toMatch(/"id":"rs_ocx_[0-9a-f]{8}/); - }); - - test("the WebSocket bounded-JSON reframe carries the same repaired ids", async () => { - globalThis.fetch = (async () => completedWithPlaceholderIds()) as typeof fetch; - const config = { providers: { deepseek: repairingProvider() } } as unknown as OcxConfig; + test("streaming terminal repair composes with canonical item-id repair", async () => { + const source = controlledSse(); + const scheduler = new ManualTerminalScheduler(); + globalThis.fetch = (async () => new Response(source.stream, { + status: 200, + headers: { "content-type": "text/event-stream" }, + })) as typeof fetch; + const config = { providers: { deepseek: deepseekProvider() } } as unknown as OcxConfig; + const abort = new AbortController(); const response = await handleResponses( new Request("http://localhost/v1/responses", { method: "POST", @@ -279,15 +424,56 @@ describe("the inbound scope survives the handleResponses replay", () => { }), config, { model: "", provider: "" }, - { inboundWire: "responses", inboundTransport: "websocket" }, + { abortSignal: abort.signal, responsesTerminalRepairScheduler: scheduler }, ); - const text = await response.text(); - expect(text).not.toContain("msg_placeholder"); - expect(text).not.toContain("rs_placeholder"); - expect(text).toMatch(/"id":"msg_ocx_[0-9a-f]{8}/); + const reader = response.body!.getReader(); + const uuidReasoning = "1b9d6bcd-bbfd-4b2d-9b9d-5c0a2fb41a1b"; + const uuidMessage = "9b1deb4d-3b7d-4bad-9bdd-2b0d7b3dcb6d"; + const uuidFunction = "550e8400-e29b-41d4-a716-446655440000"; + try { + source.push([ + sse({ type: "response.created", response: { id: "resp_ids", status: "in_progress", output: [] }, sequence_number: 0 }), + sse({ type: "response.output_item.added", item: { type: "reasoning", id: uuidReasoning, status: "in_progress", content: [] }, output_index: 0, sequence_number: 1 }), + sse({ type: "response.output_item.done", item: { type: "reasoning", id: uuidReasoning, status: "completed", content: [{ type: "reasoning_text", text: "thinking" }], summary: [] }, output_index: 0, sequence_number: 2 }), + sse({ type: "response.output_item.added", item: { type: "message", id: uuidMessage, role: "assistant", status: "in_progress", content: [] }, output_index: 1, sequence_number: 3 }), + sse({ type: "response.output_text.delta", item_id: uuidMessage, output_index: 1, delta: "hello", sequence_number: 4 }), + sse({ type: "response.output_item.done", item: { type: "message", id: uuidMessage, role: "assistant", status: "completed", content: [{ type: "output_text", text: "hello" }] }, output_index: 1, sequence_number: 5 }), + sse({ type: "response.output_item.added", item: { type: "function_call", id: uuidFunction, status: "in_progress", arguments: "", call_id: "call_stream", name: "probe" }, output_index: 2, sequence_number: 6 }), + sse({ type: "response.function_call_arguments.done", item_id: uuidFunction, output_index: 2, arguments: "{\"text\":\"OK\"}", sequence_number: 7 }), + sse({ type: "response.output_item.done", item: { type: "function_call", id: uuidFunction, status: "completed", arguments: "{\"text\":\"OK\"}", call_id: "call_stream", name: "probe" }, output_index: 2, sequence_number: 8 }), + ].join("")); + const prefix = await readUntil(reader, '"sequence_number":8'); + scheduler.advance(5_000); + const text = prefix + await drainReader(reader); + const payloads = text + .split(/\r?\n/) + .filter(line => line.startsWith("data: ") && line !== "data: [DONE]") + .map(line => JSON.parse(line.slice(6)) as Record); + const reasoningAdded = payloads.find(event => event.type === "response.output_item.added" && event.output_index === 0)!; + const reasoningDone = payloads.find(event => event.type === "response.output_item.done" && event.output_index === 0)!; + const messageAdded = payloads.find(event => event.type === "response.output_item.added" && event.output_index === 1)!; + const messageDone = payloads.find(event => event.type === "response.output_item.done" && event.output_index === 1)!; + const completed = payloads.find(event => event.type === "response.completed")!; + const output = (completed.response as { output: Array<{ id: string; call_id?: string }> }).output; + const reasoningId = (reasoningAdded.item as { id: string }).id; + const messageId = (messageAdded.item as { id: string }).id; + expect(reasoningId).toMatch(/^rs_ocx_/); + expect(messageId).toMatch(/^msg_ocx_/); + expect((reasoningDone.item as { id: string }).id).toBe(reasoningId); + expect((messageDone.item as { id: string }).id).toBe(messageId); + expect(output[0]?.id).toBe(reasoningId); + expect(output[1]?.id).toBe(messageId); + expect(output[2]).toMatchObject({ id: uuidFunction, call_id: "call_stream" }); + expect(text).not.toContain(uuidReasoning); + expect(text).not.toContain(uuidMessage); + } finally { + abort.abort("test cleanup"); + source.cancel(); + try { await reader.cancel(); } catch { /* already closed */ } + } }); - test("a provider without id repair keeps the bounded-JSON body byte-identical", async () => { + test("a JSON fallback without id repair keeps its body byte-identical", async () => { globalThis.fetch = (async () => completedWithPlaceholderIds()) as typeof fetch; const config = { providers: { deepseek: deepseekProvider() } } as unknown as OcxConfig; const response = await handleResponses( @@ -306,8 +492,8 @@ describe("the inbound scope survives the handleResponses replay", () => { }); test("an oversized upstream JSON body fails closed instead of buffering without limit", async () => { - // Review finding: the WebSocket bounded-JSON path (and every non-streaming upstream) - // materializes the whole body, so the read must have a hard byte ceiling. 33 MiB is + // Every application/json fallback materializes the whole body, so the read must + // have a hard byte ceiling. 33 MiB is // one MiB over MAX_UPSTREAM_JSON_BODY_BYTES. globalThis.fetch = (async () => new Response(" ".repeat(33 * 1024 * 1024), { status: 200, diff --git a/tests/deepseek-responses-item-id-repair.test.ts b/tests/deepseek-responses-item-id-repair.test.ts index 44fe98a59c..0dabcd388c 100644 --- a/tests/deepseek-responses-item-id-repair.test.ts +++ b/tests/deepseek-responses-item-id-repair.test.ts @@ -1,4 +1,4 @@ -import { afterEach, describe, expect, test } from "bun:test"; +import { describe, expect, test } from "bun:test"; import { enrichProviderFromRegistry, providerConfigSeed } from "../src/providers/derive"; import { getProviderRegistryEntry } from "../src/providers/registry"; import { @@ -6,8 +6,7 @@ import { hasResponsesItemIdRepair, repairResponsesJsonItemIds, } from "../src/server/responses-item-id-repair"; -import { handleResponses } from "../src/server/responses/core"; -import type { OcxConfig, OcxProviderConfig } from "../src/types"; +import type { OcxProviderConfig } from "../src/types"; import { createTestTranslatorBudget } from "./helpers/translator-budget"; const UUID_MSG = "9b1deb4d-3b7d-4bad-9bdd-2b0d7b3dcb6d"; @@ -115,47 +114,3 @@ describe("registry-derived DeepSeek repair policy (#938)", () => { expect(output[1]!.id).toMatch(/^msg_ocx_/); }); }); - -describe("bounded-JSON HTTP path carries canonical ids (#938 + #875)", () => { - const originalFetch = globalThis.fetch; - afterEach(() => { globalThis.fetch = originalFetch; }); - - test("the synthesized terminal SSE contains no upstream UUID item ids (un-enriched saved seed)", async () => { - // The live path must backfill the registry policy through routedProviderConfig — - // no manual enrichProviderFromRegistry (the ordinary saved-config shape). - const plainSeed = { ...providerConfigSeed(getProviderRegistryEntry("deepseek")!), apiKey: "sk-test" }; - expect(plainSeed.responsesItemIdRepair).toBeUndefined(); - globalThis.fetch = (async () => Response.json({ - id: "resp_deepseek", - object: "response", - status: "completed", - output: [ - { type: "reasoning", id: UUID_RS, summary: [] }, - { type: "message", id: UUID_MSG, role: "assistant", status: "completed", content: [{ type: "output_text", text: "hi", annotations: [] }] }, - { type: "function_call", id: UUID_FC, call_id: "call_keep", name: "search", arguments: "{}" }, - ], - })) as typeof fetch; - - const config = { providers: { deepseek: plainSeed } } as unknown as OcxConfig; - const response = await handleResponses( - new Request("http://localhost/v1/responses", { - method: "POST", - headers: { "content-type": "application/json" }, - body: JSON.stringify({ model: "deepseek-v4-flash", input: "ping", stream: true }), - }), - config, - { model: "", provider: "" }, - {}, - ); - expect(response.headers.get("content-type")).toContain("text/event-stream"); - const text = await response.text(); - expect(text).not.toContain(UUID_MSG); - expect(text).not.toContain(UUID_RS); - expect(text).toContain("msg_ocx_"); - expect(text).toContain("rs_ocx_"); - // function_call identity survives byte-for-byte. - expect(text).toContain(UUID_FC); - expect(text).toContain("call_keep"); - expect(text).toContain("data: [DONE]"); - }); -}); diff --git a/tests/passthrough-abort.test.ts b/tests/passthrough-abort.test.ts index c38e29f73f..2b98ce91a5 100644 --- a/tests/passthrough-abort.test.ts +++ b/tests/passthrough-abort.test.ts @@ -46,7 +46,10 @@ describe("passthrough relayWithAbort (RC2, passthrough path)", () => { capsSource.indexOf("export function selectEagerPath"), ); - expect(sseBranch).toContain("upstreamResponse.body.tee()"); + expect(sseBranch).toContain("const terminalRepairPolicy = providerModelResponsesTerminalRepair("); + expect(sseBranch).toContain("const passthroughSseBody = terminalRepairPolicy"); + expect(sseBranch).toContain(": upstreamResponse.body;"); + expect(sseBranch).toContain("passthroughSseBody.tee()"); // Rewrite traffic is derived from the finalized block chain so every // provider-specific transform participates in the platform gate. expect(sseBranch).toContain("const repairConfig = route.provider.responsesItemIdRepair;"); @@ -71,7 +74,7 @@ describe("passthrough relayWithAbort (RC2, passthrough path)", () => { expect(sseBranch).toContain("config.streamMode ?? \"auto\","); expect(selector).toContain('platform !== "win32" && platform !== "darwin"'); expect(selector).toContain('decision.reason === "config-eager"'); - expect(sseBranch).toContain("relaySseEagerBounded(upstreamResponse.body, turnAc,"); + expect(sseBranch).toContain("relaySseEagerBounded(passthroughSseBody, turnAc,"); expect(sseBranch).not.toContain("relaySseWithHeartbeat("); expect(sseBranch).not.toContain("trackStreamLifetime("); expect(logWrapper.indexOf("isNativePassthroughSseResponse(response)")).toBeGreaterThanOrEqual(0); diff --git a/tests/responses-terminal-repair.test.ts b/tests/responses-terminal-repair.test.ts new file mode 100644 index 0000000000..22feb27ec1 --- /dev/null +++ b/tests/responses-terminal-repair.test.ts @@ -0,0 +1,511 @@ +import { describe, expect, test } from "bun:test"; +import type { ResponsesTerminalRepairPolicy } from "../src/providers/registry"; +import { relaySseWithFailedTail } from "../src/server/relay"; +import { + relayResponsesSseWithTerminalRepair, + type ResponsesTerminalRepairScheduler, +} from "../src/server/responses-terminal-repair"; +import { createTestTranslatorBudget } from "./helpers/translator-budget"; + +const encoder = new TextEncoder(); +const decoder = new TextDecoder(); +const POLICY: ResponsesTerminalRepairPolicy = { graceMs: 5_000 }; + +class ManualScheduler implements ResponsesTerminalRepairScheduler { + private current = 0; + private nextId = 1; + private readonly jobs = new Map 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; + for (;;) { + const due = [...this.jobs.entries()] + .filter(([, job]) => job.at <= this.current) + .sort((left, right) => left[1].at - right[1].at); + if (due.length === 0) return; + for (const [id, job] of due) { + if (!this.jobs.delete(id)) continue; + job.callback(); + } + } + } + + pending(): number { return this.jobs.size; } +} + +function sse(event: Record): string { + return `event: ${String(event.type)}\ndata: ${JSON.stringify(event)}\n\n`; +} + +function streamFromText(text: string): ReadableStream { + return new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode(text)); + controller.close(); + }, + }); +} + +function controlledSource(): { + stream: ReadableStream; + push(text: string): void; + close(): void; + cancelled(): boolean; +} { + let controller: ReadableStreamDefaultController | null = null; + let wasCancelled = false; + return { + stream: new ReadableStream({ + start(next) { controller = next; }, + cancel() { wasCancelled = true; }, + }), + push(text) { controller?.enqueue(encoder.encode(text)); }, + close() { controller?.close(); }, + cancelled: () => wasCancelled, + }; +} + +async function readAll(stream: ReadableStream): Promise { + const reader = stream.getReader(); + let out = ""; + for (;;) { + const { done, value } = await reader.read(); + if (done) return out + decoder.decode(); + out += decoder.decode(value, { stream: true }); + } +} + +async function settle(): Promise { + await Promise.resolve(); + await Bun.sleep(0); +} + +function capturedToolCallLifecycle(): string { + return [ + sse({ + type: "response.created", + response: { id: "resp_probe", object: "response", status: "in_progress", output: [] }, + sequence_number: 0, + }), + sse({ + type: "response.output_item.added", + item: { type: "reasoning", id: "rs_probe", status: "in_progress", content: [], summary: [] }, + output_index: 0, + sequence_number: 1, + }), + sse({ + type: "response.output_item.done", + item: { + type: "reasoning", + id: "rs_probe", + status: "completed", + content: [{ type: "reasoning_text", text: "Call the probe tool." }], + summary: [], + }, + output_index: 0, + sequence_number: 2, + }), + sse({ + type: "response.output_item.added", + item: { + type: "function_call", + id: "fc_probe", + status: "in_progress", + arguments: "", + call_id: "call_probe", + name: "probe", + }, + output_index: 1, + sequence_number: 3, + }), + sse({ + type: "response.function_call_arguments.done", + arguments: "{\"text\":\"OK\"}", + item_id: "fc_probe", + output_index: 1, + sequence_number: 4, + }), + sse({ + type: "response.output_item.done", + item: { + type: "function_call", + id: "fc_probe", + status: "completed", + arguments: "{\"text\":\"OK\"}", + call_id: "call_probe", + name: "probe", + }, + output_index: 1, + sequence_number: 5, + }), + ].join(""); +} + +function completedMessageLifecycle(text = "hello"): string { + return [ + sse({ + type: "response.created", + response: { id: "resp_message", object: "response", status: "in_progress", output: [] }, + sequence_number: 0, + }), + sse({ + type: "response.output_item.added", + item: { type: "message", id: "msg_probe", role: "assistant", status: "in_progress", content: [] }, + output_index: 0, + sequence_number: 1, + }), + sse({ + type: "response.output_item.done", + item: { + type: "message", + id: "msg_probe", + role: "assistant", + status: "completed", + content: [{ type: "output_text", text, annotations: [] }], + }, + output_index: 0, + sequence_number: 2, + }), + ].join(""); +} + +function terminalTypes(text: string): string[] { + return [...text.matchAll(/"type":"(response\.(?:completed|failed|incomplete))"/g)] + .map(match => match[1]!); +} + +async function repairClosedText(text: string): Promise<{ output: string; cancelled: boolean }> { + let cancelled = false; + const source = new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode(text)); + controller.close(); + }, + cancel() { cancelled = true; }, + }); + const upstream = new AbortController(); + const repaired = relayResponsesSseWithTerminalRepair( + source, + upstream, + POLICY, + createTestTranslatorBudget(), + new ManualScheduler(), + ); + return { output: await readAll(relaySseWithFailedTail(repaired, upstream)), cancelled }; +} + +describe("DeepSeek Responses terminal repair", () => { + test("a healthy stream with a real terminal is relayed byte-identical", async () => { + const lifecycle = capturedToolCallLifecycle(); + const terminal = sse({ + type: "response.completed", + response: { id: "resp_probe", object: "response", status: "completed", output: [] }, + sequence_number: 6, + }); + const upstream = lifecycle + terminal + "data: [DONE]\n\n"; + const budget = createTestTranslatorBudget(); + + const output = await readAll(relayResponsesSseWithTerminalRepair( + streamFromText(upstream), + new AbortController(), + POLICY, + budget, + )); + + expect(output).toBe(upstream); + expect(output.match(/"type":"response\.completed"/g)?.length).toBe(1); + expect(budget.snapshot().currentBytes).toBe(0); + }); + + test("a complete terminal-less tool call commits only after the grace period", async () => { + const source = controlledSource(); + const scheduler = new ManualScheduler(); + const budget = createTestTranslatorBudget(); + const upstream = new AbortController(); + const repaired = relayResponsesSseWithTerminalRepair(source.stream, upstream, POLICY, budget, scheduler); + let resolved = false; + const outputPromise = readAll(relaySseWithFailedTail(repaired, upstream)).then(output => { + resolved = true; + return output; + }); + + source.push(capturedToolCallLifecycle()); + await settle(); + expect(scheduler.pending()).toBe(1); + scheduler.advance(4_999); + await settle(); + expect(resolved).toBe(false); + + scheduler.advance(1); + const output = await outputPromise; + expect(resolved).toBe(true); + expect(output.match(/"type":"response\.completed"/g)?.length).toBe(1); + expect(output.endsWith("data: [DONE]\n\n")).toBe(true); + expect(output).toContain('"call_id":"call_probe"'); + expect(source.cancelled()).toBe(true); + expect(scheduler.pending()).toBe(0); + expect(budget.snapshot().currentBytes).toBe(0); + }); + + test("new activity invalidates the old grace generation and waits after the new item", async () => { + const source = controlledSource(); + const scheduler = new ManualScheduler(); + const budget = createTestTranslatorBudget(); + const upstream = new AbortController(); + const repaired = relayResponsesSseWithTerminalRepair(source.stream, upstream, POLICY, budget, scheduler); + let resolved = false; + const outputPromise = readAll(relaySseWithFailedTail(repaired, upstream)).then(output => { + resolved = true; + return output; + }); + + source.push(completedMessageLifecycle("first")); + await settle(); + scheduler.advance(4_999); + source.push([ + sse({ + type: "response.output_item.added", + item: { type: "message", id: "msg_second", role: "assistant", status: "in_progress", content: [] }, + output_index: 1, + sequence_number: 3, + }), + sse({ + type: "response.output_item.done", + item: { + type: "message", + id: "msg_second", + role: "assistant", + status: "completed", + content: [{ type: "output_text", text: "second", annotations: [] }], + }, + output_index: 1, + sequence_number: 4, + }), + ].join("")); + await settle(); + scheduler.advance(1); + await settle(); + expect(resolved).toBe(false); + + scheduler.advance(4_999); + const output = await outputPromise; + expect(terminalTypes(output)).toEqual(["response.completed"]); + expect(output).toContain('"text":"first"'); + expect(output).toContain('"text":"second"'); + expect(scheduler.pending()).toBe(0); + expect(budget.snapshot().currentBytes).toBe(0); + }); + + test("EOF synthesizes completed only for a complete candidate", async () => { + const { output } = await repairClosedText(completedMessageLifecycle()); + expect(terminalTypes(output)).toEqual(["response.completed"]); + expect(output.endsWith("data: [DONE]\n\n")).toBe(true); + }); + + test("DONE is replaced by completed then one DONE only for a complete candidate", async () => { + const { output } = await repairClosedText(completedMessageLifecycle() + "data: [DONE]\n\n"); + expect(terminalTypes(output)).toEqual(["response.completed"]); + expect(output.match(/data: \[DONE\]/g)?.length).toBe(1); + expect(output.indexOf("response.completed")).toBeLessThan(output.indexOf("data: [DONE]")); + }); + + test("open output items end as incomplete, never completed", async () => { + const input = [ + sse({ type: "response.created", response: { id: "resp_open", status: "in_progress", output: [] }, sequence_number: 0 }), + sse({ + type: "response.output_item.added", + item: { type: "message", id: "msg_open", role: "assistant", status: "in_progress", content: [] }, + output_index: 0, + sequence_number: 1, + }), + ].join(""); + const { output } = await repairClosedText(input); + expect(terminalTypes(output)).toEqual(["response.incomplete"]); + expect(output).not.toContain('"type":"response.completed"'); + }); + + test("invalid function arguments end as incomplete, never completed", async () => { + const input = capturedToolCallLifecycle().replaceAll('{\\"text\\":\\"OK\\"}', "{broken"); + const { output } = await repairClosedText(input); + expect(terminalTypes(output)).toEqual(["response.incomplete"]); + }); + + test("unknown output item types taint synthetic success", async () => { + const input = [ + sse({ type: "response.created", response: { id: "resp_unknown", status: "in_progress", output: [] }, sequence_number: 0 }), + sse({ type: "response.output_item.added", item: { type: "computer_call", id: "cmp_1", status: "in_progress" }, output_index: 0, sequence_number: 1 }), + sse({ type: "response.output_item.done", item: { type: "computer_call", id: "cmp_1", status: "completed" }, output_index: 0, sequence_number: 2 }), + ].join(""); + const { output } = await repairClosedText(input); + expect(terminalTypes(output)).toEqual(["response.incomplete"]); + }); + + test("duplicate or contradictory output indices taint synthetic success", async () => { + const input = [ + sse({ type: "response.created", response: { id: "resp_duplicate", status: "in_progress", output: [] }, sequence_number: 0 }), + sse({ type: "response.output_item.added", item: { type: "reasoning", id: "rs_1", status: "in_progress", content: [] }, output_index: 0, sequence_number: 1 }), + sse({ type: "response.output_item.added", item: { type: "message", id: "msg_1", role: "assistant", status: "in_progress", content: [] }, output_index: 0, sequence_number: 2 }), + sse({ + type: "response.output_item.done", + item: { type: "message", id: "msg_1", role: "assistant", status: "completed", content: [{ type: "output_text", text: "x" }] }, + output_index: 0, + sequence_number: 3, + }), + ].join(""); + const { output } = await repairClosedText(input); + expect(terminalTypes(output)).toEqual(["response.incomplete"]); + }); + + test("real completed failed and incomplete terminals beat the timer", async () => { + for (const terminal of ["completed", "failed", "incomplete"] as const) { + const scheduler = new ManualScheduler(); + const suffix = sse({ + type: `response.${terminal}`, + response: { id: `resp_${terminal}`, status: terminal }, + sequence_number: 3, + }); + const input = completedMessageLifecycle() + suffix + "data: [DONE]\n\n"; + const budget = createTestTranslatorBudget(); + const output = await readAll(relayResponsesSseWithTerminalRepair( + streamFromText(input), + new AbortController(), + POLICY, + budget, + scheduler, + )); + scheduler.advance(5_000); + expect(terminalTypes(output)).toEqual([`response.${terminal}`]); + expect(scheduler.pending()).toBe(0); + expect(budget.snapshot().currentBytes).toBe(0); + } + }); + + test("a timer racing a real terminal commits exactly one terminal", async () => { + const source = controlledSource(); + const scheduler = new ManualScheduler(); + const budget = createTestTranslatorBudget(); + const outputPromise = readAll(relayResponsesSseWithTerminalRepair( + source.stream, + new AbortController(), + POLICY, + budget, + scheduler, + )); + source.push(completedMessageLifecycle()); + await settle(); + source.push(sse({ type: "response.completed", response: { id: "resp_message", status: "completed" }, sequence_number: 3 })); + source.close(); + scheduler.advance(5_000); + const output = await outputPromise; + expect(terminalTypes(output)).toEqual(["response.completed"]); + expect(scheduler.pending()).toBe(0); + }); + + 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)]; + const source = new ReadableStream({ + start(controller) { + for (const chunk of chunks) controller.enqueue(chunk); + controller.close(); + }, + }); + const output = await readAll(relayResponsesSseWithTerminalRepair( + source, + new AbortController(), + POLICY, + createTestTranslatorBudget(), + new ManualScheduler(), + )); + expect(output).toBe(input); + expect(terminalTypes(output)).toEqual(["response.completed"]); + }); + + test("client cancel and upstream abort suppress synthetic completion", async () => { + const cancelSource = controlledSource(); + const cancelScheduler = new ManualScheduler(); + const cancelUpstream = new AbortController(); + const cancelled = relayResponsesSseWithTerminalRepair( + cancelSource.stream, + cancelUpstream, + POLICY, + createTestTranslatorBudget(), + cancelScheduler, + ); + cancelSource.push(completedMessageLifecycle()); + await settle(); + await cancelled.cancel("client gone"); + cancelScheduler.advance(5_000); + expect(cancelSource.cancelled()).toBe(true); + expect(cancelScheduler.pending()).toBe(0); + + const abortSource = controlledSource(); + const abortScheduler = new ManualScheduler(); + const abortUpstream = new AbortController(); + const aborted = relayResponsesSseWithTerminalRepair( + abortSource.stream, + abortUpstream, + POLICY, + createTestTranslatorBudget(), + abortScheduler, + ); + abortSource.push(completedMessageLifecycle()); + await settle(); + abortUpstream.abort("shutdown"); + await settle(); + expect(abortSource.cancelled()).toBe(true); + expect(abortScheduler.pending()).toBe(0); + await aborted.cancel("test cleanup"); + }); + + test("retained-state overflow throws translation_buffer_limit and releases budget", async () => { + const budget = createTestTranslatorBudget({ maxTurnBytes: 128 }); + const source = streamFromText(completedMessageLifecycle("x".repeat(512))); + const repaired = relayResponsesSseWithTerminalRepair( + source, + new AbortController(), + POLICY, + budget, + new ManualScheduler(), + ); + await expect(readAll(repaired)).rejects.toMatchObject({ code: "translation_buffer_limit" }); + expect(budget.snapshot().currentBytes).toBe(0); + }); + + test("the wrapper does not continuously read ahead without downstream pulls", async () => { + let pulls = 0; + const source = new ReadableStream({ + pull(controller) { + pulls += 1; + if (pulls <= 5) controller.enqueue(encoder.encode(`: heartbeat ${pulls}\n\n`)); + }, + }); + const repaired = relayResponsesSseWithTerminalRepair( + source, + new AbortController(), + POLICY, + createTestTranslatorBudget(), + new ManualScheduler(), + ); + + await settle(); + // One chunk may be prefetched by each Web Stream layer; continuous read-ahead must stop there. + expect(pulls).toBeLessThanOrEqual(2); + await repaired.cancel("test cleanup"); + }); +});