diff --git a/packages/ai/src/transports/openai-responses-client.continuation.test.ts b/packages/ai/src/transports/openai-responses-client.continuation.test.ts new file mode 100644 index 000000000000..cb41908c1c6a --- /dev/null +++ b/packages/ai/src/transports/openai-responses-client.continuation.test.ts @@ -0,0 +1,355 @@ +import type { AssistantMessage, Context, Model } from "@openclaw/llm-core"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +type SdkResponse = { data: AsyncIterable; response: Response }; + +const sseState = vi.hoisted(() => ({ + clientHeaders: [] as Array>, + outcomes: [] as Array, + requests: [] as Array>, +})); + +vi.mock("openai", () => { + class MockOpenAI { + apiKey: string; + baseURL: string; + responses = { + create: (request: Record) => { + sseState.requests.push(request); + const outcome = sseState.outcomes.shift() ?? new Error("Unexpected SSE request"); + return { + withResponse: async () => { + if (outcome instanceof Error) { + throw outcome; + } + return outcome; + }, + }; + }, + }; + + constructor(options: { + apiKey?: string; + baseURL?: string; + defaultHeaders?: Record; + }) { + this.apiKey = options.apiKey ?? ""; + this.baseURL = options.baseURL ?? "https://api.openai.com/v1"; + sseState.clientHeaders.push(options.defaultHeaders ?? {}); + } + } + + return { default: MockOpenAI, AzureOpenAI: MockOpenAI }; +}); + +vi.mock("openai/resources/responses/ws.js", () => ({ + ResponsesWS: function UnexpectedResponsesWS() { + throw new Error("SSE continuation tests must not construct a WebSocket"); + }, +})); + +import { configureAiTransportHost, getAiTransportHost } from "../host.js"; +import { cleanupSessionResources } from "../session-resources.js"; +import { createOpenAIResponsesTransportStreamFn } from "./openai-responses-client.js"; + +const initialHost = getAiTransportHost(); +const model = { + id: "gpt-5.6-luna", + name: "GPT-5.6 Luna", + api: "openai-responses", + provider: "openai", + baseUrl: "https://api.openai.com/v1", + reasoning: true, + input: ["text"], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 200_000, + maxTokens: 8192, +} satisfies Model<"openai-responses">; + +function userMessage(text: string, timestamp: number) { + return { role: "user" as const, content: text, timestamp }; +} + +function completedEvent(responseId: string, content: string) { + const output = [ + { + id: `msg_${responseId}`, + type: "message", + status: "completed", + content: [ + { + annotations: [ + { + type: "url_citation", + url: "https://example.test/source", + title: "source", + start_index: 0, + end_index: content.length, + }, + ], + logprobs: [{ token: content, logprob: -0.1, bytes: [], top_logprobs: [] }], + text: content, + type: "output_text", + }, + ], + role: "assistant", + phase: "final_answer", + }, + ]; + return { + type: "response.completed", + response: { + id: responseId, + status: "completed", + output, + usage: { input_tokens: 5, output_tokens: 3, total_tokens: 8 }, + }, + }; +} + +function sdkCompletion(responseId: string, content: string): SdkResponse { + return sdkEvents(completedEvent(responseId, content)); +} + +function sdkEvents(...events: Array>): SdkResponse { + return { + data: (async function* () { + yield* events; + })(), + response: new Response(null, { status: 200 }), + }; +} + +async function run( + context: Context, + options: { + sessionId?: string; + onPayload: (payload: Record) => Record; + signal?: AbortSignal; + }, +): Promise { + const stream = await createOpenAIResponsesTransportStreamFn()(model, context, { + apiKey: "test-key", + sessionId: options.sessionId ?? "session-1", + transport: "sse", + reasoningEffort: "low", + onPayload: options.onPayload, + signal: options.signal, + } as never); + return stream.result(); +} + +describe("native OpenAI Responses SSE continuation", () => { + beforeEach(() => { + cleanupSessionResources(); + sseState.clientHeaders.length = 0; + sseState.outcomes.length = 0; + sseState.requests.length = 0; + let turn = 0; + configureAiTransportHost({ + ...initialHost, + plugin: { + ...initialHost.plugin, + resolveTransportTurnState: ({ context }) => { + turn += 1; + return { + headers: { + "x-openclaw-session-id": context.sessionId ?? "", + "x-openclaw-turn-id": `turn-${turn}`, + "x-openclaw-turn-attempt": "1", + }, + metadata: { + openclaw_session_id: context.sessionId ?? "", + openclaw_turn_id: `turn-${turn}`, + openclaw_turn_attempt: "1", + openclaw_transport: context.transport, + }, + }; + }, + }, + }); + }); + + afterEach(() => { + cleanupSessionResources(); + configureAiTransportHost(initialHost); + }); + + it("continues stateful literal SSE turns with only appended input", async () => { + sseState.outcomes.push( + sdkCompletion("resp_1", "first answer"), + sdkCompletion("resp_2", "second answer"), + ); + const firstUser = userMessage("first question", 1); + const onPayload = (payload: Record) => ({ ...payload, store: true }); + const first = await run({ messages: [firstUser], tools: [] }, { onPayload }); + const second = await run( + { messages: [firstUser, first, userMessage("second question", 2)], tools: [] }, + { onPayload }, + ); + + expect(second.stopReason).toBe("stop"); + expect(sseState.clientHeaders).toMatchObject([ + { "x-openclaw-turn-id": "turn-1" }, + { "x-openclaw-turn-id": "turn-2" }, + ]); + expect(sseState.requests[1]).toMatchObject({ + previous_response_id: "resp_1", + input: [ + { + type: "message", + role: "user", + content: [{ type: "input_text", text: "second question" }], + }, + ], + }); + }); + + it("keeps final store:false turns stateless and sends full history", async () => { + sseState.outcomes.push( + sdkCompletion("resp_1", "first answer"), + sdkCompletion("resp_2", "second answer"), + ); + const firstUser = userMessage("first question", 1); + const onPayload = (payload: Record) => ({ ...payload, store: false }); + const first = await run({ messages: [firstUser], tools: [] }, { onPayload }); + await run( + { messages: [firstUser, first, userMessage("second question", 2)], tools: [] }, + { onPayload }, + ); + + expect(sseState.requests[1]).not.toHaveProperty("previous_response_id"); + expect(sseState.requests[1]?.input).toHaveLength(3); + }); + + it("recovers a rejected continuation with full history and advances the baseline", async () => { + sseState.outcomes.push( + sdkCompletion("resp_1", "first answer"), + Object.assign(new Error("previous response not found"), { + code: "previous_response_not_found", + status: 400, + }), + sdkCompletion("resp_2", "second answer"), + sdkCompletion("resp_3", "third answer"), + ); + const onPayload = (payload: Record) => ({ ...payload, store: true }); + const firstUser = userMessage("first question", 1); + const first = await run({ messages: [firstUser], tools: [] }, { onPayload }); + const secondContext = { + messages: [firstUser, first, userMessage("second question", 2)], + tools: [], + }; + const second = await run(secondContext, { onPayload }); + await run( + { + messages: [...secondContext.messages, second, userMessage("third question", 3)], + tools: [], + }, + { onPayload }, + ); + + expect(sseState.requests[1]).toMatchObject({ previous_response_id: "resp_1" }); + expect(sseState.requests[1]?.input).toHaveLength(1); + expect(sseState.requests[2]).not.toHaveProperty("previous_response_id"); + expect(sseState.requests[2]?.input).toHaveLength(3); + expect(sseState.requests[3]).toMatchObject({ previous_response_id: "resp_2" }); + expect(sseState.requests[3]?.input).toHaveLength(1); + }); + + it("records the effective full-history compaction recovery request", async () => { + sseState.outcomes.push( + sdkCompletion("resp_1", "first answer"), + Object.assign(new Error("invalid encrypted content"), { + code: "invalid_encrypted_content", + }), + sdkCompletion("resp_2", "second answer"), + sdkCompletion("resp_3", "third answer"), + ); + const stateful = (payload: Record) => ({ ...payload, store: true }); + const withCompaction = (payload: Record) => ({ + ...payload, + store: true, + input: [ + ...((payload.input as unknown[]) ?? []), + { type: "compaction", encrypted_content: "opaque" }, + ], + }); + const firstUser = userMessage("first question", 1); + const first = await run({ messages: [firstUser], tools: [] }, { onPayload: stateful }); + const secondContext = { + messages: [firstUser, first, userMessage("second question", 2)], + tools: [], + }; + const second = await run(secondContext, { onPayload: withCompaction }); + await run( + { + messages: [...secondContext.messages, second, userMessage("third question", 3)], + tools: [], + }, + { onPayload: stateful }, + ); + + expect(sseState.requests[1]).toMatchObject({ previous_response_id: "resp_1" }); + expect(JSON.stringify(sseState.requests[1]?.input)).toContain('"compaction"'); + expect(sseState.requests[2]).not.toHaveProperty("previous_response_id"); + expect(JSON.stringify(sseState.requests[2]?.input)).not.toContain('"compaction"'); + expect(sseState.requests[3]).toMatchObject({ previous_response_id: "resp_2" }); + }); + + it.each([ + "request failure", + "continuation error without previous_response_id", + "incomplete response", + "post-dispatch stream rejection", + "abort", + ])("does not commit after %s", async (failure) => { + const controller = new AbortController(); + if (failure === "request failure") { + sseState.outcomes.push(new Error("request failed")); + } else if (failure === "continuation error without previous_response_id") { + sseState.outcomes.push( + Object.assign(new Error("previous response not found"), { + code: "previous_response_not_found", + status: 400, + }), + ); + } else if (failure === "incomplete response") { + sseState.outcomes.push( + sdkEvents({ + type: "response.incomplete", + response: { id: "resp_incomplete", status: "incomplete", output: [] }, + }), + ); + } else if (failure === "post-dispatch stream rejection") { + sseState.outcomes.push( + sdkEvents({ + type: "error", + code: "previous_response_not_found", + message: "previous response not found after stream acceptance", + }), + ); + } else { + sseState.outcomes.push({ + data: (async function* () { + controller.abort(); + yield completedEvent("resp_aborted", "ignored"); + })(), + response: new Response(null, { status: 200 }), + }); + } + sseState.outcomes.push(sdkCompletion("resp_next", "next answer")); + const onPayload = (payload: Record) => ({ ...payload, store: true }); + const sessionId = `session-${failure}`; + await run( + { messages: [userMessage("first", 1)], tools: [] }, + { + onPayload, + sessionId, + signal: failure === "abort" ? controller.signal : undefined, + }, + ); + await run({ messages: [userMessage("next", 2)], tools: [] }, { onPayload, sessionId }); + + expect(sseState.requests[1]).not.toHaveProperty("previous_response_id"); + }); +}); diff --git a/packages/ai/src/transports/openai-responses-client.ts b/packages/ai/src/transports/openai-responses-client.ts index 8f1ed359f5ef..7f753d1486e8 100644 --- a/packages/ai/src/transports/openai-responses-client.ts +++ b/packages/ai/src/transports/openai-responses-client.ts @@ -20,6 +20,10 @@ import { suppressOpenAIResponsesCompaction, type OpenAIResponsesReplayMode, } from "./openai-responses-compaction-replay.js"; +import { + claimOpenAIResponsesHttpContinuation, + type ResponsesContinuationRequest, +} from "./openai-responses-continuation.js"; import { AZURE_RESPONSES_FIRST_EVENT_TIMEOUT_MS, OpenAIResponsesWebSocketPreDispatchError, @@ -47,7 +51,7 @@ import { observeResponsesStream } from "./openai-responses-stream-observer-inter import { createOpenAIResponsesWebSocketStream, type OpenAIResponsesWebSocketMode, - supportsNativeOpenAIResponsesWebSocket, + supportsNativeOpenAIResponsesEndpoint, } from "./openai-responses-websocket.js"; import { assertCodeModeResponsesToolSurface, @@ -77,7 +81,7 @@ function resolveNativeOpenAIResponsesWebSocketMode( if (getAiTransportHost().requiresManagedTransport(model)) { return undefined; } - return supportsNativeOpenAIResponsesWebSocket({ + return supportsNativeOpenAIResponsesEndpoint({ provider: model.provider, api: model.api, baseUrl: model.baseUrl, @@ -163,6 +167,7 @@ type ResponsesTransportExecutorOptions = { outputApi?: AssistantMessage["api"]; firstEventTimeoutMs?: number; streamRequest?: boolean; + httpContinuation?: boolean; createClient: typeof createOpenAIResponsesClient; buildRequest: ( model: Model, @@ -173,7 +178,7 @@ type ResponsesTransportExecutorOptions = { ) => ReturnType; createResponseStream: ( params: ResponsesStreamParams, - ) => Promise<{ stream: AsyncIterable; response: Response }>; + ) => ReturnType; pricingOptions?: (options: OpenAIResponsesOptions | undefined) => ResponsesPricingOptions; }; @@ -201,6 +206,7 @@ function createResponsesTransportExecutor(config: ResponsesTransportExecutorOpti timestamp: Date.now(), }; let firstEventAbort: ReturnType | undefined; + let continuationClaim: ReturnType; try { const apiKey = options?.apiKey || getEnvApiKey(model.provider) || ""; const websocketMode = resolveNativeOpenAIResponsesWebSocketMode( @@ -265,6 +271,36 @@ function createResponsesTransportExecutor(config: ResponsesTransportExecutorOpti return params; }; const params = await buildRequest("checkpoint"); + const sessionId = options?.sessionId; + const httpContinuationEligible = + config.httpContinuation && + !websocketMode && + !getAiTransportHost().requiresManagedTransport(model) && + supportsNativeOpenAIResponsesEndpoint({ + provider: model.provider, + api: model.api, + baseUrl: model.baseUrl, + }); + if ( + httpContinuationEligible && + sessionId && + params.store === true && + !params.previous_response_id + ) { + continuationClaim = claimOpenAIResponsesHttpContinuation({ + sessionId, + apiKey, + baseUrl: model.baseUrl, + headers: buildOpenAIClientHeaders( + model, + context, + options?.headers, + turnState?.headers, + sessionId, + ), + request: params as ResponsesContinuationRequest, + }); + } const observePrompt = createResponsesPromptEgressObserver( responsesOptions, context.systemPrompt, @@ -296,10 +332,16 @@ function createResponsesTransportExecutor(config: ResponsesTransportExecutorOpti `baseUrl=${formatModelTransportDebugBaseUrl(model.baseUrl)} timeoutMs=${safeDebugValue(requestOptions?.timeout)} ` + `apiKey=${apiKey ? "present" : "missing"} ${summarizeResponsesPayload(params)}`, ); + let continuationBaseline: ResponsesContinuationRequest | undefined; const createSseStream = async (): Promise> => { - const { stream: responseStream, response } = await config.createResponseStream({ + const initialRequest = (continuationClaim?.request ?? params) as typeof params; + const { + stream: rawResponseStream, + response, + attempt, + } = await config.createResponseStream({ client, - request: params, + request: initialRequest, requestOptions, model, observePrompt, @@ -310,8 +352,13 @@ function createResponsesTransportExecutor(config: ResponsesTransportExecutorOpti authProfileId: responsesOptions?.authProfileId, }), }); + if (continuationClaim) { + continuationBaseline = attempt.request.previous_response_id + ? (params as ResponsesContinuationRequest) + : (attempt.request as ResponsesContinuationRequest); + } return withProviderResponseHook({ - stream: observeResponsesStream(responseStream, model, requestStartedAt), + stream: observeResponsesStream(rawResponseStream, model, requestStartedAt), signal: firstEvent.signal, abort: firstEvent.abort, hook: createOpenAIResponseHook(options?.onResponse, response, model), @@ -391,7 +438,7 @@ function createResponsesTransportExecutor(config: ResponsesTransportExecutorOpti responseStream = await createSseStream(); } try { - await processResponsesStream(responseStream, output, stream, model, { + const terminal = await processResponsesStream(responseStream, output, stream, model, { ...config.pricingOptions?.(responsesOptions), firstEventTimeoutMs: getFirstStreamEventTimeoutMs(options) ?? config.firstEventTimeoutMs, @@ -404,6 +451,15 @@ function createResponsesTransportExecutor(config: ResponsesTransportExecutorOpti }), }); finishWebSocket?.(); + if (options?.signal?.aborted) { + throw transportAbortError(options.signal); + } + if (output.stopReason === "aborted" || output.stopReason === "error") { + throw new Error("An unknown error occurred"); + } + if (continuationClaim && continuationBaseline && terminal) { + continuationClaim.commit(continuationBaseline, terminal); + } } catch (error) { finishWebSocket?.({ keep: false }); throw error; @@ -413,12 +469,6 @@ function createResponsesTransportExecutor(config: ResponsesTransportExecutorOpti `[responses] completed provider=${model.provider} api=${model.api} model=${model.id} ` + `transport=${transport} elapsedMs=${Date.now() - requestStartedAt}`, ); - if (options?.signal?.aborted) { - throw transportAbortError(options.signal); - } - if (output.stopReason === "aborted" || output.stopReason === "error") { - throw new Error("An unknown error occurred"); - } stream.push({ type: "done", reason: output.stopReason as never, message: output as never }); stream.end(); } catch (error) { @@ -433,6 +483,7 @@ function createResponsesTransportExecutor(config: ResponsesTransportExecutorOpti stream.push({ type: "error", reason: output.stopReason as never, error: output as never }); stream.end(); } finally { + continuationClaim?.release(); firstEventAbort?.dispose(); } })(); @@ -443,6 +494,7 @@ function createResponsesTransportExecutor(config: ResponsesTransportExecutorOpti export function createOpenAIResponsesTransportStreamFn(): StreamFn { return createResponsesTransportExecutor({ streamRequest: true, + httpContinuation: true, createClient: createOpenAIResponsesClient, buildRequest: buildOpenAIResponsesParams, createResponseStream: createResponsesStreamWithEncryptedContentRetry, diff --git a/packages/ai/src/transports/openai-responses-continuation.test.ts b/packages/ai/src/transports/openai-responses-continuation.test.ts new file mode 100644 index 000000000000..f3371009003d --- /dev/null +++ b/packages/ai/src/transports/openai-responses-continuation.test.ts @@ -0,0 +1,184 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { cleanupSessionResources } from "../session-resources.js"; +import { + claimOpenAIResponsesHttpContinuation, + resolveResponsesContinuationRequest, + type ResponsesContinuationRequest, + type ResponsesContinuationState, +} from "./openai-responses-continuation.js"; + +const firstUser = { + type: "message", + role: "user", + content: [{ type: "input_text", text: "first" }], +}; +const assistantOutput = { + id: "msg_1", + type: "message", + role: "assistant", + status: "completed", + phase: "final_answer", + content: [ + { + type: "output_text", + text: "answer", + annotations: [ + { + type: "url_citation", + url: "https://example.test/source", + title: "source", + start_index: 0, + end_index: 6, + }, + ], + logprobs: [{ token: "answer", logprob: -0.1, bytes: [], top_logprobs: [] }], + }, + ], +}; + +function continuationState(): ResponsesContinuationState { + return { + lastRequest: { + model: "gpt-5.6-luna", + store: true, + max_output_tokens: undefined, + metadata: { stable: "yes", openclaw_turn_id: "turn-1", openclaw_turn_attempt: "1" }, + input: [firstUser] as never, + }, + lastResponseId: "resp_1", + lastResponseItems: [assistantOutput] as never, + }; +} + +function nextRequest(phase = "final_answer"): ResponsesContinuationRequest { + return { + input: [ + firstUser, + { + type: "message", + role: "assistant", + phase, + content: [{ type: "output_text", text: "answer", annotations: [] }], + }, + { type: "message", role: "user", content: [{ type: "input_text", text: "second" }] }, + ] as never, + metadata: { openclaw_turn_attempt: "2", openclaw_turn_id: "turn-2", stable: "yes" }, + store: true, + model: "gpt-5.6-luna", + }; +} + +function claim(params: { + sessionId?: string; + authorization?: string; + turn?: string; + request?: ResponsesContinuationRequest; +}) { + return claimOpenAIResponsesHttpContinuation({ + sessionId: params.sessionId ?? "session-1", + apiKey: "api-key", + baseUrl: "https://api.openai.com/v1", + headers: { + Authorization: params.authorization ?? "Bearer tenant-a", + traceparent: `trace-${params.turn ?? "1"}`, + "x-openclaw-turn-id": `turn-${params.turn ?? "1"}`, + "x-openclaw-turn-attempt": params.turn ?? "1", + "x-stable-route": "route-a", + }, + request: params.request ?? continuationState().lastRequest, + }); +} + +afterEach(() => { + cleanupSessionResources(); + vi.useRealTimers(); +}); + +describe("OpenAI Responses continuation", () => { + it("matches JSON wire semantics and provider-only assistant replay metadata", () => { + const continued = resolveResponsesContinuationRequest(continuationState(), nextRequest()); + expect(continued).toMatchObject({ + continuationStatus: "continued", + request: { + previous_response_id: "resp_1", + input: [ + { + type: "message", + role: "user", + content: [{ type: "input_text", text: "second" }], + }, + ], + }, + }); + + expect( + resolveResponsesContinuationRequest(continuationState(), nextRequest("commentary")) + .continuationStatus, + ).toBe("history_changed"); + const explicit = { ...nextRequest(), previous_response_id: "resp_explicit" }; + expect(resolveResponsesContinuationRequest(continuationState(), explicit)).toEqual({ + request: explicit, + continuationStatus: "explicit_previous_response_id", + }); + }); + + it("ignores turn correlation headers but isolates explicit authorization", () => { + const first = claim({ turn: "1" }); + expect(first).toBeDefined(); + first?.commit(continuationState().lastRequest, { + id: "resp_1", + output: continuationState().lastResponseItems, + }); + + const sameTenant = claim({ turn: "2", request: nextRequest() }); + expect(sameTenant?.request.previous_response_id).toBe("resp_1"); + sameTenant?.commit(nextRequest(), { id: "resp_2", output: [] }); + + const rotated = claim({ + turn: "3", + authorization: "Bearer tenant-b", + request: nextRequest(), + }); + expect(rotated?.request.previous_response_id).toBeUndefined(); + rotated?.release(); + }); + + it("grants one claim and prevents a concurrent non-owner from overwriting it", () => { + const owner = claim({}); + expect(owner).toBeDefined(); + expect(claim({})).toBeUndefined(); + + owner?.commit(continuationState().lastRequest, { + id: "resp_owner", + output: continuationState().lastResponseItems, + }); + expect(claim({ request: nextRequest() })?.request.previous_response_id).toBe("resp_owner"); + }); + + it("prevents cleanup-time claims from resurrecting session state", () => { + const stale = claim({}); + cleanupSessionResources("session-1"); + stale?.commit(continuationState().lastRequest, { + id: "resp_stale", + output: continuationState().lastResponseItems, + }); + + const next = claim({ request: nextRequest() }); + expect(next?.request.previous_response_id).toBeUndefined(); + next?.release(); + }); + + it("expires completed continuation state after the bounded idle TTL", () => { + vi.useFakeTimers(); + const first = claim({}); + first?.commit(continuationState().lastRequest, { + id: "resp_expiring", + output: continuationState().lastResponseItems, + }); + vi.advanceTimersByTime(5 * 60 * 1000 + 1); + + const next = claim({ request: nextRequest() }); + expect(next?.request.previous_response_id).toBeUndefined(); + next?.release(); + }); +}); diff --git a/packages/ai/src/transports/openai-responses-continuation.ts b/packages/ai/src/transports/openai-responses-continuation.ts new file mode 100644 index 000000000000..c4144765eb71 --- /dev/null +++ b/packages/ai/src/transports/openai-responses-continuation.ts @@ -0,0 +1,214 @@ +import { stableStringify } from "@openclaw/normalization-core"; +import { isRecord } from "@openclaw/normalization-core/record-coerce"; +import type { ResponseInput, ResponseOutputItem } from "openai/resources/responses/responses.js"; +import { getAiTransportHost, resolveAiTransportHeaderSentinels } from "../host.js"; +import { registerSessionResourceCleanup } from "../session-resources.js"; +import { sha256Hex } from "./transport-utils.js"; + +const HTTP_CONTINUATION_IDLE_TTL_MS = 5 * 60 * 1000; +const TURN_HEADERS = new Set(["traceparent", "x-openclaw-turn-id", "x-openclaw-turn-attempt"]); + +export type ResponsesContinuationRequest = Record & { + input?: ResponseInput; + previous_response_id?: string; +}; +export type ResponsesContinuationState = { + lastRequest: ResponsesContinuationRequest; + lastResponseId: string; + lastResponseItems: ResponseOutputItem[]; +}; +export type ResponsesContinuationStatus = + | "continued" + | "explicit_previous_response_id" + | "history_changed" + | "history_shorter" + | "no_previous_response" + | "request_changed"; + +function jsonValuesEqual(left: object, right: object): boolean { + // Round-trip first so stable key ordering retains JSON's omitted/undefined wire semantics. + return ( + stableStringify(JSON.parse(JSON.stringify(left) as string)) === + stableStringify(JSON.parse(JSON.stringify(right) as string)) + ); +} + +function requestWithoutInput(request: ResponsesContinuationRequest): ResponsesContinuationRequest { + const { input: _input, previous_response_id: _previousResponseId, ...rest } = request; + if (!isRecord(rest.metadata)) { + return rest; + } + const metadata = Object.fromEntries( + Object.entries(rest.metadata).filter( + ([key]) => key !== "openclaw_turn_id" && key !== "openclaw_turn_attempt", + ), + ); + return { ...rest, metadata }; +} + +function normalizeAssistantReplayInput(input: readonly unknown[]): unknown[] { + return input.map((item) => { + if (!isRecord(item)) { + return item; + } + if (item.type === "reasoning") { + return { type: "reasoning" }; + } + if (item.type !== "function_call" && !(item.type === "message" && item.role === "assistant")) { + return item; + } + const { id: _id, status: _status, ...stableItem } = item; + if (item.type === "message" && Array.isArray(stableItem.content)) { + stableItem.content = stableItem.content.map((part) => { + if (!isRecord(part) || part.type !== "output_text") { + return part; + } + const { annotations: _annotations, logprobs: _logprobs, ...stablePart } = part; + return stablePart; + }); + } + return stableItem; + }); +} + +export function resolveResponsesContinuationRequest( + continuation: ResponsesContinuationState | undefined, + request: ResponsesContinuationRequest, +): { request: ResponsesContinuationRequest; continuationStatus: ResponsesContinuationStatus } { + if (!continuation) { + return { request, continuationStatus: "no_previous_response" }; + } + if (request.previous_response_id) { + return { request, continuationStatus: "explicit_previous_response_id" }; + } + if ( + !jsonValuesEqual(requestWithoutInput(request), requestWithoutInput(continuation.lastRequest)) + ) { + return { request, continuationStatus: "request_changed" }; + } + const currentInput = request.input ?? []; + const previousInput = continuation.lastRequest.input ?? []; + const baselineLength = previousInput.length + continuation.lastResponseItems.length; + if (currentInput.length < baselineLength) { + return { request, continuationStatus: "history_shorter" }; + } + if ( + !jsonValuesEqual( + normalizeAssistantReplayInput(currentInput.slice(0, previousInput.length)), + normalizeAssistantReplayInput(previousInput), + ) || + !jsonValuesEqual( + normalizeAssistantReplayInput(currentInput.slice(previousInput.length, baselineLength)), + normalizeAssistantReplayInput(continuation.lastResponseItems), + ) + ) { + return { request, continuationStatus: "history_changed" }; + } + return { + request: { + ...request, + previous_response_id: continuation.lastResponseId, + input: currentInput.slice(baselineLength), + }, + continuationStatus: "continued", + }; +} + +type HttpContinuationEntry = + | { + kind: "ready"; + sessionId: string; + generation: number; + state: ResponsesContinuationState; + idleTimer: ReturnType; + } + | { kind: "claimed"; sessionId: string; generation: number }; + +const httpContinuationEntries = new Map(); +let nextHttpContinuationGeneration = 1; + +type HttpContinuationIdentity = { + apiKey: string; + baseUrl: string; + headers: Record; +}; +type ContinuationResponse = { id: string; output: ResponseOutputItem[] }; + +function connectionIdentity(params: HttpContinuationIdentity): string { + const headers = Object.entries(resolveAiTransportHeaderSentinels(params.headers) ?? {}) + .map(([name, value]) => [name.toLowerCase(), value] as const) + .filter(([name]) => !TURN_HEADERS.has(name)) + .toSorted(([a], [b]) => a.localeCompare(b)); + return sha256Hex( + JSON.stringify([ + getAiTransportHost().resolveSecretSentinel(params.apiKey), + params.baseUrl, + headers, + ]), + ); +} + +export function claimOpenAIResponsesHttpContinuation( + params: HttpContinuationIdentity & { + sessionId: string; + request: ResponsesContinuationRequest; + }, +) { + const key = `${params.sessionId}\0${connectionIdentity(params)}`; + const previous = httpContinuationEntries.get(key); + if (previous?.kind === "claimed") { + return undefined; + } + if (previous?.kind === "ready") { + clearTimeout(previous.idleTimer); + } + const generation = nextHttpContinuationGeneration++; + const claimed = { kind: "claimed", sessionId: params.sessionId, generation } as const; + httpContinuationEntries.set(key, claimed); + const wireRequest = resolveResponsesContinuationRequest( + previous?.kind === "ready" ? previous.state : undefined, + params.request, + ).request; + return { + request: wireRequest, + commit: (effectiveRequest: ResponsesContinuationRequest, response: ContinuationResponse) => { + if (httpContinuationEntries.get(key) !== claimed) { + return; + } + const idleTimer = setTimeout(() => { + const current = httpContinuationEntries.get(key); + if (current?.kind === "ready" && current.generation === generation) { + httpContinuationEntries.delete(key); + } + }, HTTP_CONTINUATION_IDLE_TTL_MS); + idleTimer.unref?.(); + const ready = { + ...claimed, + kind: "ready", + state: { + lastRequest: effectiveRequest, + lastResponseId: response.id, + lastResponseItems: response.output, + }, + idleTimer, + } satisfies Extract; + httpContinuationEntries.set(key, ready); + }, + release: () => { + if (httpContinuationEntries.get(key) === claimed) { + httpContinuationEntries.delete(key); + } + }, + }; +} + +registerSessionResourceCleanup((sessionId) => { + for (const [key, entry] of httpContinuationEntries) { + if (!sessionId || entry.sessionId === sessionId) { + if (entry.kind === "ready") { + clearTimeout(entry.idleTimer); + } + httpContinuationEntries.delete(key); + } + } +}); diff --git a/packages/ai/src/transports/openai-responses-contracts.ts b/packages/ai/src/transports/openai-responses-contracts.ts index 54207adac0a7..b96a18a37c14 100644 --- a/packages/ai/src/transports/openai-responses-contracts.ts +++ b/packages/ai/src/transports/openai-responses-contracts.ts @@ -91,7 +91,11 @@ export type OpenAIResponsesOptions = BaseOpenAIStreamOptions & { const PROMPT_OBSERVER = Symbol("openaiResponsesPromptObserver"); export type ResponsesPromptObservation = { egress: "responses-sdk" | "responses-websocket" | "native-codex-websocket" | "native-codex-sse"; - payloadVariant: "initial" | "reasoning-stripped" | "compaction-stripped"; + payloadVariant: + | "initial" + | "reasoning-stripped" + | "compaction-stripped" + | "continuation-rejected"; promptSource: "instructions" | "input.developer" | "input.system" | "missing"; expectedChars: number; observedChars: number; @@ -131,6 +135,7 @@ export type OpenAIResponsesRequestParams = { prompt_cache_key?: string; prompt_cache_retention?: "24h"; metadata?: Record; + previous_response_id?: string; store?: boolean; max_output_tokens?: number; temperature?: number; diff --git a/packages/ai/src/transports/openai-responses-replay-internal.ts b/packages/ai/src/transports/openai-responses-replay-internal.ts index 01a352a37309..825ca3a52733 100644 --- a/packages/ai/src/transports/openai-responses-replay-internal.ts +++ b/packages/ai/src/transports/openai-responses-replay-internal.ts @@ -101,7 +101,8 @@ type ResponsesEncryptedContentRequest = { input?: ResponseInput }; type ResponsesEncryptedContentAttemptKind = | "initial" | "reasoning-stripped" - | "compaction-stripped"; + | "compaction-stripped" + | "continuation-rejected"; export type ResponsesEncryptedContentAttempt = { kind: ResponsesEncryptedContentAttemptKind; @@ -148,7 +149,7 @@ export async function resolveNextResponsesEncryptedContentAttempt< if (!isInvalidEncryptedContentError(error) || attempt.kind === "compaction-stripped") { return undefined; } - if (attempt.kind === "initial") { + if (attempt.kind === "initial" || attempt.kind === "continuation-rejected") { const reasoningStripped = stripResponsesRequestEncryptedReasoning(attempt.request); if (reasoningStripped !== attempt.request) { return { kind: "reasoning-stripped", request: reasoningStripped }; @@ -284,7 +285,11 @@ export async function createResponsesStreamWithEncryptedContentRetry(params: { buildFullHistoryRequest?: () => | OpenAIResponsesRequestParams | Promise; -}): Promise<{ stream: AsyncIterable; response: Response }> { +}): Promise<{ + stream: AsyncIterable; + response: Response; + attempt: ResponsesEncryptedContentAttempt; +}> { const sendAttempt = async ( attempt: ResponsesEncryptedContentAttempt, ) => { @@ -294,7 +299,7 @@ export async function createResponsesStreamWithEncryptedContentRetry(params: { if (attempt.kind === "compaction-stripped") { params.onCompactionRejected?.(); } - return { stream: data as unknown as AsyncIterable, response }; + return { stream: data as unknown as AsyncIterable, response, attempt }; }; let attempt: ResponsesEncryptedContentAttempt = { @@ -309,23 +314,38 @@ export async function createResponsesStreamWithEncryptedContentRetry(params: { try { return await sendAttempt(attempt); } catch (error) { - const nextAttempt = await resolveNextResponsesEncryptedContentAttempt(attempt, error, { + let nextAttempt = await resolveNextResponsesEncryptedContentAttempt(attempt, error, { buildFullHistoryRequest: params.buildFullHistoryRequest, }); + if ( + !nextAttempt && + attempt.request.previous_response_id && + error && + typeof error === "object" && + typeof (error as { status?: unknown }).status === "number" && + (error as { code?: unknown }).code === "previous_response_not_found" + ) { + const request = { + ...(params.buildFullHistoryRequest + ? await params.buildFullHistoryRequest() + : attempt.request), + }; + delete request.previous_response_id; + nextAttempt = { kind: "continuation-rejected", request }; + } if (!nextAttempt) { throw error; } - if (nextAttempt.kind === "reasoning-stripped") { - log.warn( - `[responses] retrying without encrypted reasoning content provider=${params.model.provider} ` + - `api=${params.model.api} model=${params.model.id}`, - ); - } else { - log.warn( - `[responses] retrying without encrypted compaction content provider=${params.model.provider} ` + - `api=${params.model.api} model=${params.model.id}`, - ); - } + const retryDescription = + nextAttempt.kind === "reasoning-stripped" + ? "without encrypted reasoning content" + : nextAttempt.kind === "compaction-stripped" + ? "without encrypted compaction content" + : "full history after rejected previous_response_id"; + log.warn( + `[responses] retrying ${retryDescription} provider=${params.model.provider} ` + + `api=${params.model.api} model=${params.model.id}`, + ); attempt = nextAttempt; } } diff --git a/packages/ai/src/transports/openai-responses-stream-internal.ts b/packages/ai/src/transports/openai-responses-stream-internal.ts index 105aa9f5d82d..64b42d3b0fa8 100644 --- a/packages/ai/src/transports/openai-responses-stream-internal.ts +++ b/packages/ai/src/transports/openai-responses-stream-internal.ts @@ -72,6 +72,7 @@ type OpenAIResponsesConsumedEvent = Extract< ResponseStreamEvent, { type: ResponsesConsumedEventType } >; +type CompletedResponse = Extract["response"]; type OpenAIResponsesIgnoredSdkEvent = Exclude; type ResponsesTextContentPart = | ResponseOutputMessage["content"][number] @@ -119,7 +120,7 @@ export async function processResponsesStream( stream: ResponsesEventSink, model: Model, options?: ResponsesStreamOptions, -): Promise { +) { type StreamingToolCallBlock = ToolCall & { partialJson: string }; type StreamingToolCallState = ResponsesToolCallState & { block: StreamingToolCallBlock; @@ -134,7 +135,7 @@ export async function processResponsesStream( const reasoningBlocksById = new Map(); const outputItemContentIndexes = createResponsesOutputContentIndex(); const startedTextBlocksByItemId = new Map(); - let terminalResponseEvent: "finalized" | undefined; + let terminalResponse: CompletedResponse | null | undefined; let lastTextBlock: TextBlockReference | null = null; const blocks = output.content; const compactionTracker = createCompactionTracker(output, model, options); @@ -268,9 +269,7 @@ export async function processResponsesStream( setLastTextBlock: (block) => { lastTextBlock = block; }, - markFinalized: () => { - terminalResponseEvent = "finalized"; - }, + markFinalized: () => undefined, }); const guardedStream = adaptResponsesStream( @@ -693,6 +692,7 @@ export async function processResponsesStream( if (event.type === "response.completed" || output.stopReason === "length") { recoverTerminalOutput(event.response.output ?? [], event.type === "response.completed"); } + terminalResponse = event.type === "response.completed" ? event.response : null; if ( output.stopReason === "stop" && output.content.some((block) => block.type === "toolCall") @@ -721,9 +721,10 @@ export async function processResponsesStream( if (streamingToolCalls.hasActive()) { throw new Error("Responses stream ended with unresolved tool calls"); } - if (!terminalResponseEvent) { + if (terminalResponse === undefined) { throw new Error("OpenAI Responses stream ended before a terminal response event"); } + return terminalResponse ?? undefined; } finally { for (const block of output.content) { delete (block as { partialJson?: string }).partialJson; diff --git a/packages/ai/src/transports/openai-responses-websocket-client.test.ts b/packages/ai/src/transports/openai-responses-websocket-client.test.ts index ca7f4161ea3a..3872dc9eb7a9 100644 --- a/packages/ai/src/transports/openai-responses-websocket-client.test.ts +++ b/packages/ai/src/transports/openai-responses-websocket-client.test.ts @@ -232,7 +232,6 @@ async function run( transport?: "sse" | "websocket" | "websocket-cached" | "auto"; sessionId?: string; timeoutMs?: number; - onPayload?: (payload: Record) => Record; headers?: Record; } = {}, ): Promise { @@ -242,7 +241,6 @@ async function run( transport: overrides.transport ?? "websocket-cached", reasoningEffort: "low", timeoutMs: overrides.timeoutMs, - onPayload: overrides.onPayload, headers: overrides.headers, } as never); return stream.result(); diff --git a/packages/ai/src/transports/openai-responses-websocket.test.ts b/packages/ai/src/transports/openai-responses-websocket.test.ts index d2b52bf0a50d..132768eb22b7 100644 --- a/packages/ai/src/transports/openai-responses-websocket.test.ts +++ b/packages/ai/src/transports/openai-responses-websocket.test.ts @@ -64,7 +64,7 @@ import { configureAiTransportHost, getAiTransportHost } from "../host.js"; import { cleanupSessionResources } from "../session-resources.js"; import { createOpenAIResponsesWebSocketStream, - supportsNativeOpenAIResponsesWebSocket, + supportsNativeOpenAIResponsesEndpoint, } from "./openai-responses-websocket.js"; const initialHost = getAiTransportHost(); @@ -139,7 +139,7 @@ describe("native OpenAI Responses WebSocket transport", () => { it("only enables WebSockets for the official native OpenAI Responses endpoint", () => { expect( - supportsNativeOpenAIResponsesWebSocket({ + supportsNativeOpenAIResponsesEndpoint({ provider: "openai", api: "openai-responses", baseUrl: "https://api.openai.com/v1", @@ -156,7 +156,7 @@ describe("native OpenAI Responses WebSocket transport", () => { ["different provider", "azure-openai", "https://api.openai.com/v1"], ])("rejects %s", (_name, provider, baseUrl) => { expect( - supportsNativeOpenAIResponsesWebSocket({ provider, api: "openai-responses", baseUrl }), + supportsNativeOpenAIResponsesEndpoint({ provider, api: "openai-responses", baseUrl }), ).toBe(false); }); diff --git a/packages/ai/src/transports/openai-responses-websocket.ts b/packages/ai/src/transports/openai-responses-websocket.ts index b2b9a1db6e8f..8fe1951eb5f5 100644 --- a/packages/ai/src/transports/openai-responses-websocket.ts +++ b/packages/ai/src/transports/openai-responses-websocket.ts @@ -1,15 +1,17 @@ -import { stableStringify } from "@openclaw/normalization-core"; -import { isRecord } from "@openclaw/normalization-core/record-coerce"; import type OpenAI from "openai"; import type { - ResponseInput, - ResponseOutputItem, ResponsesClientEvent, ResponsesServerEvent, } from "openai/resources/responses/responses.js"; import { ResponsesWS } from "openai/resources/responses/ws.js"; import { getAiTransportHost, resolveAiTransportHeaderSentinels } from "../host.js"; import { registerSessionResourceCleanup } from "../session-resources.js"; +import { + resolveResponsesContinuationRequest, + type ResponsesContinuationRequest, + type ResponsesContinuationState, + type ResponsesContinuationStatus, +} from "./openai-responses-continuation.js"; import { OpenAIResponsesWebSocketPostDispatchError, OpenAIResponsesWebSocketPreDispatchError, @@ -22,24 +24,13 @@ const SESSION_WEBSOCKET_CACHE_TTL_MS = 5 * 60 * 1000; const SESSION_WEBSOCKET_MAX_AGE_MS = 55 * 60 * 1000; const WEBSOCKET_OPEN_STATE = 1; -type ResponsesWebSocketRequest = Record & { - input?: ResponseInput; - previous_response_id?: string; -}; - -type CachedWebSocketContinuation = { - lastRequest: ResponsesWebSocketRequest; - lastResponseId: string; - lastResponseItems: ResponseOutputItem[]; -}; - type CachedWebSocketConnection = { socket: ResponsesWS; sessionId: string; busy: boolean; createdAt: number; idleTimer?: ReturnType; - continuation?: CachedWebSocketContinuation; + continuation?: ResponsesContinuationState; }; type ResponsesWebSocketStreamMessage = @@ -49,16 +40,9 @@ export type OpenAIResponsesWebSocketMode = "websocket" | "websocket-cached" | "a type OpenAIResponsesWebSocketStream = { stream: AsyncIterable; - request: ResponsesWebSocketRequest; + request: ResponsesContinuationRequest; reusedConnection: boolean; - continuationStatus: - | "continued" - | "explicit_previous_response_id" - | "history_changed" - | "history_shorter" - | "no_previous_response" - | "request_changed" - | "socket_not_cached"; + continuationStatus: ResponsesContinuationStatus | "socket_not_cached"; finish: (options?: { keep?: boolean }) => void; }; @@ -74,22 +58,19 @@ function isOfficialOpenAIResponsesBaseUrl(baseUrl: string | undefined): boolean } try { const url = new URL(baseUrl); - const path = url.pathname.replace(/\/+$/, ""); return ( - url.protocol === "https:" && - url.hostname === "api.openai.com" && - url.port === "" && + url.origin === "https://api.openai.com" && url.username === "" && url.password === "" && url.search === "" && url.hash === "" && - path === "/v1" + url.pathname.replace(/\/+$/, "") === "/v1" ); } catch { return false; } } -export function supportsNativeOpenAIResponsesWebSocket(params: { +export function supportsNativeOpenAIResponsesEndpoint(params: { provider: string; api: string; baseUrl?: string; @@ -283,113 +264,9 @@ function acquireWebSocket( return createCachedWebSocketLease(cacheKey, entry, false); } -function requestWithoutInput(request: ResponsesWebSocketRequest): ResponsesWebSocketRequest { - const { input: _input, previous_response_id: _previousResponseId, ...rest } = request; - if (!rest.metadata || typeof rest.metadata !== "object" || Array.isArray(rest.metadata)) { - return rest; - } - const metadata = Object.fromEntries( - Object.entries(rest.metadata as Record).filter( - ([key]) => key !== "openclaw_turn_id" && key !== "openclaw_turn_attempt", - ), - ); - return { ...rest, metadata }; -} - -function sanitizeWebSocketRequest(request: Record): ResponsesWebSocketRequest { +function sanitizeWebSocketRequest(request: Record): ResponsesContinuationRequest { const { stream: _stream, background: _background, ...websocketRequest } = request; - return websocketRequest as ResponsesWebSocketRequest; -} - -function jsonValuesEqual(left: object, right: object): boolean { - // Round-trip first so stable key ordering retains JSON's omitted/undefined wire semantics. - const leftJson = JSON.parse(JSON.stringify(left) as string); - const rightJson = JSON.parse(JSON.stringify(right) as string); - return stableStringify(leftJson) === stableStringify(rightJson); -} - -function normalizeAssistantReplayInput(input: readonly unknown[]): unknown[] { - return input.map((item) => { - if (!isRecord(item)) { - return item; - } - if (item.type === "reasoning") { - return { type: "reasoning" }; - } - if (item.type !== "function_call" && !(item.type === "message" && item.role === "assistant")) { - return item; - } - const { id: _id, status: _status, ...stableItem } = item; - if (item.type === "message" && Array.isArray(stableItem.content)) { - // Strip only provider delivery metadata that reconstructed assistant replay cannot contain. - stableItem.content = stableItem.content.map((part) => { - if (!isRecord(part) || part.type !== "output_text") { - return part; - } - const { annotations: _annotations, logprobs: _logprobs, ...stablePart } = part; - return stablePart; - }); - } - return stableItem; - }); -} - -function buildCachedWebSocketRequest( - entry: CachedWebSocketConnection, - request: ResponsesWebSocketRequest, -): Pick { - const continuation = entry.continuation; - if (!continuation) { - return { request, continuationStatus: "no_previous_response" }; - } - const rejectContinuation = ( - continuationStatus: Exclude< - OpenAIResponsesWebSocketStream["continuationStatus"], - "continued" | "no_previous_response" | "socket_not_cached" - >, - ) => { - entry.continuation = undefined; - return { request, continuationStatus }; - }; - if (request.previous_response_id) { - return rejectContinuation("explicit_previous_response_id"); - } - if ( - !jsonValuesEqual(requestWithoutInput(request), requestWithoutInput(continuation.lastRequest)) - ) { - return rejectContinuation("request_changed"); - } - - const currentInput = request.input ?? []; - const previousInput = continuation.lastRequest.input ?? []; - const baselineLength = previousInput.length + continuation.lastResponseItems.length; - if (currentInput.length < baselineLength) { - return rejectContinuation("history_shorter"); - } - if ( - !jsonValuesEqual( - normalizeAssistantReplayInput(currentInput.slice(0, previousInput.length)), - normalizeAssistantReplayInput(previousInput), - ) || - !jsonValuesEqual( - normalizeAssistantReplayInput(currentInput.slice(previousInput.length, baselineLength)), - normalizeAssistantReplayInput(continuation.lastResponseItems), - ) - ) { - return rejectContinuation("history_changed"); - } - - // Continuations are single-use. A terminal incomplete/error cannot leave an - // older response id eligible for a later, unrelated turn. - entry.continuation = undefined; - return { - request: { - ...request, - previous_response_id: continuation.lastResponseId, - input: currentInput.slice(baselineLength), - }, - continuationStatus: "continued", - }; + return websocketRequest as ResponsesContinuationRequest; } async function nextWebSocketMessage( @@ -471,14 +348,19 @@ export function createOpenAIResponsesWebSocketStream(params: { markDegraded(); throw new OpenAIResponsesWebSocketPreDispatchError(error); } - let prepared: ReturnType; + let prepared: Pick; try { - prepared = lease.entry - ? buildCachedWebSocketRequest(lease.entry, fullRequest) - : { - request: fullRequest, - continuationStatus: "socket_not_cached" as const, - }; + const continuation = lease.entry?.continuation; + if (continuation && lease.entry) { + // Consume before dispatch so incomplete/error terminals cannot reuse stale state. + lease.entry.continuation = undefined; + prepared = resolveResponsesContinuationRequest(continuation, fullRequest); + } else { + prepared = { + request: fullRequest, + continuationStatus: lease.entry ? "no_previous_response" : "socket_not_cached", + }; + } } catch (error) { void lease.iterator.return?.().catch(() => undefined); lease.release({ keep: false }); diff --git a/src/agents/openai-transport-stream.replay-and-tools.test.ts b/src/agents/openai-transport-stream.replay-and-tools.test.ts index f4c10c07f8ce..615d0ee96415 100644 --- a/src/agents/openai-transport-stream.replay-and-tools.test.ts +++ b/src/agents/openai-transport-stream.replay-and-tools.test.ts @@ -577,7 +577,11 @@ describe("openai transport stream", () => { model: makeResponsesModel({ id: "gpt-5.5", name: "GPT-5.5" }), onCompactionRejected, }), - ).resolves.toEqual({ stream: recoveredStream, response: recoveredResponse }); + ).resolves.toMatchObject({ + stream: recoveredStream, + response: recoveredResponse, + attempt: { kind: "reasoning-stripped" }, + }); expect(create).toHaveBeenCalledTimes(2); const retry = create.mock.calls[1]?.[0] as typeof request; @@ -628,7 +632,11 @@ describe("openai transport stream", () => { model: makeResponsesModel({ id: "gpt-5.5", name: "GPT-5.5" }), onCompactionRejected, }), - ).resolves.toEqual({ stream: recoveredStream, response: recoveredResponse }); + ).resolves.toMatchObject({ + stream: recoveredStream, + response: recoveredResponse, + attempt: { kind: "compaction-stripped" }, + }); expect(create).toHaveBeenCalledTimes(3); expect(JSON.stringify(create.mock.calls[1]?.[0])).not.toContain("reasoning-ciphertext"); @@ -809,7 +817,11 @@ describe("openai transport stream", () => { maxTokens: 8192, }, }), - ).resolves.toEqual({ stream: recoveredStream, response: recoveredResponse }); + ).resolves.toMatchObject({ + stream: recoveredStream, + response: recoveredResponse, + attempt: { kind: "reasoning-stripped" }, + }); expect(create).toHaveBeenCalledTimes(2); expect(create.mock.calls[0]?.[0]).toBe(request);