ceos/opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(ai): reconcile terminal response items

Browse source
This commit is contained in:
Aiden Cline 2026-07-25 14:45:15 -05:00
commit 2f6a907a1a
3 changed files with 411 additions and 26 deletions
Download patch file Download diff file Expand all files Collapse all files

188
packages/ai/src/protocols/open-responses.ts
View file

@ -280,6 +280,8 @@ export interface ParserState {
readonly hasFunctionCall: boolean
readonly completedItems: ReadonlySet<string>
readonly functionArguments: Readonly<Record<string, string>>
readonly outputIndexes: Readonly<Record<string, number>>
readonly outputSequence: ReadonlyArray<string>
readonly lifecycle: Lifecycle.State
readonly messageItems: ReadonlySet<string>
readonly messagePhase: (value: unknown) => MessagePhase | null | undefined
@ -660,6 +662,12 @@ export const terminal = (event: Event) => TERMINAL_TYPES.has(event.type)
const onOutputTextDelta = (state: ParserState, event: Event, id: string): StepResult => {
if (!event.delta) return [state, NO_EVENTS]
const key =
event.content_index === undefined
? id
: event.content_index === 0 && state.outputText[`${id}:0`] === undefined && state.outputText[id] !== undefined
? id
: `${id}:${event.content_index}`
const events: LLMEvent[] = []
const phase = state.messagePhases[id]
const metadata = phase === undefined ? undefined : providerMetadata(state, { phase })
@ -668,7 +676,7 @@ const onOutputTextDelta = (state: ParserState, event: Event, id: string): StepRe
{
...state,
lifecycle: Lifecycle.textDelta(lifecycle, events, id, event.delta),
outputText: { ...state.outputText, [id]: `${state.outputText[id] ?? ""}${event.delta}` },
outputText: { ...state.outputText, [key]: `${state.outputText[key] ?? ""}${event.delta}` },
},
events,
]
@ -694,8 +702,14 @@ const onOutputTextDone = Effect.fn("OpenResponses.onOutputTextDone")(function* (
event: Event,
id: string,
) {
const key =
event.content_index === undefined
? id
: event.content_index === 0 && state.outputText[`${id}:0`] === undefined && state.outputText[id] !== undefined
? id
: `${id}:${event.content_index}`
const suffix =
event.text === undefined ? "" : yield* authoritativeSuffix(state, "output text", id, state.outputText[id] ?? "", event.text)
event.text === undefined ? "" : yield* authoritativeSuffix(state, "output text", key, state.outputText[key] ?? "", event.text)
const [reconciled, deltaEvents] = onOutputTextDelta(state, { ...event, delta: suffix }, id)
if (reconciled.messageItems.has(id)) return [reconciled, deltaEvents] satisfies StepResult
const events: LLMEvent[] = []
@ -927,29 +941,50 @@ const onFunctionCallArgumentsDelta = Effect.fn("OpenResponses.onFunctionCallArgu
return [{ ...state, lifecycle, tools: result.tools }, events] satisfies StepResult
})
const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* (state: ParserState, event: Event) {
export const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* (
state: ParserState,
event: Event,
) {
const item = event.item
if (!item) return [state, NO_EVENTS] satisfies StepResult
if (item.type === "message" && item.id) {
const itemID = item.id
const itemPhase = state.messagePhase(item.phase)
const phase = itemPhase === undefined ? state.messagePhases[item.id] : itemPhase
const messageItems = new Set([...state.messageItems, item.id])
const messagePhases = phase === undefined ? state.messagePhases : { ...state.messagePhases, [item.id]: phase }
const text = item.content
?.filter((part) => part.type === "output_text" && part.text !== undefined)
.map((part) => part.text)
.join("")
const [reconciled, reconciledEvents] =
text === undefined
? [{ ...state, messageItems, messagePhases }, NO_EVENTS]
: yield* onOutputTextDone({ ...state, messageItems, messagePhases }, { ...event, text }, item.id)
const reconciledEvents: LLMEvent[] = []
const reconciled = yield* (item.content ?? [])
.map((part, contentIndex) => ({ part, contentIndex }))
.filter((entry) => entry.part.type === "output_text" && entry.part.text !== undefined)
.reduce<Effect.Effect<ParserState, LLMError>>(
(effect, entry) =>
effect.pipe(
Effect.flatMap(
Effect.fnUntraced(function* (current) {
const [next, nextEvents] = yield* onOutputTextDone(
current,
{ ...event, content_index: entry.contentIndex, text: entry.part.text ?? "" },
itemID,
)
reconciledEvents.push(...nextEvents)
return next
}),
),
),
Effect.succeed({ ...state, messageItems, messagePhases }),
)
const events: LLMEvent[] = []
messageItems.delete(item.id)
const { [item.id]: _phase, ...remainingPhases } = reconciled.messagePhases
const metadata = phase === undefined ? undefined : providerMetadata(state, { phase })
const lifecycle = Lifecycle.textEnd(reconciled.lifecycle, events, item.id, metadata)
if (!events.length && state.outputText[item.id] !== undefined && metadata)
if (
!events.length &&
Object.keys(state.outputText).some((id) => id === item.id || id.startsWith(`${item.id}:`)) &&
metadata
)
events.push(LLMEvent.textEnd({ id: item.id, providerMetadata: metadata }))
return [
{
@ -957,6 +992,7 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* (
lifecycle,
messageItems,
messagePhases: remainingPhases,
completedItems: new Set([...reconciled.completedItems, item.id]),
},
[...reconciledEvents, ...events],
] satisfies StepResult
@ -1007,6 +1043,47 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* (
const events: LLMEvent[] = []
const metadata = reasoningMetadata(state, item)
const summaries = item.summary?.filter((part) => part.type === "summary_text" && part.text !== undefined) ?? []
if (state.completedItems.has(item.id)) {
const reconciled = yield* summaries
.map((part, index) => ({ part, index }))
.reduce<Effect.Effect<ParserState, LLMError>>(
(effect, entry) =>
effect.pipe(
Effect.flatMap(
Effect.fnUntraced(function* (current) {
const [next, nextEvents] = yield* onReasoningDone(current, {
...event,
item_id: item.id,
summary_index: entry.index,
text: entry.part.text,
})
events.push(...nextEvents)
const endEvents: LLMEvent[] = []
const lifecycle = Lifecycle.reasoningEnd(
next.lifecycle,
endEvents,
`${item.id}:${entry.index}`,
metadata,
)
events.push(
...(endEvents.length
? endEvents
: [LLMEvent.reasoningEnd({ id: `${item.id}:${entry.index}`, providerMetadata: metadata })]),
)
return { ...next, lifecycle }
}),
),
),
Effect.succeed(state),
)
if (!summaries.length) {
const id = Object.keys(state.reasoningText)
.filter((id) => id === item.id || id.startsWith(`${item.id}:`))
.at(-1) ?? `${item.id}:0`
events.push(LLMEvent.reasoningEnd({ id, providerMetadata: metadata }))
}
return [reconciled, events] satisfies StepResult
}
if (summaries.length === 1 && !state.reasoningItems[item.id] && state.reasoningText[item.id] !== undefined) {
const [reconciled, reconciledEvents] = yield* onReasoningDone(state, {
...event,
@ -1015,7 +1092,11 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* (
})
events.push(...reconciledEvents)
return [
{ ...reconciled, lifecycle: Lifecycle.reasoningEnd(reconciled.lifecycle, events, item.id, metadata) },
{
...reconciled,
lifecycle: Lifecycle.reasoningEnd(reconciled.lifecycle, events, item.id, metadata),
completedItems: new Set([...reconciled.completedItems, item.id]),
},
events,
] satisfies StepResult
}
@ -1049,7 +1130,13 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* (
text: entry.part.text,
})
events.push(...nextEvents)
return next
const [done, doneEvents] = onReasoningSummaryPartDone(next, {
...event,
item_id: item.id,
summary_index: entry.index,
})
events.push(...doneEvents)
return done
}),
),
),
@ -1064,20 +1151,38 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* (
reconciled.lifecycle,
)
const { [item.id]: _removed, ...reasoningItems } = reconciled.reasoningItems
return [{ ...reconciled, lifecycle, reasoningItems }, events] satisfies StepResult
return [
{
...reconciled,
lifecycle,
reasoningItems,
completedItems: new Set([...reconciled.completedItems, item.id]),
},
events,
] satisfies StepResult
}
if (summaries.length) {
events.push(LLMEvent.reasoningEnd({ id: `${item.id}:${summaries.length - 1}`, providerMetadata: metadata }))
return [reconciled, events] satisfies StepResult
return [
{ ...reconciled, completedItems: new Set([...reconciled.completedItems, item.id]) },
events,
] satisfies StepResult
}
if (!reconciled.lifecycle.reasoning.has(item.id)) {
const lifecycle = Lifecycle.stepStart(reconciled.lifecycle, events)
events.push(LLMEvent.reasoningStart({ id: item.id, providerMetadata: metadata }))
events.push(LLMEvent.reasoningEnd({ id: item.id, providerMetadata: metadata }))
return [{ ...reconciled, lifecycle }, events] satisfies StepResult
return [
{ ...reconciled, lifecycle, completedItems: new Set([...reconciled.completedItems, item.id]) },
events,
] satisfies StepResult
}
return [
{ ...reconciled, lifecycle: Lifecycle.reasoningEnd(reconciled.lifecycle, events, item.id, metadata) },
{
...reconciled,
lifecycle: Lifecycle.reasoningEnd(reconciled.lifecycle, events, item.id, metadata),
completedItems: new Set([...reconciled.completedItems, item.id]),
},
events,
] satisfies StepResult
}
@ -1085,14 +1190,37 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* (
return [state, NO_EVENTS] satisfies StepResult
})
const onResponseFinish = Effect.fn("OpenResponses.onResponseFinish")(function* (state: ParserState, event: Event) {
export const onResponseFinish = Effect.fn("OpenResponses.onResponseFinish")(function* (
state: ParserState,
event: Event,
onItemDone: (state: ParserState, event: Event) => Effect.Effect<StepResult, LLMError> = onOutputItemDone,
) {
const events: LLMEvent[] = []
const reconciled = yield* (event.response?.output ?? []).reduce<Effect.Effect<ParserState, LLMError>>(
const output = event.response?.output ?? []
const terminalIndexes = Object.fromEntries(
output.flatMap((item, index) => (item.id === undefined ? [] : [[item.id, index] as const])),
)
const emitted = state.outputSequence.filter((id) => terminalIndexes[id] !== undefined)
const expected = output.flatMap((item) =>
item.id !== undefined && state.outputIndexes[item.id] !== undefined ? [item.id] : [],
)
const missingBeforeEmitted = output.some(
(item, index) =>
item.id !== undefined &&
state.outputIndexes[item.id] === undefined &&
Object.values(state.outputIndexes).some((seen) => seen > index),
)
if (missingBeforeEmitted || emitted.some((id, index) => id !== expected[index]))
return yield* ProviderShared.eventError(
state.id,
`${state.name} terminal output conflicts with the streamed output order`,
)
const reconciled = yield* output.reduce<Effect.Effect<ParserState, LLMError>>(
(effect, item) =>
effect.pipe(
Effect.flatMap(
Effect.fnUntraced(function* (current) {
const [next, nextEvents] = yield* onOutputItemDone(current, { ...event, item })
const [next, nextEvents] = yield* onItemDone(current, { ...event, item })
events.push(...nextEvents)
return next
}),
@ -1140,7 +1268,21 @@ const providerError = (state: ParserState, event: Event, fallback: string) => {
})
}
export const step = (state: ParserState, event: Event) => {
export const step = (state: ParserState, event: Event): Effect.Effect<StepResult, LLMError> => {
const outputItemID = event.item_id ?? event.item?.id
if (
outputItemID &&
event.output_index !== undefined &&
state.outputIndexes[outputItemID] === undefined
)
return step(
{
...state,
outputIndexes: { ...state.outputIndexes, [outputItemID]: event.output_index },
outputSequence: [...state.outputSequence, outputItemID],
},
{ ...event, output_index: undefined },
)
if (event.type === "response.output_text.delta" || event.type === "response.output_text.done") {
if (!event.item_id) return ProviderShared.eventError(state.id, `${event.type} is missing item_id`)
return event.type === "response.output_text.delta"
@ -1211,6 +1353,8 @@ export const initial = (request: LLMRequest, extension: Extension = BASE): Parse
hasFunctionCall: false,
completedItems: new Set<string>(),
functionArguments: {},
outputIndexes: {},
outputSequence: [],
tools: ToolStream.empty<string>(),
lifecycle: Lifecycle.initial(),
messageItems: new Set<string>(),

26
packages/ai/src/protocols/openai-responses.ts
View file

@ -194,6 +194,7 @@ const onHostedToolDone = Effect.fn("OpenAIResponses.onHostedToolDone")(function*
state: OpenResponses.ParserState,
item: HostedToolItem,
) {
if (state.completedItems.has(item.id)) return [state, []] satisfies OpenResponses.StepResult
const tool = HOSTED_TOOLS[item.type]
const providerMetadata = OpenResponses.providerMetadata(state, { itemId: item.id })
const events: LLMEvent[] = []
@ -214,10 +215,28 @@ const onHostedToolDone = Effect.fn("OpenAIResponses.onHostedToolDone")(function*
providerMetadata,
}),
)
return [{ ...state, lifecycle }, events] satisfies OpenResponses.StepResult
return [
{ ...state, lifecycle, completedItems: new Set([...state.completedItems, item.id]) },
events,
] satisfies OpenResponses.StepResult
})
const onOutputItemDone = (state: OpenResponses.ParserState, event: OpenResponses.Event) =>
event.item && isHostedToolItem(event.item)
? onHostedToolDone(state, event.item)
: OpenResponses.onOutputItemDone(state, event)
const step = (state: OpenResponses.ParserState, event: OpenResponses.Event) => {
const outputItemID = event.item_id ?? event.item?.id
if (outputItemID && event.output_index !== undefined && state.outputIndexes[outputItemID] === undefined)
return step(
{
...state,
outputIndexes: { ...state.outputIndexes, [outputItemID]: event.output_index },
outputSequence: [...state.outputSequence, outputItemID],
},
{ ...event, output_index: undefined },
)
if (event.type === "response.reasoning_text.delta" || event.type === "response.reasoning_summary.delta")
return event.item_id
? Effect.succeed(OpenResponses.onReasoningDelta(state, event, event.item_id))
@ -226,8 +245,9 @@ const step = (state: OpenResponses.ParserState, event: OpenResponses.Event) => {
return event.item_id
? OpenResponses.onReasoningDone(state, event)
: ProviderShared.eventError(ADAPTER, `${event.type} is missing item_id`)
if (event.type === "response.output_item.done" && event.item && isHostedToolItem(event.item))
return onHostedToolDone(state, event.item)
if (event.type === "response.output_item.done") return onOutputItemDone(state, event)
if (event.type === "response.completed" || event.type === "response.incomplete")
return OpenResponses.onResponseFinish(state, event, onOutputItemDone)
return OpenResponses.step(state, event)
}

223
packages/ai/test/provider/openai-responses.test.ts
View file

@ -913,6 +913,58 @@ describe("OpenAI Responses route", () => {
}),
)
it.effect("reconciles output text independently by content index", () =>
Effect.gen(function* () {
const response = yield* LLMClient.generate(request).pipe(
Effect.provide(
fixedResponse(
sseEvents(
{ type: "response.output_item.added", item: { type: "message", id: "msg_1" } },
{
type: "response.output_text.delta",
item_id: "msg_1",
content_index: 0,
delta: "First",
},
{
type: "response.output_text.done",
item_id: "msg_1",
content_index: 0,
text: "First.",
},
{
type: "response.output_text.delta",
item_id: "msg_1",
content_index: 1,
delta: "Second",
},
{
type: "response.output_text.done",
item_id: "msg_1",
content_index: 1,
text: "Second.",
},
{
type: "response.output_item.done",
item: {
type: "message",
id: "msg_1",
content: [
{ type: "output_text", text: "First." },
{ type: "output_text", text: "Second." },
],
},
},
{ type: "response.completed", response: {} },
),
),
),
)
expect(response.text).toBe("First.Second.")
}),
)
it.effect("recovers output from the authoritative terminal response", () =>
Effect.gen(function* () {
const response = yield* LLMClient.generate(request).pipe(
@ -944,6 +996,52 @@ describe("OpenAI Responses route", () => {
}),
)
it.effect("rejects terminal output that cannot be restored to authoritative order", () =>
Effect.gen(function* () {
const error = yield* LLMClient.generate(request).pipe(
Effect.provide(
fixedResponse(
sseEvents(
{
type: "response.output_item.added",
output_index: 1,
item: { type: "message", id: "msg_later" },
},
{
type: "response.output_text.delta",
item_id: "msg_later",
output_index: 1,
content_index: 0,
delta: "Later",
},
{
type: "response.completed",
response: {
output: [
{
type: "message",
id: "msg_earlier",
content: [{ type: "output_text", text: "Earlier" }],
},
{
type: "message",
id: "msg_later",
content: [{ type: "output_text", text: "Later" }],
},
],
},
},
),
),
),
Effect.flip,
)
expect(error.reason._tag).toBe("InvalidProviderOutput")
expect(error.message).toContain("terminal output conflicts with the streamed output order")
}),
)
it.effect("rejects authoritative text that conflicts with streamed deltas", () =>
Effect.gen(function* () {
const error = yield* LLMClient.generate(request).pipe(
@ -1185,6 +1283,92 @@ describe("OpenAI Responses route", () => {
}),
)
it.effect("recovers terminal-only reasoning summaries sequentially", () =>
Effect.gen(function* () {
const response = yield* LLMClient.generate(request).pipe(
Effect.provide(
fixedResponse(
sseEvents({
type: "response.completed",
response: {
output: [
{
type: "reasoning",
id: "rs_1",
encrypted_content: "encrypted-state",
summary: [
{ type: "summary_text", text: "First" },
{ type: "summary_text", text: "Second" },
],
},
],
},
}),
),
),
)
expect(response.message.content.filter((part) => part.type === "reasoning")).toEqual([
{
type: "reasoning",
text: "First",
providerMetadata: { openai: { itemId: "rs_1" } },
},
{
type: "reasoning",
text: "Second",
providerMetadata: {
openai: { itemId: "rs_1", reasoningEncryptedContent: "encrypted-state" },
},
},
])
expect(
response.events
.filter((event) => event.type.startsWith("reasoning-"))
.map((event) => `${event.type}:${event.id}`),
).toEqual([
"reasoning-start:rs_1:0",
"reasoning-delta:rs_1:0",
"reasoning-end:rs_1:0",
"reasoning-start:rs_1:1",
"reasoning-delta:rs_1:1",
"reasoning-end:rs_1:1",
])
}),
)
it.effect("updates completed reasoning metadata without adding a phantom block", () =>
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.output_item.done",
item: { type: "reasoning", id: "rs_1", encrypted_content: "stale", summary: [] },
},
{
type: "response.completed",
response: {
output: [{ type: "reasoning", id: "rs_1", encrypted_content: "final", summary: [] }],
},
},
),
),
),
)
expect(response.message.content.filter((part) => part.type === "reasoning")).toEqual([
{
type: "reasoning",
text: "",
providerMetadata: { openai: { itemId: "rs_1", reasoningEncryptedContent: "final" } },
},
])
}),
)
it.effect("preserves encrypted reasoning metadata for continuation", () =>
Effect.gen(function* () {
const response = yield* LLMClient.generate(request).pipe(
@ -1792,7 +1976,10 @@ describe("OpenAI Responses route", () => {
const body = sseEvents(
{ type: "response.output_item.added", item },
{ type: "response.output_item.done", item },
{ type: "response.completed", response: { usage: { input_tokens: 5, output_tokens: 1 } } },
{
type: "response.completed",
response: { output: [item], usage: { input_tokens: 5, output_tokens: 1 } },
},
)
const response = yield* LLMClient.generate(request).pipe(Effect.provide(fixedResponse(body)))
@ -1851,6 +2038,40 @@ describe("OpenAI Responses route", () => {
}),
)
it.effect("recovers hosted tools from the terminal response", () =>
Effect.gen(function* () {
const response = yield* LLMClient.generate(request).pipe(
Effect.provide(
fixedResponse(
sseEvents({
type: "response.completed",
response: {
output: [
{
type: "image_generation_call",
id: "ig_1",
status: "completed",
result: "AQID",
},
],
},
}),
),
),
)
expect(response.events.find(LLMEvent.is.toolResult)).toMatchObject({
id: "ig_1",
name: "image_generation",
providerExecuted: true,
result: {
type: "content",
value: [{ type: "file", uri: "data:image/png;base64,AQID", mime: "image/png" }],
},
})
}),
)
it.effect("rejects malformed image generation base64", () =>
Effect.gen(function* () {
const error = yield* LLMClient.generate(request).pipe(