From ee54ca9122cdc8705b24dc12d409d1ea294be419 Mon Sep 17 00:00:00 2001 From: Wibias <37517432+Wibias@users.noreply.github.com> Date: Wed, 12 Aug 2026 00:13:55 +0200 Subject: [PATCH] fix(deepseek): keep parallel reasoning replay on continuation --- .../src/content/docs/reference/adapters.md | 7 ++ src/adapters/openai-responses.ts | 78 ++++++++++--- structure/04_transports-and-sidecars.md | 7 ++ tests/deepseek-inbound-wire.test.ts | 107 ++++++++++++++++++ 4 files changed, 182 insertions(+), 17 deletions(-) diff --git a/docs-site/src/content/docs/reference/adapters.md b/docs-site/src/content/docs/reference/adapters.md index 84b47b3670..832b6126f8 100644 --- a/docs-site/src/content/docs/reference/adapters.md +++ b/docs-site/src/content/docs/reference/adapters.md @@ -64,6 +64,13 @@ of the HTTP retry loop. ChatGPT account id, and the OpenAI beta/originator/session headers. This is the ChatGPT-login path that also powers the [sidecars](/guides/sidecars/). +For providers that declare `requiresAdjacentResponsesToolResults` (currently DeepSeek), the adapter +normalizes an unambiguous Responses tool history so one parallel-tool-call assistant turn stays +together as a call batch followed by its matching results in call order. Hook-injected context that +interleaved the batch moves immediately after it rather than being dropped. Histories that are +missing, duplicate, or out-of-order (backward) are ambiguous and are forwarded unchanged — they are +left for the upstream to reject rather than being guessed. + ## `anthropic` **Targets:** Anthropic **Messages** (`/v1/messages`). diff --git a/src/adapters/openai-responses.ts b/src/adapters/openai-responses.ts index 7aa3fa8ec8..3e78d5e1d1 100644 --- a/src/adapters/openai-responses.ts +++ b/src/adapters/openai-responses.ts @@ -540,15 +540,27 @@ function repairOrphanedInputItems(body: unknown, dropReasoning: boolean): unknow } /** - * Make unambiguous Responses tool pairs adjacent for upstream parsers that require it. + * Make unambiguous Responses tool batches contiguous for upstream parsers that require it. * * [Decision Log] - * - 목적과 의도: Keep Codex hook-injected developer context without letting it make a strict upstream reject the matching tool result. - * - 기존 구현 및 제약 조건: The orphan repair verifies only pair presence; globally reordering valid history would change tolerant providers unnecessarily. - * - 검토한 주요 대안: Reorder every Responses request, drop the intervening message, or gate a lossless reorder behind provider capability metadata. - * - 선택한 방식: Reorder only unique call/result pairs for providers that explicitly require adjacency, preserving every intervening item immediately after the result. - * - 다른 대안 대신 이 방식을 선택한 이유: The provider gate limits semantic blast radius, while refusing ambiguous duplicate ids avoids guessing which result belongs to which call. - * - 장점, 단점 및 영향: DeepSeek receives the adjacency its parser requires; tolerant providers stay byte/order equivalent. Ambiguous duplicate ids still fail upstream rather than being silently rewritten. + * - 목적과 의도: Keep Codex hook-injected developer context without splitting a parallel tool-call + * turn away from its reasoning or letting a strict upstream reject the matching tool results. + * - 기존 구현 및 제약 조건: The orphan repair verifies only pair presence, while a pair-by-pair + * reorder turned `reasoning, call A, call B, output A, output B` into two assistant turns and made + * DeepSeek reject call B for missing reasoning (#1477). Pair-by-pair handling also skipped calls + * without a matched result, so a partially matched history could still be reordered. + * - 검토한 주요 대안: Disable parallel calls (DeepSeek always enables them); duplicate reasoning per + * call; reorder each pair; or normalize only a complete, unambiguous call/output batch. + * - 선택한 방식: Treat calls emitted before the first matched result as one batch, emit all calls + * followed by their matched outputs in call order, and preserve intervening non-tool items after + * the batch. Any missing, duplicate, backward, or otherwise ambiguous call/result history is left + * unchanged and fails closed upstream. + * - 다른 대안 대신 이 방식을 선택한 이유: Batch normalization matches the Responses parallel-call shape + * without fabricating reasoning, while the provider gate and explicit completeness check keep the + * blast radius narrow and never reorder a history we cannot prove unambiguous. + * - 장점, 단점 및 영향: DeepSeek keeps one reasoning-bearing assistant turn for parallel calls and still + * accepts hook-interleaved single calls; tolerant providers stay byte/order equivalent, and missing, + * duplicate, or backward call/result histories are not guessed. */ function normalizeResponsesToolResultAdjacency(body: unknown): unknown { if (!isPlainObject(body) || !Array.isArray(body.input)) return body; @@ -576,24 +588,56 @@ function normalizeResponsesToolResultAdjacency(body: unknown): unknown { } } - const movedOutputIndices = new Set(); - const outputAfterCall = new Map(); + // Fail closed: only normalize when every collected call has exactly one matching + // result and that result appears after its call. Missing, duplicate, or backward + // histories are ambiguous and must be left untouched for the upstream to reject. + const pairs: Array<{ callIndex: number; outputIndex: number }> = []; for (const [key, callIndices] of calls) { const outputIndices = outputs.get(key); - if (callIndices.length !== 1 || outputIndices?.length !== 1) continue; + if (callIndices.length !== 1 || outputIndices?.length !== 1) return body; const callIndex = callIndices[0]!; const outputIndex = outputIndices[0]!; - if (outputIndex === callIndex + 1) continue; - movedOutputIndices.add(outputIndex); - outputAfterCall.set(callIndex, input[outputIndex]); + if (outputIndex <= callIndex) return body; + pairs.push({ callIndex, outputIndex }); } - if (movedOutputIndices.size === 0) return body; + if (pairs.length === 0) return body; + + pairs.sort((left, right) => left.callIndex - right.callIndex); + + const movedIndices = new Set(); + const batchAt = new Map(); + for (let cursor = 0; cursor < pairs.length; ) { + const group = [pairs[cursor]!]; + let firstOutputIndex = pairs[cursor]!.outputIndex; + let next = cursor + 1; + while (next < pairs.length && pairs[next]!.callIndex < firstOutputIndex) { + group.push(pairs[next]!); + firstOutputIndex = Math.min(firstOutputIndex, pairs[next]!.outputIndex); + next += 1; + } + + const batch = [ + ...group.map(pair => input[pair.callIndex]), + ...group.map(pair => input[pair.outputIndex]), + ]; + const anchor = group[0]!.callIndex; + const alreadyContiguous = batch.every((item, offset) => input[anchor + offset] === item); + if (!alreadyContiguous) { + batchAt.set(anchor, batch); + for (const pair of group) { + movedIndices.add(pair.callIndex); + movedIndices.add(pair.outputIndex); + } + } + cursor = next; + } + if (batchAt.size === 0) return body; const normalized: unknown[] = []; for (let index = 0; index < input.length; index += 1) { - if (movedOutputIndices.has(index)) continue; - normalized.push(input[index]); - if (outputAfterCall.has(index)) normalized.push(outputAfterCall.get(index)); + const batch = batchAt.get(index); + if (batch) normalized.push(...batch); + if (!movedIndices.has(index)) normalized.push(input[index]); } return { ...body, input: normalized }; } diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index 6c88a61d07..451b26f707 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -262,6 +262,13 @@ as `response.incomplete`, never synthetic success. The repair shares the per-tur budget, preserves backpressure, and composes ahead of item-id/snapshot rewrites so HTTP/SSE and WebSocket clients observe the same canonical lifecycle. +DeepSeek also opts into a provider-scoped Responses history normalization +(`requiresAdjacentResponsesToolResults`). Only an unambiguous call/output batch is normalized: +calls emitted before the first matched result stay together as one assistant batch, followed by +their matching results in call order, and any intervening hook-injected context moves after the +batch. Missing, duplicate, or backward (out-of-order) call/result histories are ambiguous and are +left unchanged so the strict upstream rejects them rather than guessing. + `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 `response.failed` frame is sent; otherwise `response.completed` carries through the original status. diff --git a/tests/deepseek-inbound-wire.test.ts b/tests/deepseek-inbound-wire.test.ts index 301edbae11..cd508e1263 100644 --- a/tests/deepseek-inbound-wire.test.ts +++ b/tests/deepseek-inbound-wire.test.ts @@ -99,6 +99,10 @@ function deepseekProvider(): OcxProviderConfig { return { ...providerConfigSeed(getProviderRegistryEntry("deepseek")!), apiKey: "sk-test" }; } +function deepseekReasoningProvider(): OcxProviderConfig { + return { ...deepseekProvider(), preserveResponsesReasoningContent: true }; +} + describe("DeepSeek wire selection is scoped to the inbound protocol", () => { test("a Responses inbound rides the native Responses wire", () => { const resolved = resolveWireProtocolOverride("deepseek", MODEL, deepseekProvider(), "responses"); @@ -836,6 +840,109 @@ describe("stateless Responses upstreams get no stateful parameters", () => { expect(body.input).toEqual(input); }); + test("DeepSeek keeps a parallel call batch attached to one reasoning turn", () => { + const reasoning = { + type: "reasoning", + content: [{ type: "reasoning_text", text: "read both files" }], + summary: [], + }; + const callA = { type: "function_call", call_id: "call_a", name: "read_file", arguments: "{}" }; + const callB = { type: "function_call", call_id: "call_b", name: "read_file", arguments: "{}" }; + const outputA = { type: "function_call_output", call_id: "call_a", output: "A" }; + const outputB = { type: "function_call_output", call_id: "call_b", output: "B" }; + + const body = buildBody(deepseekReasoningProvider(), { + input: [reasoning, callA, callB, outputA, outputB], + }) as { input: unknown[] }; + expect(body.input).toEqual([reasoning, callA, callB, outputA, outputB]); + }); + + test("DeepSeek moves injected context after the complete parallel call and result batches", () => { + const reasoning = { + type: "reasoning", + content: [{ type: "reasoning_text", text: "read both files" }], + summary: [], + }; + const callA = { type: "function_call", call_id: "call_a", name: "read_file", arguments: "{}" }; + const callB = { type: "function_call", call_id: "call_b", name: "read_file", arguments: "{}" }; + const outputA = { type: "function_call_output", call_id: "call_a", output: "A" }; + const outputB = { type: "function_call_output", call_id: "call_b", output: "B" }; + const injected = { + type: "message", + role: "developer", + content: [{ type: "input_text", text: "[planning-with-files] ACTIVE PLAN" }], + }; + const tail = { type: "message", role: "user", content: [{ type: "input_text", text: "continue" }] }; + + const body = buildBody(deepseekReasoningProvider(), { + input: [reasoning, callA, injected, callB, outputA, outputB, tail], + }) as { input: unknown[] }; + expect(body.input).toEqual([reasoning, callA, callB, outputA, outputB, injected, tail]); + }); + + test("DeepSeek keeps sequential reasoning and tool rounds separate", () => { + const reasoningA = { + type: "reasoning", + content: [{ type: "reasoning_text", text: "first" }], + summary: [], + }; + const reasoningB = { + type: "reasoning", + content: [{ type: "reasoning_text", text: "second" }], + summary: [], + }; + const callA = { type: "function_call", call_id: "call_a", name: "read_file", arguments: "{}" }; + const callB = { type: "function_call", call_id: "call_b", name: "read_file", arguments: "{}" }; + const outputA = { type: "function_call_output", call_id: "call_a", output: "A" }; + const outputB = { type: "function_call_output", call_id: "call_b", output: "B" }; + + const body = buildBody(deepseekReasoningProvider(), { + input: [reasoningA, callA, outputA, reasoningB, callB, outputB], + }) as { input: unknown[] }; + expect(body.input).toEqual([reasoningA, callA, outputA, reasoningB, callB, outputB]); + }); + + test("DeepSeek leaves a history with a missing call result unchanged (fail closed)", () => { + const callA = { type: "function_call", call_id: "call_a", name: "read_file", arguments: "{}" }; + const callB = { type: "function_call", call_id: "call_b", name: "read_file", arguments: "{}" }; + const injected = { + type: "message", + role: "developer", + content: [{ type: "input_text", text: "[planning-with-files] ACTIVE PLAN" }], + }; + const outputB = { type: "function_call_output", call_id: "call_b", output: "B" }; + const tail = { type: "message", role: "user", content: [{ type: "input_text", text: "continue" }] }; + + const input = [callA, callB, injected, outputB, tail]; + const body = buildBody(deepseekReasoningProvider(), { input }) as { input: unknown[] }; + expect(body.input).toEqual(input); + }); + + test("DeepSeek leaves a backward call/result pair unchanged (fail closed)", () => { + const callA = { type: "function_call", call_id: "call_a", name: "read_file", arguments: "{}" }; + const outputA = { type: "function_call_output", call_id: "call_a", output: "A" }; + const injected = { + type: "message", + role: "developer", + content: [{ type: "input_text", text: "[planning-with-files] ACTIVE PLAN" }], + }; + + // outputA appears before its own callA, which is ambiguous. + const input = [outputA, injected, callA]; + const body = buildBody(deepseekReasoningProvider(), { input }) as { input: unknown[] }; + expect(body.input).toEqual(input); + }); + + test("DeepSeek leaves a duplicate call/result history unchanged (fail closed)", () => { + const callA1 = { type: "function_call", call_id: "call_a", name: "read_file", arguments: "{}" }; + const callA2 = { type: "function_call", call_id: "call_a", name: "read_file", arguments: "{}" }; + const outputA = { type: "function_call_output", call_id: "call_a", output: "A" }; + + const input = [callA1, callA2, outputA]; + const body = buildBody(deepseekReasoningProvider(), { input }) as { input: unknown[] }; + expect(body.input).toEqual(input); + }); + test("a replay miss does not forward an orphaned tool result", () => { // On a replay miss the delta can open with a function_call_output whose paired // function_call sat in the prefix that was never expanded. A stateless upstream