Add explicit LLM stream lifecycle events (#26722)

Author: kitlangton

packages/llm/example/tutorial.ts

@@ -184,7 +184,7 @@ const FakeProtocol = Protocol.make<FakeBody, string, string, void>({
   stream: {
     event: Schema.String,
     initial: () => undefined,
-    step: (_, frame) => Effect.succeed([undefined, [{ type: "text-delta", text: frame }]] as const),
+    step: (_, frame) => Effect.succeed([undefined, [{ type: "text-delta", id: "text-0", text: frame }]] as const),
     onHalt: () => [{ type: "request-finish", reason: "stop" }],
   },
 })

packages/llm/src/protocols/anthropic-messages.ts

@@ -5,10 +5,10 @@ import { Endpoint } from "../route/endpoint"
 import { Framing } from "../route/framing"
 import { Protocol } from "../route/protocol"
 import {
+  LLMEvent,
   Usage,
   type CacheHint,
   type FinishReason,
-  type LLMEvent,
   type LLMRequest,
   type ProviderMetadata,
   type ToolCallPart,
@@ -415,14 +415,13 @@ const serverToolResultEvent = (block: NonNullable<AnthropicEvent["content_block"
       ? String((block.content as Record<string, unknown>).type)
       : ""
   const isError = errorPayload.endsWith("_tool_result_error")
-  return {
-    type: "tool-result",
+  return LLMEvent.toolResult({
     id: block.tool_use_id ?? "",
     name: SERVER_TOOL_RESULT_NAMES[block.type],
     result: isError ? { type: "error", value: block.content } : { type: "json", value: block.content },
     providerExecuted: true,
     providerMetadata: anthropicMetadata({ blockType: block.type }),
-  }
+  })
 }
 
 type StepResult = readonly [ParserState, ReadonlyArray<LLMEvent>]
@@ -453,18 +452,17 @@ const onContentBlockStart = (state: ParserState, event: AnthropicEvent): StepRes
   }
 
   if (block.type === "text" && block.text) {
-    return [state, [{ type: "text-delta", text: block.text }]]
+    return [state, [LLMEvent.textDelta({ id: `text-${event.index ?? 0}`, text: block.text })]]
   }
 
   if (block.type === "thinking" && block.thinking) {
     return [
       state,
       [
-        {
-          type: "reasoning-delta",
+        LLMEvent.reasoningDelta({
+          id: `reasoning-${event.index ?? 0}`,
           text: block.thinking,
-          ...(block.signature ? { providerMetadata: anthropicMetadata({ signature: block.signature }) } : {}),
-        },
+        }),
       ],
     ]
   }
@@ -480,17 +478,17 @@ const onContentBlockDelta = Effect.fn("AnthropicMessages.onContentBlockDelta")(f
   const delta = event.delta
 
   if (delta?.type === "text_delta" && delta.text) {
-    return [state, [{ type: "text-delta", text: delta.text }]] satisfies StepResult
+    return [state, [LLMEvent.textDelta({ id: `text-${event.index ?? 0}`, text: delta.text })]] satisfies StepResult
   }
 
   if (delta?.type === "thinking_delta" && delta.thinking) {
-    return [state, [{ type: "reasoning-delta", text: delta.thinking }]] satisfies StepResult
+    return [state, [LLMEvent.reasoningDelta({ id: `reasoning-${event.index ?? 0}`, text: delta.thinking })]] satisfies StepResult
   }
 
   if (delta?.type === "signature_delta" && delta.signature) {
     return [
       state,
-      [{ type: "reasoning-delta", text: "", providerMetadata: anthropicMetadata({ signature: delta.signature }) }],
+      [LLMEvent.reasoningEnd({ id: `reasoning-${event.index ?? 0}`, providerMetadata: anthropicMetadata({ signature: delta.signature }) })],
     ] satisfies StepResult
   }
 
@@ -524,21 +522,20 @@ const onMessageDelta = (state: ParserState, event: AnthropicEvent): StepResult =
   return [
     { ...state, usage },
     [
-      {
-        type: "request-finish",
+      LLMEvent.requestFinish({
         reason: mapFinishReason(event.delta?.stop_reason),
         usage,
-        ...(event.delta?.stop_sequence
-          ? { providerMetadata: anthropicMetadata({ stopSequence: event.delta.stop_sequence }) }
-          : {}),
-      },
+        providerMetadata: event.delta?.stop_sequence
+          ? anthropicMetadata({ stopSequence: event.delta.stop_sequence })
+          : undefined,
+      }),
     ],
   ]
 }
 
 const onError = (state: ParserState, event: AnthropicEvent): StepResult => [
   state,
-  [{ type: "provider-error", message: event.error?.message ?? "Anthropic Messages stream error" }],
+  [LLMEvent.providerError({ message: event.error?.message ?? "Anthropic Messages stream error" })],
 ]
 
 const step = (state: ParserState, event: AnthropicEvent) => {

packages/llm/src/protocols/bedrock-converse.ts

@@ -3,10 +3,10 @@ import { Route, type RouteModelInput } from "../route/client"
 import { Endpoint } from "../route/endpoint"
 import { Protocol } from "../route/protocol"
 import {
+  LLMEvent,
   Usage,
   type CacheHint,
   type FinishReason,
-  type LLMEvent,
   type LLMRequest,
   type ToolCallPart,
   type ToolDefinition,
@@ -400,13 +400,26 @@ const step = (state: ParserState, event: BedrockEvent) =>
     }
 
     if (event.contentBlockDelta?.delta?.text) {
-      return [state, [{ type: "text-delta" as const, text: event.contentBlockDelta.delta.text }]] as const
+      return [
+        state,
+        [
+          LLMEvent.textDelta({
+            id: `text-${event.contentBlockDelta.contentBlockIndex}`,
+            text: event.contentBlockDelta.delta.text,
+          }),
+        ],
+      ] as const
     }
 
     if (event.contentBlockDelta?.delta?.reasoningContent?.text) {
       return [
         state,
-        [{ type: "reasoning-delta" as const, text: event.contentBlockDelta.delta.reasoningContent.text }],
+        [
+          LLMEvent.reasoningDelta({
+            id: `reasoning-${event.contentBlockDelta.contentBlockIndex}`,
+            text: event.contentBlockDelta.delta.reasoningContent.text,
+          }),
+        ],
       ] as const
     }
 
@@ -449,15 +462,15 @@ const step = (state: ParserState, event: BedrockEvent) =>
         event.modelStreamErrorException?.message ??
         event.serviceUnavailableException?.message ??
         "Bedrock Converse stream error"
-      return [state, [{ type: "provider-error" as const, message, retryable: true }]] as const
+      return [state, [LLMEvent.providerError({ message, retryable: true })]] as const
     }
 
     if (event.validationException || event.throttlingException) {
       const message =
         event.validationException?.message ?? event.throttlingException?.message ?? "Bedrock Converse error"
       return [
         state,
-        [{ type: "provider-error" as const, message, retryable: event.throttlingException !== undefined }],
+        [LLMEvent.providerError({ message, retryable: event.throttlingException !== undefined })],
       ] as const
     }
 
@@ -468,7 +481,7 @@ const framing = BedrockEventStream.framing(ADAPTER)
 
 const onHalt = (state: ParserState): ReadonlyArray<LLMEvent> =>
   state.pendingFinish
-    ? [{ type: "request-finish", reason: state.pendingFinish.reason, usage: state.pendingFinish.usage }]
+    ? [LLMEvent.requestFinish({ reason: state.pendingFinish.reason, usage: state.pendingFinish.usage })]
     : []
 
 // =============================================================================

packages/llm/src/protocols/gemini.ts

@@ -5,9 +5,9 @@ import { Endpoint } from "../route/endpoint"
 import { Framing } from "../route/framing"
 import { Protocol } from "../route/protocol"
 import {
+  LLMEvent,
   Usage,
   type FinishReason,
-  type LLMEvent,
   type LLMRequest,
   type MediaPart,
   type TextPart,
@@ -311,7 +311,7 @@ const mapFinishReason = (finishReason: string | undefined, hasToolCalls: boolean
 
 const finish = (state: ParserState): ReadonlyArray<LLMEvent> =>
   state.finishReason || state.usage
-    ? [{ type: "request-finish", reason: mapFinishReason(state.finishReason, state.hasToolCalls), usage: state.usage }]
+    ? [LLMEvent.requestFinish({ reason: mapFinishReason(state.finishReason, state.hasToolCalls), usage: state.usage })]
     : []
 
 const step = (state: ParserState, event: GeminiEvent) => {
@@ -332,14 +332,18 @@ const step = (state: ParserState, event: GeminiEvent) => {
 
   for (const part of candidate.content.parts) {
     if ("text" in part && part.text.length > 0) {
-      events.push({ type: part.thought ? "reasoning-delta" : "text-delta", text: part.text })
+      events.push(
+        part.thought
+          ? LLMEvent.reasoningDelta({ id: "reasoning-0", text: part.text })
+          : LLMEvent.textDelta({ id: "text-0", text: part.text }),
+      )
       continue
     }
 
     if ("functionCall" in part) {
       const input = part.functionCall.args
       const id = `tool_${nextToolCallId++}`
-      events.push({ type: "tool-call", id, name: part.functionCall.name, input })
+      events.push(LLMEvent.toolCall({ id, name: part.functionCall.name, input }))
       hasToolCalls = true
     }
   }

packages/llm/src/protocols/openai-chat.ts

@@ -6,9 +6,9 @@ import { Framing } from "../route/framing"
 import { HttpTransport } from "../route/transport"
 import { Protocol } from "../route/protocol"
 import {
+  LLMEvent,
   Usage,
   type FinishReason,
-  type LLMEvent,
   type LLMRequest,
   type TextPart,
   type ToolCallPart,
@@ -312,7 +312,7 @@ const step = (state: ParserState, event: OpenAIChatEvent) =>
     const toolDeltas = delta?.tool_calls ?? []
     let tools = state.tools
 
-    if (delta?.content) events.push({ type: "text-delta", text: delta.content })
+    if (delta?.content) events.push(LLMEvent.textDelta({ id: "text-0", text: delta.content }))
 
     for (const tool of toolDeltas) {
       const result = ToolStream.appendOrStart(
@@ -350,7 +350,7 @@ const finishEvents = (state: ParserState): ReadonlyArray<LLMEvent> => {
   const reason = state.finishReason === "stop" && hasToolCalls ? "tool-calls" : state.finishReason
   return [
     ...state.toolCallEvents,
-    ...(reason ? ([{ type: "request-finish", reason, usage: state.usage }] satisfies ReadonlyArray<LLMEvent>) : []),
+    ...(reason ? [LLMEvent.requestFinish({ reason, usage: state.usage })] : []),
   ]
 }
 

packages/llm/src/protocols/openai-responses.ts

@@ -6,9 +6,9 @@ import { Framing } from "../route/framing"
 import { HttpTransport, WebSocketTransport } from "../route/transport"
 import { Protocol } from "../route/protocol"
 import {
+  LLMEvent,
   Usage,
   type FinishReason,
-  type LLMEvent,
   type LLMRequest,
   type ProviderMetadata,
   type TextPart,
@@ -348,22 +348,20 @@ const hostedToolEvents = (
   const tool = HOSTED_TOOLS[item.type]
   const providerMetadata = openaiMetadata({ itemId: item.id })
   return [
-    {
-      type: "tool-call",
+    LLMEvent.toolCall({
       id: item.id,
       name: tool.name,
       input: tool.input(item),
       providerExecuted: true,
       providerMetadata,
-    },
-    {
-      type: "tool-result",
+    }),
+    LLMEvent.toolResult({
       id: item.id,
       name: tool.name,
       result: hostedToolResult(item),
       providerExecuted: true,
       providerMetadata,
-    },
+    }),
   ]
 }
 
@@ -382,12 +380,7 @@ const onOutputTextDelta = (state: ParserState, event: OpenAIResponsesEvent): Ste
   return [
     state,
     [
-      {
-        type: "text-delta",
-        id: event.item_id,
-        text: event.delta,
-        ...(event.item_id ? { providerMetadata: openaiMetadata({ itemId: event.item_id }) } : {}),
-      },
+      LLMEvent.textDelta({ id: event.item_id ?? "text-0", text: event.delta }),
     ],
   ]
 }
@@ -458,30 +451,28 @@ const onOutputItemDone = Effect.fn("OpenAIResponses.onOutputItemDone")(function*
 const onResponseFinish = (state: ParserState, event: OpenAIResponsesEvent): StepResult => [
   state,
   [
-    {
-      type: "request-finish",
+    LLMEvent.requestFinish({
       reason: mapFinishReason(event, state.hasFunctionCall),
       usage: mapUsage(event.response?.usage),
-      ...(event.response?.id || event.response?.service_tier
-        ? {
-            providerMetadata: openaiMetadata({
+      providerMetadata:
+        event.response?.id || event.response?.service_tier
+          ? openaiMetadata({
               responseId: event.response.id,
               serviceTier: event.response.service_tier,
-            }),
-          }
-        : {}),
-    },
+            })
+          : undefined,
+    }),
   ],
 ]
 
 const onResponseFailed = (state: ParserState, event: OpenAIResponsesEvent): StepResult => [
   state,
-  [{ type: "provider-error", message: event.message ?? event.code ?? "OpenAI Responses response failed" }],
+  [LLMEvent.providerError({ message: event.message ?? event.code ?? "OpenAI Responses response failed" })],
 ]
 
 const onError = (state: ParserState, event: OpenAIResponsesEvent): StepResult => [
   state,
-  [{ type: "provider-error", message: event.message ?? event.code ?? "OpenAI Responses stream error" }],
+  [LLMEvent.providerError({ message: event.message ?? event.code ?? "OpenAI Responses stream error" })],
 ]
 
 const step = (state: ParserState, event: OpenAIResponsesEvent) => {

packages/llm/src/protocols/utils/tool-stream.ts

@@ -1,5 +1,5 @@
 import { Effect } from "effect"
-import { LLMError, type ProviderMetadata, type ToolCall, type ToolInputDelta } from "../../schema"
+import { LLMError, LLMEvent, type ProviderMetadata, type ToolCall, type ToolInputDelta } from "../../schema"
 import { eventError, parseToolInput, type ToolAccumulator } from "../shared"
 
 type StreamKey = string | number
@@ -49,34 +49,24 @@ const withoutTool = <K extends StreamKey>(tools: State<K>, key: K): State<K> =>
   return next
 }
 
-const inputDelta = (tool: PendingTool, text: string): ToolInputDelta => ({
-  type: "tool-input-delta",
-  id: tool.id,
-  name: tool.name,
-  text,
-  ...(tool.providerMetadata ? { providerMetadata: tool.providerMetadata } : {}),
-})
+const inputDelta = (tool: PendingTool, text: string): ToolInputDelta =>
+  LLMEvent.toolInputDelta({
+    id: tool.id,
+    name: tool.name,
+    text,
+  })
 
 const toolCall = (route: string, tool: PendingTool, inputOverride?: string) =>
   parseToolInput(route, tool.name, inputOverride ?? tool.input).pipe(
     Effect.map(
       (input): ToolCall =>
-        tool.providerExecuted
-          ? {
-              type: "tool-call",
-              id: tool.id,
-              name: tool.name,
-              input,
-              providerExecuted: true,
-              ...(tool.providerMetadata ? { providerMetadata: tool.providerMetadata } : {}),
-            }
-          : {
-              type: "tool-call",
-              id: tool.id,
-              name: tool.name,
-              input,
-              ...(tool.providerMetadata ? { providerMetadata: tool.providerMetadata } : {}),
-            },
+        LLMEvent.toolCall({
+          id: tool.id,
+          name: tool.name,
+          input,
+          providerExecuted: tool.providerExecuted ? true : undefined,
+          providerMetadata: tool.providerMetadata,
+        }),
     ),
   )