From 93701210bd867c6efc18ab1c87cbc27bd6dd971b Mon Sep 17 00:00:00 2001 From: Aiden Cline <63023139+rekram1-node@users.noreply.github.com> Date: Mon, 26 Oct 2026 16:49:35 -0500 Subject: [PATCH] feat(ai): add opt-in raw provider events to LLM streams (#59382) --- packages/ai/README.md | 20 +++++ .../ai/src/protocols/bedrock-event-stream.ts | 1 + packages/ai/src/route/client.ts | 22 +++++- packages/ai/src/route/framing.ts | 2 + packages/ai/src/route/transport/http.ts | 1 + packages/ai/src/route/transport/index.ts | 1 + packages/ai/src/schema/events.ts | 11 +++ packages/ai/src/schema/messages.ts | 3 + .../ai/test/provider/bedrock-converse.test.ts | 22 ++++++ packages/ai/test/raw-events.test.ts | 75 +++++++++++++++++++ 10 files changed, 155 insertions(+), 3 deletions(-) create mode 100644 packages/ai/test/raw-events.test.ts diff --git a/packages/ai/README.md b/packages/ai/README.md index 612c76bca29..ebfb9ce4ed4 100644 --- a/packages/ai/README.md +++ b/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 diff --git a/packages/ai/src/protocols/bedrock-event-stream.ts b/packages/ai/src/protocols/bedrock-event-stream.ts index 6a385b442be..f730f115a34 100644 --- a/packages/ai/src/protocols/bedrock-event-stream.ts +++ b/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 => ({ id: "aws-event-stream", body: (frame) => frame.rawBody, + raw: ({ rawBody: _, ...event }) => event, frame: (bytes) => bytes.pipe( Stream.map((bytes): FrameInput => ({ _tag: "Chunk", bytes })), diff --git a/packages/ai/src/route/client.ts b/packages/ai/src/route/client.ts index 60f41f62b1c..6b763f9d77c 100644 --- a/packages/ai/src/route/client.ts +++ b/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 { - 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 + 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( input: MakeTransportInput, ): Route { @@ -456,7 +470,9 @@ function makeFromTransport( ), 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)), ), diff --git a/packages/ai/src/route/framing.ts b/packages/ai/src/route/framing.ts index 4ff92c2abc6..ea2889e5c77 100644 --- a/packages/ai/src/route/framing.ts +++ b/packages/ai/src/route/framing.ts @@ -23,6 +23,8 @@ export interface Definition { readonly frame: (bytes: Stream.Stream) => Stream.Stream /** 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 } /** diff --git a/packages/ai/src/route/transport/http.ts b/packages/ai/src/route/transport/http.ts index a9385338130..a2080994e5d 100644 --- a/packages/ai/src/route/transport/http.ts +++ b/packages/ai/src/route/transport/http.ts @@ -131,6 +131,7 @@ export const httpJson = (input: HttpJsonInput): HttpJs ), http, body: prepared.framing.body, + raw: prepared.framing.raw, } }), }) diff --git a/packages/ai/src/route/transport/index.ts b/packages/ai/src/route/transport/index.ts index 9690392dcc8..27d22e371d0 100644 --- a/packages/ai/src/route/transport/index.ts +++ b/packages/ai/src/route/transport/index.ts @@ -13,6 +13,7 @@ export interface TransportExecution { readonly frames: Stream.Stream readonly http?: HttpContext body?(frame: Frame): string | undefined + raw?(frame: Frame): unknown /** Optional successful-consumption acknowledgement. HTTP leaves this absent. */ readonly complete?: Effect.Effect } diff --git a/packages/ai/src/schema/events.ts b/packages/ai/src/schema/events.ts index 3f05398357e..adcd037e496 100644 --- a/packages/ai/src/schema/events.ts +++ b/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 +/** 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 + const llmEventTagged = Schema.Union([ CompactionPart, StepStart, @@ -330,6 +337,7 @@ const llmEventTagged = Schema.Union([ StepFinish, Finish, ProviderErrorEvent, + RawEvent, ]).pipe(Schema.toTaggedUnion("type")) type WithID = Omit & { 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 @@ -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": diff --git a/packages/ai/src/schema/messages.ts b/packages/ai/src/schema/messages.ts index a9359bef1d0..ac3bfd451aa 100644 --- a/packages/ai/src/schema/messages.ts +++ b/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 extends Schema.Class("LLM.Request")( @@ -427,6 +429,7 @@ export namespace LLMRequest { cache: request.cache, promptCacheKey: request.promptCacheKey, metadata: request.metadata, + includeRaw: request.includeRaw, }) export function update( diff --git a/packages/ai/test/provider/bedrock-converse.test.ts b/packages/ai/test/provider/bedrock-converse.test.ts index a7d09a2e983..d2dcada6319 100644 --- a/packages/ai/test/provider/bedrock-converse.test.ts +++ b/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 = [ diff --git a/packages/ai/test/raw-events.test.ts b/packages/ai/test/raw-events.test.ts new file mode 100644 index 00000000000..5748eb7d3a2 --- /dev/null +++ b/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") + }), + ) +})