fix(ai): reconcile terminal response items

packages/ai/src/protocols/open-responses.ts

@@ -193,6 +193,14 @@ const OpenResponsesUsage = Schema.Struct({
 })
 type OpenResponsesUsage = Schema.Schema.Type<typeof OpenResponsesUsage>
 
+const StreamContent = Schema.StructWithRest(
+  Schema.Struct({
+    type: Schema.String,
+    text: Schema.optional(Schema.String),
+  }),
+  [Schema.Record(Schema.String, Schema.Unknown)],
+)
+
 export const StreamItem = Schema.StructWithRest(
   Schema.Struct({
     type: Schema.String,
@@ -201,6 +209,8 @@ export const StreamItem = Schema.StructWithRest(
     name: Schema.optional(Schema.String),
     arguments: Schema.optional(Schema.String),
     encrypted_content: optionalNull(Schema.String),
+    content: optionalArray(StreamContent),
+    summary: optionalArray(StreamContent),
   }),
   [Schema.Record(Schema.String, Schema.Unknown)],
 )
@@ -221,7 +231,10 @@ export const Event = Schema.StructWithRest(
     type: Schema.String,
     delta: Schema.optional(Schema.String),
     text: Schema.optional(Schema.String),
+    arguments: Schema.optional(Schema.String),
     item_id: Schema.optional(Schema.String),
+    output_index: Schema.optional(Schema.Number),
+    content_index: Schema.optional(Schema.Number),
     summary_index: Schema.optional(Schema.Number),
     item: Schema.optional(StreamItem),
     response: Schema.optional(
@@ -232,6 +245,7 @@ export const Event = Schema.StructWithRest(
           incomplete_details: optionalNull(Schema.Struct({ reason: Schema.optional(Schema.String) })),
           usage: optionalNull(OpenResponsesUsage),
           error: optionalNull(OpenResponsesErrorPayload),
+          output: optionalArray(StreamItem),
         }),
         [Schema.Record(Schema.String, Schema.Unknown)],
       ),
@@ -264,10 +278,16 @@ export interface ParserState {
   readonly providerMetadataKey: string
   readonly tools: ToolStream.State<string>
   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
   readonly messagePhases: Readonly<Record<string, MessagePhase | null>>
+  readonly outputText: Readonly<Record<string, string>>
+  readonly reasoningText: Readonly<Record<string, string>>
   readonly reasoningItems: Readonly<Record<string, ReasoningStreamItem>>
   readonly store: boolean | undefined
 }
@@ -642,24 +662,60 @@ 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 })
   const lifecycle = Lifecycle.textStart(state.lifecycle, events, id, metadata)
   return [
-    { ...state, lifecycle: Lifecycle.textDelta(lifecycle, events, id, event.delta) },
+    {
+      ...state,
+      lifecycle: Lifecycle.textDelta(lifecycle, events, id, event.delta),
+      outputText: { ...state.outputText, [key]: `${state.outputText[key] ?? ""}${event.delta}` },
+    },
     events,
   ]
 }
 
-const onOutputTextDone = (state: ParserState, event: Event, id: string): StepResult => {
+const authoritativeSuffix = Effect.fn("OpenResponses.authoritativeSuffix")(function* (
-  if (state.messageItems.has(id)) {
+  state: ParserState,
-    if (state.lifecycle.text.has(id) || event.text === undefined) return [state, NO_EVENTS]
+  kind: string,
-    return onOutputTextDelta(state, { ...event, delta: event.text }, id)
+  id: string,
-  }
+  current: string,
+  value: string,
+) {
+  if (value.startsWith(current)) return value.slice(current.length)
+  return yield* ProviderShared.eventError(
+    state.id,
+    `${kind} ${id} completed with content that conflicts with its streamed deltas`,
+    value,
+  )
+})
+
+const onOutputTextDone = Effect.fn("OpenResponses.onOutputTextDone")(function* (
+  state: ParserState,
+  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", 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[] = []
-  return [{ ...state, lifecycle: Lifecycle.textEnd(state.lifecycle, events, id) }, events]
+  const lifecycle = Lifecycle.textEnd(reconciled.lifecycle, events, id)
-}
+  return [{ ...reconciled, lifecycle }, [...deltaEvents, ...events]] satisfies StepResult
+})
 
 export const onReasoningDelta = (state: ParserState, event: Event, itemID: string): StepResult => {
   if (!event.delta) return [state, NO_EVENTS]
@@ -670,12 +726,39 @@ export const onReasoningDelta = (state: ParserState, event: Event, itemID: strin
     {
       ...state,
       lifecycle: Lifecycle.reasoningDelta(state.lifecycle, events, id, event.delta),
+      reasoningText: { ...state.reasoningText, [id]: `${state.reasoningText[id] ?? ""}${event.delta}` },
     },
     events,
   ]
 }
 
