From 11ce936b6386f0348fd6bbdc76120e06db116fbd Mon Sep 17 00:00:00 2001 From: Aiden Cline <63023139+rekram1-node@users.noreply.github.com> Date: Sun, 1 Nov 2026 21:18:44 -0600 Subject: [PATCH] refactor(ai): drop unused Responses continuation state (#61163) --- .../src/protocols/open-responses-channel.ts | 32 ++----- .../protocols/open-responses-continuation.ts | 19 ++-- .../ai/test/provider/openai-responses.test.ts | 95 ++++++++----------- 3 files changed, 52 insertions(+), 94 deletions(-) diff --git a/packages/ai/src/protocols/open-responses-channel.ts b/packages/ai/src/protocols/open-responses-channel.ts index 2317eb0885c..ab47b0a0535 100644 --- a/packages/ai/src/protocols/open-responses-channel.ts +++ b/packages/ai/src/protocols/open-responses-channel.ts @@ -19,7 +19,6 @@ const WebSocketResponseCreate = Schema.StructWithRest(Schema.Struct({ type: Sche Schema.Record(Schema.String, Schema.Unknown), ]) const decodeMessage = ProviderShared.validateWith(Schema.decodeUnknownEffect(WebSocketResponseCreate)) -const encodeMessage = Schema.encodeSync(Schema.fromJsonString(WebSocketResponseCreate)) export interface Options { readonly id: string @@ -46,14 +45,13 @@ const message = (body: unknown) => if (!ProviderShared.isRecord(body)) return yield* ProviderShared.invalidRequest("Open Responses WebSocket body must be a JSON object") const { stream: _stream, stream_options: _streamOptions, background: _background, ...request } = body - const decoded = yield* decodeMessage({ ...request, type: "response.create" }) - return { request: decoded, message: encodeMessage(decoded) } + return yield* decodeMessage({ ...request, type: "response.create" }) }) -const driver = (options: Options, body: string): OpenResponsesContinuation.BaseDriver => { +const observer = (options: Options) => { let responseID: string | undefined let terminal = false - const observeEvent = Effect.fnUntraced(function* ( + return Effect.fnUntraced(function* ( _create: ChannelCreate, frame: string, event: OpenResponses.Event, @@ -125,22 +123,6 @@ const driver = (options: Options, body: string): OpenResponsesContinuation.BaseD } return { type: "frame", frame } }) - return { - create: () => - Effect.sync(() => { - responseID = undefined - terminal = false - return { message: body, mode: "full" } - }), - observeEvent, - observe: (create, frame) => - OpenResponses.decodeChannelEvent(frame).pipe( - Effect.mapError((cause) => - ProviderShared.eventError(options.id, `Invalid ${options.name} WebSocket event`, frame, cause), - ), - Effect.flatMap((event) => observeEvent(create, frame, event)), - ), - } } export const transport = (options: Options): Transport => { @@ -154,8 +136,7 @@ export const transport = (options: Options): Transport(options: Options): Transport>) => Readonly> | undefined -export interface BaseDriver extends WebSocketChannelDriver { - readonly observeEvent?: ( - create: ChannelCreate, - frame: string, - event: OpenResponses.Event, - ) => Effect.Effect -} - export interface DriverInput { readonly id: string readonly name: string readonly request: Readonly> - readonly message: string - readonly base: BaseDriver + readonly base: ( + create: ChannelCreate, + frame: string, + event: OpenResponses.Event, + ) => Effect.Effect readonly continuation?: Shape } @@ -170,9 +165,7 @@ export const driver = (input: DriverInput): WebSocketChannelDriver => { ProviderShared.eventError(input.id, `Invalid ${input.name} WebSocket event`, frame, cause), ), ) - const observation = yield* input.base.observeEvent - ? input.base.observeEvent(create, frame, event) - : input.base.observe(create, frame) + const observation = yield* input.base(create, frame, event) if (event.type === "response.output_item.done" && event.item) output.push(event.item) if (observation.type === "provider-failure") { const rejection = code(event) diff --git a/packages/ai/test/provider/openai-responses.test.ts b/packages/ai/test/provider/openai-responses.test.ts index 2a7e71497ff..338c30d37d9 100644 --- a/packages/ai/test/provider/openai-responses.test.ts +++ b/packages/ai/test/provider/openai-responses.test.ts @@ -1,5 +1,5 @@ import { describe, expect } from "bun:test" -import { ConfigProvider, Effect, Layer, Logger, Ref, Schema, Stream } from "effect" +import { ConfigProvider, Effect, Layer, Logger, Ref, Stream } from "effect" import { Headers, HttpClientRequest } from "effect/http" import { ai, @@ -22,8 +22,8 @@ import { LLMClient, RequestExecutor, WebSocketTransport, + type ChannelCreate, type ChannelObservation, - type WebSocketChannelDriver, } from "../../src/route.js" import { compileRequest } from "../../src/route/client.js" import * as Azure from "../../src/providers/azure.js" @@ -46,67 +46,52 @@ const model = OpenAIResponses.route const xaiModel = XAI.configure({ apiKey: "test", baseURL: "https://api.x.ai/v1" }).responses("grok-4.5") -const baseChannelDriver = (message: string): WebSocketChannelDriver => ({ - create: () => Effect.succeed({ message, mode: "full" }), - observe: (_create, frame): Effect.Effect => { - const event = ProviderShared.decodeJson(frame) - if (!ProviderShared.isRecord(event)) return Effect.die("Expected event") - if (event.type === "response.completed") return Effect.succeed({ type: "completed", frame }) - if (event.type === "response.incomplete") return Effect.succeed({ type: "incomplete", frame }) - if (event.type === "error" || event.type === "response.failed") - return Effect.succeed({ - type: "provider-failure", - error: new AIError({ - reason: new TransportError({ - message: "provider rejected request", - body: frame, - cause: new Error("provider cause"), - http: new HttpContext({ url: "https://provider.test", status: 200, headers: { "x-trace": "trace-1" } }), - transport: "websocket", - operation: "read", - phase: "receive", - }), +const baseChannelObserve = (_create: ChannelCreate, frame: string): Effect.Effect => { + const event = ProviderShared.decodeJson(frame) + if (!ProviderShared.isRecord(event)) return Effect.die("Expected event") + if (event.type === "response.completed") return Effect.succeed({ type: "completed", frame }) + if (event.type === "response.incomplete") return Effect.succeed({ type: "incomplete", frame }) + if (event.type === "error" || event.type === "response.failed") + return Effect.succeed({ + type: "provider-failure", + error: new AIError({ + reason: new TransportError({ + message: "provider rejected request", + body: frame, + cause: new Error("provider cause"), + http: new HttpContext({ url: "https://provider.test", status: 200, headers: { "x-trace": "trace-1" } }), + transport: "websocket", + operation: "read", + phase: "receive", }), - }) - return Effect.succeed({ type: "frame", frame }) - }, -}) + }), + }) + return Effect.succeed({ type: "frame", frame }) +} /** Classifies error frames the way the production channel does, so recovery can read the canonical reason. */ -const classifyingChannelDriver = (message: string): WebSocketChannelDriver => { - const base = baseChannelDriver(message) - const decodeEvent = Schema.decodeUnknownSync(OpenResponses.protocol.stream.event) - return { - ...base, - observe: (create, frame) => - base.observe(create, frame).pipe( - Effect.map((observation) => - observation.type === "provider-failure" - ? { - ...observation, - error: OpenResponses.providerFailure(decodeEvent(frame), "stream error", frame), - } - : observation, - ), - ), - } -} +const classifyingChannelObserve = (create: ChannelCreate, frame: string, event: OpenResponses.Event) => + baseChannelObserve(create, frame).pipe( + Effect.map( + (observation): ChannelObservation => + observation.type === "provider-failure" + ? { ...observation, error: OpenResponses.providerFailure(event, "stream error", frame) } + : observation, + ), + ) const continuationDriver = ( request: Readonly>, - base = baseChannelDriver, + base: OpenResponsesContinuation.DriverInput["base"] = baseChannelObserve, continuation?: OpenResponsesContinuation.Shape, -) => { - const message = ProviderShared.encodeJson(request) - return OpenResponsesContinuation.driver({ +) => + OpenResponsesContinuation.driver({ id: "openai-responses", name: "OpenAI Responses", request, - message, - base: base(message), + base, continuation, }) -} const checkpoint = (observation: ChannelObservation) => { if (observation.type !== "completed" || !observation.checkpoint) throw new Error("Expected checkpoint") @@ -948,7 +933,7 @@ describe("OpenAI Responses route", () => { store: false, input: [{ type: "message", role: "user", content: [{ type: "input_text", text: "First" }] }], } - const first = continuationDriver(firstRequest, classifyingChannelDriver) + const first = continuationDriver(firstRequest, classifyingChannelObserve) const saved = checkpoint( yield* first.observe( yield* first.create(undefined), @@ -963,7 +948,7 @@ describe("OpenAI Responses route", () => { { type: "message", role: "user", content: [{ type: "input_text", text: "Second" }] }, ], }, - classifyingChannelDriver, + classifyingChannelObserve, ) // Codex reports a stale previous_response_id as a plain invalid_request_error. const stale = ProviderShared.encodeJson({ @@ -1026,7 +1011,7 @@ describe("OpenAI Responses route", () => { const trimmed = yield* continuationDriver( secondRequest, - baseChannelDriver, + baseChannelObserve, ({ instructions: _, ...rest }) => rest, ).create(saved) expect(trimmed.mode).toBe("incremental") @@ -1039,7 +1024,7 @@ describe("OpenAI Responses route", () => { }) // Declining the continuation sends the step in full and never sends a previous_response_id. - const declined = yield* continuationDriver(secondRequest, baseChannelDriver, () => undefined).create(saved) + const declined = yield* continuationDriver(secondRequest, baseChannelObserve, () => undefined).create(saved) expect(declined.mode).toBe("full") expect(JSON.parse(declined.message)).toEqual(secondRequest) }),