diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 46231962..08faa95a 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -15,7 +15,9 @@ import { EventId, MessageId, ProjectId, + ProviderDriverKind, ProviderItemId, + RuntimeItemId, ThreadId, TurnId, } from "@t3tools/contracts"; @@ -674,16 +676,104 @@ describe("ProviderRuntimeIngestion", () => { const thread = await waitForThread(harness.engine, (entry) => entry.messages.some( (message: ProviderRuntimeTestMessage) => - message.id === "assistant:item-1" && !message.streaming, + message.id === "assistant:turn-2:item-1" && !message.streaming, ), ); const message = thread.messages.find( - (entry: ProviderRuntimeTestMessage) => entry.id === "assistant:item-1", + (entry: ProviderRuntimeTestMessage) => entry.id === "assistant:turn-2:item-1", ); expect(message?.text).toBe("hello world"); expect(message?.streaming).toBe(false); }); + it("keeps assistant replies distinct when a resumed session reuses an item id", async () => { + const harness = await createHarness(); + const firstTurnAt = "2026-07-01T21:06:10.000Z"; + const secondTurnAt = "2026-07-02T01:44:30.000Z"; + const reusedItemId = RuntimeItemId.make("assistant:session-1:segment:0"); + + harness.emit({ + type: "content.delta", + eventId: asEventId("evt-resume-turn-1"), + provider: ProviderDriverKind.make("cursor"), + createdAt: firstTurnAt, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-1"), + itemId: reusedItemId, + payload: { + streamKind: "assistant_text", + delta: "first turn reply", + }, + }); + harness.emit({ + type: "item.completed", + eventId: asEventId("evt-resume-turn-1-complete"), + provider: ProviderDriverKind.make("cursor"), + createdAt: firstTurnAt, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-1"), + itemId: reusedItemId, + payload: { + itemType: "assistant_message", + status: "completed", + }, + }); + + await waitForThread(harness.engine, (thread) => + thread.messages.some( + (message) => + message.id === "assistant:turn-1:assistant:session-1:segment:0" && + message.text === "first turn reply", + ), + ); + + harness.emit({ + type: "content.delta", + eventId: asEventId("evt-resume-turn-2"), + provider: ProviderDriverKind.make("cursor"), + createdAt: secondTurnAt, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-2"), + itemId: reusedItemId, + payload: { + streamKind: "assistant_text", + delta: "second turn reply", + }, + }); + harness.emit({ + type: "item.completed", + eventId: asEventId("evt-resume-turn-2-complete"), + provider: ProviderDriverKind.make("cursor"), + createdAt: secondTurnAt, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-2"), + itemId: reusedItemId, + payload: { + itemType: "assistant_message", + status: "completed", + }, + }); + + const thread = await waitForThread(harness.engine, (entry) => + entry.messages.some( + (message) => + message.id === "assistant:turn-2:assistant:session-1:segment:0" && + message.text === "second turn reply", + ), + ); + + expect( + thread.messages.find( + (message) => message.id === "assistant:turn-1:assistant:session-1:segment:0", + )?.text, + ).toBe("first turn reply"); + expect( + thread.messages.find( + (message) => message.id === "assistant:turn-2:assistant:session-1:segment:0", + )?.text, + ).toBe("second turn reply"); + }); + it("uses assistant item completion detail when no assistant deltas were streamed", async () => { const harness = await createHarness(); const now = new Date().toISOString(); @@ -706,11 +796,11 @@ describe("ProviderRuntimeIngestion", () => { const thread = await waitForThread(harness.engine, (entry) => entry.messages.some( (message: ProviderRuntimeTestMessage) => - message.id === "assistant:item-no-delta" && !message.streaming, + message.id === "assistant:turn-no-delta:item-no-delta" && !message.streaming, ), ); const message = thread.messages.find( - (entry: ProviderRuntimeTestMessage) => entry.id === "assistant:item-no-delta", + (entry: ProviderRuntimeTestMessage) => entry.id === "assistant:turn-no-delta:item-no-delta", ); expect(message?.text).toBe("assistant-only final text"); expect(message?.streaming).toBe(false); @@ -1339,7 +1429,8 @@ describe("ProviderRuntimeIngestion", () => { const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.make("thread-1")); expect( midThread?.messages.some( - (message: ProviderRuntimeTestMessage) => message.id === "assistant:item-buffered", + (message: ProviderRuntimeTestMessage) => + message.id === "assistant:turn-buffered:item-buffered", ), ).toBe(false); @@ -1360,11 +1451,11 @@ describe("ProviderRuntimeIngestion", () => { const thread = await waitForThread(harness.engine, (entry) => entry.messages.some( (message: ProviderRuntimeTestMessage) => - message.id === "assistant:item-buffered" && !message.streaming, + message.id === "assistant:turn-buffered:item-buffered" && !message.streaming, ), ); const message = thread.messages.find( - (entry: ProviderRuntimeTestMessage) => entry.id === "assistant:item-buffered", + (entry: ProviderRuntimeTestMessage) => entry.id === "assistant:turn-buffered:item-buffered", ); expect(message?.text).toBe("buffer me"); expect(message?.streaming).toBe(false); @@ -1425,13 +1516,14 @@ describe("ProviderRuntimeIngestion", () => { const liveThread = await waitForThread(harness.engine, (entry) => entry.messages.some( (message: ProviderRuntimeTestMessage) => - message.id === "assistant:item-streaming-mode" && + message.id === "assistant:turn-streaming-mode:item-streaming-mode" && message.streaming && message.text === "hello live", ), ); const liveMessage = liveThread.messages.find( - (entry: ProviderRuntimeTestMessage) => entry.id === "assistant:item-streaming-mode", + (entry: ProviderRuntimeTestMessage) => + entry.id === "assistant:turn-streaming-mode:item-streaming-mode", ); expect(liveMessage?.streaming).toBe(true); @@ -1453,11 +1545,12 @@ describe("ProviderRuntimeIngestion", () => { const finalThread = await waitForThread(harness.engine, (entry) => entry.messages.some( (message: ProviderRuntimeTestMessage) => - message.id === "assistant:item-streaming-mode" && !message.streaming, + message.id === "assistant:turn-streaming-mode:item-streaming-mode" && !message.streaming, ), ); const finalMessage = finalThread.messages.find( - (entry: ProviderRuntimeTestMessage) => entry.id === "assistant:item-streaming-mode", + (entry: ProviderRuntimeTestMessage) => + entry.id === "assistant:turn-streaming-mode:item-streaming-mode", ); expect(finalMessage?.text).toBe("hello live"); expect(finalMessage?.streaming).toBe(false); @@ -1513,11 +1606,12 @@ describe("ProviderRuntimeIngestion", () => { const thread = await waitForThread(harness.engine, (entry) => entry.messages.some( (message: ProviderRuntimeTestMessage) => - message.id === "assistant:item-buffer-spill" && !message.streaming, + message.id === "assistant:turn-buffer-spill:item-buffer-spill" && !message.streaming, ), ); const message = thread.messages.find( - (entry: ProviderRuntimeTestMessage) => entry.id === "assistant:item-buffer-spill", + (entry: ProviderRuntimeTestMessage) => + entry.id === "assistant:turn-buffer-spill:item-buffer-spill", ); expect(message?.text.length).toBe(oversizedText.length); expect(message?.text).toBe(oversizedText); @@ -1589,7 +1683,8 @@ describe("ProviderRuntimeIngestion", () => { thread.session?.activeTurnId === null && thread.messages.some( (message: ProviderRuntimeTestMessage) => - message.id === "assistant:item-complete-dedup" && !message.streaming, + message.id === "assistant:turn-complete-dedup:item-complete-dedup" && + !message.streaming, ), ); @@ -1603,7 +1698,7 @@ describe("ProviderRuntimeIngestion", () => { return false; } return ( - event.payload.messageId === "assistant:item-complete-dedup" && + event.payload.messageId === "assistant:turn-complete-dedup:item-complete-dedup" && event.payload.streaming === false ); }); @@ -1973,7 +2068,7 @@ describe("ProviderRuntimeIngestion", () => { (entry: ProviderRuntimeTestCheckpoint) => entry.turnId === "turn-p1", ); expect(checkpoint?.status).toBe("missing"); - expect(checkpoint?.assistantMessageId).toBe("assistant:item-p1-assistant"); + expect(checkpoint?.assistantMessageId).toBe("assistant:turn-p1:item-p1-assistant"); expect(checkpoint?.checkpointRef).toBe("provider-diff:evt-turn-diff-updated"); }); diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index e23cbd7c..9e407456 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -26,6 +26,7 @@ import { ProviderRuntimeIngestionService, type ProviderRuntimeIngestionShape, } from "../Services/ProviderRuntimeIngestion.ts"; +import { assistantMessageBaseKey, assistantMessageId } from "../assistantMessageIds.ts"; const providerTurnKey = (threadId: ThreadId, turnId: TurnId) => `${threadId}:${turnId}`; const providerCommandId = (event: ProviderRuntimeEvent, tag: string): CommandId => @@ -60,6 +61,13 @@ function toTurnId(value: TurnId | string | undefined): TurnId | undefined { return value === undefined ? undefined : TurnId.make(String(value)); } +function assistantMessageIdForEvent(event: ProviderRuntimeEvent): MessageId { + return assistantMessageId( + assistantMessageBaseKey(event.itemId, event.turnId, event.eventId), + toTurnId(event.turnId), + ); +} + function toApprovalRequestId(value: string | undefined): ApprovalRequestId | undefined { return value === undefined ? undefined : ApprovalRequestId.make(value); } @@ -1026,9 +1034,7 @@ const make = Effect.gen(function* () { event.type === "turn.proposed.delta" ? event.payload.delta : undefined; if (assistantDelta && assistantDelta.length > 0) { - const assistantMessageId = MessageId.make( - `assistant:${event.itemId ?? event.turnId ?? event.eventId}`, - ); + const assistantMessageId = assistantMessageIdForEvent(event); const turnId = toTurnId(event.turnId); if (turnId) { yield* rememberAssistantMessageId(thread.id, turnId, assistantMessageId); @@ -1069,9 +1075,7 @@ const make = Effect.gen(function* () { const assistantCompletion = event.type === "item.completed" && event.payload.itemType === "assistant_message" ? { - messageId: MessageId.make( - `assistant:${event.itemId ?? event.turnId ?? event.eventId}`, - ), + messageId: assistantMessageIdForEvent(event), fallbackText: event.payload.detail, } : undefined; @@ -1209,9 +1213,7 @@ const make = Effect.gen(function* () { if (thread.checkpoints.some((c) => c.turnId === turnId)) { // Already tracked; no-op. } else { - const assistantMessageId = MessageId.make( - `assistant:${event.itemId ?? event.turnId ?? event.eventId}`, - ); + const assistantMessageId = assistantMessageIdForEvent(event); const maxTurnCount = thread.checkpoints.reduce( (max, c) => Math.max(max, c.checkpointTurnCount), 0, diff --git a/apps/server/src/orchestration/assistantMessageIds.test.ts b/apps/server/src/orchestration/assistantMessageIds.test.ts new file mode 100644 index 00000000..c3cb5483 --- /dev/null +++ b/apps/server/src/orchestration/assistantMessageIds.test.ts @@ -0,0 +1,26 @@ +import { describe, expect, it } from "vitest"; +import { TurnId } from "@t3tools/contracts"; + +import { assistantMessageBaseKey, assistantMessageId } from "./assistantMessageIds.ts"; + +describe("assistant message ids", () => { + it("scopes reused provider item ids to their turn", () => { + const baseKey = assistantMessageBaseKey("assistant:session-1:segment:0", "turn-1", "event-1"); + + const firstTurnId = assistantMessageId(baseKey, TurnId.make("turn-1")); + const secondTurnId = assistantMessageId(baseKey, TurnId.make("turn-2")); + + expect(firstTurnId).toBe("assistant:turn-1:assistant:session-1:segment:0"); + expect(secondTurnId).toBe("assistant:turn-2:assistant:session-1:segment:0"); + expect(firstTurnId).not.toBe(secondTurnId); + }); + + it("falls back from item id to turn id and event id", () => { + expect(assistantMessageBaseKey(undefined, "turn-1", "event-1")).toBe("turn-1"); + expect(assistantMessageBaseKey(undefined, undefined, "event-1")).toBe("event-1"); + }); + + it("preserves the legacy shape when no turn id is available", () => { + expect(assistantMessageId("item-1")).toBe("assistant:item-1"); + }); +}); diff --git a/apps/server/src/orchestration/assistantMessageIds.ts b/apps/server/src/orchestration/assistantMessageIds.ts new file mode 100644 index 00000000..60bc8059 --- /dev/null +++ b/apps/server/src/orchestration/assistantMessageIds.ts @@ -0,0 +1,13 @@ +import { MessageId, type TurnId } from "@t3tools/contracts"; + +export function assistantMessageBaseKey( + itemId: string | undefined, + turnId: string | undefined, + eventId: string, +): string { + return String(itemId ?? turnId ?? eventId); +} + +export function assistantMessageId(baseKey: string, turnId?: TurnId): MessageId { + return MessageId.make(`assistant:${turnId ? `${turnId}:` : ""}${baseKey}`); +}