diff --git a/packages/llm/src/protocols/openai-responses.ts b/packages/llm/src/protocols/openai-responses.ts index 4936d31c92..33e05db9a0 100644 --- a/packages/llm/src/protocols/openai-responses.ts +++ b/packages/llm/src/protocols/openai-responses.ts @@ -48,6 +48,9 @@ const OpenAIResponsesOutputText = Schema.Struct({ text: Schema.String, }) +const OpenAIResponsesMessagePhase = Schema.Literals(["commentary", "final_answer"]) +type OpenAIResponsesMessagePhase = Schema.Schema.Type + const OpenAIResponsesReasoningSummaryText = Schema.Struct({ type: Schema.tag("summary_text"), text: Schema.String, @@ -78,7 +81,11 @@ const OpenAIResponsesFunctionCallOutput = Schema.Union([ const OpenAIResponsesInputItem = Schema.Union([ Schema.Struct({ role: Schema.tag("system"), content: Schema.String }), Schema.Struct({ role: Schema.tag("user"), content: Schema.Array(OpenAIResponsesInputContent) }), - Schema.Struct({ role: Schema.tag("assistant"), content: Schema.Array(OpenAIResponsesOutputText) }), + Schema.Struct({ + role: Schema.tag("assistant"), + content: Schema.Array(OpenAIResponsesOutputText), + phase: Schema.optional(OpenAIResponsesMessagePhase), + }), OpenAIResponsesReasoningItem, OpenAIResponsesItemReference, Schema.Struct({ @@ -195,6 +202,7 @@ const OpenAIResponsesStreamItem = Schema.Struct({ output: Schema.optional(Schema.Unknown), error: Schema.optional(Schema.Unknown), encrypted_content: optionalNull(Schema.String), + phase: optionalNull(OpenAIResponsesMessagePhase), }) type OpenAIResponsesStreamItem = Schema.Schema.Type @@ -237,6 +245,7 @@ interface ParserState { readonly tools: ToolStream.State readonly hasFunctionCall: boolean readonly lifecycle: Lifecycle.State + readonly messageItems: Readonly> readonly reasoningItems: Readonly> readonly store: boolean | undefined } @@ -298,6 +307,11 @@ const lowerReasoning = (part: ReasoningPart): OpenAIResponsesReasoningInput | un } } +const messagePhase = (part: TextPart): OpenAIResponsesMessagePhase | undefined => { + const phase = part.providerMetadata?.openai?.phase + return phase === "commentary" || phase === "final_answer" ? phase : undefined +} + const hostedToolItemID = (part: ToolResultPart) => { const openai = part.providerMetadata?.openai return ProviderShared.isRecord(openai) && typeof openai.itemId === "string" && openai.itemId.length > 0 @@ -369,16 +383,25 @@ const lowerMessages = Effect.fn("OpenAIResponses.lowerMessages")(function* (requ if (message.role === "assistant") { const content: TextPart[] = [] + let phase: OpenAIResponsesMessagePhase | undefined const reasoningItems: Record = {} const reasoningReferences = new Set() const hostedToolReferences = new Set() const flushText = () => { if (content.length === 0) return - input.push({ role: "assistant", content: content.map((part) => ({ type: "output_text", text: part.text })) }) + input.push({ + role: "assistant", + content: content.map((part) => ({ type: "output_text", text: part.text })), + ...(phase ? { phase } : {}), + }) content.splice(0, content.length) + phase = undefined } for (const part of message.content) { if (part.type === "text") { + const nextPhase = messagePhase(part) + if (content.length > 0 && phase !== nextPhase) flushText() + phase = nextPhase content.push(part) continue } @@ -641,6 +664,9 @@ const onReasoningDone = (state: ParserState, _event: OpenAIResponsesEvent): Step const reasoningMetadata = (item: OpenAIResponsesStreamItem & { id: string }) => openaiMetadata({ itemId: item.id, reasoningEncryptedContent: item.encrypted_content ?? null }) +const messageMetadata = (item: OpenAIResponsesStreamItem, id: string) => + item.phase ? openaiMetadata({ itemId: id, phase: item.phase }) : undefined + // OpenAI Responses streams reasoning items in a stable order: // `output_item.added` (reasoning) → // `reasoning_summary_part.added` (index=0) → @@ -655,6 +681,18 @@ const reasoningMetadata = (item: OpenAIResponsesStreamItem & { id: string }) => // best-effort, not guaranteed. const onOutputItemAdded = (state: ParserState, event: OpenAIResponsesEvent): StepResult => { const item = event.item + if (item?.type === "message" && item.id && item.phase) { + const events: LLMEvent[] = [] + const providerMetadata = openaiMetadata({ itemId: item.id, phase: item.phase }) + return [ + { + ...state, + lifecycle: Lifecycle.textStart(state.lifecycle, events, item.id, providerMetadata), + messageItems: { ...state.messageItems, [item.id]: providerMetadata }, + }, + events, + ] + } if (item && isReasoningItem(item)) { const events: LLMEvent[] = [] return [ @@ -812,6 +850,24 @@ const onOutputItemDone = Effect.fn("OpenAIResponses.onOutputItemDone")(function* const item = event.item if (!item) return [state, NO_EVENTS] satisfies StepResult + if (item.type === "message" && item.id) { + const events: LLMEvent[] = [] + const { [item.id]: _finished, ...messageItems } = state.messageItems + return [ + { + ...state, + lifecycle: Lifecycle.textEnd( + state.lifecycle, + events, + item.id, + messageMetadata(item, item.id) ?? state.messageItems[item.id], + ), + messageItems, + }, + events, + ] satisfies StepResult + } + if (item.type === "function_call") { if (!item.id || !item.call_id || !item.name) return [state, NO_EVENTS] satisfies StepResult const tools = state.tools[item.id] @@ -968,6 +1024,7 @@ export const protocol = Protocol.make({ hasFunctionCall: false, tools: ToolStream.empty(), lifecycle: Lifecycle.initial(), + messageItems: {}, reasoningItems: {}, store: OpenAIOptions.store(request), }), diff --git a/packages/llm/src/protocols/utils/lifecycle.ts b/packages/llm/src/protocols/utils/lifecycle.ts index eb6c95dfbd..8add02ce67 100644 --- a/packages/llm/src/protocols/utils/lifecycle.ts +++ b/packages/llm/src/protocols/utils/lifecycle.ts @@ -14,16 +14,19 @@ export const stepStart = (state: State, events: LLMEvent[]): State => { return { ...state, stepStarted: true } } -export const textDelta = (state: State, events: LLMEvent[], id: string, text: string): State => { +export const textStart = (state: State, events: LLMEvent[], id: string, providerMetadata?: ProviderMetadata): State => { + if (state.text.has(id)) return state const stepped = stepStart(state, events) - if (stepped.text.has(id)) { - events.push(LLMEvent.textDelta({ id, text })) - return stepped - } - events.push(LLMEvent.textStart({ id }), LLMEvent.textDelta({ id, text })) + events.push(LLMEvent.textStart({ id, ...(providerMetadata ? { providerMetadata } : {}) })) return { ...stepped, text: new Set([...stepped.text, id]) } } +export const textDelta = (state: State, events: LLMEvent[], id: string, text: string): State => { + const started = textStart(state, events, id) + events.push(LLMEvent.textDelta({ id, text })) + return started +} + export const reasoningStart = ( state: State, events: LLMEvent[], @@ -65,7 +68,7 @@ export const reasoningEnd = ( export const textEnd = (state: State, events: LLMEvent[], id: string, providerMetadata?: ProviderMetadata): State => { if (!state.text.has(id)) return state const stepped = stepStart(state, events) - events.push(LLMEvent.textEnd({ id, providerMetadata })) + events.push(LLMEvent.textEnd({ id, ...(providerMetadata ? { providerMetadata } : {}) })) const text = new Set(stepped.text) text.delete(id) return { ...stepped, text } diff --git a/packages/llm/test/provider/openai-responses.test.ts b/packages/llm/test/provider/openai-responses.test.ts index cd8bad51af..5c8aad7241 100644 --- a/packages/llm/test/provider/openai-responses.test.ts +++ b/packages/llm/test/provider/openai-responses.test.ts @@ -754,6 +754,76 @@ describe("OpenAI Responses route", () => { }), ) + it.effect("preserves streamed assistant message phases", () => + Effect.gen(function* () { + const response = yield* LLMClient.generate(request).pipe( + Effect.provide( + fixedResponse( + sseEvents( + { + type: "response.output_item.added", + item: { type: "message", id: "msg_commentary", phase: "commentary" }, + }, + { type: "response.output_text.delta", item_id: "msg_commentary", delta: "Checking first." }, + { type: "response.output_text.done", item_id: "msg_commentary" }, + { + type: "response.output_item.done", + item: { type: "message", id: "msg_commentary", phase: "commentary" }, + }, + { + type: "response.output_item.added", + item: { type: "message", id: "msg_final", phase: "final_answer" }, + }, + { type: "response.output_text.delta", item_id: "msg_final", delta: "Finished." }, + { type: "response.output_text.done", item_id: "msg_final" }, + { + type: "response.output_item.done", + item: { type: "message", id: "msg_final", phase: "final_answer" }, + }, + { type: "response.completed", response: { id: "resp_1" } }, + ), + ), + ), + ) + + expect(response.events.filter((event) => event.type.startsWith("text-"))).toEqual([ + { + type: "text-start", + id: "msg_commentary", + providerMetadata: { openai: { itemId: "msg_commentary", phase: "commentary" } }, + }, + { type: "text-delta", id: "msg_commentary", text: "Checking first." }, + { + type: "text-end", + id: "msg_commentary", + providerMetadata: { openai: { itemId: "msg_commentary", phase: "commentary" } }, + }, + { + type: "text-start", + id: "msg_final", + providerMetadata: { openai: { itemId: "msg_final", phase: "final_answer" } }, + }, + { type: "text-delta", id: "msg_final", text: "Finished." }, + { + type: "text-end", + id: "msg_final", + providerMetadata: { openai: { itemId: "msg_final", phase: "final_answer" } }, + }, + ]) + expect(response.message.content).toEqual([ + { + type: "text", + text: "Checking first.", + providerMetadata: { openai: { itemId: "msg_commentary", phase: "commentary" } }, + }, + { + type: "text", + text: "Finished.", + providerMetadata: { openai: { itemId: "msg_final", phase: "final_answer" } }, + }, + ]) + }), + ) it.effect("parses reasoning summary stream fixtures", () => Effect.gen(function* () { const body = sseEvents( @@ -1006,6 +1076,51 @@ describe("OpenAI Responses route", () => { }), ) + it.effect("round-trips assistant message phases", () => + Effect.gen(function* () { + const prepared = yield* LLMClient.prepare( + LLM.request({ + model, + messages: [ + Message.assistant([ + { + type: "text", + text: "Checking first.", + providerMetadata: { openai: { itemId: "msg_commentary", phase: "commentary" } }, + }, + { + type: "text", + text: "Still checking.", + providerMetadata: { openai: { itemId: "msg_commentary_2", phase: "commentary" } }, + }, + { + type: "text", + text: "Finished.", + providerMetadata: { openai: { itemId: "msg_final", phase: "final_answer" } }, + }, + ]), + ], + }), + ) + + expect(prepared.body.input).toEqual([ + { + role: "assistant", + phase: "commentary", + content: [ + { type: "output_text", text: "Checking first." }, + { type: "output_text", text: "Still checking." }, + ], + }, + { + role: "assistant", + phase: "final_answer", + content: [{ type: "output_text", text: "Finished." }], + }, + ]) + }), + ) + it.effect("references stored reasoning items by id", () => Effect.gen(function* () { const prepared = yield* LLMClient.prepare(