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

fix(ai): preserve response message phases (#38777)

Browse source
This commit is contained in:
Aiden Cline 2026-07-25 14:06:00 -05:00 • committed by GitHub
commit c5bf4edb10
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 416 additions and 54 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -55,6 +55,9 @@ const OpenResponsesOutputText = Schema.Struct({
text: Schema.String,
})
export const MessagePhase = Schema.Literals(["commentary", "final_answer"])
type MessagePhase = Schema.Schema.Type<typeof MessagePhase>
const OpenResponsesReasoningSummaryText = Schema.Struct({
type: Schema.tag("summary_text"),
text: Schema.String,
@ -86,10 +89,14 @@ const OpenResponsesFunctionCallOutput = Schema.Union([
Schema.Array(OpenResponsesFunctionCallOutputContent),
])
const OpenResponsesInputItem = Schema.Union([
export const InputItem = Schema.Union([
Schema.Struct({ role: Schema.tag("system"), content: Schema.String }),
Schema.Struct({ role: Schema.tag("user"), content: Schema.Array(OpenResponsesInputContent) }),
Schema.Struct({ role: Schema.tag("assistant"), content: Schema.Array(OpenResponsesOutputText) }),
Schema.Struct({
role: Schema.tag("assistant"),
content: Schema.Array(OpenResponsesOutputText),
phase: Schema.optionalKey(MessagePhase),
}),
OpenResponsesReasoningItem,
OpenResponsesItemReference,
Schema.Struct({
@ -104,7 +111,14 @@ const OpenResponsesInputItem = Schema.Union([
output: OpenResponsesFunctionCallOutput,
}),
])
type OpenResponsesInputItem = Schema.Schema.Type<typeof OpenResponsesInputItem>
type OpenResponsesInputItem = Schema.Schema.Type<typeof InputItem>
type LoweredInputItem =
| OpenResponsesInputItem
| {
readonly role: "assistant"
readonly content: ReadonlyArray<{ readonly type: "output_text"; readonly text: string }>
readonly phase?: MessagePhase | null
}
// Mutable counterpart of the schema reasoning item so `lowerMessages` can fold
// multiple streamed summary parts into the same item before flushing.
@ -135,7 +149,7 @@ export const ToolChoice = Schema.Union([
// transports in sync without a destructure-and-strip dance.
export const coreFields = {
model: Schema.String,
input: Schema.Array(OpenResponsesInputItem),
input: Schema.Array(InputItem),
instructions: Schema.optional(Schema.String),
tools: optionalArray(Tool),
tool_choice: Schema.optional(ToolChoice),
@ -206,6 +220,7 @@ export const Event = Schema.StructWithRest(
Schema.Struct({
type: Schema.String,
delta: Schema.optional(Schema.String),
text: Schema.optional(Schema.String),
item_id: Schema.optional(Schema.String),
summary_index: Schema.optional(Schema.Number),
item: Schema.optional(StreamItem),
@ -238,6 +253,7 @@ export interface Extension {
readonly media: ProviderShared.ValidatedMedia
readonly request: LLMRequest
}) => MediaInput | undefined
readonly messagePhase?: (value: unknown) => MessagePhase | null | undefined
}
const BASE: Extension = { id: ADAPTER, name: NAME }
@ -249,6 +265,9 @@ export interface ParserState {
readonly tools: ToolStream.State<string>
readonly hasFunctionCall: boolean
readonly lifecycle: Lifecycle.State
readonly messageItems: ReadonlySet<string>
readonly messagePhase: (value: unknown) => MessagePhase | null | undefined
readonly messagePhases: Readonly<Record<string, MessagePhase | null>>
readonly reasoningItems: Readonly<Record<string, ReasoningStreamItem>>
readonly store: boolean | undefined
}
@ -378,9 +397,9 @@ const lowerToolResultOutput = Effect.fn("OpenResponses.lowerToolResultOutput")(f
})
const lowerMessages = Effect.fn("OpenResponses.lowerMessages")(function* (request: LLMRequest, extension: Extension) {
const system: OpenResponsesInputItem[] =
const system: LoweredInputItem[] =
request.system.length === 0 ? [] : [{ role: "system", content: ProviderShared.joinText(request.system) }]
const input: OpenResponsesInputItem[] = [...system]
const input: LoweredInputItem[] = [...system]
const store = OpenResponsesOptions.resolve(request).store
const providerMetadataKey = request.model.route.providerMetadataKey ?? "openresponses"
@ -412,7 +431,27 @@ const lowerMessages = Effect.fn("OpenResponses.lowerMessages")(function* (reques
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 })) })
const groups = content.reduce<Array<{ phase: MessagePhase | null | undefined; parts: TextPart[] }>>(
(groups, part) => {
const metadata = part.providerMetadata?.[providerMetadataKey]
const phase =
ProviderShared.isRecord(metadata)
? messagePhase(metadata.phase, extension)
: undefined
const group = groups.at(-1)
if (group && group.phase === phase) group.parts.push(part)
else groups.push({ phase, parts: [part] })
return groups
},
[],
)
input.push(
...groups.map((group) => ({
role: "assistant" as const,
content: group.parts.map((part) => ({ type: "output_text" as const, text: part.text })),
...(group.phase === undefined ? {} : { phase: group.phase }),
})),
)
content.splice(0, content.length)
}
for (const part of message.content) {
@ -513,9 +552,9 @@ const lowerOptions = (request: LLMRequest) => {
}
}
export const fromRequest = Effect.fn("OpenResponses.fromRequest")(function* (
export const fromRequestWithExtension = Effect.fn("OpenResponses.fromRequestWithExtension")(function* (
request: LLMRequest,
extension: Extension = BASE,
extension: Extension,
) {
const generation = request.generation
const toolSchemaCompatibility = request.model.compatibility?.toolSchema
@ -541,6 +580,12 @@ export const fromRequest = Effect.fn("OpenResponses.fromRequest")(function* (
}
})
const decodeBody = ProviderShared.validateWith(Schema.decodeUnknownEffect(OpenResponsesBody))
export const fromRequest = Effect.fn("OpenResponses.fromRequest")(function* (request: LLMRequest) {
return yield* decodeBody(yield* fromRequestWithExtension(request, BASE))
})
// =============================================================================
// Stream Parsing
// =============================================================================
@ -595,24 +640,30 @@ const NO_EVENTS: StepResult["1"] = []
const TERMINAL_TYPES = new Set(["response.completed", "response.incomplete", "response.failed"])
export const terminal = (event: Event) => TERMINAL_TYPES.has(event.type)
const onOutputTextDelta = (state: ParserState, event: Event): StepResult => {
const onOutputTextDelta = (state: ParserState, event: Event, id: string): StepResult => {
if (!event.delta) return [state, NO_EVENTS]
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(state.lifecycle, events, event.item_id ?? "text-0", event.delta) },
{ ...state, lifecycle: Lifecycle.textDelta(lifecycle, events, id, event.delta) },
events,
]
}
const onOutputTextDone = (state: ParserState, event: Event): StepResult => {
const onOutputTextDone = (state: ParserState, event: Event, id: string): StepResult => {
if (state.messageItems.has(id)) {
if (state.lifecycle.text.has(id) || event.text === undefined) return [state, NO_EVENTS]
return onOutputTextDelta(state, { ...event, delta: event.text }, id)
}
const events: LLMEvent[] = []
return [{ ...state, lifecycle: Lifecycle.textEnd(state.lifecycle, events, event.item_id ?? "text-0") }, events]
return [{ ...state, lifecycle: Lifecycle.textEnd(state.lifecycle, events, id) }, events]
}
export const onReasoningDelta = (state: ParserState, event: Event): StepResult => {
export const onReasoningDelta = (state: ParserState, event: Event, itemID: string): StepResult => {
if (!event.delta) return [state, NO_EVENTS]
const events: LLMEvent[] = []
const itemID = event.item_id ?? "reasoning-0"
const id =
event.summary_index !== undefined || state.reasoningItems[itemID] ? `${itemID}:${event.summary_index ?? 0}` : itemID
return [
@ -643,6 +694,18 @@ const reasoningMetadata = (state: ParserState, item: StreamItem & { id: string }
// best-effort, not guaranteed.
const onOutputItemAdded = (state: ParserState, event: Event): StepResult => {
const item = event.item
if (item?.type === "message" && item.id)
return [
{
...state,
messageItems: new Set([...state.messageItems, item.id]),
messagePhases: (() => {
const phase = state.messagePhase(item.phase)
return phase === undefined ? state.messagePhases : { ...state.messagePhases, [item.id]: phase }
})(),
},
NO_EVENTS,
]
if (item && isReasoningItem(item)) {
const events: LLMEvent[] = []
return [
@ -799,7 +862,28 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* (
const item = event.item
if (!item) return [state, NO_EVENTS] satisfies StepResult
if (item.type === "message" && item.id) return onOutputTextDone(state, { ...event, item_id: item.id })
if (item.type === "message" && item.id) {
const itemPhase = state.messagePhase(item.phase)
const phase = itemPhase === undefined ? state.messagePhases[item.id] : itemPhase
const events: LLMEvent[] = []
const messageItems = new Set(state.messageItems)
messageItems.delete(item.id)
const { [item.id]: _phase, ...messagePhases } = state.messagePhases
return [
{
...state,
lifecycle: Lifecycle.textEnd(
state.lifecycle,
events,
item.id,
phase === undefined ? undefined : providerMetadata(state, { phase }),
),
messageItems,
messagePhases,
},
events,
] satisfies StepResult
}
if (item.type === "function_call") {
if (!item.id || !item.call_id || !item.name) return [state, NO_EVENTS] satisfies StepResult
@ -899,19 +983,41 @@ const providerError = (state: ParserState, event: Event, fallback: string) => {
}
export const step = (state: ParserState, event: Event) => {
if (event.type === "response.output_text.delta") return Effect.succeed(onOutputTextDelta(state, event))
if (event.type === "response.output_text.done") return Effect.succeed(onOutputTextDone(state, event))
if (event.type === "response.reasoning.delta" || event.type === "response.reasoning_summary_text.delta")
return Effect.succeed(onReasoningDelta(state, event))
if (event.type === "response.reasoning.done" || event.type === "response.reasoning_summary_text.done")
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(
event.type === "response.output_text.delta"
? onOutputTextDelta(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`)
return Effect.succeed(onReasoningDelta(state, event, event.item_id))
}
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))
}
if (event.type === "response.reasoning_summary_part.added")
return Effect.succeed(onReasoningSummaryPartAdded(state, event))
return event.item_id
? Effect.succeed(onReasoningSummaryPartAdded(state, event))
: ProviderShared.eventError(state.id, `${event.type} is missing item_id`)
if (event.type === "response.reasoning_summary_part.done")
return Effect.succeed(onReasoningSummaryPartDone(state, event))
if (event.type === "response.output_item.added") return Effect.succeed(onOutputItemAdded(state, event))
return event.item_id
? Effect.succeed(onReasoningSummaryPartDone(state, event))
: ProviderShared.eventError(state.id, `${event.type} is missing item_id`)
if (event.type === "response.output_item.added") {
if (event.item?.type === "message" && !event.item.id)
return ProviderShared.eventError(state.id, `${event.type} message is missing id`)
return Effect.succeed(onOutputItemAdded(state, event))
}
if (event.type === "response.function_call_arguments.delta") return onFunctionCallArgumentsDelta(state, event)
if (event.type === "response.output_item.done") return onOutputItemDone(state, event)
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))
if (event.type === "response.failed") return providerError(state, event, `${state.name} response failed`)
@ -933,10 +1039,18 @@ export const initial = (request: LLMRequest, extension: Extension = BASE): Parse
hasFunctionCall: false,
tools: ToolStream.empty<string>(),
lifecycle: Lifecycle.initial(),
messageItems: new Set<string>(),
messagePhase: (value) => messagePhase(value, extension),
messagePhases: {},
reasoningItems: {},
store: OpenResponsesOptions.resolve(request).store,
})
const messagePhase = (value: unknown, extension: Extension): MessagePhase | null | undefined => {
if (value === "commentary" || value === "final_answer") return value
return extension.messagePhase?.(value)
}
export const protocol = Protocol.make({
id: ADAPTER,
body: {

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

@ -35,8 +35,18 @@ const OpenAIResponsesToolChoice = Schema.Union([
Schema.Struct({ type: Schema.tag("image_generation") }),
])
const OpenAIResponsesInputItem = Schema.Union([
Schema.Struct({
role: Schema.tag("assistant"),
content: Schema.Array(Schema.Struct({ type: Schema.tag("output_text"), text: Schema.String })),
phase: Schema.optionalKey(Schema.NullOr(OpenResponses.MessagePhase)),
}),
OpenResponses.InputItem,
])
const OpenAIResponsesCoreFields = {
...OpenResponses.coreFields,
input: Schema.Array(OpenAIResponsesInputItem),
tools: optionalArray(OpenAIResponsesTools),
tool_choice: Schema.optional(OpenAIResponsesToolChoice),
}
@ -60,6 +70,7 @@ const encodeWebSocketMessage = Schema.encodeSync(Schema.fromJsonString(OpenAIRes
const extension = {
id: ADAPTER,
name: NAME,
messagePhase: (value: unknown) => (value === null ? null : undefined),
lowerMedia: ({ part, media, request }) => {
if (request.model.provider !== "xai" || media.mime !== "application/pdf") return undefined
return {
@ -102,7 +113,7 @@ const lowerToolChoice = (toolChoice: NonNullable<LLMRequest["toolChoice"]>, tool
})
const fromRequest = Effect.fn("OpenAIResponses.fromRequest")(function* (request: LLMRequest) {
const body = yield* OpenResponses.fromRequest(
const body = yield* OpenResponses.fromRequestWithExtension(
LLMRequest.update(request, { tools: [], toolChoice: undefined }),
extension,
)
@ -208,9 +219,13 @@ const onHostedToolDone = Effect.fn("OpenAIResponses.onHostedToolDone")(function*
const step = (state: OpenResponses.ParserState, event: OpenResponses.Event) => {
if (event.type === "response.reasoning_text.delta" || event.type === "response.reasoning_summary.delta")
return Effect.succeed(OpenResponses.onReasoningDelta(state, event))
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 Effect.succeed(OpenResponses.onReasoningDone(state, event))
return event.item_id
? Effect.succeed(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)
return OpenResponses.step(state, event)