From 71486566525748d72c9cf38ae1dec7c52d9907b6 Mon Sep 17 00:00:00 2001 From: ivan-mezentsev Date: Wed, 29 Apr 2026 14:53:23 +0300 Subject: [PATCH] fix(codex): stabilize copilot responses reasoning and tool replay (#1750) --- open-sse/executors/codex.ts | 63 +++++++- open-sse/handlers/chatCore.ts | 3 +- open-sse/services/responsesToolCallState.ts | 109 ++++++++++++++ open-sse/utils/stream.ts | 156 ++++++++++++++++++-- tests/unit/executor-codex.test.ts | 47 ++++++ tests/unit/stream-utilities.test.ts | 87 ++++++++++- 6 files changed, 447 insertions(+), 18 deletions(-) create mode 100644 open-sse/services/responsesToolCallState.ts diff --git a/open-sse/executors/codex.ts b/open-sse/executors/codex.ts index 42b3049a7f..b95ab31a68 100644 --- a/open-sse/executors/codex.ts +++ b/open-sse/executors/codex.ts @@ -12,6 +12,7 @@ import { import { PROVIDERS } from "../config/constants.ts"; import { getCodexClientVersion, getCodexUserAgent } from "../config/codexClient.ts"; import { getAccessToken } from "../services/tokenRefresh.ts"; +import { getRememberedResponseFunctionCalls } from "../services/responsesToolCallState.ts"; import { getThinkingBudgetConfig, ThinkingMode } from "../services/thinkingBudget.ts"; import { CORS_HEADERS } from "../utils/cors.ts"; import { createRequire } from "module"; @@ -359,16 +360,61 @@ function convertSystemToDeveloperRole(body: Record): void { * 3. Strips the "id" field from any object in input whose id matches a * server-generated prefix (rs_, fc_, resp_, msg_) — so the content is * preserved but the backend won't try to look it up - * 4. Always deletes previous_response_id (endpoint doesn't persist responses) + * 4. Rehydrates missing function_call items for stateful tool-output follow-ups + * using locally remembered response state, then deletes previous_response_id */ function stripStoredItemReferences(body: Record): void { const hasInput = Array.isArray(body.input) && body.input.length > 0; + const inputItems = Array.isArray(body.input) ? body.input : []; + const previousResponseId = typeof body.previous_response_id === "string" ? body.previous_response_id : ""; + const inputFunctionCallIds = new Set(); + const inputFunctionCallOutputIds = new Set(); - // Always strip previous_response_id IF we have input. - // The /codex/responses endpoint does not persist responses, so any reference - // to a previous response would cause a 404. However, if input is missing (e.g. Cursor - // trying to continue generation), stripping it leaves the payload empty causing a 400 Schema error. - // We leave it intact so Codex returns 404, which correctly triggers Cursor's fallback to resend history. + for (const item of inputItems) { + if (!item || typeof item !== "object" || Array.isArray(item)) continue; + const record = item as Record; + const type = typeof record.type === "string" ? record.type : ""; + const callId = typeof record.call_id === "string" ? record.call_id : ""; + if (!callId) continue; + if (type === "function_call") { + inputFunctionCallIds.add(callId); + continue; + } + if (type === "function_call_output") { + inputFunctionCallOutputIds.add(callId); + } + } + + const missingFunctionCallIds = [...inputFunctionCallOutputIds].filter( + (callId) => !inputFunctionCallIds.has(callId) + ); + + if (hasInput && previousResponseId && missingFunctionCallIds.length > 0) { + const rememberedFunctionCalls = getRememberedResponseFunctionCalls(previousResponseId); + const injectedFunctionCalls = rememberedFunctionCalls + .filter((functionCall) => missingFunctionCallIds.includes(functionCall.call_id)) + .filter((functionCall) => !inputFunctionCallIds.has(functionCall.call_id)) + .map((functionCall) => ({ + type: "function_call", + call_id: functionCall.call_id, + name: functionCall.name, + arguments: functionCall.arguments, + })); + + if (injectedFunctionCalls.length > 0) { + body.input = [...injectedFunctionCalls, ...inputItems]; + for (const functionCall of injectedFunctionCalls) { + inputFunctionCallIds.add(functionCall.call_id); + } + } + } + + // Strip previous_response_id whenever the request already carries input items. + // Codex rejects this field outright, so stateful follow-up turns must be made + // self-contained via the local function_call replay above. + // + // If input is missing entirely (e.g. Cursor trying to continue generation), keep + // previous_response_id so upstream can decide whether to fall back. if (hasInput) { delete body.previous_response_id; } @@ -1133,6 +1179,11 @@ export class CodexExecutor extends BaseExecutor { // whether the request came via native passthrough or translation. delete body.max_tokens; delete body.max_output_tokens; + // VS Code Copilot BYOK Responses requests include `truncation` (for example + // "auto" or "disabled"). The Codex /responses backend currently rejects this + // field entirely with 400 Unsupported parameter: truncation, so strip it for + // both native passthrough and translated requests. + delete body.truncation; delete body.background; // Droid CLI sends this but Codex Responses API rejects it // Inject prompt_cache_key for Codex prompt caching. diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index 6d9f8526dd..876ee51db1 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -3542,7 +3542,8 @@ export async function handleChatCore({ body, onStreamComplete, apiKeyInfo, - handleStreamFailure + handleStreamFailure, + clientResponseFormat ); } diff --git a/open-sse/services/responsesToolCallState.ts b/open-sse/services/responsesToolCallState.ts new file mode 100644 index 0000000000..06e588d77a --- /dev/null +++ b/open-sse/services/responsesToolCallState.ts @@ -0,0 +1,109 @@ +type JsonRecord = Record; + +type RememberedFunctionCall = { + call_id: string; + name: string; + arguments: string; +}; + +type RememberedResponseToolState = { + functionCalls: RememberedFunctionCall[]; + expiresAt: number; + updatedAt: number; +}; + +const RESPONSE_TOOL_CALL_TTL_MS = 30 * 60 * 1000; +const RESPONSE_TOOL_CALL_CACHE_MAX_ENTRIES = 512; + +const rememberedResponseToolCalls = new Map(); + +function toRecord(value: unknown): JsonRecord | null { + return value && typeof value === "object" && !Array.isArray(value) + ? (value as JsonRecord) + : null; +} + +function cleanupRememberedResponseToolCalls(now: number = Date.now()) { + for (const [responseId, entry] of rememberedResponseToolCalls.entries()) { + if (entry.expiresAt <= now) { + rememberedResponseToolCalls.delete(responseId); + } + } + + if (rememberedResponseToolCalls.size <= RESPONSE_TOOL_CALL_CACHE_MAX_ENTRIES) { + return; + } + + const oldestEntries = [...rememberedResponseToolCalls.entries()].sort( + (a, b) => a[1].updatedAt - b[1].updatedAt + ); + + while (rememberedResponseToolCalls.size > RESPONSE_TOOL_CALL_CACHE_MAX_ENTRIES) { + const oldest = oldestEntries.shift(); + if (!oldest) break; + rememberedResponseToolCalls.delete(oldest[0]); + } +} + +export function rememberResponseFunctionCalls(responseId: unknown, outputItems: readonly unknown[]) { + const normalizedResponseId = typeof responseId === "string" ? responseId.trim() : ""; + if (!normalizedResponseId || !Array.isArray(outputItems) || outputItems.length === 0) { + return; + } + + const functionCalls: RememberedFunctionCall[] = []; + + for (const item of outputItems) { + const record = toRecord(item); + if (!record || record.type !== "function_call") continue; + + const callId = typeof record.call_id === "string" ? record.call_id.trim() : ""; + const name = typeof record.name === "string" ? record.name.trim() : ""; + const argumentsValue = + typeof record.arguments === "string" + ? record.arguments + : JSON.stringify(record.arguments ?? {}); + + if (!callId || !name) continue; + + functionCalls.push({ + call_id: callId, + name, + arguments: argumentsValue, + }); + } + + if (functionCalls.length === 0) { + return; + } + + cleanupRememberedResponseToolCalls(); + + rememberedResponseToolCalls.set(normalizedResponseId, { + functionCalls, + updatedAt: Date.now(), + expiresAt: Date.now() + RESPONSE_TOOL_CALL_TTL_MS, + }); +} + +export function getRememberedResponseFunctionCalls( + responseId: unknown +): RememberedFunctionCall[] { + cleanupRememberedResponseToolCalls(); + + const normalizedResponseId = typeof responseId === "string" ? responseId.trim() : ""; + if (!normalizedResponseId) { + return []; + } + + const entry = rememberedResponseToolCalls.get(normalizedResponseId); + if (!entry) { + return []; + } + + return entry.functionCalls.map((functionCall) => ({ ...functionCall })); +} + +export function clearRememberedResponseFunctionCallsForTesting() { + rememberedResponseToolCalls.clear(); +} \ No newline at end of file diff --git a/open-sse/utils/stream.ts b/open-sse/utils/stream.ts index 7675ef04e3..0f8028069f 100644 --- a/open-sse/utils/stream.ts +++ b/open-sse/utils/stream.ts @@ -28,6 +28,7 @@ import { sanitizeStreamingChunk, extractThinkingFromContent, } from "../handlers/responseSanitizer.ts"; +import { rememberResponseFunctionCalls } from "../services/responsesToolCallState.ts"; import { buildErrorBody } from "./error.ts"; /** @@ -80,6 +81,7 @@ type StreamOptions = { mode?: string; targetFormat?: string; sourceFormat?: string; + clientResponseFormat?: string | null; provider?: string | null; reqLogger?: StreamLogger | null; toolNameMap?: unknown; @@ -476,6 +478,7 @@ export function createSSEStream(options: StreamOptions = {}) { mode = STREAM_MODE.TRANSLATE, targetFormat, sourceFormat, + clientResponseFormat = null, provider = null, reqLogger = null, toolNameMap = null, @@ -487,6 +490,11 @@ export function createSSEStream(options: StreamOptions = {}) { onFailure = null, } = options; + const clientExpectsResponsesStream = + (mode === STREAM_MODE.PASSTHROUGH + ? clientResponseFormat === FORMATS.OPENAI_RESPONSES + : sourceFormat === FORMATS.OPENAI_RESPONSES) === true; + let buffer = ""; let usage: UsageTokenRecord | null = null; /** Passthrough (OpenAI CC shape): saw tool_calls in stream before finish_reason */ @@ -516,6 +524,8 @@ export function createSSEStream(options: StreamOptions = {}) { // used to backfill `response.completed.response.output` when upstream returns it // empty (which happens when `store: false` — see backfillResponsesCompletedOutput). const passthroughResponsesOutputItems: unknown[] = []; + let passthroughResponsesId: string | null = null; + const passthroughResponsesReasoningSummarySeen = new Set(); const streamStartedAt = Date.now(); // Guard against duplicate [DONE] events — ensures exactly one per stream @@ -683,6 +693,101 @@ export function createSSEStream(options: StreamOptions = {}) { controller.enqueue(encoder.encode(comment)); }; + const getResponsesReasoningKey = (payload: Record): string | null => { + if (typeof payload.item_id === "string" && payload.item_id) { + return payload.item_id; + } + + const item = + payload.item && typeof payload.item === "object" && !Array.isArray(payload.item) + ? (payload.item as Record) + : null; + if (item && typeof item.id === "string" && item.id) { + return item.id; + } + + const responseId = + typeof payload.response_id === "string" && payload.response_id + ? payload.response_id + : passthroughResponsesId; + const outputIndex = + typeof payload.output_index === "number" && Number.isInteger(payload.output_index) + ? payload.output_index + : null; + + return responseId !== null && outputIndex !== null ? `${responseId}:${outputIndex}` : null; + }; + + const emitSyntheticResponsesReasoningSummary = ( + controller: TransformStreamDefaultController, + payload: Record + ) => { + const item = + payload.item && typeof payload.item === "object" && !Array.isArray(payload.item) + ? (payload.item as Record) + : null; + if (!item || item.type !== "reasoning" || !Array.isArray(item.summary)) { + return; + } + + const summaryText = item.summary + .map((part) => { + if (!part || typeof part !== "object" || Array.isArray(part)) { + return ""; + } + return typeof (part as Record).text === "string" + ? ((part as Record).text as string) + : ""; + }) + .join(""); + + if (!summaryText) { + return; + } + + const reasoningKey = getResponsesReasoningKey(payload); + if (!reasoningKey || passthroughResponsesReasoningSummarySeen.has(reasoningKey)) { + return; + } + passthroughResponsesReasoningSummarySeen.add(reasoningKey); + + const itemId = typeof item.id === "string" && item.id ? item.id : reasoningKey; + const outputIndex = + typeof payload.output_index === "number" && Number.isInteger(payload.output_index) + ? payload.output_index + : 0; + + const syntheticEvents = [ + { + event: "response.reasoning_summary_text.delta", + body: { + type: "response.reasoning_summary_text.delta", + item_id: itemId, + output_index: outputIndex, + summary_index: 0, + delta: summaryText, + }, + }, + { + event: "response.reasoning_summary_part.done", + body: { + type: "response.reasoning_summary_part.done", + item_id: itemId, + output_index: outputIndex, + summary_index: 0, + part: { type: "summary_text", text: summaryText }, + }, + }, + ]; + + for (const syntheticEvent of syntheticEvents) { + clientPayloadCollector.push(syntheticEvent.body); + const output = `event: ${syntheticEvent.event}\ndata: ${JSON.stringify(syntheticEvent.body)}\n\n`; + reqLogger?.appendConvertedChunk?.(output); + controller.enqueue(encoder.encode(output)); + } + }; + return new TransformStream( { start(controller) { @@ -809,6 +914,15 @@ export function createSSEStream(options: StreamOptions = {}) { parsed.type === "error"); if (isResponsesSSE) { + const responseId = + typeof parsed.response?.id === "string" + ? parsed.response.id + : typeof parsed.response_id === "string" + ? parsed.response_id + : null; + if (responseId) { + passthroughResponsesId = responseId; + } // Responses SSE: only extract usage, forward payload as-is const extracted = extractUsage(parsed); if (extracted) { @@ -825,10 +939,21 @@ export function createSSEStream(options: StreamOptions = {}) { if (parsed.type === "response.failed") { failurePayload = normalizeStreamFailurePayload(parsed); } + if ( + parsed.type === "response.reasoning_summary_text.delta" || + parsed.type === "response.reasoning_summary_text.done" || + parsed.type === "response.reasoning_summary_part.done" + ) { + const reasoningKey = getResponsesReasoningKey(parsed); + if (reasoningKey) { + passthroughResponsesReasoningSummarySeen.add(reasoningKey); + } + } // Capture each completed output item so the final // response.completed snapshot can be backfilled when upstream // returns an empty `output` (happens with store: false). if (parsed.type === "response.output_item.done" && parsed.item) { + emitSyntheticResponsesReasoningSummary(controller, parsed); passthroughResponsesOutputItems.push(parsed.item); } // Two transport-level fixes for Responses passthrough: @@ -1288,6 +1413,13 @@ export function createSSEStream(options: StreamOptions = {}) { } clearPendingPassthroughEvent(); + if (passthroughResponsesId && passthroughResponsesOutputItems.length > 0) { + rememberResponseFunctionCalls( + passthroughResponsesId, + passthroughResponsesOutputItems + ); + } + // Estimate usage if provider didn't return valid usage if (!hasValidUsage(usage) && totalContentLength > 0) { usage = estimateUsage(body, totalContentLength, sourceFormat || FORMATS.OPENAI); @@ -1307,10 +1439,12 @@ export function createSSEStream(options: StreamOptions = {}) { if (!doneSent) { await emitFinalSseMetadata(controller, usage); doneSent = true; - clientPayloadCollector.push({ done: true }); - const doneOutput = "data: [DONE]\n\n"; - reqLogger?.appendConvertedChunk?.(doneOutput); - controller.enqueue(encoder.encode(doneOutput)); + if (!clientExpectsResponsesStream) { + clientPayloadCollector.push({ done: true }); + const doneOutput = "data: [DONE]\n\n"; + reqLogger?.appendConvertedChunk?.(doneOutput); + controller.enqueue(encoder.encode(doneOutput)); + } } // Notify caller for call log persistence (include full response body with accumulated content) if (onComplete) { @@ -1499,10 +1633,12 @@ export function createSSEStream(options: StreamOptions = {}) { if (!doneSent) { await emitFinalSseMetadata(controller, state?.usage as Record | null); doneSent = true; - clientPayloadCollector.push({ done: true }); - const doneOutput = "data: [DONE]\n\n"; - reqLogger?.appendConvertedChunk?.(doneOutput); - controller.enqueue(encoder.encode(doneOutput)); + if (!clientExpectsResponsesStream) { + clientPayloadCollector.push({ done: true }); + const doneOutput = "data: [DONE]\n\n"; + reqLogger?.appendConvertedChunk?.(doneOutput); + controller.enqueue(encoder.encode(doneOutput)); + } } // Estimate usage if provider didn't return valid usage (for translate mode) @@ -1632,7 +1768,8 @@ export function createPassthroughStreamWithLogger( body: unknown = null, onComplete: ((payload: StreamCompletePayload) => void) | null = null, apiKeyInfo: unknown = null, - onFailure: ((payload: StreamFailurePayload) => void | Promise) | null = null + onFailure: ((payload: StreamFailurePayload) => void | Promise) | null = null, + clientResponseFormat: string | null = null ) { return createSSEStream({ mode: STREAM_MODE.PASSTHROUGH, @@ -1645,5 +1782,6 @@ export function createPassthroughStreamWithLogger( body, onComplete, onFailure, + clientResponseFormat, }); } diff --git a/tests/unit/executor-codex.test.ts b/tests/unit/executor-codex.test.ts index 5475c78e72..f39bb0a4e3 100644 --- a/tests/unit/executor-codex.test.ts +++ b/tests/unit/executor-codex.test.ts @@ -12,6 +12,10 @@ import { isCodexResponsesWebSocketRequired, parseCodexQuotaHeaders, } from "../../open-sse/executors/codex.ts"; +import { + clearRememberedResponseFunctionCallsForTesting, + rememberResponseFunctionCalls, +} from "../../open-sse/services/responsesToolCallState.ts"; import { DEFAULT_THINKING_CONFIG, setThinkingBudgetConfig, @@ -22,6 +26,7 @@ import { CODEX_CHAT_DEFAULT_INSTRUCTIONS } from "../../open-sse/config/codexInst test.afterEach(() => { setThinkingBudgetConfig(DEFAULT_THINKING_CONFIG); __setCodexWebSocketTransportForTesting(undefined); + clearRememberedResponseFunctionCallsForTesting(); }); async function withEnv(entries: Record, fn: () => any) { @@ -296,6 +301,48 @@ test("CodexExecutor.transformRequest preserves store-enabled responses state whe assert.equal(result.previous_response_id, "resp_prev_123"); }); +test("CodexExecutor.transformRequest rehydrates missing function_call items for stateful tool outputs", () => { + const executor = new CodexExecutor(); + rememberResponseFunctionCalls("resp_prev_tool_123", [ + { + type: "function_call", + call_id: "call_tool_123", + name: "workspace_read_file", + arguments: "{\"path\":\"README.md\"}", + }, + ]); + const body = { + _nativeCodexPassthrough: true, + previous_response_id: "resp_prev_tool_123", + input: [ + { + type: "function_call_output", + call_id: "call_tool_123", + output: "{\"ok\":true}", + }, + ], + stream: false, + }; + + const result = executor.transformRequest("gpt-5.5-low", body, false, { + requestEndpointPath: "/responses", + }); + + assert.equal(result.previous_response_id, undefined); + assert.equal(result.store, false); + assert.deepEqual(result.input[0], { + type: "function_call", + call_id: "call_tool_123", + name: "workspace_read_file", + arguments: "{\"path\":\"README.md\"}", + }); + assert.deepEqual(result.input[1], { + type: "function_call_output", + call_id: "call_tool_123", + output: "{\"ok\":true}", + }); +}); + test("CodexExecutor.transformRequest applies per-connection reasoning and service tier defaults", () => { const executor = new CodexExecutor(); const result = executor.transformRequest( diff --git a/tests/unit/stream-utilities.test.ts b/tests/unit/stream-utilities.test.ts index f273e5a464..4ba35d97f0 100644 --- a/tests/unit/stream-utilities.test.ts +++ b/tests/unit/stream-utilities.test.ts @@ -6,6 +6,7 @@ import { createStreamController, createDisconnectAwareStream, } from "../../open-sse/utils/streamHandler.ts"; +import { createPassthroughStreamWithLogger } from "../../open-sse/utils/stream.ts"; import { wantsProgress, createProgressTransform } from "../../open-sse/utils/progressTracker.ts"; @@ -50,6 +51,88 @@ test("createProgressTransform maps SSE text output to valid byte stream with pro assert.match(result, /done":true/); }); +test("createPassthroughStreamWithLogger omits [DONE] for Responses clients", async () => { + const transform = createPassthroughStreamWithLogger( + "codex", + null, + null, + "gpt-5.5-low", + null, + null, + null, + null, + null, + "openai-responses" + ); + + const writer = transform.writable.getWriter(); + await writer.write( + new TextEncoder().encode( + [ + "event: response.completed", + 'data: {"type":"response.completed","response":{"id":"resp_1","model":"gpt-5.5-low","status":"completed","output":[],"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}}', + "", + ].join("\n") + ) + ); + await writer.close(); + + const reader = transform.readable.getReader(); + const decoder = new TextDecoder(); + let result = ""; + + while (true) { + const { done, value } = await reader.read(); + if (done) break; + result += decoder.decode(value); + } + + assert.match(result, /event: response\.completed/); + assert.doesNotMatch(result, /data: \[DONE\]/); +}); + +test("createPassthroughStreamWithLogger synthesizes reasoning summary events from reasoning output items", async () => { + const transform = createPassthroughStreamWithLogger( + "codex", + null, + null, + "gpt-5.5-low", + null, + null, + null, + null, + null, + "openai-responses" + ); + + const writer = transform.writable.getWriter(); + await writer.write( + new TextEncoder().encode( + [ + "event: response.output_item.done", + 'data: {"type":"response.output_item.done","response_id":"resp_reasoning_1","output_index":0,"item":{"id":"rs_resp_reasoning_1_0","type":"reasoning","summary":[{"type":"summary_text","text":"Reasoning summary text"}]}}', + "", + ].join("\n") + ) + ); + await writer.close(); + + const reader = transform.readable.getReader(); + const decoder = new TextDecoder(); + let result = ""; + + while (true) { + const { done, value } = await reader.read(); + if (done) break; + result += decoder.decode(value); + } + + assert.match(result, /event: response\.reasoning_summary_text\.delta/); + assert.match(result, /"delta":"Reasoning summary text"/); + assert.match(result, /event: response\.reasoning_summary_part\.done/); + assert.match(result, /event: response\.output_item\.done/); +}); + test("createStreamController returns valid controller", () => { let completeLogged = false; let disconnectLogged = false; @@ -61,8 +144,8 @@ test("createStreamController returns valid controller", () => { }; const sc = createStreamController({ - connectionId: "conn_1", - onStreamComplete: () => {}, + provider: "test", + model: "conn_1", }); assert.equal(typeof sc.signal, "object");