fix(codex): stabilize copilot responses reasoning and tool replay (#1750)

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<string, unknown>): 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<string, unknown>): 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<string>();
+  const inputFunctionCallOutputIds = new Set<string>();
 
-  // 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<string, unknown>;
+    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.

open-sse/handlers/chatCore.ts

@@ -3542,7 +3542,8 @@ export async function handleChatCore({
       body,
       onStreamComplete,
       apiKeyInfo,
-      handleStreamFailure
+      handleStreamFailure,
+      clientResponseFormat
     );
   }
 

open-sse/services/responsesToolCallState.ts

@@ -0,0 +1,109 @@
+type JsonRecord = Record<string, unknown>;
+
+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<string, RememberedResponseToolState>();
+
+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();
+}

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<string>();
   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, unknown>): 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<string, unknown>)
+        : 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<string, unknown>
+  ) => {
+    const item =
+      payload.item && typeof payload.item === "object" && !Array.isArray(payload.item)
+        ? (payload.item as Record<string, unknown>)
+        : 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<string, unknown>).text === "string"
+          ? ((part as Record<string, unknown>).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<string, unknown> | 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<void>) | null = null
+  onFailure: ((payload: StreamFailurePayload) => void | Promise<void>) | null = null,
+  clientResponseFormat: string | null = null
 ) {
   return createSSEStream({
     mode: STREAM_MODE.PASSTHROUGH,
@@ -1645,5 +1782,6 @@ export function createPassthroughStreamWithLogger(
     body,
     onComplete,
     onFailure,
+    clientResponseFormat,
   });
 }

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<string, string | undefined>, 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(

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");