From 91f77deaa911dfcb64d4813762e885cced84f08d Mon Sep 17 00:00:00 2001 From: macodev00 Date: Fri, 25 Sep 2026 06:25:53 +0000 Subject: [PATCH] fix(server): recover assistant text when completion extends a prefix item.completed detail is a full snapshot. A nonempty buffered or already projected prefix was kept and the rest of the reply was dropped. When detail strictly extends that accumulated text, persist only the missing suffix. --- .../Layers/ProviderRuntimeIngestion.test.ts | 123 ++++++++++++++++++ .../Layers/ProviderRuntimeIngestion.ts | 49 +++++-- 2 files changed, 162 insertions(+), 10 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 9b5b56309749..ea9f9a7562ec 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -1797,6 +1797,129 @@ describe("ProviderRuntimeIngestion", () => { expect(message?.streaming).toBe(false); }); + it("appends the missing suffix when assistant completion extends a buffered prefix", async () => { + const harness = await createHarness(); + const now = "2026-01-01T00:00:00.000Z"; + const full = "Hi! I'm Codex, ready to help."; + const codex = ProviderDriverKind.make("codex"); + const threadId = asThreadId("thread-1"); + const turnId = asTurnId("turn-buffered-prefix"); + const itemId = asItemId("item-buffered-prefix"); + + await harness.emitAndDrain([ + { + type: "turn.started", + eventId: asEventId("evt-buffered-prefix-started"), + provider: codex, + createdAt: now, + threadId, + turnId, + }, + { + type: "content.delta", + eventId: asEventId("evt-buffered-prefix-delta"), + provider: codex, + createdAt: now, + threadId, + turnId, + itemId, + payload: { streamKind: "assistant_text", delta: "Hi! I" }, + }, + ]); + expect( + (await harness.readModel()).threads + .find((thread) => thread.id === threadId) + ?.messages.find((message) => message.id === `assistant:${itemId}`), + ).toBeUndefined(); + + await harness.emitAndDrain([ + { + type: "item.completed", + eventId: asEventId("evt-buffered-prefix-completed"), + provider: codex, + createdAt: now, + threadId, + turnId, + itemId, + payload: { itemType: "assistant_message", status: "completed", detail: full }, + }, + ]); + + const thread = await waitForThread(harness.readModel, (entry) => + entry.messages.some( + (message: ProviderRuntimeTestMessage) => + message.id === `assistant:${itemId}` && !message.streaming, + ), + ); + const matches = thread.messages.filter( + (message: ProviderRuntimeTestMessage) => message.id === `assistant:${itemId}`, + ); + expect(matches).toHaveLength(1); + expect(matches[0]?.text).toBe(full); + }); + + it("appends the missing suffix when assistant completion extends an already projected prefix", async () => { + const harness = await createHarness({ serverSettings: { responseStreamingMode: "token" } }); + const now = "2026-01-01T00:00:00.000Z"; + const full = "Hi! I'm Codex, ready to help."; + const codex = ProviderDriverKind.make("codex"); + const threadId = asThreadId("thread-1"); + const turnId = asTurnId("turn-projected-prefix"); + const itemId = asItemId("item-projected-prefix"); + + await harness.emitAndDrain([ + { + type: "turn.started", + eventId: asEventId("evt-projected-prefix-started"), + provider: codex, + createdAt: now, + threadId, + turnId, + }, + { + type: "content.delta", + eventId: asEventId("evt-projected-prefix-delta"), + provider: codex, + createdAt: now, + threadId, + turnId, + itemId, + payload: { streamKind: "assistant_text", delta: "Hi! I" }, + }, + ]); + await waitForThread(harness.readModel, (entry) => + entry.messages.some( + (message: ProviderRuntimeTestMessage) => + message.id === `assistant:${itemId}` && message.streaming && message.text === "Hi! I", + ), + ); + + await harness.emitAndDrain([ + { + type: "item.completed", + eventId: asEventId("evt-projected-prefix-completed"), + provider: codex, + createdAt: now, + threadId, + turnId, + itemId, + payload: { itemType: "assistant_message", status: "completed", detail: full }, + }, + ]); + + const thread = await waitForThread(harness.readModel, (entry) => + entry.messages.some( + (message: ProviderRuntimeTestMessage) => + message.id === `assistant:${itemId}` && !message.streaming && message.text === full, + ), + ); + const matches = thread.messages.filter( + (message: ProviderRuntimeTestMessage) => message.id === `assistant:${itemId}`, + ); + expect(matches).toHaveLength(1); + expect(matches[0]?.text).toBe(full); + }); + it("preserves completed tool metadata on projected tool activities", async () => { const harness = await createHarness(); const now = "2026-01-01T00:00:00.000Z"; diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 072ac5110b9d..7e9e512f29ec 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -200,6 +200,35 @@ function hasRenderableAssistantText(text: string | undefined): boolean { return (text?.trim().length ?? 0) > 0; } +/** + * `detail` on item completion is a full snapshot. When it strictly extends + * text already accumulated (projected plus still buffered), only the part + * not yet projected is safe to persist. Equal, empty, and divergent + * snapshots keep the streamed text. + */ +function assistantCompletionDelta(input: { + readonly projectedText: string; + readonly bufferedText: string; + readonly detail: string | undefined; +}): string { + const accumulated = `${input.projectedText}${input.bufferedText}`; + const detail = input.detail; + if ( + detail !== undefined && + detail.startsWith(accumulated) && + detail.length > accumulated.length + ) { + return detail.slice(input.projectedText.length); + } + if (input.bufferedText.length > 0) { + return input.bufferedText; + } + if (input.projectedText.length === 0 && (detail?.trim().length ?? 0) > 0) { + return detail; + } + return ""; +} + // An opening fence may sit at any indentation, since fences inside list // items are indented past the marker. A closing fence may be indented at most // three spaces more than its opener. Deeper lines are content in the block. @@ -1502,16 +1531,16 @@ const make = Effect.gen(function* () { commandTag: string; finalDeltaCommandTag: string; fallbackText?: string; + projectedText?: string; hasProjectedMessage?: boolean; }) => Effect.gen(function* () { const bufferedText = yield* takeBufferedAssistantText(input.messageId); - const text = - bufferedText.length > 0 - ? bufferedText - : (input.fallbackText?.trim().length ?? 0) > 0 - ? input.fallbackText! - : ""; + const text = assistantCompletionDelta({ + projectedText: input.projectedText ?? "", + bufferedText, + detail: input.fallbackText, + }); const hasRenderableText = hasRenderableAssistantText(text); const isReasoning = messageStreamRoleOf(input.messageId) === "reasoning"; @@ -2297,9 +2326,6 @@ const make = Effect.gen(function* () { streamingOnly: false, }), ]); - const shouldApplyFallbackCompletionText = - !existingAssistantMessage || existingAssistantMessage.text.length === 0; - const shouldSkipRedundantCompletion = Option.isNone(activeAssistantMessageId) && turnId !== undefined && @@ -2320,7 +2346,10 @@ const make = Effect.gen(function* () { commandTag: "assistant-complete", finalDeltaCommandTag: "assistant-delta-finalize", hasProjectedMessage: existingAssistantMessage !== undefined, - ...(assistantCompletion.fallbackText !== undefined && shouldApplyFallbackCompletionText + ...(existingAssistantMessage !== undefined + ? { projectedText: existingAssistantMessage.text } + : {}), + ...(assistantCompletion.fallbackText !== undefined ? { fallbackText: assistantCompletion.fallbackText } : {}), });