diff --git a/src/__tests__/unit/services/canvas-turn-persistence.test.ts b/src/__tests__/unit/services/canvas-turn-persistence.test.ts index 2e9b2222b5..2df612c709 100644 --- a/src/__tests__/unit/services/canvas-turn-persistence.test.ts +++ b/src/__tests__/unit/services/canvas-turn-persistence.test.ts @@ -292,6 +292,54 @@ describe("appendTurnMessages", () => { expect(notify).not.toHaveBeenCalled(); }); + test("lands a turn's rows after its own user message, not after a later one", async () => { + withLockedRows([ + { id: "turn-1-u", role: "user", content: "Q1" }, + { id: "turn-2-u", role: "user", content: "Q2 (sent while turn 1 ran)" }, + ]); + + await appendTurnMessages({ + conversationId: "conv-1", + rows, + idPrefix: "turn-1-a", + reason: "user-turn", + turnId: "turn-1", + }); + + const written = update.mock.calls[0][0].data.messages as { id: string }[]; + expect(written.map((m) => m.id)).toEqual(["turn-1-u", "turn-1-a0", "turn-2-u"]); + }); + + test("appends when the turn is already last, or when no turn is given", async () => { + withLockedRows([ + { id: "turn-0-u", role: "user", content: "Q0" }, + { id: "turn-1-u", role: "user", content: "Q1" }, + ]); + await appendTurnMessages({ + conversationId: "conv-1", + rows, + idPrefix: "turn-1-a", + reason: "user-turn", + turnId: "turn-1", + }); + const anchored = update.mock.calls[0][0].data.messages as { id: string }[]; + expect(anchored.map((m) => m.id)).toEqual(["turn-0-u", "turn-1-u", "turn-1-a0"]); + + update.mockClear(); + withLockedRows([ + { id: "turn-1-u", role: "user", content: "Q1" }, + { id: "turn-2-u", role: "user", content: "Q2" }, + ]); + await appendTurnMessages({ + conversationId: "conv-1", + rows, + idPrefix: "turn-1-a", + reason: "user-turn", + }); + const appended = update.mock.calls[0][0].data.messages as { id: string }[]; + expect(appended.map((m) => m.id)).toEqual(["turn-1-u", "turn-2-u", "turn-1-a0"]); + }); + test("no-ops when the conversation row was deleted mid-turn", async () => { withNoRow(); diff --git a/src/app/api/ask/quick/route.ts b/src/app/api/ask/quick/route.ts index 0d5e04c806..b09deec906 100644 --- a/src/app/api/ask/quick/route.ts +++ b/src/app/api/ask/quick/route.ts @@ -812,6 +812,7 @@ export async function POST(request: NextRequest) { rows, idPrefix: assistantPrefix, reason: "user-turn", + turnId: turnIdStr ?? undefined, }); } catch (err) { console.error("❌ [quick-ask] Turn persist failed:", err); @@ -831,6 +832,7 @@ export async function POST(request: NextRequest) { rows: [errorRow], idPrefix: assistantPrefix, reason: "user-turn", + turnId: turnIdStr ?? undefined, }).catch(() => {}); } }); diff --git a/src/app/api/ask/sync/route.ts b/src/app/api/ask/sync/route.ts index 64c2f43198..756fabc5f2 100644 --- a/src/app/api/ask/sync/route.ts +++ b/src/app/api/ask/sync/route.ts @@ -455,6 +455,7 @@ export async function POST(request: NextRequest) { rows, idPrefix: assistantPrefix, reason: "user-turn", + turnId, }); // Snapshot the rendered prefix for the Agent Logs detail view, and diff --git a/src/lib/mcp/orgMcpTools.ts b/src/lib/mcp/orgMcpTools.ts index 18cc1b2619..dede654169 100644 --- a/src/lib/mcp/orgMcpTools.ts +++ b/src/lib/mcp/orgMcpTools.ts @@ -418,6 +418,7 @@ export function registerOrgTools( rows, idPrefix: assistantPrefix, reason: "user-turn", + turnId, }); // Snapshot the rendered prefix for the Agent Logs detail view, diff --git a/src/services/canvas-turn-persistence.ts b/src/services/canvas-turn-persistence.ts index 37b3267e10..9cd0289663 100644 --- a/src/services/canvas-turn-persistence.ts +++ b/src/services/canvas-turn-persistence.ts @@ -398,6 +398,27 @@ export async function fetchStoredConversationMessages(args: { return Array.isArray(row.messages) ? (row.messages as unknown as StoredMessage[]) : []; } +/** + * Where a turn's rows go: after the last stored row of that turn when it + * has one (its user message, or an earlier partial write), else at the + * end. Appending blindly puts a slow reply after a message that was sent + * while it ran. + */ +export function placeTurnRows( + existing: T[], + rows: T[], + turnId?: string, +): T[] { + if (!turnId) return [...existing, ...rows]; + const prefix = `${turnId}-`; + let last = -1; + existing.forEach((m, i) => { + if (typeof m.id === "string" && m.id.startsWith(prefix)) last = i; + }); + if (last < 0 || last === existing.length - 1) return [...existing, ...rows]; + return [...existing.slice(0, last + 1), ...rows, ...existing.slice(last + 1)]; +} + /** * Append rows into a canvas conversation under the same row-level lock * the fan-out worker and the autosave PUT use, so all writers serialize @@ -414,8 +435,15 @@ export async function appendTurnMessages(args: { rows: StoredMessage[]; idPrefix: string; reason: CanvasConversationUpdateReason; + /** + * The turn these rows finish. Its user row (`${turnId}-u`) is already + * stored, so the rows land right after that turn's last row rather + * than at the end — a reply that finishes after a later message was + * sent (another tab, a shared room) still reads user → reply → user. + */ + turnId?: string; }): Promise { - const { conversationId, rows, idPrefix, reason } = args; + const { conversationId, rows, idPrefix, reason, turnId } = args; if (rows.length === 0) return false; let didAppend = false; @@ -437,7 +465,7 @@ export async function appendTurnMessages(args: { await tx.sharedConversation.update({ where: { id: conversationId }, data: { - messages: [...existing, ...rows] as unknown as never, + messages: placeTurnRows(existing, rows, turnId) as unknown as never, lastMessageAt: new Date(), }, });