-export const onReasoningDone = (state: ParserState, _event: Event): StepResult => [state, NO_EVENTS]
+export const onReasoningDone = Effect.fn("OpenResponses.onReasoningDone")(function* (
+  state: ParserState,
+  event: Event,
+) {
+  if (!event.item_id || event.text === undefined) return [state, NO_EVENTS] satisfies StepResult
+  const id =
+    event.summary_index !== undefined || state.reasoningItems[event.item_id]
+      ? `${event.item_id}:${event.summary_index ?? 0}`
+      : event.item_id
+  const suffix = yield* authoritativeSuffix(state, "reasoning", id, state.reasoningText[id] ?? "", event.text)
+  const [reconciled, events] = onReasoningDelta(state, { ...event, delta: suffix }, event.item_id)
+  const item = reconciled.reasoningItems[event.item_id]
+  if (!suffix || !item || event.summary_index === undefined) return [reconciled, events] satisfies StepResult
+  return [
+    {
+      ...reconciled,
+      reasoningItems: {
+        ...reconciled.reasoningItems,
+        [event.item_id]: {
+          ...item,
+          summaryParts: { ...item.summaryParts, [event.summary_index]: "active" },
+        },
+      },
+    },
+    events,
+  ] satisfies StepResult
+})
 
 const reasoningMetadata = (state: ParserState, item: StreamItem & { id: string }) =>
   providerMetadata(state, { itemId: item.id, reasoningEncryptedContent: item.encrypted_content ?? null })
@@ -858,38 +941,78 @@ 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 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[] = []
-    const messageItems = new Set(state.messageItems)
     messageItems.delete(item.id)
-    const { [item.id]: _phase, ...messagePhases } = state.messagePhases
+    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 &&
+      Object.keys(state.outputText).some((id) => id === item.id || id.startsWith(`${item.id}:`)) &&
+      metadata
+    )
+      events.push(LLMEvent.textEnd({ id: item.id, providerMetadata: metadata }))
     return [
       {
-        ...state,
+        ...reconciled,
-        lifecycle: Lifecycle.textEnd(
+        lifecycle,
-          state.lifecycle,
-          events,
-          item.id,
-          phase === undefined ? undefined : providerMetadata(state, { phase }),
-        ),
         messageItems,
-        messagePhases,
+        messagePhases: remainingPhases,
+        completedItems: new Set([...reconciled.completedItems, item.id]),
       },
-      events,
+      [...reconciledEvents, ...events],
     ] satisfies StepResult
   }
 
   if (item.type === "function_call") {
     if (!item.id || !item.call_id || !item.name) return [state, NO_EVENTS] satisfies StepResult
+    if (state.completedItems.has(item.id)) {
+      if (item.arguments === undefined || item.arguments === state.functionArguments[item.id])
+        return [state, NO_EVENTS] satisfies StepResult
+      return yield* ProviderShared.eventError(
+        state.id,
+        `function call arguments ${item.id} completed with content that conflicts with its previous snapshot`,
+        item.arguments,
+      )
+    }
     const tools = state.tools[item.id]
       ? state.tools
       : ToolStream.start(state.tools, item.id, { id: item.call_id, name: item.name })
+    const authoritativeArguments = item.arguments ?? tools[item.id]?.input
     const result =
       item.arguments === undefined
         ? yield* ToolStream.finish(state.id, tools, item.id)
@@ -905,6 +1028,11 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* (
         hasFunctionCall:
           resultEvents.some((event) => LLMEvent.is.toolCall(event) || LLMEvent.is.toolInputError(event)) ||
           state.hasFunctionCall,
+        completedItems: new Set([...state.completedItems, item.id]),
+        functionArguments: {
+          ...state.functionArguments,
+          ...(authoritativeArguments === undefined ? {} : { [item.id]: authoritativeArguments }),
+        },
         tools: result.tools,
       },
       events,
@@ -914,25 +1042,147 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* (
   if (isReasoningItem(item)) {
     const events: LLMEvent[] = []
     const metadata = reasoningMetadata(state, item)
-    const reasoningItem = state.reasoningItems[item.id]
+    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,
+        item_id: item.id,
+        text: summaries[0]?.text,
+      })
+      events.push(...reconciledEvents)
+      return [
+        {
+          ...reconciled,
+          lifecycle: Lifecycle.reasoningEnd(reconciled.lifecycle, events, item.id, metadata),
+          completedItems: new Set([...reconciled.completedItems, item.id]),
+        },
+        events,
+      ] satisfies StepResult
+    }
+    const needsSummaryReconciliation = summaries.some(
+      (part, index) => part.text !== state.reasoningText[`${item.id}:${index}`],
+    )
+    const seeded =
+      needsSummaryReconciliation && !state.reasoningItems[item.id]
+        ? onOutputItemAdded(state, { ...event, item })
+        : ([state, NO_EVENTS] satisfies StepResult)
+    events.push(...seeded[1])
+    const reconciled = yield* (needsSummaryReconciliation ? summaries : [])
+      .map((part, index) => ({ part, index }))
+      .reduce<Effect.Effect<ParserState, LLMError>>(
+        (effect, entry) =>
+          effect.pipe(
+            Effect.flatMap(
+              Effect.fnUntraced(function* (current) {
+                const added = current.reasoningItems[item.id]?.summaryParts[entry.index]
+                  ? ([current, NO_EVENTS] satisfies StepResult)
+                  : onReasoningSummaryPartAdded(current, {
+                      ...event,
+                      item_id: item.id,
+                      summary_index: entry.index,
+                    })
+                events.push(...added[1])
+                const [next, nextEvents] = yield* onReasoningDone(added[0], {
+                  ...event,
+                  item_id: item.id,
+                  summary_index: entry.index,
+                  text: entry.part.text,
+                })
+                events.push(...nextEvents)
+                const [done, doneEvents] = onReasoningSummaryPartDone(next, {
+                  ...event,
+                  item_id: item.id,
+                  summary_index: entry.index,
+                })
+                events.push(...doneEvents)
+                return done
+              }),
+            ),
+          ),
+        Effect.succeed(seeded[0]),
+      )
+    const reasoningItem = reconciled.reasoningItems[item.id]
     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,
