feat(ai): add opt-in raw provider events to LLM streams (#59382)

packages/ai/README.md

@@ -50,6 +50,26 @@ Run `LLM.stream(...)` instead of `generate` when you want incremental `LLMEvent`
 `LLM.request(...)`. The event stream is provider-neutral — same shape across OpenAI Chat, OpenAI Responses,
 Anthropic Messages, Gemini, Bedrock Converse, and any OpenAI-compatible deployment.
 
+### Raw provider events
+
+Set `includeRaw: true` on a request to also receive every provider frame as `{ type: "raw", data }`, in order and
+before the events parsed from that frame. `data` is the parsed JSON frame, so frames this package doesn't handle yet
+are visible too. Raw events are off by default, and `generate` ignores them. Translate a frame with a stream map:
+
+```ts
+import { Stream } from "effect"
+import { LLM, LLMEvent } from "@opencode/ai/effect"
+
+const flagged = (data: unknown) =>
+  typeof data === "object" && data !== null && "type" in data && data.type === "moderation.flagged"
+
+const events = LLM.stream({ model, prompt: "Say hello.", includeRaw: true }).pipe(
+  Stream.map((event) =>
+    event.type === "raw" && flagged(event.data) ? LLMEvent.providerError({ message: "Flagged by moderation" }) : event,
+  ),
+)
+```
+
 ### Google Interactions
 
 `Google.configure({ apiKey }).interactions(modelID)` selects the Interactions API; `.model(modelID)` still selects

packages/ai/src/protocols/bedrock-event-stream.ts

@@ -135,6 +135,7 @@ const consumeFrames = (route: string) => (state: FrameBufferState, input: FrameI
 export const framing = (route: string): Framing.Definition<Frame> => ({
   id: "aws-event-stream",
   body: (frame) => frame.rawBody,
+  raw: ({ rawBody: _, ...event }) => event,
   frame: (bytes) =>
     bytes.pipe(
       Stream.map((bytes): FrameInput => ({ _tag: "Chunk", bytes })),

packages/ai/src/route/client.ts

@@ -1,4 +1,4 @@
-import { Cause, Context, Effect, Layer, Schema, Stream } from "effect"
+import { Cause, Context, Effect, Layer, Option, Schema, Stream } from "effect"
 import { Auth } from "./auth.js"
 import { Endpoint, type EndpointPatch } from "./endpoint.js"
 import { RequestExecutor } from "./executor.js"
@@ -335,7 +335,8 @@ const requireTerminalEvent = (route: string) => (events: Stream.Stream<LLMEvent,
     let terminal = false
     return events.pipe(
       Stream.mapEffect((event) => {
-        if (terminal)
+        // Raw events carry provider frames that may trail the terminal event.
+        if (terminal && !LLMEvent.is.raw(event))
           return Effect.fail(
             ProviderShared.eventError(route, `Provider emitted ${event.type} after the terminal event`),
           )
@@ -346,6 +347,19 @@ const requireTerminalEvent = (route: string) => (events: Stream.Stream<LLMEvent,
     )
   })
 
+const decodeJsonOption = Schema.decodeUnknownOption(ProviderShared.Json)
+
+// JSON text frames are parsed; non-JSON sentinels such as `[DONE]` stay strings.
+const rawFrame = (frame: unknown) =>
+  typeof frame === "string" ? Option.getOrElse(decodeJsonOption(frame), () => frame) : frame
+
+// Transports may interleave their own marker objects with provider frames, e.g. Gemini's continuation boundary.
+const isProviderFrame = (frame: unknown) =>
+  typeof frame !== "object" ||
+  frame === null ||
+  Array.isArray(frame) ||
+  Object.getPrototypeOf(frame) === Object.prototype
+
 function makeFromTransport<Body, Prepared, Frame, Event, State>(
   input: MakeTransportInput<Body, Prepared, Frame, Event, State>,
 ): Route<Body, Prepared> {
@@ -456,7 +470,9 @@ function makeFromTransport<Body, Prepared, Frame, Event, State>(
                       ),
                       Effect.map(([next, output]) => {
                         state = next
-                        return output
+                        return request.includeRaw && isProviderFrame(frame)
+                          ? [LLMEvent.raw({ data: execution.raw?.(frame) ?? rawFrame(frame) }), ...output]
+                          : output
                       }),
                       Effect.mapError(frameError(frame, event)),
                     ),

packages/ai/src/route/framing.ts

@@ -23,6 +23,8 @@ export interface Definition<Frame> {
   readonly frame: (bytes: Stream.Stream<Uint8Array, AIError>) => Stream.Stream<Frame, AIError>
   /** Original wire representation when framing transforms the provider payload. */
   readonly body?: (frame: Frame) => string | undefined
+  /** Provider value exposed as a `raw` event when framing adds fields of its own. Defaults to the frame. */
+  readonly raw?: (frame: Frame) => unknown
 }
 
 /**

packages/ai/src/route/transport/http.ts

@@ -131,6 +131,7 @@ export const httpJson = <Body, Frame>(input: HttpJsonInput<Body, Frame>): HttpJs
         ),
         http,
         body: prepared.framing.body,
+        raw: prepared.framing.raw,
       }
     }),
 })

packages/ai/src/route/transport/index.ts

@@ -13,6 +13,7 @@ export interface TransportExecution<Frame> {
   readonly frames: Stream.Stream<Frame, AIError>
   readonly http?: HttpContext
   body?(frame: Frame): string | undefined
+  raw?(frame: Frame): unknown
   /** Optional successful-consumption acknowledgement. HTTP leaves this absent. */
   readonly complete?: Effect.Effect<void>
 }

packages/ai/src/schema/events.ts

@@ -310,6 +310,13 @@ export const ProviderErrorEvent = Schema.Struct({
 }).annotate({ identifier: "LLM.Event.ProviderError" })
 export type ProviderErrorEvent = Schema.Schema.Type<typeof ProviderErrorEvent>
 
+/** One provider frame as received, emitted before the events parsed from it when the request sets `includeRaw`. */
+export const RawEvent = Schema.Struct({
+  type: Schema.tag("raw"),
+  data: Schema.Unknown,
+}).annotate({ identifier: "LLM.Event.Raw" })
+export type RawEvent = Schema.Schema.Type<typeof RawEvent>
+
 const llmEventTagged = Schema.Union([
   CompactionPart,
   StepStart,
@@ -330,6 +337,7 @@ const llmEventTagged = Schema.Union([
   StepFinish,
   Finish,
   ProviderErrorEvent,
+  RawEvent,
 ]).pipe(Schema.toTaggedUnion("type"))
 
 type WithID<Event extends { readonly id: unknown }, ID> = Omit<Event, "type" | "id"> & { readonly id: ID | string }
@@ -384,6 +392,7 @@ export const LLMEvent = Object.assign(llmEventTagged, {
       usage: input.usage === undefined ? undefined : Usage.from(input.usage),
     }),
   providerError: ProviderErrorEvent.make,
+  raw: RawEvent.make,
   is: {
     compaction: llmEventTagged.guards.compaction,
     stepStart: llmEventTagged.guards["step-start"],
@@ -404,6 +413,7 @@ export const LLMEvent = Object.assign(llmEventTagged, {
     stepFinish: llmEventTagged.guards["step-finish"],
     finish: llmEventTagged.guards.finish,
     providerError: llmEventTagged.guards["provider-error"],
+    raw: llmEventTagged.guards.raw,
   },
 })
 export type LLMEvent = Schema.Schema.Type<typeof llmEventTagged>
@@ -646,6 +656,7 @@ const reduceToolCall = (state: ResponseState, event: ToolCall): ResponseState =>
 }
 
 const reduceResponseState = (state: ResponseState, event: LLMEvent): ResponseState => {
+  if (event.type === "raw") return state
   const next = appendEvent(state, event)
   switch (event.type) {
     case "compaction":

packages/ai/src/schema/messages.ts

@@ -395,6 +395,8 @@ const requestSchema = Schema.Struct({
   // Stable cache affinity for protocols that support provider-managed prompt caching.
   promptCacheKey: Schema.optional(Schema.String),
   metadata: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)),
+  /** Emit a `raw` event with each provider frame before the events parsed from it. */
+  includeRaw: Schema.optional(Schema.Boolean),
 })
 
 export class LLMRequest<Model extends LanguageModel = LanguageModel> extends Schema.Class<LLMRequest>("LLM.Request")(
@@ -427,6 +429,7 @@ export namespace LLMRequest {
     cache: request.cache,
     promptCacheKey: request.promptCacheKey,
     metadata: request.metadata,
+    includeRaw: request.includeRaw,
   })
 
   export function update<Model extends LanguageModel>(

packages/ai/test/provider/bedrock-converse.test.ts

@@ -849,6 +849,28 @@ describe("Bedrock Converse route", () => {
     }),
   )
 
+  it.effect("raw events carry the event payload without decoder fields or padding", () =>
+    Effect.gen(function* () {
+      const body = eventStreamBody(
+        ["contentBlockDelta", { contentBlockIndex: 0, delta: { text: "Hi" }, p: "abc" }],
+        ["messageStop", { stopReason: "end_turn" }],
+        ["metadata", { usage: { inputTokens: 5, outputTokens: 1, totalTokens: 6 } }],
+      )
+      const events = Array.from(
+        yield* LLMClient.stream(LLMRequest.update(baseRequest, { includeRaw: true })).pipe(
+          Stream.runCollect,
+          Effect.provide(fixedBytes(body)),
+        ),
+      )
+
+      expect(events.filter((event) => event.type === "raw").map((event) => event.data)).toEqual([
+        { contentBlockDelta: { contentBlockIndex: 0, delta: { text: "Hi" } } },
+        { messageStop: { stopReason: "end_turn" } },
+        { metadata: { usage: { inputTokens: 5, outputTokens: 1, totalTokens: 6 } } },
+      ])
+    }),
+  )
+
   it.effect("rejects truncated event-stream frames after message stop", () =>
     Effect.gen(function* () {
       const partialFrames = [

packages/ai/test/raw-events.test.ts

@@ -0,0 +1,75 @@
+import { describe, expect } from "bun:test"
+import { Effect, Stream } from "effect"
+import { LLM } from "../src/effect.js"
+import { Auth, LLMClient } from "../src/route.js"
+import { AnthropicMessages } from "../src/protocols/anthropic-messages.js"
+import { it } from "./lib/effect.js"
+import { fixedResponse } from "./lib/http.js"
+import { sseEvents } from "./lib/sse.js"
+
+const model = AnthropicMessages.route
+  .with({ endpoint: { baseURL: "https://api.anthropic.test/v1/" }, auth: Auth.header("x-api-key", "test") })
+  .model({ id: "claude-sonnet-4-5" })
+
+const frames = [
+  { type: "message_start", message: { usage: { input_tokens: 5 } } },
+  { type: "content_block_start", index: 0, content_block: { type: "text", text: "" } },
+  { type: "future_event", detail: "unknown to the protocol" },
+  { type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "Hello" } },
+  { type: "content_block_stop", index: 0 },
+  { type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 1 } },
+  { type: "message_stop" },
+  { type: "ping" },
+]
+
+const layer = fixedResponse(sseEvents(...frames))
+
+describe("raw provider events", () => {
+  it.effect("stream emits every frame before its own events only when requested", () =>
+    Effect.gen(function* () {
+      const raw = Array.from(
+        yield* LLMClient.stream(LLM.request({ model, prompt: "Hi", includeRaw: true })).pipe(
+          Stream.runCollect,
+          Effect.provide(layer),
+        ),
+      )
+      const plain = Array.from(
+        yield* LLMClient.stream(LLM.request({ model, prompt: "Hi" })).pipe(Stream.runCollect, Effect.provide(layer)),
+      )
+
+      expect(raw.filter((event) => event.type === "raw").map((event) => event.data)).toEqual(frames)
+      expect(
+        raw.map((event) => (event.type === "raw" ? `raw:${(event.data as { type: string }).type}` : event.type)),
+      ).toEqual([
+        "raw:message_start",
+        "raw:content_block_start",
+        "step-start",
+        "text-start",
+        "raw:future_event",
+        "raw:content_block_delta",
+        "text-delta",
+        "raw:content_block_stop",
+        "text-end",
+        "raw:message_delta",
+        "raw:message_stop",
+        "step-finish",
+        "finish",
+        "raw:ping",
+      ])
+      expect(raw.filter((event) => event.type !== "raw")).toEqual(plain)
+      expect(plain.some((event) => event.type === "raw")).toBe(false)
+    }),
+  )
+
+  it.effect("generate ignores raw events", () =>
+    Effect.gen(function* () {
+      const withRaw = yield* LLMClient.generate(LLM.request({ model, prompt: "Hi", includeRaw: true })).pipe(
+        Effect.provide(layer),
+      )
+      const plain = yield* LLMClient.generate(LLM.request({ model, prompt: "Hi" })).pipe(Effect.provide(layer))
+
+      expect(withRaw).toEqual(plain)
+      expect(withRaw.text).toBe("Hello")
+    }),
+  )
+})