From a71884dfdfff2a7756aaee48b8bf7cff6acbd2ce Mon Sep 17 00:00:00 2001 From: Shoubhit Dash Date: Thu, 20 Aug 2026 21:14:37 +0530 Subject: [PATCH] test(ai): record responses websocket flows (#43660) --- ...continues-a-tool-call-over-one-socket.json | 100 ++++++++ ...nstructs-full-context-after-reconnect.json | 119 ++++++++++ ...-from-explicit-continuation-rejection.json | 129 +++++++++++ ...penai-responses-websocket.recorded.test.ts | 216 ++++++++++++++++++ packages/ai/test/recorded-test.ts | 12 +- .../http-recorder/src/websocket/recorder.ts | 9 +- packages/http-recorder/test/websocket.test.ts | 32 +++ 7 files changed, 612 insertions(+), 5 deletions(-) create mode 100644 packages/ai/test/fixtures/recordings/openai-responses-websocket/continues-a-tool-call-over-one-socket.json create mode 100644 packages/ai/test/fixtures/recordings/openai-responses-websocket/reconstructs-full-context-after-reconnect.json create mode 100644 packages/ai/test/fixtures/recordings/openai-responses-websocket/recovers-from-explicit-continuation-rejection.json create mode 100644 packages/ai/test/provider/openai-responses-websocket.recorded.test.ts diff --git a/packages/ai/test/fixtures/recordings/openai-responses-websocket/continues-a-tool-call-over-one-socket.json b/packages/ai/test/fixtures/recordings/openai-responses-websocket/continues-a-tool-call-over-one-socket.json new file mode 100644 index 0000000000..49766db975 --- /dev/null +++ b/packages/ai/test/fixtures/recordings/openai-responses-websocket/continues-a-tool-call-over-one-socket.json @@ -0,0 +1,100 @@ +{ + "version": 1, + "metadata": { + "provider": "openai", + "protocol": "openai-responses", + "transport": "websocket", + "model": "gpt-5.5", + "tags": [ + "prefix:openai-responses-websocket", + "provider:openai", + "protocol:openai-responses", + "transport:websocket", + "tool", + "continuation" + ], + "name": "openai-responses-websocket/continues-a-tool-call-over-one-socket", + "recordedAt": "2026-08-20T00:00:00.000Z" + }, + "interactions": [ + { + "transport": "websocket", + "connection": { + "sequence": 0, + "url": "wss://api.openai.com/v1/responses", + "protocols": [], + "close": { + "code": 1000, + "reason": "" + } + }, + "events": [ + { + "direction": "client", + "kind": "text", + "body": "{\"type\":\"response.create\",\"model\":\"gpt-5.5\",\"input\":[{\"role\":\"system\",\"content\":\"Call get_weather once, then reply exactly: Paris is sunny.\"},{\"role\":\"user\",\"content\":[{\"type\":\"input_text\",\"text\":\"What is the weather in Paris?\"}]}],\"tools\":[{\"type\":\"function\",\"name\":\"get_weather\",\"description\":\"Get current weather for a city.\",\"parameters\":{\"type\":\"object\",\"properties\":{\"city\":{\"type\":\"string\"}},\"required\":[\"city\"],\"additionalProperties\":false},\"strict\":false}],\"store\":false,\"max_output_tokens\":50,\"include\":[\"reasoning.encrypted_content\"],\"reasoning\":{\"effort\":\"medium\",\"summary\":\"auto\"},\"text\":{\"verbosity\":\"low\"}}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.created\",\"response\":{\"id\":\"resp_ws_tool_1\"}}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.output_item.added\",\"item\":{\"type\":\"function_call\",\"id\":\"fc_ws_weather\",\"call_id\":\"call_ws_weather\",\"name\":\"get_weather\",\"arguments\":\"\"}}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"fc_ws_weather\",\"delta\":\"{\\\"city\\\":\\\"Paris\\\"}\"}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.output_item.done\",\"item\":{\"type\":\"function_call\",\"id\":\"fc_ws_weather\",\"call_id\":\"call_ws_weather\",\"name\":\"get_weather\",\"arguments\":\"{\\\"city\\\":\\\"Paris\\\"}\"}}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.completed\",\"response\":{\"id\":\"resp_ws_tool_1\"}}" + }, + { + "direction": "client", + "kind": "text", + "body": "{\"type\":\"response.create\",\"model\":\"gpt-5.5\",\"input\":[{\"type\":\"function_call_output\",\"call_id\":\"call_ws_weather\",\"output\":\"{\\\"temperature\\\":22,\\\"condition\\\":\\\"sunny\\\"}\"}],\"tools\":[{\"type\":\"function\",\"name\":\"get_weather\",\"description\":\"Get current weather for a city.\",\"parameters\":{\"type\":\"object\",\"properties\":{\"city\":{\"type\":\"string\"}},\"required\":[\"city\"],\"additionalProperties\":false},\"strict\":false}],\"store\":false,\"max_output_tokens\":50,\"previous_response_id\":\"resp_ws_tool_1\",\"include\":[\"reasoning.encrypted_content\"],\"reasoning\":{\"effort\":\"medium\",\"summary\":\"auto\"},\"text\":{\"verbosity\":\"low\"}}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.created\",\"response\":{\"id\":\"resp_ws_tool_2\"}}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.output_item.added\",\"item\":{\"type\":\"message\",\"id\":\"msg_ws_tool_2\",\"role\":\"assistant\",\"content\":[]}}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.output_text.delta\",\"item_id\":\"msg_ws_tool_2\",\"delta\":\"Paris is sunny.\"}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.output_text.done\",\"item_id\":\"msg_ws_tool_2\",\"text\":\"Paris is sunny.\"}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.output_item.done\",\"item\":{\"type\":\"message\",\"id\":\"msg_ws_tool_2\",\"role\":\"assistant\",\"content\":[{\"type\":\"output_text\",\"text\":\"Paris is sunny.\"}]}}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.completed\",\"response\":{\"id\":\"resp_ws_tool_2\"}}" + } + ] + } + ] +} diff --git a/packages/ai/test/fixtures/recordings/openai-responses-websocket/reconstructs-full-context-after-reconnect.json b/packages/ai/test/fixtures/recordings/openai-responses-websocket/reconstructs-full-context-after-reconnect.json new file mode 100644 index 0000000000..5a573c0376 --- /dev/null +++ b/packages/ai/test/fixtures/recordings/openai-responses-websocket/reconstructs-full-context-after-reconnect.json @@ -0,0 +1,119 @@ +{ + "version": 1, + "metadata": { + "provider": "openai", + "protocol": "openai-responses", + "transport": "websocket", + "model": "gpt-5.5", + "tags": [ + "prefix:openai-responses-websocket", + "provider:openai", + "protocol:openai-responses", + "transport:websocket", + "reconnect", + "full-context" + ], + "name": "openai-responses-websocket/reconstructs-full-context-after-reconnect", + "recordedAt": "2026-08-20T00:00:00.000Z" + }, + "interactions": [ + { + "transport": "websocket", + "connection": { + "sequence": 0, + "url": "wss://api.openai.com/v1/responses", + "protocols": [], + "close": { + "code": 1000, + "reason": "" + } + }, + "events": [ + { + "direction": "client", + "kind": "text", + "body": "{\"type\":\"response.create\",\"model\":\"gpt-5.5\",\"input\":[{\"role\":\"system\",\"content\":\"Follow the user's exact reply instruction.\"},{\"role\":\"user\",\"content\":[{\"type\":\"input_text\",\"text\":\"Reply exactly: Alpha.\"}]}],\"store\":false,\"max_output_tokens\":30,\"include\":[\"reasoning.encrypted_content\"],\"reasoning\":{\"effort\":\"medium\",\"summary\":\"auto\"},\"text\":{\"verbosity\":\"low\"}}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.created\",\"response\":{\"id\":\"resp_ws_reconnect_1\"}}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.output_item.added\",\"item\":{\"type\":\"message\",\"id\":\"msg_ws_reconnect_1\",\"role\":\"assistant\",\"content\":[]}}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.output_text.delta\",\"item_id\":\"msg_ws_reconnect_1\",\"delta\":\"Alpha.\"}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.output_text.done\",\"item_id\":\"msg_ws_reconnect_1\",\"text\":\"Alpha.\"}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.output_item.done\",\"item\":{\"type\":\"message\",\"id\":\"msg_ws_reconnect_1\",\"role\":\"assistant\",\"content\":[{\"type\":\"output_text\",\"text\":\"Alpha.\"}]}}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.completed\",\"response\":{\"id\":\"resp_ws_reconnect_1\"}}" + } + ] + }, + { + "transport": "websocket", + "connection": { + "sequence": 1, + "url": "wss://api.openai.com/v1/responses", + "protocols": [], + "close": { + "code": 1000, + "reason": "" + } + }, + "events": [ + { + "direction": "client", + "kind": "text", + "body": "{\"type\":\"response.create\",\"model\":\"gpt-5.5\",\"input\":[{\"role\":\"system\",\"content\":\"Follow the user's exact reply instruction.\"},{\"role\":\"user\",\"content\":[{\"type\":\"input_text\",\"text\":\"Reply exactly: Alpha.\"}]},{\"type\":\"message\",\"id\":\"msg_ws_reconnect_1\",\"role\":\"assistant\",\"content\":[{\"type\":\"output_text\",\"text\":\"Alpha.\"}]},{\"role\":\"user\",\"content\":[{\"type\":\"input_text\",\"text\":\"Reply exactly: Beta.\"}]}],\"store\":false,\"max_output_tokens\":30,\"include\":[\"reasoning.encrypted_content\"],\"reasoning\":{\"effort\":\"medium\",\"summary\":\"auto\"},\"text\":{\"verbosity\":\"low\"}}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.created\",\"response\":{\"id\":\"resp_ws_reconnect_2\"}}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.output_item.added\",\"item\":{\"type\":\"message\",\"id\":\"msg_ws_reconnect_2\",\"role\":\"assistant\",\"content\":[]}}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.output_text.delta\",\"item_id\":\"msg_ws_reconnect_2\",\"delta\":\"Beta.\"}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.output_text.done\",\"item_id\":\"msg_ws_reconnect_2\",\"text\":\"Beta.\"}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.output_item.done\",\"item\":{\"type\":\"message\",\"id\":\"msg_ws_reconnect_2\",\"role\":\"assistant\",\"content\":[{\"type\":\"output_text\",\"text\":\"Beta.\"}]}}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.completed\",\"response\":{\"id\":\"resp_ws_reconnect_2\"}}" + } + ] + } + ] +} diff --git a/packages/ai/test/fixtures/recordings/openai-responses-websocket/recovers-from-explicit-continuation-rejection.json b/packages/ai/test/fixtures/recordings/openai-responses-websocket/recovers-from-explicit-continuation-rejection.json new file mode 100644 index 0000000000..537ac660b7 --- /dev/null +++ b/packages/ai/test/fixtures/recordings/openai-responses-websocket/recovers-from-explicit-continuation-rejection.json @@ -0,0 +1,129 @@ +{ + "version": 1, + "metadata": { + "provider": "openai", + "protocol": "openai-responses", + "transport": "websocket", + "model": "gpt-5.5", + "tags": [ + "prefix:openai-responses-websocket", + "provider:openai", + "protocol:openai-responses", + "transport:websocket", + "continuation", + "recovery" + ], + "name": "openai-responses-websocket/recovers-from-explicit-continuation-rejection", + "recordedAt": "2026-08-20T00:00:00.000Z" + }, + "interactions": [ + { + "transport": "websocket", + "connection": { + "sequence": 0, + "url": "wss://api.openai.com/v1/responses", + "protocols": [], + "close": { + "code": 1000, + "reason": "" + } + }, + "events": [ + { + "direction": "client", + "kind": "text", + "body": "{\"type\":\"response.create\",\"model\":\"gpt-5.5\",\"input\":[{\"role\":\"system\",\"content\":\"Follow the user's exact reply instruction.\"},{\"role\":\"user\",\"content\":[{\"type\":\"input_text\",\"text\":\"Reply exactly: Ready.\"}]}],\"store\":false,\"max_output_tokens\":30,\"include\":[\"reasoning.encrypted_content\"],\"reasoning\":{\"effort\":\"medium\",\"summary\":\"auto\"},\"text\":{\"verbosity\":\"low\"}}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.created\",\"response\":{\"id\":\"resp_ws_rejection_1\"}}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.output_item.added\",\"item\":{\"type\":\"message\",\"id\":\"msg_ws_rejection_1\",\"role\":\"assistant\",\"content\":[]}}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.output_text.delta\",\"item_id\":\"msg_ws_rejection_1\",\"delta\":\"Ready.\"}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.output_text.done\",\"item_id\":\"msg_ws_rejection_1\",\"text\":\"Ready.\"}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.output_item.done\",\"item\":{\"type\":\"message\",\"id\":\"msg_ws_rejection_1\",\"role\":\"assistant\",\"content\":[{\"type\":\"output_text\",\"text\":\"Ready.\"}]}}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.completed\",\"response\":{\"id\":\"resp_ws_rejection_1\"}}" + } + ] + }, + { + "transport": "websocket", + "connection": { + "sequence": 1, + "url": "wss://api.openai.com/v1/responses", + "protocols": [], + "close": { + "code": 1000, + "reason": "" + } + }, + "events": [ + { + "direction": "client", + "kind": "text", + "body": "{\"type\":\"response.create\",\"model\":\"gpt-5.5\",\"input\":[{\"role\":\"user\",\"content\":[{\"type\":\"input_text\",\"text\":\"Reply exactly: Recovered.\"}]}],\"store\":false,\"max_output_tokens\":30,\"previous_response_id\":\"resp_ws_rejection_1\",\"include\":[\"reasoning.encrypted_content\"],\"reasoning\":{\"effort\":\"medium\",\"summary\":\"auto\"},\"text\":{\"verbosity\":\"low\"}}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"error\",\"error\":{\"code\":\"previous_response_not_found\",\"message\":\"Previous response not found\"}}" + }, + { + "direction": "client", + "kind": "text", + "body": "{\"type\":\"response.create\",\"model\":\"gpt-5.5\",\"input\":[{\"role\":\"system\",\"content\":\"Follow the user's exact reply instruction.\"},{\"role\":\"user\",\"content\":[{\"type\":\"input_text\",\"text\":\"Reply exactly: Ready.\"}]},{\"type\":\"message\",\"id\":\"msg_ws_rejection_1\",\"role\":\"assistant\",\"content\":[{\"type\":\"output_text\",\"text\":\"Ready.\"}]},{\"role\":\"user\",\"content\":[{\"type\":\"input_text\",\"text\":\"Reply exactly: Recovered.\"}]}],\"store\":false,\"max_output_tokens\":30,\"include\":[\"reasoning.encrypted_content\"],\"reasoning\":{\"effort\":\"medium\",\"summary\":\"auto\"},\"text\":{\"verbosity\":\"low\"}}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.created\",\"response\":{\"id\":\"resp_ws_rejection_2\"}}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.output_item.added\",\"item\":{\"type\":\"message\",\"id\":\"msg_ws_rejection_2\",\"role\":\"assistant\",\"content\":[]}}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.output_text.delta\",\"item_id\":\"msg_ws_rejection_2\",\"delta\":\"Recovered.\"}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.output_text.done\",\"item_id\":\"msg_ws_rejection_2\",\"text\":\"Recovered.\"}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.output_item.done\",\"item\":{\"type\":\"message\",\"id\":\"msg_ws_rejection_2\",\"role\":\"assistant\",\"content\":[{\"type\":\"output_text\",\"text\":\"Recovered.\"}]}}" + }, + { + "direction": "server", + "kind": "text", + "body": "{\"type\":\"response.completed\",\"response\":{\"id\":\"resp_ws_rejection_2\"}}" + } + ] + } + ] +} diff --git a/packages/ai/test/provider/openai-responses-websocket.recorded.test.ts b/packages/ai/test/provider/openai-responses-websocket.recorded.test.ts new file mode 100644 index 0000000000..21e8c4353f --- /dev/null +++ b/packages/ai/test/provider/openai-responses-websocket.recorded.test.ts @@ -0,0 +1,216 @@ +import { describe, expect } from "bun:test" +import { Effect, Stream } from "effect" +import { Socket } from "effect/unstable/socket" +import { LLM, LLMRequest, Message, ToolRuntime } from "../../src/index.js" +import { + LLMClient, + WebSocketTransport, + type ChannelCheckpoint, + type ChannelObservation, + type WebSocketChannelExchange, + type WebSocketChannelExecutor, + type WebSocketConnection, +} from "../../src/route.js" +import { configure } from "../../src/providers/openai.js" +import { decodeJson } from "../../src/protocols/shared.js" +import { weatherRuntimeTool, weatherTool, weatherToolName } from "../recorded-scenarios.js" +import { recordedTests } from "../recorded-test.js" + +const model = configure({ apiKey: process.env.OPENAI_API_KEY ?? "fixture" }).responses("gpt-5.5") +const recorded = recordedTests({ + prefix: "openai-responses-websocket", + provider: "openai", + protocol: "openai-responses", + requires: ["OPENAI_API_KEY"], + tags: ["transport:websocket"], + metadata: { transport: "websocket", model: model.id }, +}) + +const observationFrame = (observation: ChannelObservation) => { + if (observation.type === "frame" || observation.type === "completed" || observation.type === "incomplete") + return Effect.succeed(observation.frame) + return Effect.fail(observation.error) +} + +const terminal = (observation: ChannelObservation) => observation.type !== "frame" + +// This deliberately models only sequential test traffic. Core owns production connection pooling and recovery. +const makeChannel = Effect.gen(function* () { + const constructor = yield* Socket.WebSocketConstructor + let connection: WebSocketConnection | undefined + let checkpoint: ChannelCheckpoint | undefined + let pending: ChannelCheckpoint | undefined + let opens = 0 + const sent: unknown[] = [] + + const close = Effect.suspend(() => { + const current = connection + connection = undefined + return current ? current.close : Effect.void + }) + yield* Effect.addFinalizer(() => close) + + const executor: WebSocketChannelExecutor = { + execute: (exchange: WebSocketChannelExchange) => + Effect.gen(function* () { + if (!connection) { + connection = yield* WebSocketTransport.open(exchange.connect).pipe( + Effect.provideService(Socket.WebSocketConstructor, constructor), + ) + opens += 1 + } + const current = connection + const create = yield* exchange.driver.create(checkpoint) + if (create.mode === "full") checkpoint = undefined + pending = undefined + sent.push(decodeJson(create.message)) + yield* current.sendText(create.message) + const decoder = new TextDecoder() + return { + frames: current.messages.pipe( + Stream.map((message) => WebSocketTransport.messageText(message, decoder)), + Stream.mapEffect((frame) => exchange.driver.observe(create, frame)), + Stream.tap((observation) => + Effect.sync(() => { + if (!terminal(observation)) return + pending = observation.type === "completed" ? observation.checkpoint : undefined + if (observation.type !== "completed") checkpoint = undefined + }), + ), + Stream.takeUntil(terminal), + Stream.mapEffect(observationFrame), + ), + complete: Effect.sync(() => { + checkpoint = pending + pending = undefined + }), + } + }), + } + + return { + executor, + sent, + opens: () => opens, + reconnect: (preserveCheckpoint = false) => + close.pipe( + Effect.andThen( + Effect.sync(() => { + pending = undefined + if (!preserveCheckpoint) checkpoint = undefined + }), + ), + ), + } +}) + +describe("OpenAI Responses WebSocket recorded", () => { + recorded.effect.with("continues a tool call over one socket", { tags: ["tool", "continuation"] }, () => + Effect.gen(function* () { + const channel = yield* makeChannel + const request = LLM.request({ + id: "recorded_openai_responses_websocket_tool", + model, + system: "Call get_weather once, then reply exactly: Paris is sunny.", + prompt: "What is the weather in Paris?", + tools: [weatherTool], + generation: { maxTokens: 50 }, + cache: "none", + }) + const first = yield* LLMClient.generate(request, { webSocket: channel.executor }) + const call = first.toolCalls[0] + if (!call) yield* Effect.die("Expected get_weather tool call") + const result = yield* ToolRuntime.dispatch({ [weatherToolName]: weatherRuntimeTool }, call) + const second = yield* LLMClient.generate( + LLMRequest.update(request, { + messages: [ + ...request.messages, + first.message, + Message.tool({ id: call.id, name: call.name, result: result.result }), + ], + }), + { webSocket: channel.executor }, + ) + + expect(second.text).toBe("Paris is sunny.") + expect(channel.opens()).toBe(1) + expect(channel.sent).toHaveLength(2) + expect(channel.sent[1]).toMatchObject({ + previous_response_id: expect.any(String), + input: [{ type: "function_call_output", call_id: call.id, output: expect.any(String) }], + }) + }), + ) + + recorded.effect.with("reconstructs full context after reconnect", { tags: ["reconnect", "full-context"] }, () => + Effect.gen(function* () { + const channel = yield* makeChannel + const request = LLM.request({ + id: "recorded_openai_responses_websocket_reconnect", + model, + system: "Follow the user's exact reply instruction.", + prompt: "Reply exactly: Alpha.", + generation: { maxTokens: 30 }, + cache: "none", + }) + const first = yield* LLMClient.generate(request, { webSocket: channel.executor }) + yield* channel.reconnect() + const second = yield* LLMClient.generate( + LLMRequest.update(request, { + messages: [...request.messages, first.message, Message.user("Reply exactly: Beta.")], + }), + { webSocket: channel.executor }, + ) + + expect(first.text).toBe("Alpha.") + expect(second.text).toBe("Beta.") + expect(channel.opens()).toBe(2) + expect(channel.sent[1]).not.toHaveProperty("previous_response_id") + expect(channel.sent[1]).toMatchObject({ + input: [ + { role: "system", content: "Follow the user's exact reply instruction." }, + { role: "user", content: [{ type: "input_text", text: "Reply exactly: Alpha." }] }, + { role: "assistant", content: [{ type: "output_text", text: "Alpha." }] }, + { role: "user", content: [{ type: "input_text", text: "Reply exactly: Beta." }] }, + ], + }) + }), + ) + + recorded.effect.with("recovers from explicit continuation rejection", { tags: ["continuation", "recovery"] }, () => + Effect.gen(function* () { + const channel = yield* makeChannel + const request = LLM.request({ + id: "recorded_openai_responses_websocket_rejection", + model, + system: "Follow the user's exact reply instruction.", + prompt: "Reply exactly: Ready.", + generation: { maxTokens: 30 }, + cache: "none", + }) + const first = yield* LLMClient.generate(request, { webSocket: channel.executor }) + const continuation = LLMRequest.update(request, { + messages: [...request.messages, first.message, Message.user("Reply exactly: Recovered.")], + }) + yield* channel.reconnect(true) + const rejected = yield* LLMClient.generate(continuation, { webSocket: channel.executor }).pipe(Effect.flip) + const recovered = yield* LLMClient.generate(continuation, { webSocket: channel.executor }) + + expect(rejected).toMatchObject({ + reason: { _tag: "Transport", delivery: "rejected", recovery: "retry-full" }, + }) + expect(recovered.text).toBe("Recovered.") + expect(channel.opens()).toBe(2) + expect(channel.sent[1]).toHaveProperty("previous_response_id", expect.any(String)) + expect(channel.sent[2]).not.toHaveProperty("previous_response_id") + expect(channel.sent[2]).toMatchObject({ + input: [ + { role: "system", content: "Follow the user's exact reply instruction." }, + { role: "user", content: [{ type: "input_text", text: "Reply exactly: Ready." }] }, + { role: "assistant", content: [{ type: "output_text", text: "Ready." }] }, + { role: "user", content: [{ type: "input_text", text: "Reply exactly: Recovered." }] }, + ], + }) + }), + ) +}) diff --git a/packages/ai/test/recorded-test.ts b/packages/ai/test/recorded-test.ts index 3c7abc095d..18dffa805a 100644 --- a/packages/ai/test/recorded-test.ts +++ b/packages/ai/test/recorded-test.ts @@ -1,5 +1,7 @@ import { HttpRecorder } from "@opencode-ai/http-recorder" +import { NodeSocket } from "@effect/platform-node" import { Layer } from "effect" +import { Socket } from "effect/unstable/socket" import * as path from "node:path" import { fileURLToPath } from "node:url" import { LLMClient, RequestExecutor } from "../src/route.js" @@ -16,7 +18,7 @@ import { const __dirname = path.dirname(fileURLToPath(import.meta.url)) const FIXTURES_DIR = path.resolve(__dirname, "fixtures", "recordings") -type RecordedEnv = RequestExecutorService | LLMClientService | ImageClientService +type RecordedEnv = RequestExecutorService | LLMClientService | ImageClientService | Socket.WebSocketConstructor type RecordedTestsOptions = RecordedGroupOptions & { readonly options?: HttpRecorder.RecorderOptions @@ -69,7 +71,7 @@ export const recordedTests = (options: RecordedTestsOptions) => ...metadata, } if (recording) { - if (process.env.CI !== undefined) throw new Error("Unset CI before recording HTTP cassettes") + if (process.env.CI !== undefined) throw new Error("Unset CI before recording cassettes") HttpRecorder.removeCassetteSync(cassette, { directory: FIXTURES_DIR }) } const requestExecutor = RequestExecutor.layer.pipe( @@ -81,10 +83,16 @@ export const recordedTests = (options: RecordedTestsOptions) => }), ), ) + const webSocket = HttpRecorder.layerWebSocketConstructor(cassette, { + ...recorderOptions, + directory: FIXTURES_DIR, + metadata: recorderMetadata, + }).pipe(Layer.provide(NodeSocket.layerWebSocketConstructorWS)) return Layer.mergeAll( requestExecutor, LLMClient.layer.pipe(Layer.provide(requestExecutor)), ImageClient.layer.pipe(Layer.provide(requestExecutor)), + webSocket, ) }, }) diff --git a/packages/http-recorder/src/websocket/recorder.ts b/packages/http-recorder/src/websocket/recorder.ts index 3d232c4db9..b2b3850214 100644 --- a/packages/http-recorder/src/websocket/recorder.ts +++ b/packages/http-recorder/src/websocket/recorder.ts @@ -31,8 +31,11 @@ interface PendingRecordings { } type Frame = string | Uint8Array -const normalizeProtocols = (protocols?: string | Array): Array => - protocols === undefined ? [] : typeof protocols === "string" ? [protocols] : [...protocols] +const normalizeProtocols = (protocols: unknown): Array => { + if (typeof protocols === "string") return [protocols] + if (Array.isArray(protocols)) return protocols.filter((protocol): protocol is string => typeof protocol === "string") + return [] +} const frameFromWebSocketData = async (data: unknown): Promise => { if (typeof data === "string") return data if (data instanceof Blob) return new Uint8Array(await data.arrayBuffer()) @@ -371,7 +374,7 @@ const makeRecordingWebSocketConstructor = ( return (url, protocols) => { const sequence = nextSequence++ const requestedProtocols = normalizeProtocols(protocols) - const native = upstream(url, requestedProtocols) + const native = Reflect.apply(upstream, undefined, [url, protocols]) const events: WebSocketEvent[] = [] let opened = false let failed = false diff --git a/packages/http-recorder/test/websocket.test.ts b/packages/http-recorder/test/websocket.test.ts index 2d4d30cbf2..9e641c5c5d 100644 --- a/packages/http-recorder/test/websocket.test.ts +++ b/packages/http-recorder/test/websocket.test.ts @@ -80,6 +80,38 @@ describe("WebSocket", () => { ]) }) + test("constructor recording forwards handshake options", async () => { + using directory = tempDirectory("http-recorder-websocket-constructor-") + let received: unknown + const recorder = HttpRecorder.layerWebSocketConstructor("websocket/constructor-options", { + directory: directory.path, + }).pipe( + Layer.provide( + Layer.succeed(Socket.WebSocketConstructor, (url, options) => { + received = options + // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion -- the fixture implements the WebSocket surface used by the recorder. + return new EchoWebSocket(url) as unknown as globalThis.WebSocket + }), + ), + ) + + await Effect.runPromise( + Effect.gen(function* () { + const constructor = yield* Socket.WebSocketConstructor + const options = { headers: { authorization: "Bearer fixture" } } + const socket = Reflect.apply(constructor, undefined, ["wss://echo.example.test/options", options]) + yield* Effect.callback((resume) => { + socket.addEventListener("open", () => { + socket.close() + resume(Effect.void) + }) + }) + }).pipe(Effect.scoped, Effect.provide(recorder)), + ) + + expect(received).toEqual({ headers: { authorization: "Bearer fixture" } }) + }) + test("constructor replay validates dynamic URLs and protocols without opening a live socket", async () => { using directory = tempDirectory("http-recorder-websocket-constructor-") await seedCassetteDirectory(directory.path, "websocket/constructor", [