+          reconciled.lifecycle,
         )
-      const { [item.id]: _removed, ...reasoningItems } = state.reasoningItems
+      const { [item.id]: _removed, ...reasoningItems } = reconciled.reasoningItems
-      return [{ ...state, lifecycle, reasoningItems }, events] satisfies StepResult
+      return [
+        {
+          ...reconciled,
+          lifecycle,
+          reasoningItems,
+          completedItems: new Set([...reconciled.completedItems, item.id]),
+        },
+        events,
+      ] satisfies StepResult
     }
-    if (!state.lifecycle.reasoning.has(item.id)) {
+    if (summaries.length) {
-      const lifecycle = Lifecycle.stepStart(state.lifecycle, events)
+      events.push(LLMEvent.reasoningEnd({ id: `${item.id}:${summaries.length - 1}`, providerMetadata: metadata }))
+      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 [{ ...state, lifecycle }, events] satisfies StepResult
+      return [
+        { ...reconciled, lifecycle, completedItems: new Set([...reconciled.completedItems, item.id]) },
+        events,
+      ] satisfies StepResult
     }
     return [
-      { ...state, lifecycle: Lifecycle.reasoningEnd(state.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
   }
@@ -940,11 +1190,47 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* (
   return [state, NO_EVENTS] satisfies StepResult
 })
 
-const onResponseFinish = (state: ParserState, event: Event): StepResult => {
+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 lifecycle = Lifecycle.finish(state.lifecycle, events, {
+  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* onItemDone(current, { ...event, item })
+            events.push(...nextEvents)
+            return next
+          }),
+        ),
+      ),
+    Effect.succeed(state),
+  )
+  const lifecycle = Lifecycle.finish(reconciled.lifecycle, events, {
     reason: {
-      normalized: mapFinishReason(event, state.hasFunctionCall),
+      normalized: mapFinishReason(event, reconciled.hasFunctionCall),
       raw: event.response?.incomplete_details?.reason,
     },
     usage: mapUsage(event.response?.usage, state.providerMetadataKey),
@@ -956,8 +1242,8 @@ const onResponseFinish = (state: ParserState, event: Event): StepResult => {
           })
         : undefined,
   })
-  return [{ ...state, lifecycle }, events]
+  return [{ ...reconciled, lifecycle }, events] satisfies StepResult
-}
+})
 
 // Build a single human-readable message from whatever the provider supplied.
 // When both code and message are present, prefix the code so consumers see
