refactor(ai): drop unused Responses continuation state (#61163)

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 = <Body>(options: Options): Transport<Body, Prepared, string> => {
@@ -154,8 +136,7 @@ export const transport = <Body>(options: Options): Transport<Body, Prepared, str
         const channel =
           input.webSocket && (options.enabled?.(parts.url) ?? true)
             ? yield* Effect.gen(function* () {
-                const create = yield* message(parts.jsonBody)
-                const base = driver(options, create.message)
+                const request = yield* message(parts.jsonBody)
                 return {
                   url: yield* WebSocketTransport.toWebSocketUrl(options.url?.(parts.url) ?? parts.url),
                   headers,
@@ -163,9 +144,8 @@ export const transport = <Body>(options: Options): Transport<Body, Prepared, str
                   driver: OpenResponsesContinuation.driver({
                     id: options.id,
                     name: options.name,
-                    request: create.request,
-                    message: create.message,
-                    base,
+                    request,
+                    base: observer(options),
                     continuation: options.continuation,
                   }),
                 }

packages/ai/src/protocols/open-responses-continuation.ts

@@ -25,20 +25,15 @@ interface CheckpointValue {
  */
 export type Shape = (request: Readonly<Record<string, unknown>>) => Readonly<Record<string, unknown>> | undefined
 
-export interface BaseDriver extends WebSocketChannelDriver {
-  readonly observeEvent?: (
-    create: ChannelCreate,
-    frame: string,
-    event: OpenResponses.Event,
-  ) => Effect.Effect<ChannelObservation, AIError>
-}
-
 export interface DriverInput {
   readonly id: string
   readonly name: string
   readonly request: Readonly<Record<string, unknown>>
-  readonly message: string
-  readonly base: BaseDriver
+  readonly base: (
+    create: ChannelCreate,
+    frame: string,
+    event: OpenResponses.Event,
+  ) => Effect.Effect<ChannelObservation, AIError>
   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)

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<ChannelObservation, AIError> => {
-    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<ChannelObservation, AIError> => {
+  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<Record<string, unknown>>,
-  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)
     }),