Mirror_Repos/anomalyco_opencode
2
0
Fork 0
Code Issues Pull Requests Actions 106 Packages Projects Releases Wiki Activity

fix(ai): preserve response message phases

Browse Source
This commit is contained in:
Aiden Cline
2026-07-22 23:28:53 -05:00
parent 92cede0541
commit 4375ce1dce
3 changed files with 184 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/llm/src/protocols/openai-responses.ts
+59 -2
View File
@@ -48,6 +48,9 @@ const OpenAIResponsesOutputText = Schema.Struct({
text: Schema.String,
})
const OpenAIResponsesMessagePhase = Schema.Literals(["commentary", "final_answer"])
type OpenAIResponsesMessagePhase = Schema.Schema.Type<typeof OpenAIResponsesMessagePhase>
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<typeof OpenAIResponsesStreamItem>
@@ -237,6 +245,7 @@ interface ParserState {
readonly tools: ToolStream.State<string>
readonly hasFunctionCall: boolean
readonly lifecycle: Lifecycle.State
readonly messageItems: Readonly<Record<string, ProviderMetadata>>
readonly reasoningItems: Readonly<Record<string, ReasoningStreamItem>>
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<string, OpenAIResponsesReasoningReplay> = {}
const reasoningReferences = new Set<string>()
const hostedToolReferences = new Set<string>()
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<string>(),
lifecycle: Lifecycle.initial(),
messageItems: {},
reasoningItems: {},
store: OpenAIOptions.store(request),
}),
packages/llm/src/protocols/utils/lifecycle.ts
+10 -7
View File
@@ -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 }
packages/llm/test/provider/openai-responses.test.ts
+115
View File
@@ -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<OpenAIResponses.OpenAIResponsesBody>(
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<OpenAIResponses.OpenAIResponsesBody>(