@@ -982,14 +1268,26 @@ 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 Effect.succeed(
+    return event.type === "response.output_text.delta"
-      event.type === "response.output_text.delta"
+      ? Effect.succeed(onOutputTextDelta(state, event, event.item_id))
-        ? onOutputTextDelta(state, event, event.item_id)
+      : onOutputTextDone(state, event, event.item_id)
-        : onOutputTextDone(state, event, event.item_id),
-    )
   }
   if (event.type === "response.reasoning.delta" || event.type === "response.reasoning_summary_text.delta") {
     if (!event.item_id) return ProviderShared.eventError(state.id, `${event.type} is missing item_id`)
@@ -997,7 +1295,7 @@ export const step = (state: ParserState, event: Event) => {
   }
   if (event.type === "response.reasoning.done" || event.type === "response.reasoning_summary_text.done") {
     if (!event.item_id) return ProviderShared.eventError(state.id, `${event.type} is missing item_id`)
-    return Effect.succeed(onReasoningDone(state, event))
+    return onReasoningDone(state, event)
   }
   if (event.type === "response.reasoning_summary_part.added")
     return event.item_id
@@ -1013,13 +1311,29 @@ export const step = (state: ParserState, event: Event) => {
     return Effect.succeed(onOutputItemAdded(state, event))
   }
   if (event.type === "response.function_call_arguments.delta") return onFunctionCallArgumentsDelta(state, event)
+  if (event.type === "response.function_call_arguments.done") {
+    if (!event.item_id) return ProviderShared.eventError(state.id, `${event.type} is missing item_id`)
+    if (event.arguments === undefined) return ProviderShared.eventError(state.id, `${event.type} is missing arguments`)
+    const tool = state.tools[event.item_id]
+    if (!tool)
+      return ProviderShared.eventError(state.id, `${state.name} completed tool arguments are missing their tool call`)
+    return Effect.succeed(
+      [
+        {
+          ...state,
+          tools: { ...state.tools, [event.item_id]: { ...tool, input: event.arguments } },
+        },
+        NO_EVENTS,
+      ] satisfies StepResult,
+    )
+  }
   if (event.type === "response.output_item.done") {
     if (event.item?.type === "message" && !event.item.id)
       return ProviderShared.eventError(state.id, `${event.type} message is missing id`)
     return onOutputItemDone(state, event)
   }
   if (event.type === "response.completed" || event.type === "response.incomplete")
-    return Effect.succeed(onResponseFinish(state, event))
+    return onResponseFinish(state, event)
   if (event.type === "response.failed") return providerError(state, event, `${state.name} response failed`)
   if (event.type === "error") return providerError(state, event, `${state.name} stream error`)
   return Effect.succeed<StepResult>([state, NO_EVENTS])
@@ -1037,11 +1351,17 @@ export const initial = (request: LLMRequest, extension: Extension = BASE): Parse
   name: extension.name,
   providerMetadataKey: request.model.route.providerMetadataKey ?? "openresponses",
   hasFunctionCall: false,
+  completedItems: new Set<string>(),
+  functionArguments: {},
+  outputIndexes: {},
+  outputSequence: [],
   tools: ToolStream.empty<string>(),
   lifecycle: Lifecycle.initial(),
   messageItems: new Set<string>(),
   messagePhase: (value) => messagePhase(value, extension),
   messagePhases: {},
+  outputText: {},
+  reasoningText: {},
   reasoningItems: {},
   store: OpenResponsesOptions.resolve(request).store,
 })

packages/ai/src/protocols/openai-responses.ts

@@ -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,20 +215,39 @@ 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))
       : ProviderShared.eventError(ADAPTER, `${event.type} is missing item_id`)
   if (event.type === "response.reasoning_text.done" || event.type === "response.reasoning_summary.done")
     return event.item_id
-      ? Effect.succeed(OpenResponses.onReasoningDone(state, event))
+      ? 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))
+  if (event.type === "response.output_item.done") return onOutputItemDone(state, event)
-    return onHostedToolDone(state, event.item)
+  if (event.type === "response.completed" || event.type === "response.incomplete")
+    return OpenResponses.onResponseFinish(state, event, onOutputItemDone)
   return OpenResponses.step(state, event)
 }
 

packages/ai/test/provider/openai-responses.test.ts

@@ -882,6 +882,185 @@ describe("OpenAI Responses route", () => {
     }),
   )
 
