feat(copilot): add seq ordinal to copilot_messages for order-preserving reads

copilot_messages had no column preserving message order: created_at (set
from each message's timestamp) ties at millisecond granularity in 58% of
chats, and some chats have out-of-order timestamps within their array. The
only other tiebreaker, id, is a random UUID — so ORDER BY created_at, id
renders same-timestamp user/assistant pairs swapped. This blocks the R+1
read cutover.

Add an integer seq = the message's 0-based index within the chat's JSONB
array (ground-truth order), backfilled inline in migration 0219 (no script
for self-hosters or us). Reads will use ORDER BY seq NULLS LAST, created_at,
id at cutover; reads still come from JSONB after this PR.

Design:
- seq is a tiebreaker, not the sole sort key (concurrent-append/NULL safety).
- Nullable now; defer NOT NULL so rolling-deploy old pods don't fail inserts.
- replace (update-messages snapshot) overwrites seq = array index
  (re-densifies after a mid-conversation delete); append preserves existing
  seq via COALESCE and assigns base+idx from a single MAX(seq) read (never
  MAX+i in SQL — multi-row batches would collide).
- Dedupe message ids before insert (87 prod chats carry dup ids; a repeated
  id in one INSERT...ON CONFLICT would otherwise throw).
- Backfill picks first-occurrence per (chat,id), gap-free via ROW_NUMBER;
  validated on staging data (0-based, contiguous, 0 bad ranges).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: waleedlatif1

apps/sim/lib/copilot/chat/messages-dual-write.test.ts

@@ -26,6 +26,12 @@ const assistantMsg: PersistedMessage = {
   timestamp: '2026-01-01T00:00:01.000Z',
 }
 
+/** The first arg passed to the most recent `.values(...)` call. */
+function lastValuesRows() {
+  const calls = dbChainMockFns.values.mock.calls
+  return calls[calls.length - 1][0] as Array<Record<string, unknown>>
+}
+
 describe('messages-dual-write', () => {
   beforeEach(() => {
     vi.clearAllMocks()
@@ -43,7 +49,7 @@ describe('messages-dual-write', () => {
 
       expect(dbChainMockFns.insert).toHaveBeenCalledTimes(1)
       expect(dbChainMockFns.values).toHaveBeenCalledTimes(1)
-      const rows = dbChainMockFns.values.mock.calls[0][0]
+      const rows = lastValuesRows()
       expect(rows).toHaveLength(2)
 
       expect(rows[0]).toMatchObject({
@@ -54,22 +60,33 @@ describe('messages-dual-write', () => {
         model: null,
         streamId: null,
       })
-      expect(rows[0].createdAt).toEqual(new Date(userMsg.timestamp))
-      expect(rows[0].updatedAt).toEqual(new Date(userMsg.timestamp))
+      expect(rows[0].createdAt as Date).toEqual(new Date(userMsg.timestamp))
+      expect(rows[0].updatedAt as Date).toEqual(new Date(userMsg.timestamp))
 
       expect(rows[1]).toMatchObject({
         chatId: 'chat-1',
         messageId: 'msg-asst-1',
         role: 'assistant',
         content: assistantMsg,
       })
-      expect(rows[1].createdAt).toEqual(new Date(assistantMsg.timestamp))
+      expect(rows[1].createdAt as Date).toEqual(new Date(assistantMsg.timestamp))
+    })
+
+    it('assigns seq as 0-based array index when the chat has no prior rows', async () => {
+      await appendCopilotChatMessages('chat-1', [userMsg, assistantMsg])
+      const rows = lastValuesRows()
+      expect(rows[0].seq).toBe(0)
+      expect(rows[1].seq).toBe(1)
     })
 
-    it('preserves per-message ordering via timestamp', async () => {
+    it('continues seq from MAX(seq)+1 when the chat already has rows', async () => {
+      // MAX(seq) read resolves through the select().from().where() chain.
+      dbChainMockFns.where.mockResolvedValueOnce([{ maxSeq: 4 }])
+
       await appendCopilotChatMessages('chat-1', [userMsg, assistantMsg])
-      const rows = dbChainMockFns.values.mock.calls[0][0]
-      expect(rows[0].createdAt.getTime()).toBeLessThan(rows[1].createdAt.getTime())
+      const rows = lastValuesRows()
+      expect(rows[0].seq).toBe(5)
+      expect(rows[1].seq).toBe(6)
     })
 
     it('passes chatModel and streamId options to every row', async () => {
@@ -78,14 +95,14 @@ describe('messages-dual-write', () => {
         streamId: 'stream-xyz',
       })
 
-      const rows = dbChainMockFns.values.mock.calls[0][0]
+      const rows = lastValuesRows()
       expect(rows[0].model).toBe('claude-sonnet-4-5')
       expect(rows[0].streamId).toBe('stream-xyz')
       expect(rows[1].model).toBe('claude-sonnet-4-5')
       expect(rows[1].streamId).toBe('stream-xyz')
     })
 
-    it('uses ON CONFLICT DO UPDATE with chat_id + message_id target', async () => {
+    it('uses ON CONFLICT DO UPDATE that PRESERVES existing seq', async () => {
       await appendCopilotChatMessages('chat-1', [userMsg])
 
       expect(dbChainMockFns.onConflictDoUpdate).toHaveBeenCalledTimes(1)
@@ -96,6 +113,15 @@ describe('messages-dual-write', () => {
       expect(conflictArg.set).toHaveProperty('model')
       expect(conflictArg.set).toHaveProperty('streamId')
       expect(conflictArg.set).toHaveProperty('updatedAt')
+      // append must not renumber existing rows -> COALESCE(existing, excluded)
+      expect(conflictArg.set.seq.strings.join('')).toContain('COALESCE(')
+    })
+
+    it('collapses duplicate message ids to a single row', async () => {
+      await appendCopilotChatMessages('chat-1', [userMsg, { ...userMsg, content: 'dupe' }])
+      const rows = lastValuesRows()
+      expect(rows).toHaveLength(1)
+      expect(rows[0].messageId).toBe('msg-user-1')
     })
 
     it('swallows DB errors so the legacy JSONB write stays canonical', async () => {
@@ -120,25 +146,43 @@ describe('messages-dual-write', () => {
       expect(dbChainMockFns.delete).toHaveBeenCalledTimes(1)
       expect(dbChainMockFns.insert).toHaveBeenCalledTimes(1)
 
-      const rows = dbChainMockFns.values.mock.calls[0][0]
+      const rows = lastValuesRows()
       expect(rows).toHaveLength(2)
-      expect(rows.map((r: { messageId: string }) => r.messageId)).toEqual([
-        'msg-user-1',
-        'msg-asst-1',
-      ])
+      expect(rows.map((r) => r.messageId)).toEqual(['msg-user-1', 'msg-asst-1'])
 
       expect(dbChainMockFns.onConflictDoUpdate).toHaveBeenCalledTimes(1)
       const conflictArg = dbChainMockFns.onConflictDoUpdate.mock.calls[0][0]
       expect(conflictArg.set).toHaveProperty('streamId')
       expect(conflictArg.set).toHaveProperty('model')
     })
 
+    it('assigns seq as the snapshot array index (0-based)', async () => {
+      await replaceCopilotChatMessages('chat-1', [userMsg, assistantMsg])
+      const rows = lastValuesRows()
+      expect(rows[0].seq).toBe(0)
+      expect(rows[1].seq).toBe(1)
+    })
+
+    it('OVERWRITES seq on conflict so positions re-densify after a delete', async () => {
+      await replaceCopilotChatMessages('chat-1', [userMsg])
+      const conflictArg = dbChainMockFns.onConflictDoUpdate.mock.calls[0][0]
+      // authoritative snapshot -> seq = excluded.seq (no COALESCE preserve)
+      expect(conflictArg.set.seq.strings.join('')).toBe('excluded.seq')
+    })
+
+    it('collapses duplicate message ids to a single row', async () => {
+      await replaceCopilotChatMessages('chat-1', [userMsg, { ...userMsg, content: 'dupe' }])
+      const rows = lastValuesRows()
+      expect(rows).toHaveLength(1)
+      expect(rows[0].seq).toBe(0)
+    })
+
     it('passes chatModel to every row in the snapshot', async () => {
       await replaceCopilotChatMessages('chat-1', [userMsg], {
         chatModel: 'gpt-4o-mini',
       })
 
-      const rows = dbChainMockFns.values.mock.calls[0][0]
+      const rows = lastValuesRows()
       expect(rows[0].model).toBe('gpt-4o-mini')
     })
 

apps/sim/lib/copilot/chat/messages-dual-write.ts

@@ -7,9 +7,27 @@ import type { PersistedMessage } from '@/lib/copilot/chat/persisted-message'
 
 const logger = createLogger('CopilotMessagesDualWrite')
 
+/**
+ * Collapse duplicate message ids (keeping the first occurrence), preserving
+ * order. A single `INSERT ... ON CONFLICT` cannot touch the same conflict
+ * target twice, so a snapshot/array carrying a repeated id would otherwise
+ * throw "ON CONFLICT DO UPDATE command cannot affect row a second time".
+ */
+function dedupeById(messages: PersistedMessage[]): PersistedMessage[] {
+  const seen = new Set<string>()
+  const out: PersistedMessage[] = []
+  for (const m of messages) {
+    if (seen.has(m.id)) continue
+    seen.add(m.id)
+    out.push(m)
+  }
+  return out
+}
+
 function toRow(
   chatId: string,
   message: PersistedMessage,
+  seq: number,
   options?: { chatModel?: string | null; streamId?: string | null }
 ): typeof copilotMessages.$inferInsert {
   const ts = new Date(message.timestamp)
@@ -18,6 +36,7 @@ function toRow(
     messageId: message.id,
     role: message.role,
     content: message,
+    seq,
     model: options?.chatModel ?? null,
     streamId: options?.streamId ?? null,
     createdAt: ts,
@@ -37,16 +56,28 @@ export async function appendCopilotChatMessages(
 ): Promise<void> {
   if (messages.length === 0) return
   try {
+    const deduped = dedupeById(messages)
+    // Assign seq as base + array index. Base is computed in JS from a single
+    // MAX(seq) read — never `MAX(seq)+i` in SQL, since a multi-row INSERT
+    // evaluates every row's subquery against the same pre-insert state and the
+    // rows would collide. Existing rows keep their seq on conflict (COALESCE);
+    // the authoritative renumber happens via replaceCopilotChatMessages.
+    const [maxRow] = await db
+      .select({ maxSeq: sql<number | null>`max(${copilotMessages.seq})` })
+      .from(copilotMessages)
+      .where(eq(copilotMessages.chatId, chatId))
+    const base = (maxRow?.maxSeq ?? -1) + 1
     await db
       .insert(copilotMessages)
-      .values(messages.map((m) => toRow(chatId, m, options)))
+      .values(deduped.map((m, i) => toRow(chatId, m, base + i, options)))
       .onConflictDoUpdate({
         target: [copilotMessages.chatId, copilotMessages.messageId],
         set: {
           content: sql`excluded.content`,
           role: sql`excluded.role`,
           model: sql`COALESCE(excluded.model, ${copilotMessages.model})`,
           streamId: sql`COALESCE(excluded.stream_id, ${copilotMessages.streamId})`,
+          seq: sql`COALESCE(${copilotMessages.seq}, excluded.seq)`,
           updatedAt: sql`now()`,
         },
       })
@@ -69,7 +100,8 @@ export async function replaceCopilotChatMessages(
   options?: { chatModel?: string | null }
 ): Promise<void> {
   try {
-    const newMessageIds = messages.map((m) => m.id)
+    const deduped = dedupeById(messages)
+    const newMessageIds = deduped.map((m) => m.id)
     await db.transaction(async (tx) => {
       // Drop rows for messages not in the new snapshot.
       await tx
@@ -82,19 +114,22 @@ export async function replaceCopilotChatMessages(
               )
             : eq(copilotMessages.chatId, chatId)
         )
-      if (messages.length === 0) return
-      // Upsert remaining rows. ON CONFLICT preserves existing stream_id / model
+      if (deduped.length === 0) return
+      // Upsert remaining rows. The snapshot is authoritative on order, so seq is
+      // overwritten with the array index — this re-densifies positions after a
+      // mid-conversation delete. ON CONFLICT preserves existing stream_id / model
       // so a snapshot save doesn't clobber metadata set during streaming.
       await tx
         .insert(copilotMessages)
-        .values(messages.map((m) => toRow(chatId, m, options)))
+        .values(deduped.map((m, i) => toRow(chatId, m, i, options)))
         .onConflictDoUpdate({
           target: [copilotMessages.chatId, copilotMessages.messageId],
           set: {
             content: sql`excluded.content`,
             role: sql`excluded.role`,
             model: sql`COALESCE(excluded.model, ${copilotMessages.model})`,
             streamId: sql`COALESCE(excluded.stream_id, ${copilotMessages.streamId})`,
+            seq: sql`excluded.seq`,
             updatedAt: sql`now()`,
           },
         })

packages/db/migrations/0219_yielding_the_captain.sql

@@ -0,0 +1,19 @@
+ALTER TABLE "copilot_messages" ADD COLUMN "seq" integer;--> statement-breakpoint
+WITH ordered AS (
+  SELECT c."id" AS chat_id, elem.value->>'id' AS message_id, elem.ord AS ord
+  FROM "copilot_chats" c
+  CROSS JOIN LATERAL jsonb_array_elements(c."messages") WITH ORDINALITY AS elem(value, ord)
+  WHERE jsonb_typeof(c."messages") = 'array' AND jsonb_array_length(c."messages") > 0
+),
+first_occurrence AS (
+  SELECT chat_id, message_id, MIN(ord) AS first_ord FROM ordered GROUP BY chat_id, message_id
+),
+ranked AS (
+  SELECT chat_id, message_id,
+         (ROW_NUMBER() OVER (PARTITION BY chat_id ORDER BY first_ord) - 1) AS seq
+  FROM first_occurrence
+)
+UPDATE "copilot_messages" m SET "seq" = r.seq
+FROM ranked r
+WHERE m."chat_id" = r.chat_id AND m."message_id" = r.message_id;--> statement-breakpoint
+CREATE INDEX "copilot_messages_chat_seq_idx" ON "copilot_messages" USING btree ("chat_id","seq") WHERE "copilot_messages"."deleted_at" IS NULL;