From 1f259ece5337810d931314ce25b380a5bf685150 Mon Sep 17 00:00:00 2001 From: Aiden Cline Date: Wed, 2 Sep 2026 21:17:51 -0500 Subject: [PATCH] refactor(ai): drop responses replay tombstones --- packages/ai/src/protocols/open-responses.ts | 27 +------ .../provider/open-responses-finals.test.ts | 26 ------- .../provider/open-responses-lifecycle.test.ts | 74 +------------------ .../ai/test/provider/openai-responses.test.ts | 17 +---- 4 files changed, 8 insertions(+), 136 deletions(-) diff --git a/packages/ai/src/protocols/open-responses.ts b/packages/ai/src/protocols/open-responses.ts index d0499c49dd23..96bd37da73b6 100644 --- a/packages/ai/src/protocols/open-responses.ts +++ b/packages/ai/src/protocols/open-responses.ts @@ -423,14 +423,10 @@ export interface ParserState { readonly name: string readonly providerMetadataKey: string readonly tools: ToolStream.State - // Item ids are response-scoped identities. Keep completed ids tombstoned so - // reconnect replay cannot reopen fragments already emitted downstream. - readonly completedTools: ReadonlySet readonly hasFunctionCall: boolean readonly lifecycle: Lifecycle.State readonly outputItems: Readonly> readonly message: { readonly id: string; readonly phase: MessagePhase | null | undefined } | undefined - readonly completedMessages: ReadonlySet readonly reasoningItems: Readonly> } @@ -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, @@ -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, @@ -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) @@ -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 = Array.isArray(item.content) ? item.content : [] const content: string[] = [] for (const part of parts) { @@ -1221,8 +1207,7 @@ 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 @@ -1230,7 +1215,6 @@ const onOutputItemDone = Effect.fn("OpenResponses.onOutputItemDone")(function* ( 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 @@ -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 @@ -1518,11 +1501,9 @@ export const initial = (request: LLMRequest, adapter: ProviderAdapter = BASE_ADA providerMetadataKey: metadataKey(request.model), hasFunctionCall: false, tools: ToolStream.empty(), - completedTools: new Set(), lifecycle: Lifecycle.initial(), outputItems: {}, message: undefined, - completedMessages: new Set(), reasoningItems: {}, }) diff --git a/packages/ai/test/provider/open-responses-finals.test.ts b/packages/ai/test/provider/open-responses-finals.test.ts index b8ad309c63d4..d46a335ed7a4 100644 --- a/packages/ai/test/provider/open-responses-finals.test.ts +++ b/packages/ai/test/provider/open-responses-finals.test.ts @@ -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", () => { diff --git a/packages/ai/test/provider/open-responses-lifecycle.test.ts b/packages/ai/test/provider/open-responses-lifecycle.test.ts index fd29b09593f9..aabe79683772 100644 --- a/packages/ai/test/provider/open-responses-lifecycle.test.ts +++ b/packages/ai/test/provider/open-responses-lifecycle.test.ts @@ -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", @@ -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, ) @@ -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", () => @@ -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", @@ -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 }, diff --git a/packages/ai/test/provider/openai-responses.test.ts b/packages/ai/test/provider/openai-responses.test.ts index 0dff1271a1c1..8cbeb04ebfcc 100644 --- a/packages/ai/test/provider/openai-responses.test.ts +++ b/packages/ai/test/provider/openai-responses.test.ts @@ -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( @@ -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" } }, ), ),