feat(ai): continue stateful OpenAI SSE turns (#122194)

* feat(ai): continue stateful OpenAI SSE turns

Use the prior OpenAI Responses ID for native stateful HTTP/SSE turns, retry missing server state with full history in the same turn, and keep stateless/custom routes unchanged.

Co-authored-by: Markus Hartung <mail@hartmark.se>

* test(ai): assert Responses retry attempt metadata

---------

Co-authored-by: Peter Steinberger <steipete@gmail.com>

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<unknown>; response: Response };
+
+const sseState = vi.hoisted(() => ({
+  clientHeaders: [] as Array<Record<string, string>>,
+  outcomes: [] as Array<Error | SdkResponse>,
+  requests: [] as Array<Record<string, unknown>>,
+}));
+
+vi.mock("openai", () => {
+  class MockOpenAI {
+    apiKey: string;
+    baseURL: string;
+    responses = {
+      create: (request: Record<string, unknown>) => {
+        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<string, string>;
+    }) {
+      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<Record<string, unknown>>): SdkResponse {
+  return {
+    data: (async function* () {
+      yield* events;
+    })(),
+    response: new Response(null, { status: 200 }),
+  };
+}
+
+async function run(
+  context: Context,
+  options: {
+    sessionId?: string;
+    onPayload: (payload: Record<string, unknown>) => Record<string, unknown>;
+    signal?: AbortSignal;
+  },
+): Promise<AssistantMessage> {
+  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<string, unknown>) => ({ ...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<string, unknown>) => ({ ...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<string, unknown>) => ({ ...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<string, unknown>) => ({ ...payload, store: true });
+    const withCompaction = (payload: Record<string, unknown>) => ({
+      ...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<string, unknown>) => ({ ...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");
+  });
+});

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<typeof buildOpenAIResponsesParams>;
   createResponseStream: (
     params: ResponsesStreamParams,
-  ) => Promise<{ stream: AsyncIterable<unknown>; response: Response }>;
+  ) => ReturnType<typeof createResponsesStreamWithEncryptedContentRetry>;
   pricingOptions?: (options: OpenAIResponsesOptions | undefined) => ResponsesPricingOptions;
 };
 
@@ -201,6 +206,7 @@ function createResponsesTransportExecutor(config: ResponsesTransportExecutorOpti
         timestamp: Date.now(),
       };
       let firstEventAbort: ReturnType<typeof createFirstStreamEventAbortController> | undefined;
+      let continuationClaim: ReturnType<typeof claimOpenAIResponsesHttpContinuation>;
       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<AsyncIterable<unknown>> => {
-          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,

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();
+  });
+});

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<string, unknown> & {
+  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<typeof setTimeout>;
+    }
+  | { kind: "claimed"; sessionId: string; generation: number };
+
+const httpContinuationEntries = new Map<string, HttpContinuationEntry>();
+let nextHttpContinuationGeneration = 1;
+
+type HttpContinuationIdentity = {
+  apiKey: string;
+  baseUrl: string;
+  headers: Record<string, string>;
+};
+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<HttpContinuationEntry, { kind: "ready" }>;
+      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);
+    }
+  }
+});

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<string, string>;
+  previous_response_id?: string;
   store?: boolean;
   max_output_tokens?: number;
   temperature?: number;

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<TRequest extends ResponsesEncryptedContentRequest> = {
   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<OpenAIResponsesRequestParams>;
-}): Promise<{ stream: AsyncIterable<unknown>; response: Response }> {
+}): Promise<{
+  stream: AsyncIterable<unknown>;
+  response: Response;
+  attempt: ResponsesEncryptedContentAttempt<OpenAIResponsesRequestParams>;
+}> {
   const sendAttempt = async (
     attempt: ResponsesEncryptedContentAttempt<OpenAIResponsesRequestParams>,
   ) => {
@@ -294,7 +299,7 @@ export async function createResponsesStreamWithEncryptedContentRetry(params: {
     if (attempt.kind === "compaction-stripped") {
       params.onCompactionRejected?.();
     }
-    return { stream: data as unknown as AsyncIterable<unknown>, response };
+    return { stream: data as unknown as AsyncIterable<unknown>, response, attempt };
   };
 
   let attempt: ResponsesEncryptedContentAttempt<OpenAIResponsesRequestParams> = {
@@ -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;
     }
   }

packages/ai/src/transports/openai-responses-stream-internal.ts

@@ -72,6 +72,7 @@ type OpenAIResponsesConsumedEvent = Extract<
   ResponseStreamEvent,
   { type: ResponsesConsumedEventType }
 >;
+type CompletedResponse = Extract<ResponseStreamEvent, { type: "response.completed" }>["response"];
 type OpenAIResponsesIgnoredSdkEvent = Exclude<ResponseStreamEvent, OpenAIResponsesConsumedEvent>;
 type ResponsesTextContentPart =
   | ResponseOutputMessage["content"][number]
@@ -119,7 +120,7 @@ export async function processResponsesStream<TApi extends Api>(
   stream: ResponsesEventSink,
   model: Model<TApi>,
   options?: ResponsesStreamOptions,
-): Promise<void> {
+) {
   type StreamingToolCallBlock = ToolCall & { partialJson: string };
   type StreamingToolCallState = ResponsesToolCallState & {
     block: StreamingToolCallBlock;
@@ -134,7 +135,7 @@ export async function processResponsesStream<TApi extends Api>(
   const reasoningBlocksById = new Map<string, ResponsesThinkingBlock>();
   const outputItemContentIndexes = createResponsesOutputContentIndex();
   const startedTextBlocksByItemId = new Map<string, TextBlockReference>();
-  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<TApi extends Api>(
       setLastTextBlock: (block) => {
         lastTextBlock = block;
       },
-      markFinalized: () => {
-        terminalResponseEvent = "finalized";
-      },
+      markFinalized: () => undefined,
     });
 
   const guardedStream = adaptResponsesStream(
@@ -693,6 +692,7 @@ export async function processResponsesStream<TApi extends Api>(
         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<TApi extends Api>(
     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;

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<string, unknown>) => Record<string, unknown>;
     headers?: Record<string, string>;
   } = {},
 ): Promise<AssistantMessage> {
@@ -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();

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

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<string, unknown> & {
-  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<typeof setTimeout>;
-  continuation?: CachedWebSocketContinuation;
+  continuation?: ResponsesContinuationState;
 };
 
 type ResponsesWebSocketStreamMessage =
@@ -49,16 +40,9 @@ export type OpenAIResponsesWebSocketMode = "websocket" | "websocket-cached" | "a
 
 type OpenAIResponsesWebSocketStream = {
   stream: AsyncIterable<unknown>;
-  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<string, unknown>).filter(
-      ([key]) => key !== "openclaw_turn_id" && key !== "openclaw_turn_attempt",
-    ),
-  );
-  return { ...rest, metadata };
-}
-
-function sanitizeWebSocketRequest(request: Record<string, unknown>): ResponsesWebSocketRequest {
+function sanitizeWebSocketRequest(request: Record<string, unknown>): 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<OpenAIResponsesWebSocketStream, "continuationStatus" | "request"> {
-  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<typeof buildCachedWebSocketRequest>;
+  let prepared: Pick<OpenAIResponsesWebSocketStream, "continuationStatus" | "request">;
   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 });

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