Skip to content

refactor(ai): drop responses replay tombstones #46965

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rekram1-node merged 1 commit into from
Sep 3, 2026
+8 −136
Merged

refactor(ai): drop responses replay tombstones #46965

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 4 additions & 23 deletions packages/ai/src/protocols/open-responses.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -423,14 +423,10 @@ export interface ParserState {
readonly name: string
readonly providerMetadataKey: string
readonly tools: ToolStream.State<string>
// Item ids are response-scoped identities. Keep completed ids tombstoned so
// reconnect replay cannot reopen fragments already emitted downstream.
readonly completedTools: ReadonlySet<string>
readonly hasFunctionCall: boolean
readonly lifecycle: Lifecycle.State
readonly outputItems: Readonly<Record<number, string>>
readonly message: { readonly id: string; readonly phase: MessagePhase | null | undefined } | undefined
readonly completedMessages: ReadonlySet<string>
readonly reasoningItems: Readonly<Record<string, ReasoningStreamItem>>
}

Expand Down Expand Up @@ -1042,16 +1038,12 @@ const onOutputItemAdded = (state: ParserState, event: NormalizedEvent): StepResu
const item = event.item
if (!item) return [state, NO_EVENTS]
if (item.type === "message") {
if (state.completedMessages.has(item.id)) return [state, NO_EVENTS]
const phase = messagePhase(item.phase)
const completedMessages = new Set(state.completedMessages)
if (state.message !== undefined && state.message.id !== item.id) completedMessages.add(state.message.id)
// A new message closes earlier messages, including ones that never streamed.
const events: LLMEvent[] = []
const lifecycle = [...state.lifecycle.text]
.filter((id) => id !== item.id)
.reduce((lifecycle, id) => {
completedMessages.add(id)
const openPhase = state.message?.id === id ? state.message.phase : undefined
return Lifecycle.textEnd(
lifecycle,
Expand All @@ -1064,7 +1056,6 @@ const onOutputItemAdded = (state: ParserState, event: NormalizedEvent): StepResu
{
...state,
lifecycle,
completedMessages,
message: {
id: item.id,
phase: phase === undefined && state.message?.id === item.id ? state.message.phase : phase,
Expand Down Expand Up @@ -1094,7 +1085,7 @@ const onOutputItemAdded = (state: ParserState, event: NormalizedEvent): StepResu
]
}
if (item.type !== "function_call" || !item.call_id) return [state, NO_EVENTS]
if (state.tools[item.id] !== undefined || state.completedTools.has(item.id)) return [state, NO_EVENTS]
if (state.tools[item.id] !== undefined) return [state, NO_EVENTS]
const metadata = providerMetadata(state, { itemId: item.id })
const events: LLMEvent[] = []
const lifecycle = Lifecycle.stepStart(state.lifecycle, events)
Expand Down Expand Up @@ -1198,14 +1189,9 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* (
}

if (item.type === "message") {
if (state.completedMessages.has(item.id)) return [state, NO_EVENTS] satisfies StepResult
const completedMessages = new Set(state.completedMessages)
completedMessages.add(item.id)
if (state.message !== undefined && state.message.id !== item.id)
return [{ ...state, completedMessages }, NO_EVENTS] satisfies StepResult
const message = state.message
const active = state.message?.id === item.id
const itemPhase = messagePhase(item.phase)
const phase = itemPhase === undefined ? message?.phase : itemPhase
const phase = itemPhase === undefined && active ? state.message?.phase : itemPhase
const parts: ReadonlyArray<unknown> = Array.isArray(item.content) ? item.content : []
const content: string[] = []
for (const part of parts) {
Expand All @@ -1221,16 +1207,14 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* (
{
...state,
lifecycle: Lifecycle.textEnd(lifecycle, events, item.id, metadata, text),
completedMessages,
message: undefined,
message: active ? undefined : state.message,
},
events,
] satisfies StepResult
}

if (item.type === "function_call") {
if (!item.call_id || !item.name) return [state, NO_EVENTS] satisfies StepResult
if (state.completedTools.has(item.id)) return [state, NO_EVENTS] satisfies StepResult
const metadata = providerMetadata(state, { itemId: item.id })
const registered = state.tools[item.id] !== undefined
const tools = registered
Expand All @@ -1257,7 +1241,6 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* (
resultEvents.some((event) => LLMEvent.is.toolCall(event) || LLMEvent.is.toolInputError(event)) ||
state.hasFunctionCall,
tools: result.tools,
completedTools: new Set([...state.completedTools, item.id]),
},
events,
] satisfies StepResult
Expand Down Expand Up @@ -1518,11 +1501,9 @@ export const initial = (request: LLMRequest, adapter: ProviderAdapter = BASE_ADA
providerMetadataKey: metadataKey(request.model),
hasFunctionCall: false,
tools: ToolStream.empty<string>(),
completedTools: new Set<string>(),
lifecycle: Lifecycle.initial(),
outputItems: {},
message: undefined,
completedMessages: new Set<string>(),
reasoningItems: {},
})

Expand Down
26 changes: 0 additions & 26 deletions packages/ai/test/provider/open-responses-finals.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,32 +82,6 @@ describe("Open Responses completed item text", () => {
expect(response.events.filter(LLMEvent.is.textStart)).toEqual([])
}),
)

it.effect("assembles a done-only message once across replayed item events", () =>
Effect.gen(function* () {
const item = {
type: "message",
id: "msg_1",
content: [{ type: "output_text", text: "Recovered" }],
}
const response = yield* generate(
{ type: "response.output_text.delta", item_id: "msg_1", delta: "Ignored after resume" },
{ type: "response.output_item.done", item },
{ type: "response.output_item.added", item },
{ type: "response.output_item.done", item },
completed,
)
expect(response.text).toBe("Recovered")
expect(response.message.content).toEqual([
{
type: "text",
text: "Recovered",
providerMetadata: { "openai-compatible": { itemId: "msg_1" } },
},
])
expect(response.events.filter(LLMEvent.is.textEnd)).toHaveLength(1)
}),
)
})

describe("Open Responses completed item reasoning", () => {
Expand Down
74 changes: 3 additions & 71 deletions packages/ai/test/provider/open-responses-lifecycle.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -217,7 +217,7 @@ describe("Open Responses basic-item lifecycles", () => {
}),
)

it.effect("preserves non-empty done-only message content without replaying duplicates", () =>
it.effect("preserves non-empty done-only message content", () =>
Effect.gen(function* () {
const text = {
type: "message",
Expand All @@ -230,17 +230,11 @@ describe("Open Responses basic-item lifecycles", () => {
content: [{ type: "refusal", refusal: "Done-only refusal." }],
}
const events = yield* collect(
{ type: "response.output_item.done", item: text },
{ type: "response.output_item.done", item: text },
{
type: "response.output_item.done",
item: { type: "message", id: "msg_empty", content: [{ type: "output_text", text: "" }] },
},
{
type: "response.output_item.done",
item: { type: "message", id: "msg_empty", content: [{ type: "output_text", text: "Late" }] },
},
{ type: "response.output_item.done", item: refusal },
{ type: "response.output_item.done", item: refusal },
completed,
)
Expand Down Expand Up @@ -272,63 +266,6 @@ describe("Open Responses basic-item lifecycles", () => {
}),
)

it.effect("treats a repeated message lifecycle as replay", () =>
Effect.gen(function* () {
const events = yield* collect(
{ type: "response.output_item.added", item: { type: "message", id: "msg_1", phase: "commentary" } },
{ type: "response.output_text.delta", item_id: "msg_1", delta: "First" },
{ type: "response.output_item.done", item: { type: "message", id: "msg_1" } },
{ type: "response.output_item.added", item: { type: "message", id: "msg_1" } },
{ type: "response.output_text.delta", item_id: "msg_1", delta: "Second" },
{ type: "response.output_item.done", item: { type: "message", id: "msg_1" } },
completed,
)
expect(events.filter(LLMEvent.is.textEnd)).toEqual([
{
type: "text-end",
id: "msg_1",
providerMetadata: { "openai-compatible": { itemId: "msg_1", phase: "commentary" } },
},
])
expect(events.filter(LLMEvent.is.textDelta).map((event) => event.text)).toEqual(["First"])
}),
)

it.effect("ignores a stale done-only message while another message is active", () =>
Effect.gen(function* () {
const events = yield* collect(
{ type: "response.output_item.added", item: { type: "message", id: "msg_1", phase: "commentary" } },
{ type: "response.output_text.delta", item_id: "msg_1", delta: "Draft" },
{
type: "response.output_item.done",
item: { type: "message", id: "msg_2", content: [{ type: "output_text", text: "Recovered" }] },
},
{
type: "response.output_item.done",
item: { type: "message", id: "msg_1", content: [{ type: "output_text", text: "Final" }] },
},
{
type: "response.output_item.done",
item: { type: "message", id: "msg_2", content: [{ type: "output_text", text: "Late" }] },
},
completed,
)
expect(events.filter((event) => event.type.startsWith("text-"))).toEqual([
{
type: "text-start",
id: "msg_1",
providerMetadata: { "openai-compatible": { itemId: "msg_1", phase: "commentary" } },
},
{ type: "text-delta", id: "msg_1", text: "Draft" },
{
type: "text-end",
id: "msg_1",
text: "Final",
providerMetadata: { "openai-compatible": { itemId: "msg_1", phase: "commentary" } },
},
])
}),
)
// Captured from Bedrock Mantle (openai.gpt-oss-120b): the terminal function_call
// items rename `id` to `item_id` and carry a stray `output_index`.
it.effect("recovers a terminal function_call id from its output slot", () =>
Expand Down Expand Up @@ -419,7 +356,7 @@ describe("Open Responses basic-item lifecycles", () => {
}),
)

it.effect("opens and closes a done-only tool once", () =>
it.effect("opens and closes a done-only tool", () =>
Effect.gen(function* () {
const item = {
type: "function_call",
Expand All @@ -428,12 +365,7 @@ describe("Open Responses basic-item lifecycles", () => {
name: "lookup",
arguments: '{"query":"weather"}',
}
const events = yield* collect(
{ type: "response.output_item.done", item },
{ type: "response.output_item.done", item },
{ type: "response.output_item.added", item },
completed,
)
const events = yield* collect({ type: "response.output_item.done", item }, completed)
const providerMetadata = { "openai-compatible": { itemId: "fc_1" } }
expect(events.filter((event) => event.type.startsWith("tool-"))).toEqual([
{ type: "tool-input-start", id: "call_1", name: "lookup", providerMetadata },
Expand Down
17 changes: 1 addition & 16 deletions packages/ai/test/provider/openai-responses.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2850,7 +2850,7 @@ describe("OpenAI Responses route", () => {
}),
)

it.effect("ignores duplicate item boundary events", () =>
it.effect("ignores duplicate item start events", () =>
Effect.gen(function* () {
const response = yield* LLMClient.generate(request).pipe(
Effect.provide(
Expand Down Expand Up @@ -2881,21 +2881,6 @@ describe("OpenAI Responses route", () => {
arguments: '{"query":"weather"}',
},
},
{
type: "response.output_item.done",
item: {
type: "function_call",
id: "fc_1",
call_id: "call_1",
name: "lookup",
arguments: '{"query":"weather"}',
},
},
// A completed item that is re-added stays closed.
{
type: "response.output_item.added",
item: { type: "function_call", id: "fc_1", call_id: "call_1", name: "lookup", arguments: "" },
},
{ type: "response.completed", response: { id: "resp_1" } },
),
),
Expand Down
Loading