diff --git a/packages/ai/src/protocols/open-responses.ts b/packages/ai/src/protocols/open-responses.ts index f767fb2e9b75..267e67089392 100644 --- a/packages/ai/src/protocols/open-responses.ts +++ b/packages/ai/src/protocols/open-responses.ts @@ -287,6 +287,7 @@ export const Event = Schema.StructWithRest( delta: Schema.optional(Schema.String), text: Schema.optional(Schema.String), item_id: Schema.optional(Schema.String), + output_index: Schema.optional(Schema.Number), summary_index: Schema.optional(Schema.Number), item: Schema.optional(StreamItem), response: Schema.optional( @@ -297,6 +298,7 @@ export const Event = Schema.StructWithRest( incomplete_details: optionalNull(Schema.Struct({ reason: Schema.optional(Schema.String) })), usage: optionalNull(OpenResponsesUsage), error: optionalNull(OpenResponsesErrorPayload), + output: Schema.optional(Schema.Array(StreamItem)), }), [Schema.Record(Schema.String, Schema.Unknown)], ), @@ -342,13 +344,14 @@ export interface ParserState { readonly messageItems: ReadonlySet readonly messagePhases: Readonly> readonly reasoningItems: Readonly> - readonly store: boolean | undefined + readonly reasoningIndexes: Readonly> } -type ReasoningSummaryStatus = "active" | "can-conclude" | "concluded" +type ReasoningSummaryStatus = "active" | "concluded" interface ReasoningStreamItem { readonly encryptedContent: string | null | undefined + readonly blockIDs?: ReadonlyArray // Keyed by the wire protocol's numeric `summary_index`. JS object keys coerce to // strings, but typing the map as `Record` documents intent // and matches the wire field. @@ -857,6 +860,10 @@ const onOutputItemAdded = (state: ParserState, event: Event): StepResult => { ...state.reasoningItems, [item.id]: { encryptedContent: item.encrypted_content, summaryParts: { 0: "active" } }, }, + reasoningIndexes: + event.output_index === undefined + ? state.reasoningIndexes + : { ...state.reasoningIndexes, [event.output_index]: item.id }, }, events, ] @@ -890,23 +897,11 @@ const onReasoningSummaryPartAdded = (state: ParserState, event: Event): StepResu if (event.summary_index === 0) return [state, NO_EVENTS] const events: LLMEvent[] = [] - const closed = Object.entries(item.summaryParts) - .filter((entry) => entry[1] === "can-conclude") - .reduce( - (lifecycle, entry) => - Lifecycle.reasoningEnd( - lifecycle, - events, - `${event.item_id}:${entry[0]}`, - providerMetadata(state, { itemId: event.item_id }), - ), - state.lifecycle, - ) return [ { ...state, lifecycle: Lifecycle.reasoningStart( - closed, + state.lifecycle, events, `${event.item_id}:${event.summary_index}`, providerMetadata(state, { itemId: event.item_id, reasoningEncryptedContent: item.encryptedContent ?? null }), @@ -916,11 +911,7 @@ const onReasoningSummaryPartAdded = (state: ParserState, event: Event): StepResu [event.item_id]: { ...item, summaryParts: { - ...Object.fromEntries( - Object.entries(item.summaryParts).map((entry) => - entry[1] === "can-conclude" ? [entry[0], "concluded" as const] : entry, - ), - ), + ...item.summaryParts, [event.summary_index]: "active", }, }, @@ -938,22 +929,19 @@ const onReasoningSummaryPartDone = (state: ParserState, event: Event): StepResul return [ { ...state, - lifecycle: - state.store !== false - ? Lifecycle.reasoningEnd( - state.lifecycle, - events, - `${event.item_id}:${event.summary_index}`, - providerMetadata(state, { itemId: event.item_id }), - ) - : state.lifecycle, + lifecycle: Lifecycle.reasoningEnd( + state.lifecycle, + events, + `${event.item_id}:${event.summary_index}`, + providerMetadata(state, { itemId: event.item_id }), + ), reasoningItems: { ...state.reasoningItems, [event.item_id]: { ...item, summaryParts: { ...item.summaryParts, - [event.summary_index]: state.store !== false ? "concluded" : "can-conclude", + [event.summary_index]: "concluded", }, }, }, @@ -1039,27 +1027,74 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* ( } if (isReasoningItem(item)) { + if ( + event.output_index !== undefined && + state.reasoningIndexes[event.output_index] !== undefined && + state.reasoningIndexes[event.output_index] !== item.id + ) + return [state, NO_EVENTS] satisfies StepResult const events: LLMEvent[] = [] const metadata = reasoningMetadata(state, item) const reasoningItem = state.reasoningItems[item.id] + const reasoningIndexes = + event.output_index === undefined + ? state.reasoningIndexes + : Object.fromEntries( + Object.entries(state.reasoningIndexes).filter((entry) => entry[0] !== `${event.output_index}`), + ) if (reasoningItem) { - const lifecycle = Object.entries(reasoningItem.summaryParts) - .filter((entry) => entry[1] === "active" || entry[1] === "can-conclude") - .reduce( - (lifecycle, entry) => Lifecycle.reasoningEnd(lifecycle, events, `${item.id}:${entry[0]}`, metadata), - state.lifecycle, - ) - const { [item.id]: _removed, ...reasoningItems } = state.reasoningItems - return [{ ...state, lifecycle, reasoningItems }, events] satisfies StepResult + const openParts = Object.entries(reasoningItem.summaryParts).filter((entry) => entry[1] === "active") + const lifecycle = openParts.reduce( + (lifecycle, entry) => Lifecycle.reasoningEnd(lifecycle, events, `${item.id}:${entry[0]}`, metadata), + state.lifecycle, + ) + if (typeof item.encrypted_content === "string" && openParts.length === 0) { + const blockID = Object.keys(reasoningItem.summaryParts) + .map((index) => `${item.id}:${index}`) + .at(-1) + if (blockID) events.push(LLMEvent.reasoningMetadata({ id: blockID, providerMetadata: metadata })) + } + const reasoningItems = + typeof item.encrypted_content === "string" + ? Object.fromEntries(Object.entries(state.reasoningItems).filter((entry) => entry[0] !== item.id)) + : { + ...state.reasoningItems, + [item.id]: { + ...reasoningItem, + encryptedContent: item.encrypted_content, + summaryParts: Object.fromEntries( + Object.keys(reasoningItem.summaryParts).map((index) => [index, "concluded" as const]), + ), + }, + } + return [{ ...state, lifecycle, reasoningItems, reasoningIndexes }, events] satisfies StepResult } if (!state.lifecycle.reasoning.has(item.id)) { const lifecycle = Lifecycle.stepStart(state.lifecycle, events) events.push(LLMEvent.reasoningStart({ id: item.id, providerMetadata: metadata })) events.push(LLMEvent.reasoningEnd({ id: item.id, providerMetadata: metadata })) - return [{ ...state, lifecycle }, events] satisfies StepResult + return [ + { + ...state, + lifecycle, + reasoningIndexes, + reasoningItems: + typeof item.encrypted_content === "string" + ? state.reasoningItems + : { + ...state.reasoningItems, + [item.id]: { encryptedContent: item.encrypted_content, blockIDs: [item.id], summaryParts: {} }, + }, + }, + events, + ] satisfies StepResult } return [ - { ...state, lifecycle: Lifecycle.reasoningEnd(state.lifecycle, events, item.id, metadata) }, + { + ...state, + lifecycle: Lifecycle.reasoningEnd(state.lifecycle, events, item.id, metadata), + reasoningIndexes, + }, events, ] satisfies StepResult } @@ -1077,7 +1112,27 @@ const onResponseFinish = Effect.fn("OpenResponses.onResponseFinish")(function* ( const hasFunctionCall = pending.events.some((event) => LLMEvent.is.toolCall(event) || LLMEvent.is.toolInputError(event)) || state.hasFunctionCall - const lifecycle = Lifecycle.finish(state.lifecycle, events, { + const terminalReasoning = new Map( + (event.response?.output ?? []).filter(isReasoningItem).map((item) => [item.id, item]), + ) + const reasoningLifecycle = Object.entries(state.reasoningItems).reduce((lifecycle, [id, item]) => { + const terminal = terminalReasoning.get(id) + if (!terminal) return lifecycle + const metadata = providerMetadata(state, { + itemId: id, + reasoningEncryptedContent: terminal.encrypted_content ?? null, + }) + const blockID = + item.blockIDs?.at(-1) ?? + Object.keys(item.summaryParts) + .map((index) => `${id}:${index}`) + .at(-1) + if (!blockID) return lifecycle + if (lifecycle.reasoning.has(blockID)) return Lifecycle.reasoningEnd(lifecycle, events, blockID, metadata) + events.push(LLMEvent.reasoningMetadata({ id: blockID, providerMetadata: metadata })) + return lifecycle + }, state.lifecycle) + const lifecycle = Lifecycle.finish(reasoningLifecycle, events, { reason: { normalized: mapFinishReason(event, hasFunctionCall), raw: event.response?.incomplete_details?.reason, @@ -1091,7 +1146,10 @@ const onResponseFinish = Effect.fn("OpenResponses.onResponseFinish")(function* ( }) : undefined, }) - return [{ ...state, lifecycle, hasFunctionCall, tools: pending.tools }, events] satisfies StepResult + return [ + { ...state, lifecycle, hasFunctionCall, tools: pending.tools, reasoningItems: {}, reasoningIndexes: {} }, + events, + ] satisfies StepResult }) // Build the prettiest summary available from whatever the provider supplied. @@ -1210,7 +1268,7 @@ export const initial = (request: LLMRequest, extension: Extension = BASE): Parse messageItems: new Set(), messagePhases: {}, reasoningItems: {}, - store: OpenResponsesOptions.resolve(request).store, + reasoningIndexes: {}, }) export const protocol = Protocol.make({ diff --git a/packages/ai/src/schema/events.ts b/packages/ai/src/schema/events.ts index f0857840d41e..00c0e1d9333d 100644 --- a/packages/ai/src/schema/events.ts +++ b/packages/ai/src/schema/events.ts @@ -138,6 +138,13 @@ export const ReasoningEnd = Schema.Struct({ }).annotate({ identifier: "LLM.Event.ReasoningEnd" }) export type ReasoningEnd = Schema.Schema.Type +export const ReasoningMetadata = Schema.Struct({ + type: Schema.tag("reasoning-metadata"), + id: ContentBlockID, + providerMetadata: ProviderMetadata, +}).annotate({ identifier: "LLM.Event.ReasoningMetadata" }) +export type ReasoningMetadata = Schema.Schema.Type + export const ToolInputStart = Schema.Struct({ type: Schema.tag("tool-input-start"), id: ToolCallID, @@ -242,6 +249,7 @@ const llmEventTagged = Schema.Union([ ReasoningStart, ReasoningDelta, ReasoningEnd, + ReasoningMetadata, ToolInputStart, ToolInputDelta, ToolInputEnd, @@ -278,6 +286,8 @@ export const LLMEvent = Object.assign(llmEventTagged, { ReasoningDelta.make({ ...input, id: contentBlockID(input.id) }), reasoningEnd: (input: WithID) => ReasoningEnd.make({ ...input, id: contentBlockID(input.id) }), + reasoningMetadata: (input: WithID) => + ReasoningMetadata.make({ ...input, id: contentBlockID(input.id) }), toolInputStart: (input: WithID) => ToolInputStart.make({ ...input, id: toolCallID(input.id) }), toolInputDelta: (input: WithID) => @@ -312,6 +322,7 @@ export const LLMEvent = Object.assign(llmEventTagged, { reasoningStart: llmEventTagged.guards["reasoning-start"], reasoningDelta: llmEventTagged.guards["reasoning-delta"], reasoningEnd: llmEventTagged.guards["reasoning-end"], + reasoningMetadata: llmEventTagged.guards["reasoning-metadata"], toolInputStart: llmEventTagged.guards["tool-input-start"], toolInputDelta: llmEventTagged.guards["tool-input-delta"], toolInputEnd: llmEventTagged.guards["tool-input-end"], @@ -483,6 +494,18 @@ const reduceReasoningEnd = (state: ResponseState, event: ReasoningEnd): Response } } +const reduceReasoningMetadata = (state: ResponseState, event: ReasoningMetadata): ResponseState => { + const current = state.reasoningParts[event.id] + if (!current) return state + return { + ...replaceContent(state, current.contentIndex, reasoningContent(current.text, event.providerMetadata)), + reasoningParts: { + ...state.reasoningParts, + [event.id]: { ...current, providerMetadata: event.providerMetadata }, + }, + } +} + const reduceToolInputStart = (state: ResponseState, event: ToolInputStart): ResponseState => ({ ...state, toolInputs: { @@ -552,6 +575,8 @@ const reduceResponseState = (state: ResponseState, event: LLMEvent): ResponseSta return reduceReasoningDelta(next, event) case "reasoning-end": return reduceReasoningEnd(next, event) + case "reasoning-metadata": + return reduceReasoningMetadata(next, event) case "tool-input-start": return reduceToolInputStart(next, event) case "tool-input-delta": diff --git a/packages/ai/test/lib/tool-runtime.ts b/packages/ai/test/lib/tool-runtime.ts index 629720cd191a..36bad625ea56 100644 --- a/packages/ai/test/lib/tool-runtime.ts +++ b/packages/ai/test/lib/tool-runtime.ts @@ -82,16 +82,25 @@ const indexStep = (event: LLMEvent, index: number): LLMEvent => { const stepState = (events: ReadonlyArray) => { const assistantContent: ContentPart[] = [] + const reasoningIndexes = new Map() const toolCalls: ToolCallPart[] = [] let reason: Extract["reason"] = { normalized: "unknown" } let usage: Usage | undefined let providerMetadata: ProviderMetadata | undefined for (const event of events) { - if (event.type === "text-delta" || event.type === "reasoning-delta") { - appendText(assistantContent, event.type === "text-delta" ? "text" : "reasoning", event.text) - } else if (event.type === "text-end" || event.type === "reasoning-end") { - appendText(assistantContent, event.type === "text-end" ? "text" : "reasoning", "", event.providerMetadata) + if (event.type === "text-delta") { + appendText(assistantContent, "text", event.text) + } else if (event.type === "reasoning-delta") { + appendReasoning(assistantContent, reasoningIndexes, event.id, event.text, event.providerMetadata) + } else if (event.type === "text-end") { + appendText(assistantContent, "text", "", event.providerMetadata) + } else if (event.type === "reasoning-end") { + appendReasoning(assistantContent, reasoningIndexes, event.id, "", event.providerMetadata) + } else if (event.type === "reasoning-metadata") { + const index = reasoningIndexes.get(event.id) + const reasoning = index === undefined ? undefined : assistantContent[index] + if (reasoning?.type === "reasoning") reasoning.providerMetadata = event.providerMetadata } else if (event.type === "tool-call") { assistantContent.push(event) if (!event.providerExecuted) toolCalls.push(event) @@ -114,6 +123,28 @@ const stepState = (events: ReadonlyArray) => { return { assistantContent, toolCalls, reason, usage, providerMetadata } } +const appendReasoning = ( + content: ContentPart[], + indexes: Map, + id: string, + text: string, + providerMetadata?: ProviderMetadata, +) => { + const index = indexes.get(id) + if (index === undefined) { + indexes.set(id, content.length) + content.push({ type: "reasoning", text, providerMetadata }) + return + } + const current = content[index] + if (current?.type !== "reasoning") return + content[index] = { + ...current, + text: `${current.text}${text}`, + providerMetadata: providerMetadata ?? current.providerMetadata, + } +} + const appendText = ( content: ContentPart[], type: "text" | "reasoning", diff --git a/packages/ai/test/provider/openai-responses.test.ts b/packages/ai/test/provider/openai-responses.test.ts index 3587efc1f6bf..6860afe2cc14 100644 --- a/packages/ai/test/provider/openai-responses.test.ts +++ b/packages/ai/test/provider/openai-responses.test.ts @@ -2141,6 +2141,281 @@ describe("OpenAI Responses route", () => { }), ) + it.effect("backfills encrypted reasoning from the terminal response", () => + Effect.gen(function* () { + const response = yield* LLMClient.generate(request).pipe( + Effect.provide( + fixedResponse( + sseEvents( + { type: "response.output_item.added", item: { type: "reasoning", id: "rs_1" } }, + { type: "response.reasoning_summary_text.delta", item_id: "rs_1", delta: "thinking" }, + { + type: "response.output_item.done", + item: { type: "reasoning", id: "rs_1", encrypted_content: null }, + }, + { + type: "response.completed", + response: { + id: "resp_1", + output: [{ type: "reasoning", id: "rs_1", encrypted_content: "terminal-state" }], + }, + }, + ), + ), + ), + ) + + expect(response.events.filter((event) => event.type === "reasoning-end")).toEqual([ + { + type: "reasoning-end", + id: "rs_1:0", + providerMetadata: { openai: { itemId: "rs_1", reasoningEncryptedContent: null } }, + }, + ]) + expect(response.events).toContainEqual({ + type: "reasoning-metadata", + id: "rs_1:0", + providerMetadata: { openai: { itemId: "rs_1", reasoningEncryptedContent: "terminal-state" } }, + }) + expect(response.message.content).toContainEqual({ + type: "reasoning", + text: "thinking", + providerMetadata: { openai: { itemId: "rs_1", reasoningEncryptedContent: "terminal-state" } }, + }) + + const prepared = yield* compileRequest(LLM.request({ model, messages: [response.message] })) + expect(prepared.body.input).toEqual([ + { + type: "reasoning", + id: "rs_1", + summary: [{ type: "summary_text", text: "thinking" }], + encrypted_content: "terminal-state", + }, + ]) + }), + ) + + it.effect("backfills done-only reasoning from the terminal response", () => + Effect.gen(function* () { + const response = yield* LLMClient.generate(request).pipe( + Effect.provide( + fixedResponse( + sseEvents( + { + type: "response.output_item.done", + item: { type: "reasoning", id: "rs_done", encrypted_content: null }, + }, + { + type: "response.completed", + response: { + id: "resp_1", + output: [{ type: "reasoning", id: "rs_done", encrypted_content: "terminal-state" }], + }, + }, + ), + ), + ), + ) + + expect(response.message.content).toEqual([ + { + type: "reasoning", + text: "", + providerMetadata: { openai: { itemId: "rs_done", reasoningEncryptedContent: "terminal-state" } }, + }, + ]) + expect(response.events.filter((event) => event.type === "reasoning-end")).toHaveLength(1) + }), + ) + + it.effect("ends reasoning before later tool and text output", () => + Effect.gen(function* () { + const response = yield* LLMClient.generate(request).pipe( + Effect.provide( + fixedResponse( + sseEvents( + { type: "response.output_item.added", item: { type: "reasoning", id: "rs_1" } }, + { type: "response.reasoning_summary_text.delta", item_id: "rs_1", delta: "thinking" }, + { + type: "response.output_item.done", + item: { type: "reasoning", id: "rs_1", encrypted_content: null }, + }, + { + type: "response.output_item.added", + item: { type: "function_call", id: "fc_1", call_id: "call_1", name: "lookup" }, + }, + { + type: "response.output_item.done", + item: { + type: "function_call", + id: "fc_1", + call_id: "call_1", + name: "lookup", + arguments: '{"query":"Effect"}', + }, + }, + { type: "response.output_item.added", item: { type: "message", id: "msg_1" } }, + { type: "response.output_text.delta", item_id: "msg_1", delta: "Found it" }, + { type: "response.output_item.done", item: { type: "message", id: "msg_1" } }, + { + type: "response.completed", + response: { + id: "resp_1", + output: [{ type: "reasoning", id: "rs_1", encrypted_content: "terminal-state" }], + }, + }, + ), + ), + ), + ) + + const reasoningEnd = response.events.findIndex((event) => event.type === "reasoning-end") + expect(reasoningEnd).toBeLessThan(response.events.findIndex((event) => event.type === "tool-input-start")) + expect(reasoningEnd).toBeLessThan(response.events.findIndex((event) => event.type === "text-start")) + expect(response.events.filter((event) => event.type === "reasoning-end")).toHaveLength(1) + expect(response.message.content[0]).toEqual({ + type: "reasoning", + text: "thinking", + providerMetadata: { openai: { itemId: "rs_1", reasoningEncryptedContent: "terminal-state" } }, + }) + }), + ) + + it.effect("uses terminal reasoning instead of output_item.added encryption", () => + Effect.gen(function* () { + const response = yield* LLMClient.generate(request).pipe( + Effect.provide( + fixedResponse( + sseEvents( + { + type: "response.output_item.added", + item: { type: "reasoning", id: "rs_1", encrypted_content: "provisional-state" }, + }, + { type: "response.reasoning_summary_text.delta", item_id: "rs_1", delta: "thinking" }, + { + type: "response.completed", + response: { + id: "resp_1", + output: [{ type: "reasoning", id: "rs_1", encrypted_content: "terminal-state" }], + }, + }, + ), + ), + ), + ) + + expect(response.message.content).toEqual([ + { + type: "reasoning", + text: "thinking", + providerMetadata: { openai: { itemId: "rs_1", reasoningEncryptedContent: "terminal-state" } }, + }, + ]) + }), + ) + + it.effect("backfills reasoning from an incomplete terminal response", () => + Effect.gen(function* () { + const response = yield* LLMClient.generate(request).pipe( + Effect.provide( + fixedResponse( + sseEvents( + { type: "response.output_item.added", item: { type: "reasoning", id: "rs_1" } }, + { type: "response.reasoning_summary_text.delta", item_id: "rs_1", delta: "thinking" }, + { + type: "response.output_item.done", + item: { type: "reasoning", id: "rs_1", encrypted_content: null }, + }, + { + type: "response.incomplete", + response: { + id: "resp_1", + incomplete_details: { reason: "max_output_tokens" }, + output: [{ type: "reasoning", id: "rs_1", encrypted_content: "terminal-state" }], + }, + }, + ), + ), + ), + ) + + expect(response.finishReason).toEqual({ normalized: "length", raw: "max_output_tokens" }) + expect(response.message.content[0]).toMatchObject({ + providerMetadata: { openai: { reasoningEncryptedContent: "terminal-state" } }, + }) + }), + ) + + it.effect("does not backfill reasoning from a mismatched terminal item", () => + Effect.gen(function* () { + const response = yield* LLMClient.generate(request).pipe( + Effect.provide( + fixedResponse( + sseEvents( + { type: "response.output_item.added", item: { type: "reasoning", id: "rs_1" } }, + { type: "response.reasoning_summary_text.delta", item_id: "rs_1", delta: "thinking" }, + { + type: "response.output_item.done", + item: { type: "reasoning", id: "rs_1", encrypted_content: null }, + }, + { + type: "response.completed", + response: { + id: "resp_1", + output: [{ type: "reasoning", id: "rs_other", encrypted_content: "other-state" }], + }, + }, + ), + ), + ), + ) + + expect(response.events.some((event) => event.type === "reasoning-metadata")).toBe(false) + expect(response.message.content[0]).toMatchObject({ + providerMetadata: { openai: { itemId: "rs_1", reasoningEncryptedContent: null } }, + }) + }), + ) + + it.effect("ignores a reasoning done item that mismatches its output index", () => + Effect.gen(function* () { + const response = yield* LLMClient.generate(request).pipe( + Effect.provide( + fixedResponse( + sseEvents( + { + type: "response.output_item.added", + output_index: 0, + item: { type: "reasoning", id: "rs_1" }, + }, + { type: "response.reasoning_summary_text.delta", item_id: "rs_1", delta: "thinking" }, + { + type: "response.output_item.done", + output_index: 0, + item: { type: "reasoning", id: "rs_other", encrypted_content: "wrong-state" }, + }, + { + type: "response.completed", + response: { + id: "resp_1", + output: [{ type: "reasoning", id: "rs_1", encrypted_content: "terminal-state" }], + }, + }, + ), + ), + ), + ) + + expect(response.message.content).toEqual([ + { + type: "reasoning", + text: "thinking", + providerMetadata: { openai: { itemId: "rs_1", reasoningEncryptedContent: "terminal-state" } }, + }, + ]) + }), + ) + it.effect("streams each reasoning summary part as a separate block", () => Effect.gen(function* () { const response = yield* LLMClient.generate( @@ -2188,6 +2463,11 @@ describe("OpenAI Responses route", () => { { type: "reasoning-end", id: "rs_1:1", + providerMetadata: { openai: { itemId: "rs_1" } }, + }, + { + type: "reasoning-metadata", + id: "rs_1:1", providerMetadata: { openai: { itemId: "rs_1", reasoningEncryptedContent: "encrypted-state" } }, }, { type: "step-finish", index: 0, reason: { normalized: "stop", raw: undefined } }, @@ -2196,7 +2476,7 @@ describe("OpenAI Responses route", () => { }), ) - it.effect("closes reasoning summary parts when storage is not disabled", () => + it.effect("backfills terminal reasoning when storage is enabled", () => Effect.gen(function* () { const response = yield* LLMClient.generate(LLMRequest.update(request, { providerOptions: { store: true } })).pipe( Effect.provide( @@ -2216,7 +2496,13 @@ describe("OpenAI Responses route", () => { type: "response.output_item.done", item: { type: "reasoning", id: "rs_1", encrypted_content: null }, }, - { type: "response.completed", response: { id: "resp_1" } }, + { + type: "response.completed", + response: { + id: "resp_1", + output: [{ type: "reasoning", id: "rs_1", encrypted_content: "terminal-state" }], + }, + }, ), ), ), @@ -2224,8 +2510,17 @@ describe("OpenAI Responses route", () => { expect(response.events.filter((event) => event.type === "reasoning-end")).toEqual([ { type: "reasoning-end", id: "rs_1:0", providerMetadata: { openai: { itemId: "rs_1" } } }, - { type: "reasoning-end", id: "rs_1:1", providerMetadata: { openai: { itemId: "rs_1" } } }, + { + type: "reasoning-end", + id: "rs_1:1", + providerMetadata: { openai: { itemId: "rs_1" } }, + }, ]) + expect(response.events).toContainEqual({ + type: "reasoning-metadata", + id: "rs_1:1", + providerMetadata: { openai: { itemId: "rs_1", reasoningEncryptedContent: "terminal-state" } }, + }) }), ) diff --git a/packages/client/src/effect/api/api.ts b/packages/client/src/effect/api/api.ts index a2c002a0f1b6..c868c787fb72 100644 --- a/packages/client/src/effect/api/api.ts +++ b/packages/client/src/effect/api/api.ts @@ -709,6 +709,20 @@ export type SessionLogOutput = readonly state?: SessionMessage.ProviderState | undefined } } + | { + readonly id: Event.ID + readonly created: number + readonly metadata?: { readonly [x: string]: unknown } | undefined + readonly type: "session.reasoning.state.updated" + readonly durable: { readonly aggregateID: string; readonly seq: Event.Seq; readonly version: Event.Version } + readonly location?: Location.Ref | undefined + readonly data: { + readonly sessionID: Session.ID + readonly assistantMessageID: SessionMessage.ID + readonly ordinal: number + readonly state: SessionMessage.ProviderState + } + } | { readonly id: Event.ID readonly created: number diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index 2af0d4d96f2c..3913cbe72fa3 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -1263,6 +1263,16 @@ export type SessionToolCalled = { } } +export type SessionReasoningStateUpdated = { + id: string + created: number + metadata?: { [x: string]: any } + type: "session.reasoning.state.updated" + durable: { aggregateID: string; seq: number; version: 1 } + location?: LocationRef + data: { sessionID: string; assistantMessageID: string; ordinal: number; state: SessionMessageProviderState1 } +} + export type ToolContent1 = ToolTextContent | ToolFileContent1 export type ModelCompatibility = { @@ -1975,6 +1985,7 @@ export type SessionEventDurable = | SessionTextEnded | SessionReasoningStarted | SessionReasoningEnded + | SessionReasoningStateUpdated | SessionToolInputStarted | SessionToolInputEnded | SessionToolCalled @@ -2069,6 +2080,7 @@ export type V2Event = | SessionReasoningStarted | SessionReasoningDelta | SessionReasoningEnded + | SessionReasoningStateUpdated | SessionToolInputStarted | SessionToolInputDelta | SessionToolInputEnded diff --git a/packages/client/src/solid/data.ts b/packages/client/src/solid/data.ts index 2c3f35dc8200..878c22786be4 100644 --- a/packages/client/src/solid/data.ts +++ b/packages/client/src/solid/data.ts @@ -904,6 +904,14 @@ export function createData(config: CreateDataInput) { } }) return + case "session.reasoning.state.updated": + message.update(event.data.sessionID, (draft, index) => { + const match = message + .assistant(draft, index, event.data.assistantMessageID) + ?.content.filter((item) => item.type === "reasoning")[event.data.ordinal] + if (match?.type === "reasoning") match.state = event.data.state + }) + return case "session.retry.scheduled": message.update(event.data.sessionID, (draft, index) => { const currentAssistant = message.assistant(draft, index, event.data.assistantMessageID) diff --git a/packages/core/src/session/message-updater.ts b/packages/core/src/session/message-updater.ts index f3df3a8151a5..a85c93462fd0 100644 --- a/packages/core/src/session/message-updater.ts +++ b/packages/core/src/session/message-updater.ts @@ -373,6 +373,14 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) { } }) }, + "session.reasoning.state.updated": (event) => { + return updateOwnedAssistant(event.data.assistantMessageID, (draft) => { + const match = draft.content.filter((item): item is DraftReasoning => item.type === "reasoning")[ + event.data.ordinal + ] + if (match) match.state = event.data.state + }) + }, "session.retry.scheduled": (event) => { return updateOwnedAssistant(event.data.assistantMessageID, (draft) => { draft.retry = { diff --git a/packages/core/src/session/projector.ts b/packages/core/src/session/projector.ts index 164db2d508c0..77fd7cfd076d 100644 --- a/packages/core/src/session/projector.ts +++ b/packages/core/src/session/projector.ts @@ -661,6 +661,7 @@ const layer = Layer.effectDiscard( yield* bus.project(SessionEvent.Tool.Failed, (event) => run(db, event)) yield* bus.project(SessionEvent.Reasoning.Started, (event) => run(db, event)) yield* bus.project(SessionEvent.Reasoning.Ended, (event) => run(db, event)) + yield* bus.project(SessionEvent.Reasoning.StateUpdated, (event) => run(db, event)) yield* bus.project(SessionEvent.RetryScheduled, (event) => run(db, event)) yield* bus.project(SessionEvent.Compaction.Started, (event) => run(db, event)) yield* bus.project(SessionEvent.Compaction.Ended, (event) => diff --git a/packages/core/src/session/runner/publish-llm-event.ts b/packages/core/src/session/runner/publish-llm-event.ts index 8fa3e70da077..1c3118fac5bc 100644 --- a/packages/core/src/session/runner/publish-llm-event.ts +++ b/packages/core/src/session/runner/publish-llm-event.ts @@ -181,7 +181,7 @@ export const createLLMEventPublisher = (bus: Pick, inp state === undefined ? current.state : { ...current.state, ...state }, ) chunks.delete(id) - return undefined + return current.ordinal }) const flush = Effect.fnUntraced(function* () { for (const id of Array.from(chunks.keys())) yield* end(id) @@ -235,6 +235,7 @@ export const createLLMEventPublisher = (bus: Pick, inp }), true, ) + const completedReasoning = new Map() const toolInput = fragments("tool input", (id, value) => Effect.gen(function* () { const tool = tools.get(id) @@ -413,8 +414,21 @@ export const createLLMEventPublisher = (bus: Pick, inp yield* reasoning.append(event.id, event.text, providerState(event.providerMetadata)) return case "reasoning-end": - yield* reasoning.end(event.id, providerState(event.providerMetadata)) + completedReasoning.set(event.id, yield* reasoning.end(event.id, providerState(event.providerMetadata))) return + case "reasoning-metadata": { + const ordinal = completedReasoning.get(event.id) + if (ordinal === undefined) return yield* Effect.die(new Error(`Reasoning metadata before end: ${event.id}`)) + const state = providerState(event.providerMetadata) + if (state === undefined) return + yield* bus.publish(SessionEvent.Reasoning.StateUpdated, { + sessionID: input.sessionID, + assistantMessageID: yield* currentAssistantMessageID(), + ordinal, + state, + }) + return + } case "tool-input-start": outputStarted = true yield* startToolInput(event) diff --git a/packages/core/test/session-runner-tool-events.test.ts b/packages/core/test/session-runner-tool-events.test.ts index 1ed0f16fac53..fad63eb2db6c 100644 --- a/packages/core/test/session-runner-tool-events.test.ts +++ b/packages/core/test/session-runner-tool-events.test.ts @@ -203,6 +203,53 @@ test("reasoning state from start, empty delta, and end is merged", async () => { }) }) +test("reasoning metadata updates completed reasoning state", async () => { + const { published, publisher } = capture("openai") + await Effect.runPromise( + Effect.forEach( + [ + LLMEvent.reasoningStart({ id: "reasoning" }), + LLMEvent.reasoningEnd({ + id: "reasoning", + providerMetadata: { openai: { itemId: "rs_1", reasoningEncryptedContent: null } }, + }), + LLMEvent.reasoningMetadata({ + id: "reasoning", + providerMetadata: { openai: { itemId: "rs_1", reasoningEncryptedContent: "terminal-state" } }, + }), + ], + publisher.publish, + { discard: true }, + ), + ) + + expect(published.slice(-2).map((event) => event.type)).toEqual([ + "session.reasoning.ended.1", + "session.reasoning.state.updated.1", + ]) + expect(published.at(-1)?.data).toMatchObject({ + ordinal: 0, + state: { itemId: "rs_1", reasoningEncryptedContent: "terminal-state" }, + }) +}) + +test("reasoning metadata ignores unrelated provider state", async () => { + const { published, publisher } = capture("openai") + await Effect.runPromise( + Effect.forEach( + [ + LLMEvent.reasoningStart({ id: "reasoning" }), + LLMEvent.reasoningEnd({ id: "reasoning" }), + LLMEvent.reasoningMetadata({ id: "reasoning", providerMetadata: { anthropic: { signature: "ignored" } } }), + ], + publisher.publish, + { discard: true }, + ), + ) + + expect(published.some((event) => event.type === "session.reasoning.state.updated.1")).toBe(false) +}) + it.effect("batches text deltas and flushes pending text before the terminal event", () => Effect.gen(function* () { const { published, publisher } = capture() diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index 5f706b69b608..8c4a01728818 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -2903,6 +2903,13 @@ describe("SessionRunnerLLM", () => { }), LLMEvent.reasoningDelta({ id: "reasoning-openai", text: "Encrypted thought" }), LLMEvent.reasoningEnd({ + id: "reasoning-openai", + providerMetadata: { + openai: { itemId: "rs_1", reasoningEncryptedContent: null }, + anthropic: { ignored: true }, + }, + }), + LLMEvent.reasoningMetadata({ id: "reasoning-openai", providerMetadata: { openai: { itemId: "rs_1", reasoningEncryptedContent: "encrypted-state" }, diff --git a/packages/schema/src/session-event.ts b/packages/schema/src/session-event.ts index 6db772c0757b..f35c9a189861 100644 --- a/packages/schema/src/session-event.ts +++ b/packages/schema/src/session-event.ts @@ -415,6 +415,18 @@ export namespace Reasoning { }, }) export type Ended = typeof Ended.Type + + export const StateUpdated = Event.durable({ + type: "session.reasoning.state.updated", + ...options, + schema: { + ...Base, + assistantMessageID: SessionMessage.ID, + ordinal: NonNegativeInt, + state: SessionMessage.ProviderState, + }, + }) + export type StateUpdated = typeof StateUpdated.Type } export namespace Tool { @@ -624,6 +636,7 @@ export const Definitions = Event.inventory( Reasoning.Started, Reasoning.Delta, Reasoning.Ended, + Reasoning.StateUpdated, Tool.Input.Started, Tool.Input.Delta, Tool.Input.Ended, diff --git a/packages/schema/test/event-manifest.test.ts b/packages/schema/test/event-manifest.test.ts index 556a91e695a6..336a1f34f1fd 100644 --- a/packages/schema/test/event-manifest.test.ts +++ b/packages/schema/test/event-manifest.test.ts @@ -111,6 +111,7 @@ describe("public event manifest", () => { "session.tool.failed.2", "session.reasoning.started.1", "session.reasoning.ended.1", + "session.reasoning.state.updated.1", "session.retry.scheduled.1", "session.compaction.started.1", "session.compaction.ended.1",