mirrors/opencode
1
0
Fork 0
mirror of https://github.com/anomalyco/opencode.git synced 2026-11-13 19:46:16 +03:00
Code Issues Packages Projects Releases Wiki Activity

refactor(ai): untrace stream event handlers and inner protocol helpers (#53232)

Browse Source
This commit is contained in:
Aiden Cline
2026-10-04 19:43:29 -05:00
committed by GitHub
Unverified
parent 238d4cf5ca
commit 44193bd71f
18 changed files with 36 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/ai/src/protocols/anthropic-messages.ts
+3 -3
View File
@@ -1300,7 +1300,7 @@ const onContentBlockStart = (
return [{ ...state, lifecycle: Lifecycle.stepStart(state.lifecycle, events) }, [...events, result]]
}
const onContentBlockDelta = Effect.fn("AnthropicMessages.onContentBlockDelta")(function* (
const onContentBlockDelta = Effect.fnUntraced(function* (
state: ParserState,
event: AnthropicEvent & { readonly delta: AnthropicStreamDelta },
) {
@@ -1368,7 +1368,7 @@ const onContentBlockDelta = Effect.fn("AnthropicMessages.onContentBlockDelta")(f
return [state, NO_EVENTS] satisfies StepResult
})
const onContentBlockStop = Effect.fn("AnthropicMessages.onContentBlockStop")(function* (
const onContentBlockStop = Effect.fnUntraced(function* (
state: ParserState,
event: AnthropicEvent,
) {
@@ -1439,7 +1439,7 @@ const onMessageDelta = (
]
}
const onMessageStop = Effect.fn("AnthropicMessages.onMessageStop")(function* (state: ParserState) {
const onMessageStop = Effect.fnUntraced(function* (state: ParserState) {
if (Object.keys(state.compactions).length)
return yield* ProviderShared.eventError(ADAPTER, "Response ended with an incomplete compaction block")
const result = yield* ToolStream.finishAll(ADAPTER, state.tools)
packages/ai/src/protocols/cartesia-speech.ts
+3 -3
View File
@@ -75,7 +75,7 @@ const usesSse = (request: MediaProtocol.Addressed<Request>) => request.mode ===
const CONTAINERS: Readonly<Record<string, "raw" | "wav" | "mp3">> = { pcm: "raw", wav: "wav", mp3: "mp3" }
const outputFormat = Effect.fn("CartesiaSpeech.outputFormat")(function* (request: MediaProtocol.Addressed<Request>) {
const outputFormat = Effect.fnUntraced(function* (request: MediaProtocol.Addressed<Request>) {
const sse = usesSse(request)
const format = request.format ?? (sse ? "pcm" : "mp3")
const container = CONTAINERS[format]
@@ -121,7 +121,7 @@ const fromRequest = Effect.fn("CartesiaSpeech.fromRequest")(function* (request:
// 6. Stream parsing
// ---------------------------------------------------------------------------
const onEvent = Effect.fn("CartesiaSpeech.onEvent")(function* (state: State, frame: string) {
const onEvent = Effect.fnUntraced(function* (state: State, frame: string) {
const event = yield* decodeEvent(frame)
if (event.type === "chunk" && event.data !== undefined) return SpeechStream.delta(state, event.data)
if (event.type === "timestamps" && event.word_timestamps !== undefined) {
@@ -140,7 +140,7 @@ const onEvent = Effect.fn("CartesiaSpeech.onEvent")(function* (state: State, fra
return [state, []] as const
})
const finish = Effect.fn("CartesiaSpeech.finish")(function* (
const finish = Effect.fnUntraced(function* (
state: State,
context: MediaProtocol.ResponseContext<Request>,
) {
packages/ai/src/protocols/cohere-chat.ts
+1 -1
View File
@@ -252,7 +252,7 @@ const mapUsage = (usage: typeof NativeUsage.Type) =>
})
// Lifecycle deltas open blocks on demand and ends are no-ops for closed blocks, so content-start needs no handling.
const step = Effect.fn("CohereChat.step")(function* (state: State, event: Event) {
const step = Effect.fnUntraced(function* (state: State, event: Event) {
const events: LLMEvent[] = []
switch (event.type) {
case "message-start":
packages/ai/src/protocols/elevenlabs-speech.ts
+3 -3
View File
@@ -95,7 +95,7 @@ const OUTPUT_FORMATS: Readonly<Record<string, string>> = {
}
/** WAV is served only by the non-streaming endpoints. */
const outputFormat = Effect.fn("ElevenLabsSpeech.outputFormat")(function* (request: MediaProtocol.Addressed<Request>) {
const outputFormat = Effect.fnUntraced(function* (request: MediaProtocol.Addressed<Request>) {
const format = request.providerOptions?.outputFormat ?? OUTPUT_FORMATS[request.format ?? "mp3"]
if (format === undefined)
return yield* route.unsupported(
@@ -138,7 +138,7 @@ const path = (request: MediaProtocol.Addressed<Request>) =>
// 6. Stream parsing
// ---------------------------------------------------------------------------
const onRecord = Effect.fn("ElevenLabsSpeech.onRecord")(function* (state: State, frame: string) {
const onRecord = Effect.fnUntraced(function* (state: State, frame: string) {
const record = yield* decodeRecord(frame)
const [next, events] = SpeechStream.delta(state, record.audio_base64)
const alignment = record.alignment
@@ -169,7 +169,7 @@ const describeOutput = (format: string) => {
return encoding === undefined ? SpeechStream.container(codec, sampleRate) : SpeechStream.pcm(encoding, sampleRate)
}
const finish = Effect.fn("ElevenLabsSpeech.finish")(function* (
const finish = Effect.fnUntraced(function* (
state: State,
context: MediaProtocol.ResponseContext<Request>,
) {
packages/ai/src/protocols/google-interactions.ts
+4 -4
View File
@@ -364,7 +364,7 @@ const mapUsage = (usage: RawUsage | undefined, key: string) => {
})
}
const onStart = Effect.fn("GoogleInteractions.onStart")(function* (
const onStart = Effect.fnUntraced(function* (
state: ParserState,
index: number,
step: OutputStep,
@@ -402,7 +402,7 @@ const onStart = Effect.fn("GoogleInteractions.onStart")(function* (
return [{ ...state, lifecycle, tools, steps: { ...state.steps, [index]: step } }, events] satisfies StepResult
})
const onDelta = Effect.fn("GoogleInteractions.onDelta")(function* (
const onDelta = Effect.fnUntraced(function* (
state: ParserState,
index: number,
delta: typeof Delta.Type,
@@ -447,7 +447,7 @@ const onDelta = Effect.fn("GoogleInteractions.onDelta")(function* (
return yield* ProviderShared.eventError(ADAPTER, `Unsupported Interactions delta: ${delta.type}`, encodeJson(delta))
})
const onStop = Effect.fn("GoogleInteractions.onStop")(function* (state: ParserState, index: number) {
const onStop = Effect.fnUntraced(function* (state: ParserState, index: number) {
const step = state.steps[index]
if (!step) return yield* ProviderShared.eventError(ADAPTER, "Interactions step.stop without step.start")
const events: LLMEvent[] = []
@@ -465,7 +465,7 @@ const onStop = Effect.fn("GoogleInteractions.onStop")(function* (state: ParserSt
return [{ ...state, tools: result.tools }, result.events ?? []] satisfies StepResult
})
const step = Effect.fn("GoogleInteractions.step")(function* (state: ParserState, event: Event) {
const step = Effect.fnUntraced(function* (state: ParserState, event: Event) {
switch (event.event_type) {
case "step.start":
return yield* onStart(state, event.index, event.step)
packages/ai/src/protocols/google-speech.ts
+1 -1
View File
@@ -94,7 +94,7 @@ const fromRequest = Effect.fn("GoogleSpeech.fromRequest")(function* (request: Me
// 6. Stream parsing
// ---------------------------------------------------------------------------
const step = Effect.fn("GoogleSpeech.step")(function* (state: State, frame: string) {
const step = Effect.fnUntraced(function* (state: State, frame: string) {
const chunk = yield* decodeChunk(frame)
const blocked = GeminiGenerateContent.blocked(route.name, chunk, frame)
if (blocked !== undefined) return yield* blocked
packages/ai/src/protocols/google-transcription.ts
+1 -1
View File
@@ -138,7 +138,7 @@ const turn = (part: Schema.Schema.Type<typeof AudioTranscription>) => {
}
}
const step = Effect.fn("GoogleTranscription.step")(function* (state: State, frame: string) {
const step = Effect.fnUntraced(function* (state: State, frame: string) {
const chunk = yield* decodeChunk(frame)
const blocked = GeminiGenerateContent.blocked(route.name, chunk, frame)
if (blocked !== undefined) return yield* blocked
packages/ai/src/protocols/meta-responses.ts
+3 -3
View File
@@ -95,7 +95,7 @@ const HOSTED_TOOLS = {
image_generation_call: {
name: "image_generation",
input: () => ({}),
result: Effect.fn("MetaResponses.imageResult")(function* (raw: ResponsesHostedTools.Item) {
result: Effect.fnUntraced(function* (raw: ResponsesHostedTools.Item) {
const item = yield* Schema.decodeUnknownEffect(ImageItem)(raw).pipe(
Effect.mapError((cause) =>
ProviderShared.eventError(
@@ -136,7 +136,7 @@ const HOSTED_TOOLS = {
},
} satisfies ResponsesHostedTools.Definitions
const onEvent = Effect.fn("MetaResponses.onEvent")(function* (
const onEvent = Effect.fnUntraced(function* (
state: OpenResponses.ParserState,
input: OpenResponses.Event,
) {
@@ -173,7 +173,7 @@ const onEvent = Effect.fn("MetaResponses.onEvent")(function* (
] satisfies OpenResponses.StepResult
})
const step = Effect.fn("MetaResponses.step")(function* (state: ParserState, input: OpenResponses.Event) {
const step = Effect.fnUntraced(function* (state: ParserState, input: OpenResponses.Event) {
const completedItems = new Set(state.completedItems)
const event = OpenResponses.normalize(state, input)
if (event.type === "response.output_item.done" && event.item && completedItems.has(event.item.id))
packages/ai/src/protocols/mistral-chat.ts
+3 -3
View File
@@ -596,7 +596,7 @@ const toolText = (tool: MistralToolDelta) => {
return value === null || value === undefined ? "" : ProviderShared.encodeJson(value)
}
const appendTools = Effect.fn("MistralChat.appendTools")(function* (
const appendTools = Effect.fnUntraced(function* (
initial: ParserState,
events: LLMEvent[],
deltas: ReadonlyArray<MistralToolDelta>,
@@ -662,7 +662,7 @@ const hasLateContent = (event: MistralEvent) => {
)
}
const step = Effect.fn("MistralChat.step")(function* (state: ParserState, event: MistralEvent) {
const step = Effect.fnUntraced(function* (state: ParserState, event: MistralEvent) {
if (event.error) {
const body = ProviderShared.encodeJson(event)
return yield* new AIError({
@@ -726,7 +726,7 @@ const step = Effect.fn("MistralChat.step")(function* (state: ParserState, event:
] as const
})
const finishEvents = Effect.fn("MistralChat.finishEvents")(function* (state: ParserState) {
const finishEvents = Effect.fnUntraced(function* (state: ParserState) {
if (!state.finishReason)
return yield* new AIError({
reason: new InvalidProviderOutputError({
packages/ai/src/protocols/open-responses.ts
+3 -3
View File
@@ -1143,7 +1143,7 @@ const onReasoningSummaryPartDone = (state: ParserState, event: Event): StepResul
]
}
const onFunctionCallArgumentsDelta = Effect.fn("OpenResponses.onFunctionCallArgumentsDelta")(function* (
const onFunctionCallArgumentsDelta = Effect.fnUntraced(function* (
state: ParserState,
event: Event,
) {
@@ -1174,7 +1174,7 @@ const onFunctionCallArgumentsDelta = Effect.fn("OpenResponses.onFunctionCallArgu
return [{ ...state, lifecycle, tools: result.tools }, events] satisfies StepResult
})
const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* (
const onOutputItemDone = Effect.fnUntraced(function* (
state: ParserState,
item: NormalizedEvent["item"],
) {
@@ -1310,7 +1310,7 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* (
return [state, NO_EVENTS] satisfies StepResult
})
const onResponseFinish = Effect.fn("OpenResponses.onResponseFinish")(function* (state: ParserState, event: Event) {
const onResponseFinish = Effect.fnUntraced(function* (state: ParserState, event: Event) {
let current = state
const events: LLMEvent[] = []
if (event.type === "response.completed") {
packages/ai/src/protocols/openai-chat.ts
+2 -2
View File
@@ -855,7 +855,7 @@ export const fromRequest = Effect.fn("OpenAIChat.fromRequest")(function* (
// Streaming parsers are small state machines: every event returns a new state
// plus the common `LLMEvent`s produced by that event. Tool calls are accumulated
// because OpenAI streams JSON arguments across multiple deltas.
const mapFinishReason = Effect.fn("OpenAIChat.mapFinishReason")(function* (event: OpenAIChatEvent, reason: string) {
const mapFinishReason = Effect.fnUntraced(function* (event: OpenAIChatEvent, reason: string) {
switch (reason) {
case "error":
return yield* new AIError({
@@ -1214,7 +1214,7 @@ const step = (state: ParserState, event: OpenAIChatEvent) =>
] as const
})
const finishEvents = Effect.fn("OpenAIChat.finishEvents")(function* (state: ParserState) {
const finishEvents = Effect.fnUntraced(function* (state: ParserState) {
if (state.finishReason === undefined && state.requireFinishReason)
return yield* new AIError({
reason: new InvalidProviderOutputError({
packages/ai/src/protocols/openai-images.ts
+2 -2
View File
@@ -208,7 +208,7 @@ const eventImage = (frame: string, label: string, data: string, format: string,
info: info(format, size),
})
const onEvent = Effect.fn("OpenAIImages.onEvent")(function* (state: State, frame: string) {
const onEvent = Effect.fnUntraced(function* (state: State, frame: string) {
const event = yield* decodeEvent(frame)
const format = event.output_format
if ("partial_image_index" in event) {
@@ -229,7 +229,7 @@ const onEvent = Effect.fn("OpenAIImages.onEvent")(function* (state: State, frame
] as const
})
const onDocument = Effect.fn("OpenAIImages.onDocument")(function* (frame: Exclude<Frame, string>) {
const onDocument = Effect.fnUntraced(function* (frame: Exclude<Frame, string>) {
const invalid = (message: string, cause?: unknown) => route.frameError(message, frame.document, cause)
const decoded = yield* decodeDocument(frame.document).pipe(
Effect.mapError((cause) => invalid(`${route.name} returned an invalid response`, cause)),
packages/ai/src/protocols/openai-responses.ts
+1 -1
View File
@@ -237,7 +237,7 @@ const checkpointBody = {
}),
}
const hostedToolResult = Effect.fn("OpenAIResponses.hostedToolResult")(function* (item: ResponsesHostedTools.Item) {
const hostedToolResult = Effect.fnUntraced(function* (item: ResponsesHostedTools.Item) {
const isError = item.error !== undefined && item.error !== null
if (item.type === "image_generation_call" && item.result) {
yield* Effect.fromResult(Encoding.decodeBase64(item.result)).pipe(
packages/ai/src/protocols/openai-speech.ts
+1 -1
View File
@@ -94,7 +94,7 @@ const fromRequest = Effect.fn("OpenAISpeech.fromRequest")(function* (request: Me
const isSse = (body: MediaProtocol.Body) => body.type === "json" && body.value.stream_format === "sse"
const onEvent = Effect.fn("OpenAISpeech.onEvent")(function* (state: State, frame: string) {
const onEvent = Effect.fnUntraced(function* (state: State, frame: string) {
const event = yield* decodeEvent(frame)
if (event.type === "speech.audio.delta") return SpeechStream.delta(state, event.audio)
const usage = event.usage
packages/ai/src/protocols/openai-transcription.ts
+1 -1
View File
@@ -210,7 +210,7 @@ const segment = (value: Schema.Schema.Type<typeof Segment>): TranscriptionSegmen
speaker: value.speaker,
})
const onEvent = Effect.fn("OpenAITranscription.onEvent")(function* (state: State, frame: string) {
const onEvent = Effect.fnUntraced(function* (state: State, frame: string) {
if (!EVENT_TYPES.has((yield* decodeEventType(frame)).type)) return [state, []] as const
const event = yield* decodeEvent(frame)
if (event.type === "error")
packages/ai/src/protocols/stability-images.ts
+1 -1
View File
@@ -82,7 +82,7 @@ const endpoint = (model: string) => (model.startsWith("sd3") ? "sd3" : model)
const RESERVED_FORM_FIELDS = new Set(["image", "prompt", "mode", "model"])
const form = Effect.fn("StabilityImages.form")(function* (
const form = Effect.fnUntraced(function* (
identity: MediaProtocol.Identity,
fields: Record<string, unknown>,
native: Record<string, unknown> | undefined,
packages/ai/src/protocols/utils/responses-checkpoint.ts
+2 -2
View File
@@ -11,7 +11,7 @@ interface State {
readonly responseID?: string
}
const onOutputItem = Effect.fn("ResponsesCheckpoint.onOutputItem")(function* (
const onOutputItem = Effect.fnUntraced(function* (
state: State,
input: OpenResponses.Event,
) {
@@ -63,7 +63,7 @@ export const make = <Body>(body: RouteBody<Body>): TriggerCompactOperation =>
checkpoints: {},
}),
terminal: OpenResponses.terminal,
step: Effect.fn("ResponsesCheckpoint.step")(function* (state: State, event: OpenResponses.Event) {
step: Effect.fnUntraced(function* (state: State, event: OpenResponses.Event) {
if (event.response?.id && state.responseID && event.response.id !== state.responseID)
return yield* ProviderShared.eventError(source.id, "Compaction response ID changed during execution")
if (event.type === "response.created") return [{ ...state, responseID: event.response?.id }, []] as const
packages/ai/src/protocols/utils/responses-hosted-tools.ts
+1 -1
View File
@@ -33,7 +33,7 @@ export const onDone: (
state: OpenResponses.ParserState,
item: Item,
tools: Definitions,
) => Effect.Effect<OpenResponses.StepResult, AIError> = Effect.fn("ResponsesHostedTools.onDone")(
) => Effect.Effect<OpenResponses.StepResult, AIError> = Effect.fnUntraced(
function* (state, item, tools) {
const tool = tools[item.type]
if (!tool) return [state, []] satisfies OpenResponses.StepResult