From 75c37dcaf644a68d0318915495f9055169c2a018 Mon Sep 17 00:00:00 2001 From: mouzhi Date: Mon, 3 Aug 2026 16:23:16 +0800 Subject: [PATCH 1/6] fix deepseek responses uuid item ids for codex Make openai-responses passthrough optionally rewrite non-canonical message/reasoning ids, fold reasoning_text into encrypted_content, and keep SSE rewrite consumers alive when events are dropped so DeepSeek Responses works with Codex CLI/TUI. --- src/config.ts | 1 + src/server/relay-eager.ts | 16 +- src/server/responses-item-id-repair.ts | 304 +++++++++++++++--- src/server/sse-payload-rewrite.ts | 79 +++-- src/types.ts | 2 + .../responses-item-id-repair-deepseek.test.ts | 95 ++++++ tests/responses-item-id-repair.test.ts | 1 + 7 files changed, 430 insertions(+), 68 deletions(-) create mode 100644 tests/responses-item-id-repair-deepseek.test.ts diff --git a/src/config.ts b/src/config.ts index fe523323ad..7f5333261e 100644 --- a/src/config.ts +++ b/src/config.ts @@ -567,6 +567,7 @@ const providerConfigSchema = z.object({ message: z.array(z.string().min(1)).optional(), reasoning: z.array(z.string().min(1)).optional(), repairMissingTerminalIds: z.boolean().optional(), + rewriteNonCanonicalIds: z.boolean().optional(), }).strict().optional(), }).passthrough(); diff --git a/src/server/relay-eager.ts b/src/server/relay-eager.ts index ee8ac88e7e..57faeabdd2 100644 --- a/src/server/relay-eager.ts +++ b/src/server/relay-eager.ts @@ -123,14 +123,16 @@ export function relaySseEagerBounded( if (!next) break; const payload = sseDataPayload(next.block); const rewrittenPayload = payload === null ? null : rewrite!(payload); + frameBuffer = next.rest; + // Returning null from rewrite drops the entire SSE event. + if (payload !== null && rewrittenPayload === null) continue; // Replace only on an actual change: replaceSseDataPayload collapses // multi-data-line events and normalizes newline style even when the // payload is identical, which corrupts valid streams. - const block = payload !== null && rewrittenPayload !== payload - ? replaceSseDataPayload(next.block, rewrittenPayload!) + const block = payload !== null && rewrittenPayload !== null && rewrittenPayload !== payload + ? replaceSseDataPayload(next.block, rewrittenPayload) : next.block; out += block + next.delimiter; - frameBuffer = next.rest; } if (rewriteBudget) { const remaining = rewriteEncoder!.encode(frameBuffer).byteLength; @@ -149,6 +151,14 @@ export function relaySseEagerBounded( const payload = sseDataPayload(tail); if (payload !== null) { const rewrittenPayload = rewrite(payload); + if (rewrittenPayload === null) { + frameBuffer = ""; + if (rewriteBudget && frameBufferBytes > 0) { + rewriteBudget.releaseRetained(frameBufferBytes, { kind: "live_transient" }); + } + frameBufferBytes = 0; + return rewriteEncoder!.encode(""); + } if (rewrittenPayload !== payload) tail = replaceSseDataPayload(tail, rewrittenPayload); } frameBuffer = ""; diff --git a/src/server/responses-item-id-repair.ts b/src/server/responses-item-id-repair.ts index 1f0397e8e6..47792bb353 100644 --- a/src/server/responses-item-id-repair.ts +++ b/src/server/responses-item-id-repair.ts @@ -1,5 +1,6 @@ import { randomUUID } from "node:crypto"; import type { ResponsesItemIdRepairConfig } from "../types"; +import { encodeReasoningEnvelope } from "../responses/reasoning-envelope"; import { relaySseWithPayloadRewrite, type SsePayloadRewrite } from "./sse-payload-rewrite"; import type { TranslatorBudget } from "../lib/translator-budget"; @@ -7,10 +8,16 @@ type RepairableItemType = "message" | "reasoning"; interface ResponsesItemIdRepairState { readonly repairMissingTerminalIds: boolean; + readonly rewriteNonCanonicalIds: boolean; readonly placeholders: Record>; readonly outputIds: Record>; + readonly rawToCanonical: Map; + readonly reasoningTextByOutputIndex: Map; + readonly responseIdMap: Map; readonly scope: string; readonly budget?: TranslatorBudget; + sawCompleted: boolean; + sawDoneTrailer: boolean; } const REPAIRABLE_PREFIXES: Record = { @@ -50,9 +57,58 @@ function mintCanonicalId(type: RepairableItemType, scope: string, outputIndex: n return `${REPAIRABLE_PREFIXES[type]}ocx_${scope}_${outputIndex}`; } +function isCanonicalItemId(type: RepairableItemType, id: string): boolean { + return id.startsWith(REPAIRABLE_PREFIXES[type]); +} + +function shouldRewriteRawId( + state: ResponsesItemIdRepairState, + type: RepairableItemType, + rawId: string, +): boolean { + if (state.placeholders[type].has(rawId)) return true; + if (state.rewriteNonCanonicalIds && !isCanonicalItemId(type, rawId)) return true; + return false; +} + +function mapRawId( + state: ResponsesItemIdRepairState, + type: RepairableItemType, + outputIndex: number, + rawId: string | undefined, +): string | null { + const existing = state.outputIds[type].get(outputIndex); + if (existing) { + if (rawId && shouldRewriteRawId(state, type, rawId) && !state.rawToCanonical.has(rawId)) { + state.rawToCanonical.set(rawId, existing); + } + return existing; + } + + if (!rawId) { + if (!state.repairMissingTerminalIds) return null; + const minted = mintCanonicalId(type, state.scope, outputIndex); + state.budget?.chargeRetained(new TextEncoder().encode(JSON.stringify([outputIndex, minted])).byteLength, { kind: "item_ids" }); + state.outputIds[type].set(outputIndex, minted); + return minted; + } + + const mapped = shouldRewriteRawId(state, type, rawId) + ? mintCanonicalId(type, state.scope, outputIndex) + : state.repairMissingTerminalIds + ? rawId + : null; + if (!mapped) return null; + state.budget?.chargeRetained(new TextEncoder().encode(JSON.stringify([outputIndex, rawId, mapped])).byteLength, { kind: "item_ids" }); + state.outputIds[type].set(outputIndex, mapped); + if (mapped !== rawId) state.rawToCanonical.set(rawId, mapped); + return mapped; +} + function createRepairState(config: ResponsesItemIdRepairConfig, budget?: TranslatorBudget): ResponsesItemIdRepairState { - const state = { + const state: ResponsesItemIdRepairState = { repairMissingTerminalIds: config.repairMissingTerminalIds === true, + rewriteNonCanonicalIds: config.rewriteNonCanonicalIds === true, placeholders: { message: new Set(config.message ?? []), reasoning: new Set(config.reasoning ?? []), @@ -61,13 +117,19 @@ function createRepairState(config: ResponsesItemIdRepairConfig, budget?: Transla message: new Map(), reasoning: new Map(), }, + rawToCanonical: new Map(), + reasoningTextByOutputIndex: new Map(), + responseIdMap: new Map(), scope: randomUUID().replace(/-/g, ""), budget, + sawCompleted: false, + sawDoneTrailer: false, }; budget?.chargeRetained(new TextEncoder().encode(JSON.stringify({ message: [...state.placeholders.message], reasoning: [...state.placeholders.reasoning], scope: state.scope, + rewriteNonCanonicalIds: state.rewriteNonCanonicalIds, })).byteLength, { kind: "item_ids" }); return state; } @@ -79,26 +141,81 @@ function rememberMappedId( ): string | null { const type = repairableItemType(item); if (!type) return null; - const existing = state.outputIds[type].get(outputIndex); - if (existing) return existing; const rawId = typeof item.id === "string" ? item.id : undefined; - if (!rawId) return null; - const mapped = state.placeholders[type].has(rawId) - ? mintCanonicalId(type, state.scope, outputIndex) - : state.repairMissingTerminalIds - ? rawId - : null; - if (!mapped) return null; - state.budget?.chargeRetained(new TextEncoder().encode(JSON.stringify([outputIndex, rawId, mapped])).byteLength, { kind: "item_ids" }); - state.outputIds[type].set(outputIndex, mapped); - return mapped; + return mapRawId(state, type, outputIndex, rawId); +} + +function mapResponseId(state: ResponsesItemIdRepairState, rawId: string | undefined): string | undefined { + if (!rawId) return rawId; + if (!state.rewriteNonCanonicalIds) return rawId; + if (rawId.startsWith("resp_")) return rawId; + const existing = state.responseIdMap.get(rawId); + if (existing) return existing; + const minted = `resp_ocx_${state.scope}`; + state.responseIdMap.set(rawId, minted); + return minted; +} + +function normalizeReasoningItem( + state: ResponsesItemIdRepairState, + outputIndex: number, + item: Record, + terminal: boolean, +): Record { + const mapped = rememberMappedId(state, outputIndex, item) ?? item.id; + const text = state.reasoningTextByOutputIndex.get(outputIndex) + ?? (Array.isArray(item.content) + ? item.content + .filter((part): part is Record => isPlainObject(part) && part.type === "reasoning_text" && typeof part.text === "string") + .map(part => String(part.text)) + .join("") + : ""); + if (text) state.reasoningTextByOutputIndex.set(outputIndex, text); + const next: Record = { + type: "reasoning", + id: mapped, + summary: Array.isArray(item.summary) ? item.summary : [], + }; + if (terminal || text) { + next.encrypted_content = encodeReasoningEnvelope({ txt: text || " " }); + } + return next; +} + +function normalizeMessageItem( + state: ResponsesItemIdRepairState, + outputIndex: number, + item: Record, +): Record { + const mapped = rememberMappedId(state, outputIndex, item); + const next: Record = { ...item }; + if (mapped) next.id = mapped; + if (Array.isArray(next.content)) { + next.content = next.content.map(part => { + if (!isPlainObject(part)) return part; + if (part.type !== "output_text") return part; + return { + type: "output_text", + text: typeof part.text === "string" ? part.text : "", + annotations: Array.isArray(part.annotations) ? part.annotations : [], + }; + }); + } + return next; } function rewriteOutputItem( state: ResponsesItemIdRepairState, outputIndex: number, item: Record, + terminal = false, ): { item: Record; changed: boolean } { + if (state.rewriteNonCanonicalIds && item.type === "reasoning") { + return { item: normalizeReasoningItem(state, outputIndex, item, terminal), changed: true }; + } + if (state.rewriteNonCanonicalIds && item.type === "message") { + return { item: normalizeMessageItem(state, outputIndex, item), changed: true }; + } const mapped = rememberMappedId(state, outputIndex, item); if (!mapped) return { item, changed: false }; const currentId = typeof item.id === "string" ? item.id : undefined; @@ -112,35 +229,80 @@ function rewriteItemIdField( event: Record, outputIndex: number, ): { event: Record; changed: boolean } { + const currentId = typeof event.item_id === "string" ? event.item_id : undefined; + const reverseMapped = currentId ? state.rawToCanonical.get(currentId) : undefined; + if (reverseMapped && reverseMapped !== currentId) { + return { event: { ...event, item_id: reverseMapped }, changed: true }; + } + const eventType = typeof event.type === "string" ? ITEM_ID_EVENT_TYPES[event.type] : undefined; - if (!eventType) return { event, changed: false }; - const mapped = state.outputIds[eventType].get(outputIndex); + let mapped: string | null | undefined; + if (eventType === "message" && ( + event.type === "response.content_part.added" + || event.type === "response.content_part.done" + )) { + mapped = state.outputIds.message.get(outputIndex) + ?? state.outputIds.reasoning.get(outputIndex) + ?? null; + } else if (eventType) { + mapped = state.outputIds[eventType].get(outputIndex); + if (!mapped && currentId && state.rewriteNonCanonicalIds) { + mapped = mapRawId(state, eventType, outputIndex, currentId); + } + } else { + return { event, changed: false }; + } if (!mapped) return { event, changed: false }; - const currentId = typeof event.item_id === "string" ? event.item_id : undefined; if (currentId === mapped) return { event, changed: false }; if (currentId === undefined && !state.repairMissingTerminalIds) return { event, changed: false }; - return { event: { ...event, item_id: mapped }, changed: true }; + const next: Record = { ...event, item_id: mapped }; + if (state.rewriteNonCanonicalIds && "logprobs" in next) delete next.logprobs; + return { event: next, changed: true }; } function rewriteResponseSnapshot( state: ResponsesItemIdRepairState, response: Record, ): { response: Record; changed: boolean } { - if (!Array.isArray(response.output)) return { response, changed: false }; let changed = false; + let next = response; + if (typeof response.id === "string") { + const mapped = mapResponseId(state, response.id); + if (mapped && mapped !== response.id) { + next = { ...next, id: mapped }; + changed = true; + } + } + if (!Array.isArray(response.output)) return { response: next, changed }; const output = response.output.map((item, outputIndex) => { if (!isPlainObject(item)) return item; - const rewritten = rewriteOutputItem(state, outputIndex, item); + const rewritten = rewriteOutputItem(state, outputIndex, item, true); changed = changed || rewritten.changed; return rewritten.item; }); - return changed ? { response: { ...response, output }, changed: true } : { response, changed: false }; + return changed ? { response: { ...next, output }, changed: true } : { response: next, changed }; +} + +function accumulateReasoningText( + state: ResponsesItemIdRepairState, + outputIndex: number, + delta: string, +): void { + state.reasoningTextByOutputIndex.set( + outputIndex, + (state.reasoningTextByOutputIndex.get(outputIndex) ?? "") + delta, + ); } function repairEventPayload( payload: string, state: ResponsesItemIdRepairState, -): string { +): string | null { + if (payload.trim() === "[DONE]") { + state.sawDoneTrailer = true; + return payload; + } + let event: unknown; try { event = JSON.parse(payload); @@ -149,11 +311,41 @@ function repairEventPayload( } if (!isPlainObject(event)) return payload; + const type = typeof event.type === "string" ? event.type : undefined; + const outputIndex = asOutputIndex(event.output_index); + + if (state.rewriteNonCanonicalIds) { + if (type === "response.in_progress") return null; + if (type === "response.reasoning_text.delta") { + if (outputIndex !== null && typeof event.delta === "string") { + accumulateReasoningText(state, outputIndex, event.delta); + } + return null; + } + if (type === "response.reasoning_text.done") { + if (outputIndex !== null && typeof event.text === "string") { + state.reasoningTextByOutputIndex.set(outputIndex, event.text); + } + return null; + } + if ( + (type === "response.content_part.added" || type === "response.content_part.done") + && isPlainObject(event.part) + && event.part.type === "reasoning_text" + ) { + if (outputIndex !== null && typeof event.part.text === "string" && event.part.text) { + state.reasoningTextByOutputIndex.set(outputIndex, event.part.text); + } + return null; + } + } + let changed = false; let nextEvent = event; - const outputIndex = asOutputIndex(event.output_index); + if (outputIndex !== null && isPlainObject(event.item)) { - const rewritten = rewriteOutputItem(state, outputIndex, event.item); + const terminal = type === "response.output_item.done"; + const rewritten = rewriteOutputItem(state, outputIndex, event.item, terminal); if (rewritten.changed) { nextEvent = { ...nextEvent, item: rewritten.item }; changed = true; @@ -173,6 +365,30 @@ function repairEventPayload( changed = true; } } + + if (type === "response.completed") state.sawCompleted = true; + + if (state.rewriteNonCanonicalIds && isPlainObject(nextEvent.part) && nextEvent.part.type === "output_text") { + nextEvent = { + ...nextEvent, + part: { + type: "output_text", + text: typeof nextEvent.part.text === "string" ? nextEvent.part.text : "", + annotations: Array.isArray(nextEvent.part.annotations) ? nextEvent.part.annotations : [], + }, + }; + if ("logprobs" in nextEvent) { + const { logprobs: _lp, ...rest } = nextEvent; + nextEvent = rest; + } + changed = true; + } + if (state.rewriteNonCanonicalIds && "logprobs" in nextEvent) { + const { logprobs: _lp, ...rest } = nextEvent; + nextEvent = rest; + changed = true; + } + if (!changed) return payload; const rewritten = JSON.stringify(nextEvent); const bytes = new TextEncoder().encode(rewritten).byteLength; @@ -183,30 +399,30 @@ function repairEventPayload( } /** - * [Decision Log] - * - 목적과 의도: 일부 openai-responses 호환 게이트웨이가 재사용/누락하는 message·reasoning item id를 - * downstream SSE에서만 선택적으로 보정해 Codex Desktop 카드 상관관계를 안정화한다. - * - 기존 구현 및 제약 조건: 기본 passthrough는 바이트 단위 그대로 relay되고, local replay 상태는 raw - * upstream 응답을 기억한다. function_call id / call_id는 upstream 의미가 있으므로 절대 바꾸면 안 된다. - * - 검토한 주요 대안: 모든 passthrough SSE를 항상 재작성하기, raw inspect 분기까지 함께 재작성하기, - * function_call 포함 전체 item id를 정규화하기. - * - 선택한 방식: provider-local opt-in 설정이 있을 때만 client-facing SSE 분기에 한정해 exact - * message/reasoning placeholder id와 missing terminal id를 item type + output_index 기준으로 보정하고, - * event-level item_id는 명시적인 message/reasoning lifecycle allowlist에서만 바꾼다. - * - 다른 대안 대신 이 방식을 선택한 이유: disabled-by-default byte-for-byte passthrough를 유지하면서, - * previous_response_id replay는 raw upstream snapshot을 계속 사용해 synthetic id가 upstream으로 - * 역류하지 않게 막을 수 있다. - * - 장점, 단점 및 영향: 기본 경로는 변하지 않는다. malformed stream이 output_index를 다른 item - * type에 재사용해도 function_call id/call_id는 보존된다. opt-in 게이트웨이는 sequential streams에서도 - * 고유한 canonical id를 얻지만, 보정이 필요한 경우에만 JS stream 재작성 비용을 지불한다. + * Opt-in client-facing SSE repair for openai-responses gateways. + * rewriteNonCanonicalIds rewrites DeepSeek-style UUID item/response ids, converts + * reasoning_text streams into Codex-friendly encrypted_content reasoning items, and + * ensures a terminal [DONE] trailer so CLI/TUI turns finish. */ -/** Stateful payload rewrite for composition with other client-facing SSE transforms. */ export function createResponsesItemIdPayloadRewrite( config: ResponsesItemIdRepairConfig, budget?: TranslatorBudget, ): SsePayloadRewrite { const state = createRepairState(config, budget); - return (payload) => repairEventPayload(payload, state); + const rewrite: SsePayloadRewrite = (payload) => repairEventPayload(payload, state); + (rewrite as SsePayloadRewrite & { __state?: ResponsesItemIdRepairState }).__state = state; + return rewrite; +} + +export function createResponsesItemIdDoneTrailer( + rewrite: SsePayloadRewrite, +): () => string | undefined { + return () => { + const state = (rewrite as SsePayloadRewrite & { __state?: ResponsesItemIdRepairState }).__state; + if (!state) return undefined; + if (state.sawCompleted && !state.sawDoneTrailer) return "data: [DONE]\n\n"; + return undefined; + }; } export function relaySseWithResponsesItemIdRepair( @@ -214,11 +430,15 @@ export function relaySseWithResponsesItemIdRepair( config: ResponsesItemIdRepairConfig, budget: TranslatorBudget, ): ReadableStream { - return relaySseWithPayloadRewrite(body, createResponsesItemIdPayloadRewrite(config, budget), budget); + const rewrite = createResponsesItemIdPayloadRewrite(config, budget); + return relaySseWithPayloadRewrite(body, rewrite, budget, { + trailer: createResponsesItemIdDoneTrailer(rewrite), + }); } export function hasResponsesItemIdRepair(config: ResponsesItemIdRepairConfig | undefined): boolean { return config?.repairMissingTerminalIds === true + || config?.rewriteNonCanonicalIds === true || (config?.message?.length ?? 0) > 0 || (config?.reasoning?.length ?? 0) > 0; } diff --git a/src/server/sse-payload-rewrite.ts b/src/server/sse-payload-rewrite.ts index 712853cee1..26bbef7feb 100644 --- a/src/server/sse-payload-rewrite.ts +++ b/src/server/sse-payload-rewrite.ts @@ -7,7 +7,7 @@ import type { TranslatorBudget } from "../lib/translator-budget"; * parse/stringify pass so a tee'd stream is not re-framed twice per event. */ -export type SsePayloadRewrite = (payload: string) => string; +export type SsePayloadRewrite = (payload: string) => string | null; /** Split one complete SSE event block while retaining its original blank-line delimiter. */ export function nextSseBlock(buffer: string): { block: string; delimiter: string; rest: string } | null { @@ -55,8 +55,11 @@ export function composeSsePayloadRewrites(...rewrites: SsePayloadRewrite[]): Sse if (rewrites.length === 0) return (payload) => payload; if (rewrites.length === 1) return rewrites[0]!; return (payload) => { - let next = payload; - for (const rewrite of rewrites) next = rewrite(next); + let next: string | null = payload; + for (const rewrite of rewrites) { + if (next === null) return null; + next = rewrite(next); + } return next; }; } @@ -65,10 +68,16 @@ export function composeSsePayloadRewrites(...rewrites: SsePayloadRewrite[]): Sse * Relay an SSE body through a single JS pull wrapper, rewriting each event's data payload in place. * Non-data fields and framing are preserved; invalid JSON payloads are left to the rewrite callback. */ +export interface RelaySseRewriteOptions { + /** Optional trailer emitted once after the upstream body ends (e.g. `data: [DONE]\n\n`). */ + trailer?: string | (() => string | undefined); +} + export function relaySseWithPayloadRewrite( body: ReadableStream, rewrite: SsePayloadRewrite, translatorBudget: TranslatorBudget, + options?: RelaySseRewriteOptions, ): ReadableStream { const reader = body.getReader(); const decoder = new TextDecoder(); @@ -126,41 +135,65 @@ export function relaySseWithPayloadRewrite( const emitProcessedBlocks = ( controller: ReadableStreamDefaultController, flushFinal = false, - ): void => { + ): number => { + let enqueued = 0; let next: { block: string; delimiter: string; rest: string } | null; while ((next = nextSseBlock(buffer))) { replaceBuffer(next.rest); const payload = sseDataPayload(next.block); - const rewrittenPayload = payload ? rewrite(payload) : undefined; - const block = payload && rewrittenPayload !== undefined && rewrittenPayload !== payload - ? replaceSseDataPayload(next.block, rewrittenPayload) - : next.block; - enqueueText(controller, block + next.delimiter); + if (payload) { + const rewrittenPayload = rewrite(payload); + if (rewrittenPayload === null) continue; + const block = rewrittenPayload !== payload + ? replaceSseDataPayload(next.block, rewrittenPayload) + : next.block; + enqueueText(controller, block + next.delimiter); + enqueued += 1; + } else { + enqueueText(controller, next.block + next.delimiter); + enqueued += 1; + } } if (flushFinal && buffer.length > 0) { const payload = sseDataPayload(buffer); - const rewrittenPayload = payload ? rewrite(payload) : undefined; - const block = payload && rewrittenPayload !== undefined && rewrittenPayload !== payload - ? replaceSseDataPayload(buffer, rewrittenPayload) - : buffer; - enqueueText(controller, block); + if (payload) { + const rewrittenPayload = rewrite(payload); + if (rewrittenPayload !== null) { + const block = rewrittenPayload !== payload + ? replaceSseDataPayload(buffer, rewrittenPayload) + : buffer; + enqueueText(controller, block); + enqueued += 1; + } + } else { + enqueueText(controller, buffer); + enqueued += 1; + } releaseBuffer(); } + return enqueued; }; return new ReadableStream({ async pull(controller) { try { - const { done, value } = await reader.read(); - if (done) { - appendBuffer(decoder.decode()); - emitProcessedBlocks(controller, true); - releaseBuffer(); - controller.close(); - return; + // Keep reading while rewrites drop events so the consumer is never left waiting + // on a stream that still has upstream data but produced zero client-facing bytes. + while (true) { + const { done, value } = await reader.read(); + if (done) { + appendBuffer(decoder.decode()); + emitProcessedBlocks(controller, true); + const trailer = typeof options?.trailer === "function" ? options.trailer() : options?.trailer; + if (trailer) enqueueText(controller, trailer); + releaseBuffer(); + controller.close(); + return; + } + appendBuffer(decoder.decode(value, { stream: true })); + const enqueued = emitProcessedBlocks(controller); + if (enqueued > 0) return; } - appendBuffer(decoder.decode(value, { stream: true })); - emitProcessedBlocks(controller); } catch (error) { releaseBuffer(); try { await reader.cancel(error); } catch { /* already closed */ } diff --git a/src/types.ts b/src/types.ts index 0bbdc7e416..d21b45c444 100644 --- a/src/types.ts +++ b/src/types.ts @@ -908,6 +908,8 @@ export interface ResponsesItemIdRepairConfig { reasoning?: string[]; /** Backfill missing `output_item.done` / terminal snapshot ids from the matching output_index. */ repairMissingTerminalIds?: boolean; + /** Rewrite non-canonical message/reasoning item ids (e.g. DeepSeek UUIDs) to msg_/rs_ ids and normalize reasoning_text streams for Codex. */ + rewriteNonCanonicalIds?: boolean; } export interface OcxProviderConfig { diff --git a/tests/responses-item-id-repair-deepseek.test.ts b/tests/responses-item-id-repair-deepseek.test.ts new file mode 100644 index 0000000000..d973f57605 --- /dev/null +++ b/tests/responses-item-id-repair-deepseek.test.ts @@ -0,0 +1,95 @@ +import { describe, expect, test } from "bun:test"; +import { + hasResponsesItemIdRepair, + relaySseWithResponsesItemIdRepair as relaySseWithResponsesItemIdRepairProduction, +} from "../src/server/responses-item-id-repair"; +import { finalizeTranslatorBudgetResponse } from "../src/lib/translator-budget"; +import { createTestTranslatorBudget } from "./helpers/translator-budget"; + +function relaySseWithResponsesItemIdRepair( + body: ReadableStream, + config: Parameters[1], +): ReadableStream { + const budget = createTestTranslatorBudget(); + return finalizeTranslatorBudgetResponse( + new Response(relaySseWithResponsesItemIdRepairProduction(body, config, budget)), + budget, + ).body!; +} + +function streamFromText(text: string): ReadableStream { + const chunk = new TextEncoder().encode(text); + let sent = false; + return new ReadableStream({ + pull(controller) { + if (sent) { controller.close(); return; } + sent = true; + controller.enqueue(chunk); + }, + }); +} + +async function readAll(stream: ReadableStream): Promise { + const reader = stream.getReader(); + const decoder = new TextDecoder(); + let text = ""; + while (true) { + const { done, value } = await reader.read(); + if (done) break; + text += decoder.decode(value, { stream: true }); + } + return text; +} + +async function parseSse(text: string): Promise[]> { + return text + .trim() + .split(/\r?\n\r?\n/) + .map(block => block.split(/\r?\n/).find(line => line.startsWith("data:"))?.slice(5).trim()) + .filter((payload): payload is string => !!payload && payload !== "[DONE]") + .map(payload => JSON.parse(payload) as Record); +} + +describe("DeepSeek Responses item-id repair", () => { + test("rewrites UUID ids and folds reasoning_text into encrypted_content", async () => { + const upstream = [ + `event: response.created\ndata: {"type":"response.created","response":{"id":"58786d76-18be-43d9-b6f5-8922796fbe28","status":"in_progress","output":[]}}\n\n`, + `event: response.in_progress\ndata: {"type":"response.in_progress","response":{"id":"58786d76-18be-43d9-b6f5-8922796fbe28","status":"in_progress","output":[]}}\n\n`, + `event: response.output_item.added\ndata: {"type":"response.output_item.added","output_index":0,"item":{"type":"reasoning","id":"8da7b778-aff2-4d83-bfc3-8cd09ee79b34","status":"in_progress","content":[],"summary":[]}}\n\n`, + `event: response.reasoning_text.delta\ndata: {"type":"response.reasoning_text.delta","output_index":0,"item_id":"8da7b778-aff2-4d83-bfc3-8cd09ee79b34","delta":"think"}\n\n`, + `event: response.reasoning_text.done\ndata: {"type":"response.reasoning_text.done","output_index":0,"item_id":"8da7b778-aff2-4d83-bfc3-8cd09ee79b34","text":"think"}\n\n`, + `event: response.output_item.done\ndata: {"type":"response.output_item.done","output_index":0,"item":{"type":"reasoning","id":"8da7b778-aff2-4d83-bfc3-8cd09ee79b34","status":"completed","content":[{"type":"reasoning_text","text":"think"}],"summary":[]}}\n\n`, + `event: response.output_item.added\ndata: {"type":"response.output_item.added","output_index":1,"item":{"type":"message","id":"1dbc05ae-0b41-40fd-961e-7a84deebe064","status":"in_progress","role":"assistant","content":[]}}\n\n`, + `event: response.output_text.delta\ndata: {"type":"response.output_text.delta","output_index":1,"item_id":"1dbc05ae-0b41-40fd-961e-7a84deebe064","delta":"OK","logprobs":[]}\n\n`, + `event: response.output_item.done\ndata: {"type":"response.output_item.done","output_index":1,"item":{"type":"message","id":"1dbc05ae-0b41-40fd-961e-7a84deebe064","status":"completed","role":"assistant","content":[{"type":"output_text","text":"OK","annotations":[],"logprobs":[]}]}}\n\n`, + `event: response.completed\ndata: {"type":"response.completed","response":{"id":"58786d76-18be-43d9-b6f5-8922796fbe28","status":"completed","output":[{"type":"reasoning","id":"8da7b778-aff2-4d83-bfc3-8cd09ee79b34","content":[{"type":"reasoning_text","text":"think"}],"summary":[]},{"type":"message","id":"1dbc05ae-0b41-40fd-961e-7a84deebe064","role":"assistant","content":[{"type":"output_text","text":"OK","annotations":[],"logprobs":[]}]}]}}\n\n`, + ].join(""); + + const repaired = await readAll(relaySseWithResponsesItemIdRepair(streamFromText(upstream), { + rewriteNonCanonicalIds: true, + repairMissingTerminalIds: true, + })); + const events = await parseSse(repaired); + + expect(repaired).toContain("data: [DONE]"); + expect(events.some(e => e.type === "response.in_progress")).toBe(false); + expect(events.some(e => String(e.type).startsWith("response.reasoning_text"))).toBe(false); + + const reasoningAdded = events.find(e => e.type === "response.output_item.added" && (e.item as any)?.type === "reasoning")!.item as Record; + const messageAdded = events.find(e => e.type === "response.output_item.added" && (e.item as any)?.type === "message")!.item as Record; + const textDelta = events.find(e => e.type === "response.output_text.delta")!; + const completed = events.find(e => e.type === "response.completed")!.response as { id: string; output: Record[] }; + + expect(reasoningAdded.id).toMatch(/^rs_ocx_[0-9a-f]+_0$/); + expect(messageAdded.id).toMatch(/^msg_ocx_[0-9a-f]+_1$/); + expect(textDelta.item_id).toBe(messageAdded.id); + expect(completed.id).toMatch(/^resp_ocx_[0-9a-f]+$/); + expect(completed.output[0].id).toBe(reasoningAdded.id); + expect(String(completed.output[0].encrypted_content)).toMatch(/^ocxr1:/); + expect(completed.output[1].id).toBe(messageAdded.id); + }); + + test("opt-in flag is recognized", () => { + expect(hasResponsesItemIdRepair({ rewriteNonCanonicalIds: true })).toBe(true); + }); +}); diff --git a/tests/responses-item-id-repair.test.ts b/tests/responses-item-id-repair.test.ts index e708e6eb3a..fae1d643e1 100644 --- a/tests/responses-item-id-repair.test.ts +++ b/tests/responses-item-id-repair.test.ts @@ -185,6 +185,7 @@ describe("Responses passthrough item-id repair", () => { expect(hasResponsesItemIdRepair(undefined)).toBe(false); expect(hasResponsesItemIdRepair({})).toBe(false); expect(hasResponsesItemIdRepair({ repairMissingTerminalIds: true })).toBe(true); + expect(hasResponsesItemIdRepair({ rewriteNonCanonicalIds: true })).toBe(true); expect(hasResponsesItemIdRepair({ message: ["msg_0"] })).toBe(true); }); }); From 2739386382ddd5460e1e12ca5e7d578558d1b507 Mon Sep 17 00:00:00 2001 From: mouzhi Date: Mon, 3 Aug 2026 16:40:48 +0800 Subject: [PATCH 2/6] address review feedback for deepseek responses repair Make response ids unique, mint missing item ids under rewriteNonCanonicalIds, share rewrite/trailer state through one factory, pass trailers through the Windows eager path, and add focused regression tests. Also link the PR to #938. --- src/server/relay-eager.ts | 8 +++ src/server/responses-item-id-repair.ts | 54 +++++++++++-------- src/server/responses/core.ts | 23 +++++--- src/server/sse-payload-rewrite.ts | 4 +- .../responses-item-id-repair-deepseek.test.ts | 43 ++++++++++++++- 5 files changed, 102 insertions(+), 30 deletions(-) diff --git a/src/server/relay-eager.ts b/src/server/relay-eager.ts index 57faeabdd2..6e3dbbcbe4 100644 --- a/src/server/relay-eager.ts +++ b/src/server/relay-eager.ts @@ -58,6 +58,8 @@ export type EagerRelayHooks = { }; export type EagerRelayOptions = { + /** Optional trailer emitted after rewrite flush (e.g. data: [DONE]). */ + trailer?: string | (() => string | undefined); /** Bounded client queue in bytes; producer pauses above it. Default 8 MiB. */ maxQueueBytes?: number; /** Transient-budget owner for the inline-rewrite frame buffer. */ @@ -232,6 +234,12 @@ export function relaySseEagerBounded( queuedBytes += tail.byteLength; try { controllerRef?.enqueue(tail); } catch { /* client already gone */ } } + const trailer = typeof opts?.trailer === "function" ? opts.trailer() : opts?.trailer; + if (trailer && !cancelled) { + const trailerBytes = new TextEncoder().encode(trailer); + queuedBytes += trailerBytes.byteLength; + try { controllerRef?.enqueue(trailerBytes); } catch { /* client already gone */ } + } } if (!hooks.sawTerminal() && !cancelled && !upstream.signal.aborted) { syntheticKind = "incomplete"; diff --git a/src/server/responses-item-id-repair.ts b/src/server/responses-item-id-repair.ts index 47792bb353..fbd79d999a 100644 --- a/src/server/responses-item-id-repair.ts +++ b/src/server/responses-item-id-repair.ts @@ -86,7 +86,9 @@ function mapRawId( } if (!rawId) { - if (!state.repairMissingTerminalIds) return null; + // When non-canonical rewrite is active, mint a request-local id even if the upstream + // omitted the terminal id, so Codex lifecycle correlation stays stable. + if (!state.repairMissingTerminalIds && !state.rewriteNonCanonicalIds) return null; const minted = mintCanonicalId(type, state.scope, outputIndex); state.budget?.chargeRetained(new TextEncoder().encode(JSON.stringify([outputIndex, minted])).byteLength, { kind: "item_ids" }); state.outputIds[type].set(outputIndex, minted); @@ -151,7 +153,8 @@ function mapResponseId(state: ResponsesItemIdRepairState, rawId: string | undefi if (rawId.startsWith("resp_")) return rawId; const existing = state.responseIdMap.get(rawId); if (existing) return existing; - const minted = `resp_ocx_${state.scope}`; + // Keep each distinct upstream response id unique within the request-local scope. + const minted = `resp_ocx_${state.scope}_${state.responseIdMap.size}`; state.responseIdMap.set(rawId, minted); return minted; } @@ -220,7 +223,7 @@ function rewriteOutputItem( if (!mapped) return { item, changed: false }; const currentId = typeof item.id === "string" ? item.id : undefined; if (currentId === mapped) return { item, changed: false }; - if (currentId === undefined && !state.repairMissingTerminalIds) return { item, changed: false }; + if (currentId === undefined && !state.repairMissingTerminalIds && !state.rewriteNonCanonicalIds) return { item, changed: false }; return { item: { ...item, id: mapped }, changed: true }; } @@ -254,7 +257,7 @@ function rewriteItemIdField( } if (!mapped) return { event, changed: false }; if (currentId === mapped) return { event, changed: false }; - if (currentId === undefined && !state.repairMissingTerminalIds) return { event, changed: false }; + if (currentId === undefined && !state.repairMissingTerminalIds && !state.rewriteNonCanonicalIds) return { event, changed: false }; const next: Record = { ...event, item_id: mapped }; if (state.rewriteNonCanonicalIds && "logprobs" in next) delete next.logprobs; return { event: next, changed: true }; @@ -403,26 +406,35 @@ function repairEventPayload( * rewriteNonCanonicalIds rewrites DeepSeek-style UUID item/response ids, converts * reasoning_text streams into Codex-friendly encrypted_content reasoning items, and * ensures a terminal [DONE] trailer so CLI/TUI turns finish. + * + * The rewrite and trailer share one request-local state object via closure so + * composition cannot lose the trailer state. */ -export function createResponsesItemIdPayloadRewrite( +export interface ResponsesItemIdRepairHandlers { + rewrite: SsePayloadRewrite; + trailer: () => string | undefined; +} + +export function createResponsesItemIdRepairHandlers( config: ResponsesItemIdRepairConfig, budget?: TranslatorBudget, -): SsePayloadRewrite { +): ResponsesItemIdRepairHandlers { const state = createRepairState(config, budget); - const rewrite: SsePayloadRewrite = (payload) => repairEventPayload(payload, state); - (rewrite as SsePayloadRewrite & { __state?: ResponsesItemIdRepairState }).__state = state; - return rewrite; + return { + rewrite: (payload) => repairEventPayload(payload, state), + trailer: () => { + if (state.sawCompleted && !state.sawDoneTrailer) return "data: [DONE]\n\n"; + return undefined; + }, + }; } -export function createResponsesItemIdDoneTrailer( - rewrite: SsePayloadRewrite, -): () => string | undefined { - return () => { - const state = (rewrite as SsePayloadRewrite & { __state?: ResponsesItemIdRepairState }).__state; - if (!state) return undefined; - if (state.sawCompleted && !state.sawDoneTrailer) return "data: [DONE]\n\n"; - return undefined; - }; +/** Backward-compatible helper used by existing tests and callers. */ +export function createResponsesItemIdPayloadRewrite( + config: ResponsesItemIdRepairConfig, + budget?: TranslatorBudget, +): SsePayloadRewrite { + return createResponsesItemIdRepairHandlers(config, budget).rewrite; } export function relaySseWithResponsesItemIdRepair( @@ -430,9 +442,9 @@ export function relaySseWithResponsesItemIdRepair( config: ResponsesItemIdRepairConfig, budget: TranslatorBudget, ): ReadableStream { - const rewrite = createResponsesItemIdPayloadRewrite(config, budget); - return relaySseWithPayloadRewrite(body, rewrite, budget, { - trailer: createResponsesItemIdDoneTrailer(rewrite), + const handlers = createResponsesItemIdRepairHandlers(config, budget); + return relaySseWithPayloadRewrite(body, handlers.rewrite, budget, { + trailer: handlers.trailer, }); } diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 2271b04fd0..b17bf92f5b 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -147,7 +147,7 @@ import { relaySseEagerBounded } from "../relay-eager"; import { isWin32EagerRewrite, selectEagerPath } from "../../lib/bun-stream-caps"; import { cancelBodyOnAbort } from "../../lib/abort"; import { - createResponsesItemIdPayloadRewrite, + createResponsesItemIdRepairHandlers, hasResponsesItemIdRepair, } from "../responses-item-id-repair"; import { @@ -1851,12 +1851,13 @@ async function handleResponsesInner( if (isEventStream && upstreamResponse.body) { const repairConfig = route.provider.responsesItemIdRepair; const needsClientRewrite = imageGenCallAliases.size > 0 || hasResponsesItemIdRepair(repairConfig); + const itemIdRepair = hasResponsesItemIdRepair(repairConfig) + ? createResponsesItemIdRepairHandlers(repairConfig!, translatorBudget) + : undefined; // Compose opt-in payload rewrites into one parse/stringify pass (image-gen restore first). const payloadRewrites = [ createImageGenCallRestoreRewrite(imageGenCallAliases), - hasResponsesItemIdRepair(repairConfig) - ? createResponsesItemIdPayloadRewrite(repairConfig!, translatorBudget) - : undefined, + itemIdRepair?.rewrite, ].filter((rewrite): rewrite is NonNullable => rewrite !== undefined); // #864: win32 rewrite traffic must never enter the tee()+JS-pull chain // (Bun#32111 JS-sink segfault — text frames pass, the terminal block is @@ -1920,7 +1921,12 @@ async function handleResponsesInner( }, onClientCancel: () => options.onNativePassthroughCancel?.(), onDone: () => unregisterTurn(turnAc), - }, win32EagerRewrite ? { rewriteBudget: translatorBudget } : undefined); + }, win32EagerRewrite + ? { + rewriteBudget: translatorBudget, + ...(itemIdRepair ? { trailer: itemIdRepair.trailer } : {}), + } + : undefined); // selectEagerPath admits only no-rewrite traffic on both eligible platforms; // win32 rewrite traffic reaches this relay too, but with the payload rewrite // applied inline — never via an image/item-id JS pull wrapper (#32111, #864). @@ -1994,7 +2000,12 @@ async function handleResponsesInner( // relay is established practice (relayWithAbort, relaySseWithHeartbeat) and lets a // mid-stream reset end with a clean response.failed terminal instead of a raw socket error. const rewrittenBody = payloadRewrites.length > 0 - ? relaySseWithPayloadRewrite(nativeBody, composeSsePayloadRewrites(...payloadRewrites), translatorBudget) + ? relaySseWithPayloadRewrite( + nativeBody, + composeSsePayloadRewrites(...payloadRewrites), + translatorBudget, + itemIdRepair ? { trailer: itemIdRepair.trailer } : undefined, + ) : nativeBody; const clientBody = process.platform === "win32" && !needsClientRewrite ? nativeBody diff --git a/src/server/sse-payload-rewrite.ts b/src/server/sse-payload-rewrite.ts index 26bbef7feb..8fe0a95afc 100644 --- a/src/server/sse-payload-rewrite.ts +++ b/src/server/sse-payload-rewrite.ts @@ -141,7 +141,7 @@ export function relaySseWithPayloadRewrite( while ((next = nextSseBlock(buffer))) { replaceBuffer(next.rest); const payload = sseDataPayload(next.block); - if (payload) { + if (payload !== null) { const rewrittenPayload = rewrite(payload); if (rewrittenPayload === null) continue; const block = rewrittenPayload !== payload @@ -156,7 +156,7 @@ export function relaySseWithPayloadRewrite( } if (flushFinal && buffer.length > 0) { const payload = sseDataPayload(buffer); - if (payload) { + if (payload !== null) { const rewrittenPayload = rewrite(payload); if (rewrittenPayload !== null) { const block = rewrittenPayload !== payload diff --git a/tests/responses-item-id-repair-deepseek.test.ts b/tests/responses-item-id-repair-deepseek.test.ts index d973f57605..b2487d567a 100644 --- a/tests/responses-item-id-repair-deepseek.test.ts +++ b/tests/responses-item-id-repair-deepseek.test.ts @@ -83,12 +83,53 @@ describe("DeepSeek Responses item-id repair", () => { expect(reasoningAdded.id).toMatch(/^rs_ocx_[0-9a-f]+_0$/); expect(messageAdded.id).toMatch(/^msg_ocx_[0-9a-f]+_1$/); expect(textDelta.item_id).toBe(messageAdded.id); - expect(completed.id).toMatch(/^resp_ocx_[0-9a-f]+$/); + expect(completed.id).toMatch(/^resp_ocx_[0-9a-f]+_\d+$/); expect(completed.output[0].id).toBe(reasoningAdded.id); expect(String(completed.output[0].encrypted_content)).toMatch(/^ocxr1:/); expect(completed.output[1].id).toBe(messageAdded.id); }); + + test("mints unique response ids for distinct upstream response ids", async () => { + const upstream = [ + `event: response.completed\ndata: {"type":"response.completed","response":{"id":"11111111-1111-1111-1111-111111111111","status":"completed","output":[]}}\n\n`, + `event: response.completed\ndata: {"type":"response.completed","response":{"id":"22222222-2222-2222-2222-222222222222","status":"completed","output":[]}}\n\n`, + ].join(""); + const events = await parseSse(await readAll(relaySseWithResponsesItemIdRepair(streamFromText(upstream), { + rewriteNonCanonicalIds: true, + }))); + const first = (events[0].response as { id: string }).id; + const second = (events[1].response as { id: string }).id; + expect(first).toMatch(/^resp_ocx_[0-9a-f]+_0$/); + expect(second).toMatch(/^resp_ocx_[0-9a-f]+_1$/); + expect(first).not.toBe(second); + }); + + test("mints reasoning ids when rewriteNonCanonicalIds is enabled without repairMissingTerminalIds", async () => { + const upstream = [ + `event: response.output_item.added\ndata: {"type":"response.output_item.added","output_index":0,"item":{"type":"reasoning","summary":[]}}\n\n`, + `event: response.output_item.done\ndata: {"type":"response.output_item.done","output_index":0,"item":{"type":"reasoning","summary":[]}}\n\n`, + ].join(""); + const events = await parseSse(await readAll(relaySseWithResponsesItemIdRepair(streamFromText(upstream), { + rewriteNonCanonicalIds: true, + }))); + const added = events[0].item as Record; + const done = events[1].item as Record; + expect(added.id).toMatch(/^rs_ocx_[0-9a-f]+_0$/); + expect(done.id).toBe(added.id); + }); + + test("does not duplicate an upstream [DONE] trailer", async () => { + const upstream = [ + `event: response.completed\ndata: {"type":"response.completed","response":{"id":"58786d76-18be-43d9-b6f5-8922796fbe28","status":"completed","output":[]}}\n\n`, + `data: [DONE]\n\n`, + ].join(""); + const repaired = await readAll(relaySseWithResponsesItemIdRepair(streamFromText(upstream), { + rewriteNonCanonicalIds: true, + })); + expect(repaired.match(/data: \[DONE\]/g)).toHaveLength(1); + }); + test("opt-in flag is recognized", () => { expect(hasResponsesItemIdRepair({ rewriteNonCanonicalIds: true })).toBe(true); }); From aaaad0475801724655e7a01bdb8249a5c4664c31 Mon Sep 17 00:00:00 2001 From: mouzhi Date: Mon, 3 Aug 2026 16:56:15 +0800 Subject: [PATCH 3/6] address remaining review for deepseek responses repair Budget responseIdMap and reasoning text, emit [DONE] for failed/incomplete, preserve plaintext reasoning for DeepSeek replay, alias rewritten response ids into local previous_response_id state, document rewriteNonCanonicalIds, and cover eager trailer plus budget regressions. --- .../ja/reference/configuration/providers.md | 5 +- .../ko/reference/configuration/providers.md | 5 +- .../docs/reference/configuration/providers.md | 8 +- .../ru/reference/configuration/providers.md | 5 +- .../reference/configuration/providers.md | 7 +- src/server/responses-item-id-repair.ts | 94 ++++++++++++++----- src/server/responses/core.ts | 15 ++- .../responses-item-id-repair-deepseek.test.ts | 63 ++++++++++++- 8 files changed, 162 insertions(+), 40 deletions(-) diff --git a/docs-site/src/content/docs/ja/reference/configuration/providers.md b/docs-site/src/content/docs/ja/reference/configuration/providers.md index 39a4513973..b3303ca71a 100644 --- a/docs-site/src/content/docs/ja/reference/configuration/providers.md +++ b/docs-site/src/content/docs/ja/reference/configuration/providers.md @@ -84,7 +84,7 @@ namespace 付き combo alias はその namespace prefix に selector を再利 | `noTopPModels?` | `string[]` |発信者指定の`top_p`を拒否するモデル。 | | `noPenaltyModels?` | `string[]` |存在/周波数ペナルティを拒否するモデル。 | | `parallelToolCalls?` | `boolean` |並列ツール呼び出しを切り替えます。 OpenAI Chat はデフォルトでオンになっています。非チャット アダプターは明示的な `true` でのみアドバタイズします。 | -| `responsesItemIdRepair?` | `{ message?: string[]; reasoning?: string[]; repairMissingTerminalIds?: boolean }` |正確なプレースホルダー ID および欠落している端末 ID に対するダウンストリーム SSE 修復はデフォルトで無効になっています。関数呼び出し ID は決して書き換えられません。 | +| `responsesItemIdRepair?` | `{ message?: string[]; reasoning?: string[]; repairMissingTerminalIds?: boolean; rewriteNonCanonicalIds?: boolean }` |正確なプレースホルダー ID、欠落している端末 ID、および非標準の UUID 形式 response/item ID（`rewriteNonCanonicalIds`）に対するダウンストリーム SSE 修復はデフォルトで無効です。関数呼び出し ID は決して書き換えられません。 | | `autoToolChoiceOnlyModels?` | `string[]` | `tool_choice` が `auto` または `none` のみを受け入れるモデル。強制的な選択は格下げされます。 | | `preserveReasoningContentModels?` | `string[]` |チャット履歴に以前のアシスタント `reasoning_content` が必要なモデル。 | | `thinkingToggleModels?` | `string[]` |エフォート ラダーではなく `thinking.enabled` を使用してモデルをチャットします。 | @@ -197,7 +197,8 @@ Anthropic アカウント ポリシーのリスクを理解していない限り "responsesItemIdRepair": { "reasoning": ["rs_0"], "message": ["msg_0"], - "repairMissingTerminalIds": true + "repairMissingTerminalIds": true, + "rewriteNonCanonicalIds": true } } } diff --git a/docs-site/src/content/docs/ko/reference/configuration/providers.md b/docs-site/src/content/docs/ko/reference/configuration/providers.md index f3d08deefb..38657de56d 100644 --- a/docs-site/src/content/docs/ko/reference/configuration/providers.md +++ b/docs-site/src/content/docs/ko/reference/configuration/providers.md @@ -84,7 +84,7 @@ target도 selector로 재사용할 수 없습니다. raw account id와 email은 | `noTopPModels?` | `string[]` | 호출자가 지정한 `top_p`를 거부하는 모델입니다. | | `noPenaltyModels?` | `string[]` | presence/frequency penalty를 허용하지 않는 모델입니다. | | `parallelToolCalls?` | `boolean` | 병렬 도구 호출을 켜거나 끕니다. OpenAI Chat은 기본으로 켜져 있고, 비-chat 어댑터는 명시적으로 `true`일 때만 이를 노출합니다. | -| `responsesItemIdRepair?` | `{ message?: string[]; reasoning?: string[]; repairMissingTerminalIds?: boolean }` | 기본값이 꺼진 downstream SSE 복구입니다. 정확한 자리표시자 id와 누락된 종료 id를 복구합니다. function-call id는 다시 쓰지 않습니다. | +| `responsesItemIdRepair?` | `{ message?: string[]; reasoning?: string[]; repairMissingTerminalIds?: boolean; rewriteNonCanonicalIds?: boolean }` | 기본값이 꺼진 downstream SSE 복구입니다. 정확한 자리표시자 id, 누락된 종료 id, 비표준 UUID 스타일 response/item id(`rewriteNonCanonicalIds`)를 복구합니다. function-call id는 다시 쓰지 않습니다. | | `autoToolChoiceOnlyModels?` | `string[]` | `tool_choice`가 `auto` 또는 `none`만 받는 모델입니다. 강제 선택은 낮은 수준으로 바뀝니다. | | `preserveReasoningContentModels?` | `string[]` | chat 기록에서 이전 assistant `reasoning_content`가 필요한 모델입니다. | | `thinkingToggleModels?` | `string[]` | effort 계층 대신 `thinking.enabled`를 쓰는 chat 모델입니다. | @@ -200,7 +200,8 @@ Anthropic 계정 정책 위험을 이해하지 못한다면 이 기능은 꺼두 "responsesItemIdRepair": { "reasoning": ["rs_0"], "message": ["msg_0"], - "repairMissingTerminalIds": true + "repairMissingTerminalIds": true, + "rewriteNonCanonicalIds": true } } } diff --git a/docs-site/src/content/docs/reference/configuration/providers.md b/docs-site/src/content/docs/reference/configuration/providers.md index a04a142efc..2e0f553f28 100644 --- a/docs-site/src/content/docs/reference/configuration/providers.md +++ b/docs-site/src/content/docs/reference/configuration/providers.md @@ -91,7 +91,7 @@ differing backup and rewrites known legacy namespaced selected ids to bare ids. | `noTopPModels?` | `string[]` | Models that reject caller-specified `top_p`. | | `noPenaltyModels?` | `string[]` | Models that reject presence/frequency penalties. | | `parallelToolCalls?` | `boolean` | Toggle parallel tool calls. OpenAI Chat defaults on; non-chat adapters advertise only on explicit `true`. | -| `responsesItemIdRepair?` | `{ message?: string[]; reasoning?: string[]; repairMissingTerminalIds?: boolean }` | Disabled-by-default downstream SSE repair for exact placeholder ids and missing terminal ids. Function-call ids are never rewritten. | +| `responsesItemIdRepair?` | `{ message?: string[]; reasoning?: string[]; repairMissingTerminalIds?: boolean; rewriteNonCanonicalIds?: boolean }` | Disabled-by-default downstream SSE repair for exact placeholder ids, missing terminal ids, and non-canonical UUID-style response/item ids (`rewriteNonCanonicalIds`). Function-call ids are never rewritten. | | `autoToolChoiceOnlyModels?` | `string[]` | Models whose `tool_choice` accepts only `auto` or `none`; forced choices are downgraded. | | `preserveReasoningContentModels?` | `string[]` | Models requiring prior assistant `reasoning_content` in chat history. | | `thinkingToggleModels?` | `string[]` | Chat models using `thinking.enabled` rather than an effort ladder. | @@ -237,15 +237,15 @@ For a broken `openai-responses` gateway, repair belongs on the provider object: "responsesItemIdRepair": { "reasoning": ["rs_0"], "message": ["msg_0"], - "repairMissingTerminalIds": true + "repairMissingTerminalIds": true, + "rewriteNonCanonicalIds": true } } } } ``` -Placeholder lists are exact matches. Leave the field unset for normal/stateful Responses providers -so passthrough stays byte-for-byte identical. +Placeholder lists are exact matches. Set `rewriteNonCanonicalIds` for gateways such as DeepSeek Responses that emit UUID item/response ids and raw `reasoning_text` streams; OpenCodex rewrites those ids to Codex-friendly prefixes, folds reasoning into `encrypted_content` (while keeping plaintext `content` for replay), and ensures a terminal `[DONE]`. Leave the field unset for normal/stateful Responses providers so passthrough stays byte-for-byte identical. ## Cursor provider (`adapter: "cursor"`) diff --git a/docs-site/src/content/docs/ru/reference/configuration/providers.md b/docs-site/src/content/docs/ru/reference/configuration/providers.md index 18d884525f..61f8b1d4d4 100644 --- a/docs-site/src/content/docs/ru/reference/configuration/providers.md +++ b/docs-site/src/content/docs/ru/reference/configuration/providers.md @@ -94,7 +94,7 @@ cross-route credential fallback не существует. Строки API GPT- | `noTopPModels?` | `string[]` | Модели, отвергающие переданный вызывающей стороной `top_p`. | | `noPenaltyModels?` | `string[]` | Модели, отвергающие penalty presence/frequency. | | `parallelToolCalls?` | `boolean` | Переключатель parallel tool call'ов. Для OpenAI Chat по умолчанию включено; не-chat adapter'ы рекламируют это только при явном `true`. | -| `responsesItemIdRepair?` | `{ message?: string[]; reasoning?: string[]; repairMissingTerminalIds?: boolean }` | По умолчанию выключенная downstream SSE-repair для exact placeholder-id и отсутствующих terminal-id. Function-call id никогда не переписываются. | +| `responsesItemIdRepair?` | `{ message?: string[]; reasoning?: string[]; repairMissingTerminalIds?: boolean; rewriteNonCanonicalIds?: boolean }` | По умолчанию выключенная downstream SSE-repair для exact placeholder-id, отсутствующих terminal-id и неканонических UUID-style response/item id (`rewriteNonCanonicalIds`). Function-call id никогда не переписываются. | | `autoToolChoiceOnlyModels?` | `string[]` | Модели, у которых `tool_choice` принимает только `auto` или `none`; forced choice понижается. | | `preserveReasoningContentModels?` | `string[]` | Модели, которым нужен предыдущий assistant `reasoning_content` в chat history. | | `thinkingToggleModels?` | `string[]` | Chat-модели, использующие `thinking.enabled` вместо effort-ladder. | @@ -245,7 +245,8 @@ Beijing, а `alibaba-token-plan-intl` обслуживает междунаро "responsesItemIdRepair": { "reasoning": ["rs_0"], "message": ["msg_0"], - "repairMissingTerminalIds": true + "repairMissingTerminalIds": true, + "rewriteNonCanonicalIds": true } } } diff --git a/docs-site/src/content/docs/zh-cn/reference/configuration/providers.md b/docs-site/src/content/docs/zh-cn/reference/configuration/providers.md index 3702c00467..8dd1d98421 100644 --- a/docs-site/src/content/docs/zh-cn/reference/configuration/providers.md +++ b/docs-site/src/content/docs/zh-cn/reference/configuration/providers.md @@ -83,7 +83,7 @@ pool account id（不能是内部 `__main__`），或用 `"@main"` 表示 Codex | `noTopPModels?` | `string[]` | 会拒绝调用方指定 `top_p` 的模型。 | | `noPenaltyModels?` | `string[]` | 会拒绝 presence/frequency penalty 的模型。 | | `parallelToolCalls?` | `boolean` | 切换并行工具调用。OpenAI Chat 默认开启；非 chat 适配器只有显式 `true` 时才会声明支持。 | -| `responsesItemIdRepair?` | `{ message?: string[]; reasoning?: string[]; repairMissingTerminalIds?: boolean }` | 默认关闭的下游 SSE 修复，用于精确占位 id 和缺失的终止 id。function-call id 永远不会被重写。 | +| `responsesItemIdRepair?` | `{ message?: string[]; reasoning?: string[]; repairMissingTerminalIds?: boolean; rewriteNonCanonicalIds?: boolean }` | 默认关闭的下游 SSE 修复，用于精确占位 id、缺失的终止 id，以及非标准 UUID 风格的 response/item id（`rewriteNonCanonicalIds`）。function-call id 永远不会被重写。 | | `autoToolChoiceOnlyModels?` | `string[]` | `tool_choice` 只接受 `auto` 或 `none` 的模型；强制选择会被降级。 | | `preserveReasoningContentModels?` | `string[]` | 需要在聊天历史中保留先前 assistant `reasoning_content` 的模型。 | | `thinkingToggleModels?` | `string[]` | 使用 `thinking.enabled` 而不是 effort 阶梯的 chat 模型。 | @@ -193,14 +193,15 @@ affinity。这些策略不能规避 provider enforcement。 "responsesItemIdRepair": { "reasoning": ["rs_0"], "message": ["msg_0"], - "repairMissingTerminalIds": true + "repairMissingTerminalIds": true, + "rewriteNonCanonicalIds": true } } } } ``` -占位列表必须精确匹配。对于正常/有状态的 Responses 提供者，请保持该字段未设置，以便转发能保持逐字节一致。 +占位列表必须精确匹配。对 DeepSeek Responses 这类会返回 UUID item/response id 与原始 `reasoning_text` 流的网关，请设置 `rewriteNonCanonicalIds`：OpenCodex 会把 id 改写成 Codex 友好前缀，把 reasoning 折叠进 `encrypted_content`（同时保留明文 `content` 便于回放），并补齐终端 `[DONE]`。对于正常/有状态的 Responses 提供者，请保持该字段未设置，以便转发能保持逐字节一致。 ## Cursor 提供者（`adapter: "cursor"`） diff --git a/src/server/responses-item-id-repair.ts b/src/server/responses-item-id-repair.ts index fbd79d999a..af9d0dd316 100644 --- a/src/server/responses-item-id-repair.ts +++ b/src/server/responses-item-id-repair.ts @@ -16,7 +16,8 @@ interface ResponsesItemIdRepairState { readonly responseIdMap: Map; readonly scope: string; readonly budget?: TranslatorBudget; - sawCompleted: boolean; + /** True once a terminal response event is observed (completed/failed/incomplete). */ + sawTerminal: boolean; sawDoneTrailer: boolean; } @@ -41,6 +42,12 @@ const ITEM_ID_EVENT_TYPES: Readonly> = { "response.reasoning_text.done": "reasoning", }; +const TERMINAL_RESPONSE_TYPES = new Set([ + "response.completed", + "response.failed", + "response.incomplete", +]); + function isPlainObject(value: unknown): value is Record { return !!value && typeof value === "object" && !Array.isArray(value); } @@ -124,7 +131,7 @@ function createRepairState(config: ResponsesItemIdRepairConfig, budget?: Transla responseIdMap: new Map(), scope: randomUUID().replace(/-/g, ""), budget, - sawCompleted: false, + sawTerminal: false, sawDoneTrailer: false, }; budget?.chargeRetained(new TextEncoder().encode(JSON.stringify({ @@ -155,10 +162,53 @@ function mapResponseId(state: ResponsesItemIdRepairState, rawId: string | undefi if (existing) return existing; // Keep each distinct upstream response id unique within the request-local scope. const minted = `resp_ocx_${state.scope}_${state.responseIdMap.size}`; + state.budget?.chargeRetained( + new TextEncoder().encode(JSON.stringify([rawId, minted])).byteLength, + { kind: "item_ids" }, + ); state.responseIdMap.set(rawId, minted); return minted; } +function releaseReasoningText(state: ResponsesItemIdRepairState, outputIndex: number): void { + const existing = state.reasoningTextByOutputIndex.get(outputIndex); + if (existing === undefined) return; + state.budget?.releaseRetained(new TextEncoder().encode(existing).byteLength, { kind: "reasoning" }); + state.reasoningTextByOutputIndex.delete(outputIndex); +} + +function setReasoningText( + state: ResponsesItemIdRepairState, + outputIndex: number, + text: string, +): void { + const existing = state.reasoningTextByOutputIndex.get(outputIndex); + if (existing === text) return; + if (existing !== undefined) { + state.budget?.releaseRetained(new TextEncoder().encode(existing).byteLength, { kind: "reasoning" }); + } + if (!text) { + state.reasoningTextByOutputIndex.delete(outputIndex); + return; + } + state.budget?.chargeRetained(new TextEncoder().encode(text).byteLength, { kind: "reasoning" }); + state.reasoningTextByOutputIndex.set(outputIndex, text); +} + +function accumulateReasoningText( + state: ResponsesItemIdRepairState, + outputIndex: number, + delta: string, +): void { + if (!delta) return; + // Charge only the newly retained delta bytes; release the whole entry when the stream ends. + state.budget?.chargeRetained(new TextEncoder().encode(delta).byteLength, { kind: "reasoning" }); + state.reasoningTextByOutputIndex.set( + outputIndex, + (state.reasoningTextByOutputIndex.get(outputIndex) ?? "") + delta, + ); +} + function normalizeReasoningItem( state: ResponsesItemIdRepairState, outputIndex: number, @@ -173,15 +223,24 @@ function normalizeReasoningItem( .map(part => String(part.text)) .join("") : ""); - if (text) state.reasoningTextByOutputIndex.set(outputIndex, text); + if (text && !state.reasoningTextByOutputIndex.has(outputIndex)) { + // Snapshot content-derived text into the budgeted map only when we still need it later. + if (!terminal) setReasoningText(state, outputIndex, text); + } const next: Record = { type: "reasoning", id: mapped, summary: Array.isArray(item.summary) ? item.summary : [], }; if (terminal || text) { + // Codex consumes encrypted_content; DeepSeek tool-call continuations need plaintext + // reasoning_text after sanitizeReasoningInputContent strips the proxy-only ocxr1 envelope. next.encrypted_content = encodeReasoningEnvelope({ txt: text || " " }); + if (text) { + next.content = [{ type: "reasoning_text", text }]; + } } + if (terminal) releaseReasoningText(state, outputIndex); return next; } @@ -286,17 +345,6 @@ function rewriteResponseSnapshot( return changed ? { response: { ...next, output }, changed: true } : { response: next, changed }; } -function accumulateReasoningText( - state: ResponsesItemIdRepairState, - outputIndex: number, - delta: string, -): void { - state.reasoningTextByOutputIndex.set( - outputIndex, - (state.reasoningTextByOutputIndex.get(outputIndex) ?? "") + delta, - ); -} - function repairEventPayload( payload: string, state: ResponsesItemIdRepairState, @@ -327,7 +375,7 @@ function repairEventPayload( } if (type === "response.reasoning_text.done") { if (outputIndex !== null && typeof event.text === "string") { - state.reasoningTextByOutputIndex.set(outputIndex, event.text); + setReasoningText(state, outputIndex, event.text); } return null; } @@ -337,7 +385,7 @@ function repairEventPayload( && event.part.type === "reasoning_text" ) { if (outputIndex !== null && typeof event.part.text === "string" && event.part.text) { - state.reasoningTextByOutputIndex.set(outputIndex, event.part.text); + setReasoningText(state, outputIndex, event.part.text); } return null; } @@ -369,7 +417,7 @@ function repairEventPayload( } } - if (type === "response.completed") state.sawCompleted = true; + if (type && TERMINAL_RESPONSE_TYPES.has(type)) state.sawTerminal = true; if (state.rewriteNonCanonicalIds && isPlainObject(nextEvent.part) && nextEvent.part.type === "output_text") { nextEvent = { @@ -380,10 +428,6 @@ function repairEventPayload( annotations: Array.isArray(nextEvent.part.annotations) ? nextEvent.part.annotations : [], }, }; - if ("logprobs" in nextEvent) { - const { logprobs: _lp, ...rest } = nextEvent; - nextEvent = rest; - } changed = true; } if (state.rewriteNonCanonicalIds && "logprobs" in nextEvent) { @@ -413,6 +457,11 @@ function repairEventPayload( export interface ResponsesItemIdRepairHandlers { rewrite: SsePayloadRewrite; trailer: () => string | undefined; + /** + * Map a provider-raw response id to the client-visible id emitted on the repaired + * stream. Used to alias local previous_response_id continuation state. + */ + clientResponseId: (rawId: string | undefined) => string | undefined; } export function createResponsesItemIdRepairHandlers( @@ -423,9 +472,10 @@ export function createResponsesItemIdRepairHandlers( return { rewrite: (payload) => repairEventPayload(payload, state), trailer: () => { - if (state.sawCompleted && !state.sawDoneTrailer) return "data: [DONE]\n\n"; + if (state.sawTerminal && !state.sawDoneTrailer) return "data: [DONE]\n\n"; return undefined; }, + clientResponseId: (rawId) => mapResponseId(state, rawId), }; } diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index b17bf92f5b..c607dcd7b7 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -1643,9 +1643,19 @@ async function handleResponsesInner( // recording it would let a later expansion rehydrate the chain Codex just replaced. const passthroughRecordEligible = parsed._compactionRequest !== true && (!parsed.previousResponseId || parsed._previousResponseInputExpanded === true); + // When item-id repair rewrites response ids, the inspector still sees the raw + // upstream UUID while Codex chains previous_response_id with the client-visible + // resp_ocx_* id. Alias both keys into the local continuation cache. + let mapClientResponseId: ((rawId: string | undefined) => string | undefined) | undefined; const rememberPassthroughResponse = passthroughRecordEligible - ? (response: { id?: unknown; output?: unknown; status?: unknown }) => - rememberResponseState(parsed._rawBody, response, undefined, { force: true }) + ? (response: { id?: unknown; output?: unknown; status?: unknown }) => { + rememberResponseState(parsed._rawBody, response, undefined, { force: true }); + if (typeof response.id !== "string" || !mapClientResponseId) return; + const clientId = mapClientResponseId(response.id); + if (clientId && clientId !== response.id) { + rememberResponseState(parsed._rawBody, { ...response, id: clientId }, undefined, { force: true }); + } + } : undefined; if (parsed.previousResponseId && !parsed._previousResponseInputExpanded) { console.warn( @@ -1854,6 +1864,7 @@ async function handleResponsesInner( const itemIdRepair = hasResponsesItemIdRepair(repairConfig) ? createResponsesItemIdRepairHandlers(repairConfig!, translatorBudget) : undefined; + if (itemIdRepair) mapClientResponseId = itemIdRepair.clientResponseId; // Compose opt-in payload rewrites into one parse/stringify pass (image-gen restore first). const payloadRewrites = [ createImageGenCallRestoreRewrite(imageGenCallAliases), diff --git a/tests/responses-item-id-repair-deepseek.test.ts b/tests/responses-item-id-repair-deepseek.test.ts index b2487d567a..0aefa816c9 100644 --- a/tests/responses-item-id-repair-deepseek.test.ts +++ b/tests/responses-item-id-repair-deepseek.test.ts @@ -1,16 +1,18 @@ import { describe, expect, test } from "bun:test"; import { + createResponsesItemIdRepairHandlers, hasResponsesItemIdRepair, relaySseWithResponsesItemIdRepair as relaySseWithResponsesItemIdRepairProduction, } from "../src/server/responses-item-id-repair"; -import { finalizeTranslatorBudgetResponse } from "../src/lib/translator-budget"; +import { finalizeTranslatorBudgetResponse, isTranslatorBudgetExceededError } from "../src/lib/translator-budget"; import { createTestTranslatorBudget } from "./helpers/translator-budget"; +import { relaySseEagerBounded } from "../src/server/relay-eager"; function relaySseWithResponsesItemIdRepair( body: ReadableStream, config: Parameters[1], + budget = createTestTranslatorBudget(), ): ReadableStream { - const budget = createTestTranslatorBudget(); return finalizeTranslatorBudgetResponse( new Response(relaySseWithResponsesItemIdRepairProduction(body, config, budget)), budget, @@ -86,10 +88,13 @@ describe("DeepSeek Responses item-id repair", () => { expect(completed.id).toMatch(/^resp_ocx_[0-9a-f]+_\d+$/); expect(completed.output[0].id).toBe(reasoningAdded.id); expect(String(completed.output[0].encrypted_content)).toMatch(/^ocxr1:/); + expect((completed.output[0].content as Array<{ type: string; text: string }>)[0]).toEqual({ + type: "reasoning_text", + text: "think", + }); expect(completed.output[1].id).toBe(messageAdded.id); }); - test("mints unique response ids for distinct upstream response ids", async () => { const upstream = [ `event: response.completed\ndata: {"type":"response.completed","response":{"id":"11111111-1111-1111-1111-111111111111","status":"completed","output":[]}}\n\n`, @@ -130,6 +135,58 @@ describe("DeepSeek Responses item-id repair", () => { expect(repaired.match(/data: \[DONE\]/g)).toHaveLength(1); }); + test("emits [DONE] for response.failed and response.incomplete", async () => { + for (const type of ["response.failed", "response.incomplete"] as const) { + const upstream = `event: ${type}\ndata: {"type":"${type}","response":{"id":"58786d76-18be-43d9-b6f5-8922796fbe28","status":"${type === "response.failed" ? "failed" : "incomplete"}","output":[]}}\n\n`; + const repaired = await readAll(relaySseWithResponsesItemIdRepair(streamFromText(upstream), { + rewriteNonCanonicalIds: true, + })); + expect(repaired.match(/data: \[DONE\]/g)).toHaveLength(1); + } + }); + + test("charges responseIdMap through TranslatorBudget", () => { + const budget = createTestTranslatorBudget({ maxTurnBytes: 256 }); + const handlers = createResponsesItemIdRepairHandlers({ rewriteNonCanonicalIds: true }, budget); + let overflowed = false; + try { + for (let i = 0; i < 200; i++) { + const raw = `${i.toString(16).padStart(8, "0")}-0000-4000-8000-${i.toString(16).padStart(12, "0")}`; + handlers.rewrite(JSON.stringify({ + type: "response.completed", + response: { id: raw, status: "completed", output: [] }, + })); + } + } catch (error) { + overflowed = isTranslatorBudgetExceededError(error); + } + expect(overflowed).toBe(true); + }); + + test("eager relay emits exactly one [DONE] trailer without upstream DONE", async () => { + const budget = createTestTranslatorBudget(); + const handlers = createResponsesItemIdRepairHandlers({ rewriteNonCanonicalIds: true }, budget); + const upstream = [ + `event: response.completed\ndata: {"type":"response.completed","response":{"id":"58786d76-18be-43d9-b6f5-8922796fbe28","status":"completed","output":[]}}\n\n`, + ].join(""); + const body = streamFromText(upstream); + const ac = new AbortController(); + const relayed = relaySseEagerBounded(body, ac, { + inspectChunk: () => {}, + finishInspection: () => {}, + sawTerminal: () => true, + onSynthetic: () => {}, + onClientCancel: () => {}, + onDone: () => {}, + rewritePayload: handlers.rewrite, + }, { + rewriteBudget: budget, + trailer: handlers.trailer, + }); + const repaired = await readAll(relayed); + expect(repaired.match(/data: \[DONE\]/g)).toHaveLength(1); + }); + test("opt-in flag is recognized", () => { expect(hasResponsesItemIdRepair({ rewriteNonCanonicalIds: true })).toBe(true); }); From 9f25360369713a36699ae41f43064758b3c73369 Mon Sep 17 00:00:00 2001 From: mouzhi Date: Mon, 3 Aug 2026 22:25:41 +0800 Subject: [PATCH 4/6] fix type-scoped aliases and enable built-in deepseek repair Scope rawToCanonical by message/reasoning, charge newly retained aliases, and turn on rewriteNonCanonicalIds for the built-in DeepSeek preset so the native Responses route gets the #938 UUID repair by default. --- src/providers/derive.ts | 1 + src/providers/registry.ts | 10 ++++ src/server/responses-item-id-repair.ts | 39 ++++++++++----- .../responses-item-id-repair-deepseek.test.ts | 48 +++++++++++++++++++ 4 files changed, 86 insertions(+), 12 deletions(-) diff --git a/src/providers/derive.ts b/src/providers/derive.ts index ba4fc62440..617e141d0a 100644 --- a/src/providers/derive.ts +++ b/src/providers/derive.ts @@ -269,6 +269,7 @@ export function enrichProviderFromRegistry(name: string, prov: OcxProviderConfig // the entry so an explicit user value stays distinguishable from the default. if (prov.supportsServiceTier === undefined && entry.supportsServiceTier !== undefined) prov.supportsServiceTier = entry.supportsServiceTier; if (prov.preserveResponsesReasoningContent === undefined && entry.preserveResponsesReasoningContent !== undefined) prov.preserveResponsesReasoningContent = entry.preserveResponsesReasoningContent; + if (prov.responsesItemIdRepair === undefined && entry.responsesItemIdRepair !== undefined) prov.responsesItemIdRepair = { ...entry.responsesItemIdRepair }; if (!prov.autoToolChoiceOnlyModels && seed.autoToolChoiceOnlyModels) prov.autoToolChoiceOnlyModels = [...seed.autoToolChoiceOnlyModels]; if (!prov.preserveReasoningContentModels && seed.preserveReasoningContentModels) prov.preserveReasoningContentModels = [...seed.preserveReasoningContentModels]; if (!prov.reasoningSplitModels && seed.reasoningSplitModels) prov.reasoningSplitModels = [...seed.reasoningSplitModels]; diff --git a/src/providers/registry.ts b/src/providers/registry.ts index 91d3ad07dd..514fdfee5f 100644 --- a/src/providers/registry.ts +++ b/src/providers/registry.ts @@ -184,6 +184,8 @@ export interface ProviderRegistryEntry { supportsServiceTier?: boolean; /** Registry default for plaintext reasoning replay; see `OcxProviderConfig.preserveResponsesReasoningContent`. Registry-only like `supportsServiceTier`. */ preserveResponsesReasoningContent?: boolean; + /** Registry default for client-facing Responses item-id repair (fill-only into runtime provider config). */ + responsesItemIdRepair?: import("../types").ResponsesItemIdRepairConfig; modelDiscovery?: ProviderModelDiscoverySpec; contextWindow?: number; modelContextWindows?: Record; @@ -1093,6 +1095,14 @@ export const PROVIDER_REGISTRY: readonly ProviderRegistryEntry[] = [ // route REQUIRES replay on tool-call continuations is an inference from the // Chat Thinking-Mode docs, not a confirmed Responses contract.) preserveResponsesReasoningContent: true, + // DeepSeek Responses emits UUID item/response ids and raw reasoning_text streams that + // leave Codex App/CLI stuck on Thinking/Working even after HTTP 200 (#938 family). + // Enable the client-facing SSE repair by default for the built-in provider so Codex + // Desktop/CLI do not need a hand-edited provider config for the native route. + responsesItemIdRepair: { + rewriteNonCanonicalIds: true, + repairMissingTerminalIds: true, + }, // "The API is stateless: responses and conversations are not stored on the // server." https://api-docs.deepseek.com/api/create-response/ statelessResponses: true, diff --git a/src/server/responses-item-id-repair.ts b/src/server/responses-item-id-repair.ts index af9d0dd316..85c01894a4 100644 --- a/src/server/responses-item-id-repair.ts +++ b/src/server/responses-item-id-repair.ts @@ -11,7 +11,7 @@ interface ResponsesItemIdRepairState { readonly rewriteNonCanonicalIds: boolean; readonly placeholders: Record>; readonly outputIds: Record>; - readonly rawToCanonical: Map; + readonly rawToCanonical: Record>; readonly reasoningTextByOutputIndex: Map; readonly responseIdMap: Map; readonly scope: string; @@ -86,8 +86,13 @@ function mapRawId( ): string | null { const existing = state.outputIds[type].get(outputIndex); if (existing) { - if (rawId && shouldRewriteRawId(state, type, rawId) && !state.rawToCanonical.has(rawId)) { - state.rawToCanonical.set(rawId, existing); + if (rawId && shouldRewriteRawId(state, type, rawId) && !state.rawToCanonical[type].has(rawId)) { + // Charge every newly retained raw alias, even when the output_index is already mapped. + state.budget?.chargeRetained( + new TextEncoder().encode(JSON.stringify([type, outputIndex, rawId, existing])).byteLength, + { kind: "item_ids" }, + ); + state.rawToCanonical[type].set(rawId, existing); } return existing; } @@ -110,7 +115,7 @@ function mapRawId( if (!mapped) return null; state.budget?.chargeRetained(new TextEncoder().encode(JSON.stringify([outputIndex, rawId, mapped])).byteLength, { kind: "item_ids" }); state.outputIds[type].set(outputIndex, mapped); - if (mapped !== rawId) state.rawToCanonical.set(rawId, mapped); + if (mapped !== rawId) state.rawToCanonical[type].set(rawId, mapped); return mapped; } @@ -126,7 +131,10 @@ function createRepairState(config: ResponsesItemIdRepairConfig, budget?: Transla message: new Map(), reasoning: new Map(), }, - rawToCanonical: new Map(), + rawToCanonical: { + message: new Map(), + reasoning: new Map(), + }, reasoningTextByOutputIndex: new Map(), responseIdMap: new Map(), scope: randomUUID().replace(/-/g, ""), @@ -292,27 +300,34 @@ function rewriteItemIdField( outputIndex: number, ): { event: Record; changed: boolean } { const currentId = typeof event.item_id === "string" ? event.item_id : undefined; - const reverseMapped = currentId ? state.rawToCanonical.get(currentId) : undefined; - if (reverseMapped && reverseMapped !== currentId) { - return { event: { ...event, item_id: reverseMapped }, changed: true }; + const eventType = typeof event.type === "string" ? ITEM_ID_EVENT_TYPES[event.type] : undefined; + if (!eventType) return { event, changed: false }; + + // Alias lookup is type-scoped so a shared placeholder cannot rewrite reasoning -> msg_*. + if (currentId) { + const reverseMapped = state.rawToCanonical[eventType].get(currentId); + if (reverseMapped && reverseMapped !== currentId) { + const next: Record = { ...event, item_id: reverseMapped }; + if (state.rewriteNonCanonicalIds && "logprobs" in next) delete next.logprobs; + return { event: next, changed: true }; + } } - const eventType = typeof event.type === "string" ? ITEM_ID_EVENT_TYPES[event.type] : undefined; let mapped: string | null | undefined; if (eventType === "message" && ( event.type === "response.content_part.added" || event.type === "response.content_part.done" )) { + // content_part events can belong to either a message or a reasoning item at the + // same output_index; prefer the already-mapped type without inventing a cross-type alias. mapped = state.outputIds.message.get(outputIndex) ?? state.outputIds.reasoning.get(outputIndex) ?? null; - } else if (eventType) { + } else { mapped = state.outputIds[eventType].get(outputIndex); if (!mapped && currentId && state.rewriteNonCanonicalIds) { mapped = mapRawId(state, eventType, outputIndex, currentId); } - } else { - return { event, changed: false }; } if (!mapped) return { event, changed: false }; if (currentId === mapped) return { event, changed: false }; diff --git a/tests/responses-item-id-repair-deepseek.test.ts b/tests/responses-item-id-repair-deepseek.test.ts index 0aefa816c9..45b12b020d 100644 --- a/tests/responses-item-id-repair-deepseek.test.ts +++ b/tests/responses-item-id-repair-deepseek.test.ts @@ -187,6 +187,54 @@ describe("DeepSeek Responses item-id repair", () => { expect(repaired.match(/data: \[DONE\]/g)).toHaveLength(1); }); + test("keeps shared placeholder aliases type-scoped", async () => { + const upstream = [ + 'event: response.output_item.added\ndata: {"type":"response.output_item.added","output_index":0,"item":{"type":"reasoning","id":"shared_placeholder"}}\n\n', + 'event: response.reasoning_summary_text.delta\ndata: {"type":"response.reasoning_summary_text.delta","output_index":0,"item_id":"shared_placeholder","delta":"r"}\n\n', + 'event: response.output_item.added\ndata: {"type":"response.output_item.added","output_index":1,"item":{"type":"message","id":"shared_placeholder","role":"assistant"}}\n\n', + 'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","output_index":1,"item_id":"shared_placeholder","delta":"m"}\n\n', + ].join(""); + const events = await parseSse(await readAll(relaySseWithResponsesItemIdRepair(streamFromText(upstream), { + reasoning: ["shared_placeholder"], + message: ["shared_placeholder"], + }))); + const reasoningId = (events[0].item as Record).id; + const messageId = (events[2].item as Record).id; + expect(reasoningId).toMatch(/^rs_ocx_[0-9a-f]+_0$/); + expect(messageId).toMatch(/^msg_ocx_[0-9a-f]+_1$/); + expect(reasoningId).not.toBe(messageId); + expect(events[1].item_id).toBe(reasoningId); + expect(events[3].item_id).toBe(messageId); + }); + + test("charges newly retained raw aliases after an output_index is already mapped", () => { + const budget = createTestTranslatorBudget({ maxTurnBytes: 512 }); + const handlers = createResponsesItemIdRepairHandlers({ + rewriteNonCanonicalIds: true, + }, budget); + // Establish one mapped reasoning id first. + handlers.rewrite(JSON.stringify({ + type: "response.output_item.added", + output_index: 0, + item: { type: "reasoning", id: "11111111-1111-4111-8111-111111111111" }, + })); + let overflowed = false; + try { + for (let i = 0; i < 400; i++) { + const raw = `${i.toString(16).padStart(8, "0")}-0000-4000-8000-${i.toString(16).padStart(12, "0")}`; + handlers.rewrite(JSON.stringify({ + type: "response.reasoning_summary_text.delta", + output_index: 0, + item_id: raw, + delta: "x", + })); + } + } catch (error) { + overflowed = isTranslatorBudgetExceededError(error); + } + expect(overflowed).toBe(true); + }); + test("opt-in flag is recognized", () => { expect(hasResponsesItemIdRepair({ rewriteNonCanonicalIds: true })).toBe(true); }); From eea5e3bf233128d7bebd7bf32fe950e3c4e088c1 Mon Sep 17 00:00:00 2001 From: mouzhi Date: Mon, 3 Aug 2026 22:52:37 +0800 Subject: [PATCH 5/6] polish deepseek responses repair review details Deep-clone registry repair arrays, preserve reasoning status metadata, flush retained reasoning into terminal snapshots, document rewrite behavior in ja/ko/ru, and cover composition trailer plus residual reasoning flush. --- .../ja/reference/configuration/providers.md | 2 +- .../ko/reference/configuration/providers.md | 2 +- .../ru/reference/configuration/providers.md | 2 +- src/providers/derive.ts | 9 ++- src/server/responses-item-id-repair.ts | 64 +++++++++++++++++-- src/server/responses/core.ts | 3 + .../responses-item-id-repair-deepseek.test.ts | 35 ++++++++++ tests/sse-payload-rewrite.test.ts | 13 ++-- 8 files changed, 114 insertions(+), 16 deletions(-) diff --git a/docs-site/src/content/docs/ja/reference/configuration/providers.md b/docs-site/src/content/docs/ja/reference/configuration/providers.md index b3303ca71a..27e25fe9e8 100644 --- a/docs-site/src/content/docs/ja/reference/configuration/providers.md +++ b/docs-site/src/content/docs/ja/reference/configuration/providers.md @@ -205,7 +205,7 @@ Anthropic アカウント ポリシーのリスクを理解していない限り } ``` -プレースホルダー リストは完全に一致します。通常/ステートフル応答プロバイダーのフィールドを未設定のままにして、パススルーがバイトごとに同一になるようにします。 +プレースホルダー リストは完全に一致します。DeepSeek Responses のように UUID 形式の response/item ID と生の `reasoning_text` ストリームを返すゲートウェイでは `rewriteNonCanonicalIds` を有効にします。OpenCodex はそれらの ID を Codex 向けプレフィックスへ書き換え、reasoning を `encrypted_content` に正規化しつつ replay 用の平文 `content` を保持し、終端の `[DONE]` を補完します。通常/ステートフルな Responses プロバイダーではこのフィールドを未設定のままにして、パススルーがバイト単位で同一になるようにします。 ## Cursor プロバイダー (`adapter: "cursor"`) diff --git a/docs-site/src/content/docs/ko/reference/configuration/providers.md b/docs-site/src/content/docs/ko/reference/configuration/providers.md index 38657de56d..41e9f48217 100644 --- a/docs-site/src/content/docs/ko/reference/configuration/providers.md +++ b/docs-site/src/content/docs/ko/reference/configuration/providers.md @@ -208,7 +208,7 @@ Anthropic 계정 정책 위험을 이해하지 못한다면 이 기능은 꺼두 } ``` -자리표시자 목록은 정확히 일치해야 합니다. 일반적인 상태 유지형 Responses 공급자에서는 필드를 설정하지 마십시오. 그래야 passthrough가 바이트 단위로 동일하게 유지됩니다. +자리표시자 목록은 정확히 일치해야 합니다. DeepSeek Responses처럼 UUID 형식 response/item id와 원본 `reasoning_text` 스트림을 반환하는 게이트웨이에서는 `rewriteNonCanonicalIds`를 켜십시오. OpenCodex는 해당 id를 Codex 친화적 prefix로 다시 쓰고, reasoning을 `encrypted_content`로 정규화하면서 replay용 평문 `content`는 유지하며, 종료용 `[DONE]`을 보장합니다. 일반/상태 유지형 Responses 공급자에서는 필드를 설정하지 마십시오. 그래야 passthrough가 바이트 단위로 동일하게 유지됩니다. ## Cursor 공급자 (`adapter: "cursor"`) diff --git a/docs-site/src/content/docs/ru/reference/configuration/providers.md b/docs-site/src/content/docs/ru/reference/configuration/providers.md index 61f8b1d4d4..d77dfc1a56 100644 --- a/docs-site/src/content/docs/ru/reference/configuration/providers.md +++ b/docs-site/src/content/docs/ru/reference/configuration/providers.md @@ -253,7 +253,7 @@ Beijing, а `alibaba-token-plan-intl` обслуживает междунаро } ``` -Списки placeholder'ов сравниваются по exact-match. Для обычных/stateful Responses-провайдеров это +Списки placeholder'ов сравниваются по exact-match. Для gateway вроде DeepSeek Responses, которые отдают UUID-style response/item id и сырые `reasoning_text` потоки, включайте `rewriteNonCanonicalIds`: OpenCodex переписывает id в Codex-friendly prefix, нормализует reasoning в `encrypted_content` (сохраняя plaintext `content` для replay) и гарантирует terminal `[DONE]`. Для обычных/stateful Responses-провайдеров это поле оставляйте unset, чтобы passthrough оставался byte-for-byte идентичным. ## Провайдер Cursor (`adapter: "cursor"`) diff --git a/src/providers/derive.ts b/src/providers/derive.ts index 617e141d0a..e40fc7e283 100644 --- a/src/providers/derive.ts +++ b/src/providers/derive.ts @@ -269,7 +269,14 @@ export function enrichProviderFromRegistry(name: string, prov: OcxProviderConfig // the entry so an explicit user value stays distinguishable from the default. if (prov.supportsServiceTier === undefined && entry.supportsServiceTier !== undefined) prov.supportsServiceTier = entry.supportsServiceTier; if (prov.preserveResponsesReasoningContent === undefined && entry.preserveResponsesReasoningContent !== undefined) prov.preserveResponsesReasoningContent = entry.preserveResponsesReasoningContent; - if (prov.responsesItemIdRepair === undefined && entry.responsesItemIdRepair !== undefined) prov.responsesItemIdRepair = { ...entry.responsesItemIdRepair }; + if (prov.responsesItemIdRepair === undefined && entry.responsesItemIdRepair !== undefined) { + const repair = entry.responsesItemIdRepair; + prov.responsesItemIdRepair = { + ...repair, + ...(repair.message ? { message: [...repair.message] } : {}), + ...(repair.reasoning ? { reasoning: [...repair.reasoning] } : {}), + }; + } if (!prov.autoToolChoiceOnlyModels && seed.autoToolChoiceOnlyModels) prov.autoToolChoiceOnlyModels = [...seed.autoToolChoiceOnlyModels]; if (!prov.preserveReasoningContentModels && seed.preserveReasoningContentModels) prov.preserveReasoningContentModels = [...seed.preserveReasoningContentModels]; if (!prov.reasoningSplitModels && seed.reasoningSplitModels) prov.reasoningSplitModels = [...seed.reasoningSplitModels]; diff --git a/src/server/responses-item-id-repair.ts b/src/server/responses-item-id-repair.ts index 85c01894a4..3f0a4755a0 100644 --- a/src/server/responses-item-id-repair.ts +++ b/src/server/responses-item-id-repair.ts @@ -235,11 +235,12 @@ function normalizeReasoningItem( // Snapshot content-derived text into the budgeted map only when we still need it later. if (!terminal) setReasoningText(state, outputIndex, text); } - const next: Record = { - type: "reasoning", - id: mapped, - summary: Array.isArray(item.summary) ? item.summary : [], - }; + // Preserve upstream metadata such as status; only proxy-owned reasoning fields are rebuilt. + const next: Record = { ...item, type: "reasoning" }; + if (mapped !== undefined) next.id = mapped; + next.summary = Array.isArray(item.summary) ? item.summary : []; + delete next.content; + delete next.encrypted_content; if (terminal || text) { // Codex consumes encrypted_content; DeepSeek tool-call continuations need plaintext // reasoning_text after sanitizeReasoningInputContent strips the proxy-only ocxr1 envelope. @@ -337,6 +338,44 @@ function rewriteItemIdField( return { event: next, changed: true }; } +function flushRetainedReasoningIntoResponse( + state: ResponsesItemIdRepairState, + response: Record, +): { response: Record; changed: boolean } { + if (!state.rewriteNonCanonicalIds || state.reasoningTextByOutputIndex.size === 0) { + return { response, changed: false }; + } + const output = Array.isArray(response.output) ? [...response.output] : []; + let changed = false; + for (const [outputIndex, text] of [...state.reasoningTextByOutputIndex.entries()]) { + if (!text) { + releaseReasoningText(state, outputIndex); + continue; + } + const existing = isPlainObject(output[outputIndex]) ? output[outputIndex] as Record : null; + if (existing?.type === "reasoning") { + const rewritten = normalizeReasoningItem(state, outputIndex, existing, true); + output[outputIndex] = rewritten; + changed = true; + continue; + } + // No reasoning item was emitted for this retained stream; synthesize one so the + // terminal snapshot still carries the accumulated reasoning text. + const synthetic = normalizeReasoningItem(state, outputIndex, { + type: "reasoning", + summary: [], + content: [{ type: "reasoning_text", text }], + }, true); + if (outputIndex < output.length) output[outputIndex] = synthetic; + else { + while (output.length < outputIndex) output.push({ type: "message", role: "assistant", content: [] }); + output.push(synthetic); + } + changed = true; + } + return changed ? { response: { ...response, output }, changed: true } : { response, changed: false }; +} + function rewriteResponseSnapshot( state: ResponsesItemIdRepairState, response: Record, @@ -425,11 +464,22 @@ function repairEventPayload( } } if (isPlainObject(event.response)) { - const rewritten = rewriteResponseSnapshot(state, event.response); - if (rewritten.changed) { + let response = event.response; + if (type && TERMINAL_RESPONSE_TYPES.has(type) && state.rewriteNonCanonicalIds) { + const flushed = flushRetainedReasoningIntoResponse(state, response); + response = flushed.response; + changed = changed || flushed.changed; + } + const rewritten = rewriteResponseSnapshot(state, response); + if (rewritten.changed || response !== event.response) { nextEvent = { ...nextEvent, response: rewritten.response }; changed = true; } + } else if (type && TERMINAL_RESPONSE_TYPES.has(type) && state.rewriteNonCanonicalIds && state.reasoningTextByOutputIndex.size > 0) { + // Terminal events without a response snapshot still need retained reasoning released. + for (const outputIndex of [...state.reasoningTextByOutputIndex.keys()]) { + releaseReasoningText(state, outputIndex); + } } if (type && TERMINAL_RESPONSE_TYPES.has(type)) state.sawTerminal = true; diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index c607dcd7b7..384a9e1769 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -1646,6 +1646,9 @@ async function handleResponsesInner( // When item-id repair rewrites response ids, the inspector still sees the raw // upstream UUID while Codex chains previous_response_id with the client-visible // resp_ocx_* id. Alias both keys into the local continuation cache. + // mapClientResponseId is assigned only on the SSE rewrite path below. The JSON + // response path currently does not apply Responses item-id repair; if JSON repair + // is added later, initialize this alias before that branch records continuation state. let mapClientResponseId: ((rawId: string | undefined) => string | undefined) | undefined; const rememberPassthroughResponse = passthroughRecordEligible ? (response: { id?: unknown; output?: unknown; status?: unknown }) => { diff --git a/tests/responses-item-id-repair-deepseek.test.ts b/tests/responses-item-id-repair-deepseek.test.ts index 45b12b020d..db6ddc9cc6 100644 --- a/tests/responses-item-id-repair-deepseek.test.ts +++ b/tests/responses-item-id-repair-deepseek.test.ts @@ -187,6 +187,41 @@ describe("DeepSeek Responses item-id repair", () => { expect(repaired.match(/data: \[DONE\]/g)).toHaveLength(1); }); + test("preserves reasoning status metadata", async () => { + const upstream = [ + `event: response.output_item.added\ndata: {"type":"response.output_item.added","output_index":0,"item":{"type":"reasoning","id":"8da7b778-aff2-4d83-bfc3-8cd09ee79b34","status":"in_progress","summary":[]}}\n\n`, + `event: response.reasoning_text.delta\ndata: {"type":"response.reasoning_text.delta","output_index":0,"item_id":"8da7b778-aff2-4d83-bfc3-8cd09ee79b34","delta":"think"}\n\n`, + `event: response.output_item.done\ndata: {"type":"response.output_item.done","output_index":0,"item":{"type":"reasoning","id":"8da7b778-aff2-4d83-bfc3-8cd09ee79b34","status":"completed","summary":[]}}\n\n`, + `event: response.completed\ndata: {"type":"response.completed","response":{"id":"58786d76-18be-43d9-b6f5-8922796fbe28","status":"completed","output":[{"type":"reasoning","id":"8da7b778-aff2-4d83-bfc3-8cd09ee79b34","status":"completed","summary":[]}]}}\n\n`, + ].join(""); + const events = await parseSse(await readAll(relaySseWithResponsesItemIdRepair(streamFromText(upstream), { + rewriteNonCanonicalIds: true, + }))); + const added = events.find(e => e.type === "response.output_item.added")!.item as Record; + const done = events.find(e => e.type === "response.output_item.done")!.item as Record; + const completed = events.find(e => e.type === "response.completed")!.response as { output: Record[] }; + expect(added.status).toBe("in_progress"); + expect(done.status).toBe("completed"); + expect(completed.output[0].status).toBe("completed"); + expect(String(done.encrypted_content)).toMatch(/^ocxr1:/); + }); + + test("flushes retained reasoning text into terminal snapshot when item.done is missing", async () => { + const upstream = [ + `event: response.output_item.added\ndata: {"type":"response.output_item.added","output_index":0,"item":{"type":"reasoning","id":"8da7b778-aff2-4d83-bfc3-8cd09ee79b34","status":"in_progress","summary":[]}}\n\n`, + `event: response.reasoning_text.delta\ndata: {"type":"response.reasoning_text.delta","output_index":0,"item_id":"8da7b778-aff2-4d83-bfc3-8cd09ee79b34","delta":"orphan"}\n\n`, + `event: response.completed\ndata: {"type":"response.completed","response":{"id":"58786d76-18be-43d9-b6f5-8922796fbe28","status":"completed","output":[]}}\n\n`, + ].join(""); + const events = await parseSse(await readAll(relaySseWithResponsesItemIdRepair(streamFromText(upstream), { + rewriteNonCanonicalIds: true, + }))); + const completed = events.find(e => e.type === "response.completed")!.response as { output: Record[] }; + expect(completed.output).toHaveLength(1); + expect(completed.output[0].type).toBe("reasoning"); + expect(String(completed.output[0].encrypted_content)).toMatch(/^ocxr1:/); + expect((completed.output[0].content as Array<{ text: string }>)[0].text).toBe("orphan"); + }); + test("keeps shared placeholder aliases type-scoped", async () => { const upstream = [ 'event: response.output_item.added\ndata: {"type":"response.output_item.added","output_index":0,"item":{"type":"reasoning","id":"shared_placeholder"}}\n\n', diff --git a/tests/sse-payload-rewrite.test.ts b/tests/sse-payload-rewrite.test.ts index c6c72f615b..cffed752d9 100644 --- a/tests/sse-payload-rewrite.test.ts +++ b/tests/sse-payload-rewrite.test.ts @@ -3,7 +3,7 @@ */ import { describe, expect, test } from "bun:test"; import { createImageGenCallRestoreRewrite } from "../src/server/responses-image-gen-repair"; -import { createResponsesItemIdPayloadRewrite } from "../src/server/responses-item-id-repair"; +import { createResponsesItemIdRepairHandlers } from "../src/server/responses-item-id-repair"; import { composeSsePayloadRewrites, relaySseWithPayloadRewrite, @@ -51,7 +51,7 @@ describe("SSE payload rewrite composition", () => { const imageGen = createImageGenCallRestoreRewrite( new Map([["image_gen__imagegen", { namespace: "image_gen", name: "imagegen" }]]), )!; - const itemId = createResponsesItemIdPayloadRewrite({ + const itemId = createResponsesItemIdRepairHandlers({ message: ["msg_0"], repairMissingTerminalIds: true, }); @@ -63,13 +63,16 @@ describe("SSE payload rewrite composition", () => { }, (payload) => { itemIdCalls += 1; - return itemId(payload); + return itemId.rewrite(payload); }, ); const budget = createTestTranslatorBudget(); - const out = await readAll(relaySseWithPayloadRewrite(streamFromText(upstream), composed, budget)); + const out = await readAll(relaySseWithPayloadRewrite(streamFromText(upstream), composed, budget, { + trailer: itemId.trailer, + })); budget.dispose(); + expect(out).toContain("data: [DONE]"); expect(imageGenCalls).toBe(3); expect(itemIdCalls).toBe(3); expect(imageGenCalls).toBe(itemIdCalls); @@ -78,7 +81,7 @@ describe("SSE payload rewrite composition", () => { .trim() .split(/\r?\n\r?\n/) .map(block => block.split(/\r?\n/).find(line => line.startsWith("data:"))?.slice(5).trim()) - .filter((payload): payload is string => !!payload) + .filter((payload): payload is string => !!payload && payload !== "[DONE]") .map(payload => JSON.parse(payload) as Record); const messageAdded = events[0].item as Record; From 570a52c2d8ce680150563fd58693ca04e334ee5e Mon Sep 17 00:00:00 2001 From: mouzhi Date: Tue, 4 Aug 2026 00:54:03 +0800 Subject: [PATCH 6/6] fix reasoning flush overwrite and continuity alias notes Do not overwrite existing assistant messages when synthesizing retained reasoning, dual-write rewritten response ids for previous_response_id continuity, fix Russian docs agreement, and add regressions. --- .../ru/reference/configuration/providers.md | 2 +- src/server/responses-item-id-repair.ts | 10 ++++-- src/server/responses/core.ts | 5 +++ .../responses-item-id-repair-deepseek.test.ts | 36 +++++++++++++++++++ 4 files changed, 49 insertions(+), 4 deletions(-) diff --git a/docs-site/src/content/docs/ru/reference/configuration/providers.md b/docs-site/src/content/docs/ru/reference/configuration/providers.md index d77dfc1a56..319afb6a12 100644 --- a/docs-site/src/content/docs/ru/reference/configuration/providers.md +++ b/docs-site/src/content/docs/ru/reference/configuration/providers.md @@ -253,7 +253,7 @@ Beijing, а `alibaba-token-plan-intl` обслуживает междунаро } ``` -Списки placeholder'ов сравниваются по exact-match. Для gateway вроде DeepSeek Responses, которые отдают UUID-style response/item id и сырые `reasoning_text` потоки, включайте `rewriteNonCanonicalIds`: OpenCodex переписывает id в Codex-friendly prefix, нормализует reasoning в `encrypted_content` (сохраняя plaintext `content` для replay) и гарантирует terminal `[DONE]`. Для обычных/stateful Responses-провайдеров это +Списки placeholder'ов сравниваются по exact-match. Для шлюзов вроде DeepSeek Responses, которые возвращают UUID-style response/item id и сырые `reasoning_text` потоки, включайте `rewriteNonCanonicalIds`: OpenCodex переписывает id в Codex-friendly prefix, нормализует reasoning в `encrypted_content` (сохраняя plaintext `content` для replay) и гарантирует terminal `[DONE]`. Для обычных/stateful Responses-провайдеров это поле оставляйте unset, чтобы passthrough оставался byte-for-byte идентичным. ## Провайдер Cursor (`adapter: "cursor"`) diff --git a/src/server/responses-item-id-repair.ts b/src/server/responses-item-id-repair.ts index 3f0a4755a0..f5da33d846 100644 --- a/src/server/responses-item-id-repair.ts +++ b/src/server/responses-item-id-repair.ts @@ -361,15 +361,19 @@ function flushRetainedReasoningIntoResponse( } // No reasoning item was emitted for this retained stream; synthesize one so the // terminal snapshot still carries the accumulated reasoning text. + // Never overwrite a non-reasoning item (e.g. assistant message) at the same index. const synthetic = normalizeReasoningItem(state, outputIndex, { type: "reasoning", summary: [], content: [{ type: "reasoning_text", text }], }, true); - if (outputIndex < output.length) output[outputIndex] = synthetic; - else { + if (!existing) { while (output.length < outputIndex) output.push({ type: "message", role: "assistant", content: [] }); - output.push(synthetic); + if (outputIndex < output.length) output[outputIndex] = synthetic; + else output.push(synthetic); + } else { + // Keep the existing assistant/message/tool item and insert reasoning just before it. + output.splice(outputIndex, 0, synthetic); } changed = true; } diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 384a9e1769..9713730d6e 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -1652,8 +1652,13 @@ async function handleResponsesInner( let mapClientResponseId: ((rawId: string | undefined) => string | undefined) | undefined; const rememberPassthroughResponse = passthroughRecordEligible ? (response: { id?: unknown; output?: unknown; status?: unknown }) => { + // Always record the inspector-visible id first (usually the raw upstream UUID). rememberResponseState(parsed._rawBody, response, undefined, { force: true }); if (typeof response.id !== "string" || !mapClientResponseId) return; + // rewriteNonCanonicalIds rewrites client-facing response ids to resp_ocx_*. + // Codex continues with previous_response_id using that client-visible id, so the + // local replay cache must also be keyed by the rewritten id or multi-turn context + // is dropped on the next request. const clientId = mapClientResponseId(response.id); if (clientId && clientId !== response.id) { rememberResponseState(parsed._rawBody, { ...response, id: clientId }, undefined, { force: true }); diff --git a/tests/responses-item-id-repair-deepseek.test.ts b/tests/responses-item-id-repair-deepseek.test.ts index db6ddc9cc6..853d5863c3 100644 --- a/tests/responses-item-id-repair-deepseek.test.ts +++ b/tests/responses-item-id-repair-deepseek.test.ts @@ -95,6 +95,21 @@ describe("DeepSeek Responses item-id repair", () => { expect(completed.output[1].id).toBe(messageAdded.id); }); + test("clientResponseId dual-maps raw and rewritten response ids", () => { + const handlers = createResponsesItemIdRepairHandlers({ rewriteNonCanonicalIds: true }); + const raw = "58786d76-18be-43d9-b6f5-8922796fbe28"; + // Simulate rewrite seeing the completed response first. + handlers.rewrite(JSON.stringify({ + type: "response.completed", + response: { id: raw, status: "completed", output: [] }, + })); + const clientId = handlers.clientResponseId(raw); + expect(clientId).toMatch(/^resp_ocx_[0-9a-f]+_\d+$/); + expect(clientId).not.toBe(raw); + // Stable across repeated lookups so continuation aliasing can dual-write safely. + expect(handlers.clientResponseId(raw)).toBe(clientId); + }); + test("mints unique response ids for distinct upstream response ids", async () => { const upstream = [ `event: response.completed\ndata: {"type":"response.completed","response":{"id":"11111111-1111-1111-1111-111111111111","status":"completed","output":[]}}\n\n`, @@ -222,6 +237,27 @@ describe("DeepSeek Responses item-id repair", () => { expect((completed.output[0].content as Array<{ text: string }>)[0].text).toBe("orphan"); }); + test("does not overwrite an existing assistant message when flushing retained reasoning", async () => { + const upstream = [ + `event: response.output_item.added\ndata: {"type":"response.output_item.added","output_index":0,"item":{"type":"message","id":"1dbc05ae-0b41-40fd-961e-7a84deebe064","role":"assistant","content":[]}}\n\n`, + // Retain reasoning text under output_index 0 even though the terminal snapshot has a message there. + `event: response.reasoning_text.delta\ndata: {"type":"response.reasoning_text.delta","output_index":0,"item_id":"8da7b778-aff2-4d83-bfc3-8cd09ee79b34","delta":"keep-me"}\n\n`, + `event: response.output_item.done\ndata: {"type":"response.output_item.done","output_index":0,"item":{"type":"message","id":"1dbc05ae-0b41-40fd-961e-7a84deebe064","role":"assistant","content":[{"type":"output_text","text":"OK","annotations":[]}]}}\n\n`, + `event: response.completed\ndata: {"type":"response.completed","response":{"id":"58786d76-18be-43d9-b6f5-8922796fbe28","status":"completed","output":[{"type":"message","id":"1dbc05ae-0b41-40fd-961e-7a84deebe064","role":"assistant","content":[{"type":"output_text","text":"OK","annotations":[]}]}]}}\n\n`, + ].join(""); + const events = await parseSse(await readAll(relaySseWithResponsesItemIdRepair(streamFromText(upstream), { + rewriteNonCanonicalIds: true, + }))); + const completed = events.find(e => e.type === "response.completed")!.response as { output: Record[] }; + expect(completed.output.length).toBeGreaterThanOrEqual(2); + const reasoning = completed.output.find(item => item.type === "reasoning"); + const message = completed.output.find(item => item.type === "message"); + expect(reasoning).toBeTruthy(); + expect(message).toBeTruthy(); + expect((message!.content as Array<{ text: string }>)[0].text).toBe("OK"); + expect(String(reasoning!.encrypted_content)).toMatch(/^ocxr1:/); + }); + test("keeps shared placeholder aliases type-scoped", async () => { const upstream = [ 'event: response.output_item.added\ndata: {"type":"response.output_item.added","output_index":0,"item":{"type":"reasoning","id":"shared_placeholder"}}\n\n',