refactor(llm): share provider stream parsing

2026-04-26 09:20:24 -04:00
parent 0f4e54d6e8
commit 9a05675200
4 changed files with 107 additions and 149 deletions
@@ -1,11 +1,9 @@
-import { Cause, Effect, Schema, Stream } from "effect"
-import * as Sse from "effect/unstable/encoding/Sse"
+import { Effect, Schema, Stream } from "effect"
 import { HttpClientRequest, type HttpClientResponse } from "effect/unstable/http"
 import { Adapter } from "../adapter"
 import { capabilities, model as llmModel, type ModelInput } from "../llm"
 import {
  InvalidRequestError,
-  ProviderChunkError,
  Usage,
  type CacheHint,
  type FinishReason,
@@ -16,6 +14,9 @@ import {
  type ToolDefinition,
  type ToolResultPart,
 } from "../schema"
+import { ProviderShared } from "./shared"
+
+const ADAPTER = "anthropic-messages"

 export type AnthropicMessagesModelInput = Omit<ModelInput, "provider" | "protocol" | "headers"> & {
  readonly apiKey?: string
@@ -151,9 +152,6 @@ interface ParserState {
  readonly usage?: Usage
 }

-const Json = Schema.fromJsonString(Schema.Unknown)
-const decodeJson = Schema.decodeUnknownSync(Json)
-const encodeJson = Schema.encodeSync(Json)
 const AnthropicChunkJson = Schema.fromJsonString(AnthropicChunk)
 const AnthropicTargetJson = Schema.fromJsonString(AnthropicMessagesTarget)
 const decodeChunk = Schema.decodeUnknownSync(AnthropicChunkJson)
@@ -170,7 +168,7 @@ const text = (values: ReadonlyArray<{ readonly text: string }>) => values.map((p

 const resultText = (part: ToolResultPart) => {
  if (part.result.type === "text" || part.result.type === "error") return String(part.result.value)
-  return encodeJson(part.result.value)
+  return ProviderShared.encodeJson(part.result.value)
 }

 const lowerTool = (tool: ToolDefinition): AnthropicTool => ({
@@ -327,37 +325,13 @@ const mergeUsage = (left: Usage | undefined, right: Usage | undefined) => {
  })
 }

-const chunkError = (message: string, raw?: string) => new ProviderChunkError({ adapter: "anthropic-messages", message, raw })
-
-const streamError = (cause: Cause.Cause<unknown>) => {
-  const failed = cause.reasons.find(Cause.isFailReason)?.error
-  if (failed instanceof ProviderChunkError) return failed
-  return chunkError("Failed to read Anthropic Messages stream", Cause.pretty(cause))
-}
-
-const parseJson = (input: string, message: string) => {
-  try {
-    return decodeJson(input)
-  } catch {
-    throw chunkError(message, input)
-  }
-}
-
-const parseChunk = (data: string) => {
-  try {
-    return decodeChunk(data)
-  } catch {
-    throw chunkError("Invalid Anthropic Messages stream chunk", data)
-  }
-}
-
 const finishToolCall = (tool: ToolAccumulator | undefined) => {
  if (!tool) return []
  return [{
    type: "tool-call" as const,
    id: tool.id,
    name: tool.name,
-    input: parseJson(tool.input || "{}", `Invalid JSON input for Anthropic Messages tool call ${tool.name}`),
+    input: ProviderShared.parseJson(ADAPTER, tool.input || "{}", `Invalid JSON input for Anthropic Messages tool call ${tool.name}`),
  }]
 }

@@ -400,7 +374,7 @@ const processChunk = (state: ParserState, chunk: AnthropicChunk): readonly [Pars
  if (chunk.type === "content_block_delta" && chunk.delta?.type === "input_json_delta" && chunk.index !== undefined) {
    if (!chunk.delta.partial_json) return [state, []]
    const current = state.tools[chunk.index]
-    if (!current) throw chunkError("Anthropic Messages tool argument delta is missing its tool call")
+    if (!current) throw ProviderShared.chunkError(ADAPTER, "Anthropic Messages tool argument delta is missing its tool call")
    const next = { ...current, input: `${current.input}${chunk.delta.partial_json ?? ""}` }
    return [{ ...state, tools: { ...state.tools, [chunk.index]: next } }, [
      { type: "tool-input-delta", id: next.id, name: next.name, text: chunk.delta.partial_json ?? "" },
@@ -426,24 +400,18 @@ const processChunk = (state: ParserState, chunk: AnthropicChunk): readonly [Pars
 }

 const events = (response: HttpClientResponse.HttpClientResponse) =>
-  response.stream.pipe(
-    Stream.mapError((error) => chunkError("Failed to read Anthropic Messages stream", String(error))),
-    Stream.decodeText(),
-    Stream.pipeThroughChannel(Sse.decode()),
-    Stream.filter((event) => event.data.length > 0 && event.data !== "[DONE]"),
-    Stream.mapEffect((event) =>
-      Effect.try({
-        try: () => parseChunk(event.data),
-        catch: (error) =>
-          error instanceof ProviderChunkError ? error : chunkError("Invalid Anthropic Messages stream chunk", event.data),
-      }),
-    ),
-    Stream.mapAccum((): ParserState => ({ tools: {} }), processChunk),
-    Stream.catchCause((cause) => Stream.fail(streamError(cause))),
-  )
+  ProviderShared.sse({
+    adapter: ADAPTER,
+    response,
+    readError: "Failed to read Anthropic Messages stream",
+    invalidChunk: "Invalid Anthropic Messages stream chunk",
+    decodeChunk,
+    initial: (): ParserState => ({ tools: {} }),
+    process: processChunk,
+  })

 export const adapter = Adapter.define<AnthropicMessagesDraft, AnthropicMessagesTarget, LLMEvent>({
-  id: "anthropic-messages",
+  id: ADAPTER,
  protocol: "anthropic-messages",
  redact: (target) => target,
  prepare,
@@ -1,11 +1,9 @@
-import { Cause, Effect, Schema, Stream } from "effect"
-import * as Sse from "effect/unstable/encoding/Sse"
+import { Effect, Schema, Stream } from "effect"
 import { HttpClientRequest, type HttpClientResponse } from "effect/unstable/http"
 import { Adapter } from "../adapter"
 import { capabilities, model as llmModel, type ModelInput } from "../llm"
 import {
  InvalidRequestError,
-  ProviderChunkError,
  Usage,
  type FinishReason,
  type ContentPart,
@@ -16,6 +14,9 @@ import {
  type ToolDefinition,
  type ToolResultPart,
 } from "../schema"
+import { ProviderShared } from "./shared"
+
+const ADAPTER = "openai-chat"

 export type OpenAIChatModelInput = Omit<ModelInput, "provider" | "protocol" | "headers"> & {
  readonly apiKey?: string
@@ -131,9 +132,6 @@ const OpenAIChatChunk = Schema.Struct({
 })
 type OpenAIChatChunk = Schema.Schema.Type<typeof OpenAIChatChunk>

-const Json = Schema.fromJsonString(Schema.Unknown)
-const decodeJson = Schema.decodeUnknownSync(Json)
-const encodeJson = Schema.encodeSync(Json)
 const OpenAIChatChunkJson = Schema.fromJsonString(OpenAIChatChunk)
 const OpenAIChatTargetJson = Schema.fromJsonString(OpenAIChatTarget)
 const decodeChunk = Schema.decodeUnknownSync(OpenAIChatChunkJson)
@@ -161,7 +159,7 @@ const text = (values: ReadonlyArray<{ readonly text: string }>) => values.map((p

 const resultText = (part: ToolResultPart) => {
  if (part.result.type === "text" || part.result.type === "error") return String(part.result.value)
-  return encodeJson(part.result.value)
+  return ProviderShared.encodeJson(part.result.value)
 }

 const lowerTool = (tool: ToolDefinition): OpenAIChatTool => ({
@@ -188,7 +186,7 @@ const lowerToolCall = (part: ToolCallPart): OpenAIChatAssistantToolCall => ({
  type: "function",
  function: {
    name: part.name,
-    arguments: encodeJson(part.input),
+    arguments: ProviderShared.encodeJson(part.input),
  },
 })

@@ -286,35 +284,11 @@ const mapUsage = (usage: OpenAIChatChunk["usage"]): Usage | undefined => {
  })
 }

-const chunkError = (message: string, raw?: string) => new ProviderChunkError({ adapter: "openai-chat", message, raw })
-
-const streamError = (cause: Cause.Cause<unknown>) => {
-  const failed = cause.reasons.find(Cause.isFailReason)?.error
-  if (failed instanceof ProviderChunkError) return failed
-  return chunkError("Failed to read OpenAI Chat stream", Cause.pretty(cause))
-}
-
-const parseJson = (input: string, message: string) => {
-  try {
-    return decodeJson(input)
-  } catch {
-    throw chunkError(message, input)
-  }
-}
-
-const parseChunk = (data: string) => {
-  try {
-    return decodeChunk(data)
-  } catch {
-    throw chunkError("Invalid OpenAI Chat stream chunk", data)
-  }
-}
-
 const pushToolDelta = (tools: Record<number, ToolAccumulator>, delta: OpenAIChatToolCallDelta) => {
  const current = tools[delta.index]
  const id = delta.id ?? current?.id
  const name = delta.function?.name ?? current?.name
-  if (!id || !name) throw chunkError("OpenAI Chat tool call delta is missing id or name")
+  if (!id || !name) throw ProviderShared.chunkError(ADAPTER, "OpenAI Chat tool call delta is missing id or name")

  return {
    id,
@@ -328,7 +302,7 @@ const finishToolCalls = (state: ParserState) =>
    type: "tool-call" as const,
    id: tool.id,
    name: tool.name,
-    input: parseJson(tool.input || "{}", `Invalid JSON input for OpenAI Chat tool call ${tool.name}`),
+    input: ProviderShared.parseJson(ADAPTER, tool.input || "{}", `Invalid JSON input for OpenAI Chat tool call ${tool.name}`),
  }))

 const processChunk = (state: ParserState, chunk: OpenAIChatChunk): readonly [ParserState, ReadonlyArray<LLMEvent>] => {
@@ -363,24 +337,19 @@ const finishEvents = (state: ParserState): ReadonlyArray<LLMEvent> => {
 }

 const events = (response: HttpClientResponse.HttpClientResponse) =>
-  response.stream.pipe(
-    Stream.mapError((error) => chunkError("Failed to read OpenAI Chat stream", String(error))),
-    Stream.decodeText(),
-    Stream.pipeThroughChannel(Sse.decode()),
-    Stream.filter((event) => event.data.length > 0 && event.data !== "[DONE]"),
-    Stream.mapEffect((event) =>
-      Effect.try({
-        try: () => parseChunk(event.data),
-        catch: (error) =>
-          error instanceof ProviderChunkError ? error : chunkError("Invalid OpenAI Chat stream chunk", event.data),
-      }),
-    ),
-    Stream.mapAccum((): ParserState => ({ tools: {} }), processChunk, { onHalt: finishEvents }),
-    Stream.catchCause((cause) => Stream.fail(streamError(cause))),
-  )
+  ProviderShared.sse({
+    adapter: ADAPTER,
+    response,
+    readError: "Failed to read OpenAI Chat stream",
+    invalidChunk: "Invalid OpenAI Chat stream chunk",
+    decodeChunk,
+    initial: (): ParserState => ({ tools: {} }),
+    process: processChunk,
+    onHalt: finishEvents,
+  })

 export const adapter = Adapter.define<OpenAIChatDraft, OpenAIChatTarget, LLMEvent>({
-  id: "openai-chat",
+  id: ADAPTER,
  protocol: "openai-chat",
  redact: (target) => target,
  prepare,
@@ -1,11 +1,9 @@
-import { Cause, Effect, Schema, Stream } from "effect"
-import * as Sse from "effect/unstable/encoding/Sse"
+import { Effect, Schema, Stream } from "effect"
 import { HttpClientRequest, type HttpClientResponse } from "effect/unstable/http"
 import { Adapter } from "../adapter"
 import { capabilities, model as llmModel, type ModelInput } from "../llm"
 import {
  InvalidRequestError,
-  ProviderChunkError,
  Usage,
  type FinishReason,
  type LLMEvent,
@@ -15,6 +13,9 @@ import {
  type ToolDefinition,
  type ToolResultPart,
 } from "../schema"
+import { ProviderShared } from "./shared"
+
+const ADAPTER = "openai-responses"

 export type OpenAIResponsesModelInput = Omit<ModelInput, "provider" | "protocol" | "headers"> & {
  readonly apiKey?: string
@@ -111,9 +112,6 @@ const OpenAIResponsesChunk = Schema.Struct({
 })
 type OpenAIResponsesChunk = Schema.Schema.Type<typeof OpenAIResponsesChunk>

-const Json = Schema.fromJsonString(Schema.Unknown)
-const decodeJson = Schema.decodeUnknownSync(Json)
-const encodeJson = Schema.encodeSync(Json)
 const OpenAIResponsesChunkJson = Schema.fromJsonString(OpenAIResponsesChunk)
 const OpenAIResponsesTargetJson = Schema.fromJsonString(OpenAIResponsesTarget)
 const decodeChunk = Schema.decodeUnknownSync(OpenAIResponsesChunkJson)
@@ -138,7 +136,7 @@ const text = (values: ReadonlyArray<{ readonly text: string }>) => values.map((p

 const resultText = (part: ToolResultPart) => {
  if (part.result.type === "text" || part.result.type === "error") return String(part.result.value)
-  return encodeJson(part.result.value)
+  return ProviderShared.encodeJson(part.result.value)
 }

 const lowerTool = (tool: ToolDefinition): OpenAIResponsesTool => ({
@@ -162,7 +160,7 @@ const lowerToolCall = (part: ToolCallPart): OpenAIResponsesInputItem => ({
  type: "function_call",
  call_id: part.id,
  name: part.name,
-  arguments: encodeJson(part.input),
+  arguments: ProviderShared.encodeJson(part.input),
 })

 const lowerMessages = Effect.fn("OpenAIResponses.lowerMessages")(function* (request: LLMRequest) {
@@ -252,33 +250,9 @@ const mapFinishReason = (chunk: OpenAIResponsesChunk): FinishReason => {
  return "unknown"
 }

-const chunkError = (message: string, raw?: string) => new ProviderChunkError({ adapter: "openai-responses", message, raw })
-
-const streamError = (cause: Cause.Cause<unknown>) => {
-  const failed = cause.reasons.find(Cause.isFailReason)?.error
-  if (failed instanceof ProviderChunkError) return failed
-  return chunkError("Failed to read OpenAI Responses stream", Cause.pretty(cause))
-}
-
-const parseJson = (input: string, message: string) => {
-  try {
-    return decodeJson(input)
-  } catch {
-    throw chunkError(message, input)
-  }
-}
-
-const parseChunk = (data: string) => {
-  try {
-    return decodeChunk(data)
-  } catch {
-    throw chunkError("Invalid OpenAI Responses stream chunk", data)
-  }
-}
-
 const pushToolDelta = (tools: Record<string, ToolAccumulator>, itemId: string, delta: string) => {
  const current = tools[itemId]
-  if (!current) throw chunkError("OpenAI Responses tool argument delta is missing its tool call")
+  if (!current) throw ProviderShared.chunkError(ADAPTER, "OpenAI Responses tool argument delta is missing its tool call")
  return {
    ...current,
    input: `${current.input}${delta}`,
@@ -292,7 +266,7 @@ const finishToolCall = (tools: Record<string, ToolAccumulator>, item: NonNullabl
    type: "tool-call" as const,
    id: item.call_id,
    name: item.name,
-    input: parseJson(input || "{}", `Invalid JSON input for OpenAI Responses tool call ${item.name}`),
+    input: ProviderShared.parseJson(ADAPTER, input || "{}", `Invalid JSON input for OpenAI Responses tool call ${item.name}`),
  }]
 }

@@ -337,24 +311,18 @@ const processChunk = (state: ParserState, chunk: OpenAIResponsesChunk): readonly
 }

 const events = (response: HttpClientResponse.HttpClientResponse) =>
-  response.stream.pipe(
-    Stream.mapError((error) => chunkError("Failed to read OpenAI Responses stream", String(error))),
-    Stream.decodeText(),
-    Stream.pipeThroughChannel(Sse.decode()),
-    Stream.filter((event) => event.data.length > 0 && event.data !== "[DONE]"),
-    Stream.mapEffect((event) =>
-      Effect.try({
-        try: () => parseChunk(event.data),
-        catch: (error) =>
-          error instanceof ProviderChunkError ? error : chunkError("Invalid OpenAI Responses stream chunk", event.data),
-      }),
-    ),
-    Stream.mapAccum((): ParserState => ({ tools: {} }), processChunk),
-    Stream.catchCause((cause) => Stream.fail(streamError(cause))),
-  )
+  ProviderShared.sse({
+    adapter: ADAPTER,
+    response,
+    readError: "Failed to read OpenAI Responses stream",
+    invalidChunk: "Invalid OpenAI Responses stream chunk",
+    decodeChunk,
+    initial: (): ParserState => ({ tools: {} }),
+    process: processChunk,
+  })

 export const adapter = Adapter.define<OpenAIResponsesDraft, OpenAIResponsesTarget, LLMEvent>({
-  id: "openai-responses",
+  id: ADAPTER,
  protocol: "openai-responses",
  redact: (target) => target,
  prepare,
@@ -0,0 +1,53 @@
+import { Cause, Effect, Schema, Stream } from "effect"
+import * as Sse from "effect/unstable/encoding/Sse"
+import type { HttpClientResponse } from "effect/unstable/http"
+import { ProviderChunkError } from "../schema"
+
+export const Json = Schema.fromJsonString(Schema.Unknown)
+export const decodeJson = Schema.decodeUnknownSync(Json)
+export const encodeJson = Schema.encodeSync(Json)
+
+export const chunkError = (adapter: string, message: string, raw?: string) =>
+  new ProviderChunkError({ adapter, message, raw })
+
+export const parseJson = (adapter: string, input: string, message: string) => {
+  try {
+    return decodeJson(input)
+  } catch {
+    throw chunkError(adapter, message, input)
+  }
+}
+
+const streamError = (adapter: string, message: string, cause: Cause.Cause<unknown>) => {
+  const failed = cause.reasons.find(Cause.isFailReason)?.error
+  if (failed instanceof ProviderChunkError) return failed
+  return chunkError(adapter, message, Cause.pretty(cause))
+}
+
+export const sse = <Chunk, State, Event>(input: {
+  readonly adapter: string
+  readonly response: HttpClientResponse.HttpClientResponse
+  readonly readError: string
+  readonly invalidChunk: string
+  readonly decodeChunk: (data: string) => Chunk
+  readonly initial: () => State
+  readonly process: (state: State, chunk: Chunk) => readonly [State, ReadonlyArray<Event>]
+  readonly onHalt?: (state: State) => ReadonlyArray<Event>
+}): Stream.Stream<Event, ProviderChunkError> =>
+  input.response.stream.pipe(
+    Stream.mapError((error) => chunkError(input.adapter, input.readError, String(error))),
+    Stream.decodeText(),
+    Stream.pipeThroughChannel(Sse.decode()),
+    Stream.filter((event) => event.data.length > 0 && event.data !== "[DONE]"),
+    Stream.mapEffect((event) =>
+      Effect.try({
+        try: () => input.decodeChunk(event.data),
+        catch: (error) =>
+          error instanceof ProviderChunkError ? error : chunkError(input.adapter, input.invalidChunk, event.data),
+      }),
+    ),
+    Stream.mapAccum(input.initial, input.process, input.onHalt ? { onHalt: input.onHalt } : undefined),
+    Stream.catchCause((cause) => Stream.fail(streamError(input.adapter, input.readError, cause))),
+  )
+
+export * as ProviderShared from "./shared"