+  it.effect("reconciles authoritative output text done values", () =>
+    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", delta: "Hello" },
+              { type: "response.output_text.done", item_id: "msg_1", text: "Hello!" },
+              {
+                type: "response.output_item.done",
+                item: {
+                  type: "message",
+                  id: "msg_1",
+                  content: [{ type: "output_text", text: "Hello!" }],
+                },
+              },
+              { type: "response.completed", response: { id: "resp_1" } },
+            ),
+          ),
+        ),
+      )
+
+      expect(response.text).toBe("Hello!")
+      expect(response.events.filter(LLMEvent.is.textDelta)).toEqual([
+        { type: "text-delta", id: "msg_1", text: "Hello" },
+        { type: "text-delta", id: "msg_1", text: "!" },
+      ])
+    }),
+  )
+
+  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(
+        Effect.provide(
+          fixedResponse(
+            sseEvents(
+              {
+                type: "response.completed",
+                response: {
+                  id: "resp_1",
+                  output: [
+                    {
+                      type: "message",
+                      id: "msg_1",
+                      content: [{ type: "output_text", text: "Terminal only." }],
+                    },
+                  ],
+                },
+              },
+            ),
+          ),
+        ),
+      )
+
+      expect(response.text).toBe("Terminal only.")
+      expect(response.events.filter(LLMEvent.is.textDelta)).toEqual([
+        { type: "text-delta", id: "msg_1", text: "Terminal only." },
+      ])
+    }),
+  )
+
+  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(
+        Effect.provide(
+          fixedResponse(
+            sseEvents(
+              { type: "response.output_text.delta", item_id: "msg_1", delta: "Hello" },
+              { type: "response.output_text.done", item_id: "msg_1", text: "Goodbye" },
+            ),
+          ),
+        ),
+        Effect.flip,
+      )
+
+      expect(error.reason._tag).toBe("InvalidProviderOutput")
+      expect(error.message).toContain("completed with content that conflicts with its streamed deltas")
+    }),
+  )
+
   it.effect("preserves and replays assistant message phases", () =>
     Effect.gen(function* () {
       const response = yield* LLMClient.generate(request).pipe(
@@ -1086,6 +1265,110 @@ describe("OpenAI Responses route", () => {
     }),
   )
 
+  it.effect("reconciles authoritative reasoning done values", () =>
+    Effect.gen(function* () {
+      const response = yield* LLMClient.generate(request).pipe(
+        Effect.provide(
+          fixedResponse(
+            sseEvents(
+              { type: "response.reasoning_summary_text.delta", item_id: "rs_1", delta: "thinking" },
+              { type: "response.reasoning_summary_text.done", item_id: "rs_1", text: "thinking..." },
+              { type: "response.completed", response: { id: "resp_1" } },
+            ),
+          ),
+        ),
+      )
+
+      expect(response.reasoning).toBe("thinking...")
+    }),
+  )
+
+  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(
@@ -1568,6 +1851,57 @@ describe("OpenAI Responses route", () => {
     }),
   )
 
+  it.effect("uses authoritative function argument done values", () =>
+    Effect.gen(function* () {
+      const response = yield* LLMClient.generate(
+        LLMRequest.update(request, {
+          tools: [ToolDefinition.make({ name: "lookup", description: "Lookup data", inputSchema: { type: "object" } })],
+        }),
+      ).pipe(
+        Effect.provide(
+          fixedResponse(
+            sseEvents(
+              {
+                type: "response.output_item.added",
+                item: { type: "function_call", id: "item_1", call_id: "call_1", name: "lookup", arguments: "" },
+              },
+              { type: "response.function_call_arguments.delta", item_id: "item_1", delta: '{"query":"stale"}' },
+              {
+                type: "response.function_call_arguments.done",
+                item_id: "item_1",
+                arguments: '{"query":"authoritative"}',
+              },
+              {
+                type: "response.output_item.done",
+                item: { type: "function_call", id: "item_1", call_id: "call_1", name: "lookup" },
+              },
+              {
+                type: "response.completed",
+                response: {
+                  output: [
+                    {
+                      type: "function_call",
+                      id: "item_1",
+                      call_id: "call_1",
+                      name: "lookup",
+                      arguments: '{"query":"authoritative"}',
+                    },
+                  ],
+                },
+              },
+            ),
+          ),
+        ),
+      )
+
+      expect(response.events.find(LLMEvent.is.toolCall)).toMatchObject({
+        id: "call_1",
+        input: { query: "authoritative" },
+      })
+      expect(response.events.filter(LLMEvent.is.toolCall)).toHaveLength(1)
+    }),
+  )
+
   it.effect("emits malformed final function arguments as an unexecuted tool error", () =>
     Effect.gen(function* () {
       const body = sseEvents(
@@ -1642,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)))
 
@@ -1701,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(