fix(collab-doc): order streaming merges by streamedAt so a late snapshot can't regress a newer durable write

Author: waleedlatif1

apps/realtime/src/handlers/file-doc.test.ts

@@ -565,6 +565,34 @@ describe('setupWorkspaceFileDocHandlers', () => {
     expect(mockFetchFileDocMerge).not.toHaveBeenCalled()
   })
 
+  it('orders a streaming merge by streamedAt but never records it (stays behind the durable version)', async () => {
+    mockFetchFileDocSeed.mockResolvedValue(seedResult('# Original')) // seed version 1
+    const { io } = createIo()
+    const { handlers } = setup('socket-1', io)
+    await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 })
+    await flushMicrotasks()
+
+    mockFetchFileDocMerge.mockResolvedValue(Y.encodeStateAsUpdate(new Y.Doc()))
+
+    // A newer durable version lands and is recorded as the synced version.
+    expect(await applyMarkdownToLiveFileDoc('file-1', '# durable', 100)).toBe('applied')
+
+    // A delayed streaming snapshot OLDER than that durable version — e.g. a throttled copilot snapshot
+    // from another process arriving late — is stale, so it can't regress the doc back toward its content.
+    // (`version` omitted, `streamedAt` passed as the 4th arg.)
+    expect(await applyMarkdownToLiveFileDoc('file-1', '# older stream', undefined, 50)).toBe(
+      'stale'
+    )
+
+    // A streaming snapshot NEWER than the durable version applies (it advances the live view)...
+    expect(await applyMarkdownToLiveFileDoc('file-1', '# newer stream', undefined, 200)).toBe(
+      'applied'
+    )
+    // ...but it is never RECORDED as the synced version: a durable write between the two (150) still
+    // applies. Had the streaming 200 been recorded, 150 would have been rejected as stale.
+    expect(await applyMarkdownToLiveFileDoc('file-1', '# durable again', 150)).toBe('applied')
+  })
+
   it('serializes concurrent merges for the same file (second waits for the first)', async () => {
     mockFetchFileDocSeed.mockResolvedValue(seedResult('# Original'))
     const { io } = createIo()

apps/realtime/src/handlers/file-doc.ts

@@ -540,14 +540,15 @@ const fileDocMergeChains = new Map<string, Promise<unknown>>()
 export function applyMarkdownToLiveFileDoc(
   fileId: string,
   markdown: string,
-  version?: number
+  version?: number,
+  streamedAt?: number
 ): Promise<'applied' | 'no-live-room' | 'merge-unavailable' | 'stale'> {
   const name = roomName(fileDocRoom(fileId))
   const prior = fileDocMergeChains.get(name) ?? Promise.resolve()
   // `.catch` so a failed prior merge doesn't reject this one — each merge is independent.
   const run = prior
     .catch(() => {})
-    .then(() => mergeMarkdownIntoRoom(name, fileId, markdown, version))
+    .then(() => mergeMarkdownIntoRoom(name, fileId, markdown, version, streamedAt))
   fileDocMergeChains.set(
     name,
     run.finally(() => {
@@ -561,14 +562,17 @@ async function mergeMarkdownIntoRoom(
   name: string,
   fileId: string,
   markdown: string,
-  version?: number
+  version?: number,
+  streamedAt?: number
 ): Promise<'applied' | 'no-live-room' | 'merge-unavailable' | 'stale'> {
   const store = getFileDocStore()
 
   // The durable version this merge carries is now incorporated in the live doc — record it (cluster-wide
   // in Redis for multi-task, plus this task's room) so the persist If-Match guard treats this write as
   // synced rather than an out-of-band conflict. AWAITED so the version is durable before the merge lock
-  // releases, so the next lock holder's staleness check (below) reads a consistent value.
+  // releases, so the next lock holder's staleness check (below) reads a consistent value. Only a durable
+  // `version` is recorded — a streaming `streamedAt` orders the merge (below) but is never a checkpoint,
+  // so the synced version stays pinned to the last durable write.
   const recordVersion = async () => {
     if (version === undefined) return
     const room = fileDocRooms.get(name)
@@ -579,12 +583,16 @@ async function mergeMarkdownIntoRoom(
     await store.setSyncedVersion(name, version)
   }
 
-  // A versioned (durable) merge that is not NEWER than the version the doc already incorporates is
-  // stale — a newer durable write already landed (possibly on another process, out of dispatch order).
-  // Diffing toward its older markdown would regress the live doc while the monotonic token stays high,
-  // so a later persist could write that stale content back over the durable file. Skip it. Streaming
-  // (versionless) merges carry no durable version and are ordered per-process by the caller.
-  const isStale = (current: number): boolean => version !== undefined && version <= current
+  // Position this merge on the file's monotonic version line: a durable write by its `version`, a
+  // streaming snapshot by its `streamedAt` production time. A merge not NEWER than the version the doc
+  // already incorporates is stale — a newer durable write already landed (possibly on another process,
+  // out of dispatch order). Diffing toward its older markdown would regress the live doc while the
+  // monotonic token stays high, so a later persist could write that stale content back over the durable
+  // file. Skip it. This is what stops a delayed streaming snapshot from clobbering a newer durable merge
+  // across processes (the per-process caller chain cannot). A merge with neither key is never stale.
+  const orderingVersion = version ?? streamedAt
+  const isStale = (current: number): boolean =>
+    orderingVersion !== undefined && orderingVersion <= current
 
   if (store.enabled) {
     // Serialize merges to this file ACROSS tasks — the per-file chain above only covers this process.

apps/realtime/src/routes/http.ts

@@ -209,16 +209,18 @@ export function createHttpHandler(roomManager: IRoomManager, logger: Logger) {
     if (req.method === 'POST' && req.url === '/api/file-doc/apply-edit') {
       try {
         const body = await readRequestBody(req)
-        const { fileId, markdown, version } = JSON.parse(body)
+        const { fileId, markdown, version, streamedAt } = JSON.parse(body)
         if (!isNonEmptyString(fileId) || typeof markdown !== 'string') {
           return sendError(res, 'Invalid fileId or markdown', 400)
         }
         // `version` (the durable updatedAt this markdown was written with) records that the live doc now
         // incorporates that durable version, so the persist If-Match guard won't flag it as a conflict.
+        // `streamedAt` orders a streaming snapshot on the same version line without recording it.
         const result = await applyMarkdownToLiveFileDoc(
           fileId,
           markdown,
-          typeof version === 'number' ? version : undefined
+          typeof version === 'number' ? version : undefined,
+          typeof streamedAt === 'number' ? streamedAt : undefined
         )
         res.writeHead(200, { 'Content-Type': 'application/json' })
         res.end(JSON.stringify({ applied: result === 'applied' }))

apps/sim/lib/copilot/request/go/file-preview-adapter.test.ts

@@ -11,7 +11,13 @@ import {
 
 const { mergeEditIntoLiveFileDocMock, isLiveDocMergeInFlightMock } = vi.hoisted(() => ({
   mergeEditIntoLiveFileDocMock:
-    vi.fn<(fileId: string, markdown: string, version?: number) => Promise<void>>(),
+    vi.fn<
+      (
+        fileId: string,
+        markdown: string,
+        order?: { version?: number; streamedAt?: number }
+      ) => Promise<void>
+    >(),
   isLiveDocMergeInFlightMock: vi.fn<(fileId: string) => boolean>(),
 }))
 
@@ -136,15 +142,17 @@ describe('processFilePreviewStreamEvent — live-doc streaming merge', () => {
 
     expect(mergeEditIntoLiveFileDocMock).toHaveBeenCalledTimes(2)
     const [first, second] = mergeEditIntoLiveFileDocMock.mock.calls
-    // A full-file snapshot (base + streamed), never a diff; it grows across deltas; no version arg on
-    // either streaming merge — the durable version rides the final edit_content write.
+    // A full-file snapshot (base + streamed), never a diff; it grows across deltas. Each streaming merge
+    // carries `streamedAt` (its wall-clock time) to order it — never `version`, which rides the final
+    // edit_content write; so the relay orders the snapshot without recording it as a durable checkpoint.
     expect(first[0]).toBe('file-grow')
-    expect(first).toHaveLength(2)
     expect(first[1]).toContain('Base.')
     expect(first[1]).toContain('Hello')
-    expect(second).toHaveLength(2)
+    expect(typeof first[2]?.streamedAt).toBe('number')
+    expect(first[2]?.version).toBeUndefined()
     expect(second[1]).toContain('Hello world')
     expect(second[1].length).toBeGreaterThan(first[1].length)
+    expect(typeof second[2]?.streamedAt).toBe('number')
   })
 
   it('throttles merges: two deltas within LIVE_DOC_MERGE_THROTTLE_MS yield one merge', async () => {

apps/sim/lib/copilot/request/go/file-preview-adapter.ts

@@ -654,10 +654,12 @@ export async function processFilePreviewStreamEvent(input: {
           // Stream the growing content into the file's LIVE collaborative Y.Doc (when a room is open)
           // so collaborators watching the file see the copilot write stream in via Yjs — the AI as a
           // CRDT peer, applied by the relay as a minimal `updateYFragment` diff. Fire-and-forget so a
-          // slow relay never stalls the stream; `mergeEditIntoLiveFileDoc` serializes ordering per file
-          // and treats a versionless call as a non-durable preview merge (the final `edit_content`
-          // write carries the real version and reconciles the durable file). No-op for `create` (never
-          // streams here) and for a file with no open room (the relay reports `applied: false`).
+          // slow relay never stalls the stream. Pass `streamedAt` (this snapshot's wall-clock time) so
+          // the relay orders it on the file's version line — a delayed snapshot older than a newer
+          // durable write, even from another process, is dropped rather than regressing the doc — but
+          // never records it as a durable checkpoint (the final `edit_content` write carries the real
+          // version and reconciles the durable file). No-op for `create` (never streams here) and for a
+          // file with no open room (the relay reports `applied: false`).
           //
           // Gates: markdown only (non-markdown has no collaborative room). Only `append`/`patch` stream
           // — they build on the existing content, so they need the base loaded (a base-less snapshot
@@ -675,7 +677,9 @@ export async function processFilePreviewStreamEvent(input: {
             now - currentPreview.lastLiveMergeAt >= LIVE_DOC_MERGE_THROTTLE_MS
           const nextLiveMergeAt = dueForLiveMerge ? now : currentPreview.lastLiveMergeAt
           if (dueForLiveMerge && nextSession.fileId) {
-            void mergeEditIntoLiveFileDoc(nextSession.fileId, nextSession.previewText)
+            void mergeEditIntoLiveFileDoc(nextSession.fileId, nextSession.previewText, {
+              streamedAt: now,
+            })
           }
 
           if (

apps/sim/lib/realtime/notify.test.ts

@@ -17,26 +17,43 @@ describe('mergeEditIntoLiveFileDoc', () => {
     const fetchMock = vi.fn().mockResolvedValue({ ok: true })
     vi.stubGlobal('fetch', fetchMock)
 
-    await mergeEditIntoLiveFileDoc('file-1', '# hello', 42)
+    await mergeEditIntoLiveFileDoc('file-1', '# hello', { version: 42 })
 
     expect(fetchMock).toHaveBeenCalledWith(
       'http://realtime/api/file-doc/apply-edit',
       expect.objectContaining({
         method: 'POST',
         headers: expect.objectContaining({ 'x-api-key': 'secret' }),
+        // A durable write sends `version`; the undefined `streamedAt` is dropped by JSON.stringify.
         body: JSON.stringify({ fileId: 'file-1', markdown: '# hello', version: 42 }),
       })
     )
   })
 
+  it('sends streamedAt (not version) for a streaming snapshot', async () => {
+    const fetchMock = vi.fn().mockResolvedValue({ ok: true })
+    vi.stubGlobal('fetch', fetchMock)
+
+    await mergeEditIntoLiveFileDoc('file-1', '# hello', { streamedAt: 1234 })
+
+    // The relay orders the snapshot by streamedAt without recording it — version stays absent on the wire.
+    expect(fetchMock.mock.calls[0][1].body).toBe(
+      JSON.stringify({ fileId: 'file-1', markdown: '# hello', streamedAt: 1234 })
+    )
+  })
+
   it('never throws when the realtime call fails (best-effort)', async () => {
     vi.stubGlobal('fetch', vi.fn().mockRejectedValue(new Error('socket pod down')))
-    await expect(mergeEditIntoLiveFileDoc('file-1', '# hello', 42)).resolves.toBeUndefined()
+    await expect(
+      mergeEditIntoLiveFileDoc('file-1', '# hello', { version: 42 })
+    ).resolves.toBeUndefined()
   })
 
   it('never throws on a non-2xx response', async () => {
     vi.stubGlobal('fetch', vi.fn().mockResolvedValue({ ok: false, status: 503 }))
-    await expect(mergeEditIntoLiveFileDoc('file-1', '# hello', 42)).resolves.toBeUndefined()
+    await expect(
+      mergeEditIntoLiveFileDoc('file-1', '# hello', { version: 42 })
+    ).resolves.toBeUndefined()
   })
 
   it('reports isLiveDocMergeInFlight while a merge runs and clears when it settles', async () => {
@@ -68,7 +85,7 @@ describe('mergeEditIntoLiveFileDoc', () => {
 
     const stream = mergeEditIntoLiveFileDoc('file-durable', 'partial') // versionless, in flight
     await Promise.resolve()
-    const durable = mergeEditIntoLiveFileDoc('file-durable', 'final content', 100) // versioned
+    const durable = mergeEditIntoLiveFileDoc('file-durable', 'final content', { version: 100 }) // versioned
     await Promise.resolve()
     await Promise.resolve()
 
@@ -107,8 +124,8 @@ describe('mergeEditIntoLiveFileDoc', () => {
     await flush()
     // Two durable writes arrive while the streaming merge is in flight — both must chain, not both
     // resume-and-fire concurrently.
-    const a = mergeEditIntoLiveFileDoc('file-order', 'a', 1)
-    const b = mergeEditIntoLiveFileDoc('file-order', 'b', 2)
+    const a = mergeEditIntoLiveFileDoc('file-order', 'a', { version: 1 })
+    const b = mergeEditIntoLiveFileDoc('file-order', 'b', { version: 2 })
     await flush()
     expect(applied).toEqual(['stream']) // A and B queued behind streaming
 

apps/sim/lib/realtime/notify.ts

@@ -125,26 +125,40 @@ export async function notifyFolderResourceChanged(
  * streaming caller fires and forgets it. Bounded to {@link APPLY_EDIT_TIMEOUT_MS}, so it adds latency
  * only when the socket pod is unreachable.
  *
- * `version` is the durable `contentUpdatedAt` (epoch ms) this markdown was written with, for a durable
- * write. Omit it for a STREAMING intermediate merge (the copilot stream mid-flight): intermediate
- * content advances the live doc for viewers but is not a durable checkpoint, so the relay leaves its
- * synced version pinned to the last durable write — which is exactly the copilot tool's final
- * `edit_content` write, carrying the real version, that reconciles the durable file.
+ * `order` positions this merge on the file's monotonic version line so a stale write never regresses
+ * the doc — the relay drops any merge not NEWER than the version the doc already incorporates:
+ * - `version` — a DURABLE write's `contentUpdatedAt` (epoch ms). Both orders the merge AND is recorded
+ *   as the doc's synced version (the persist If-Match guard), so a later persist treats this write as
+ *   synced rather than an out-of-band conflict.
+ * - `streamedAt` — the wall-clock time (epoch ms) a STREAMING snapshot was produced (the copilot stream
+ *   mid-flight). Orders the merge so a delayed snapshot older than a newer durable write — possibly from
+ *   another app process — is dropped, but is NEVER recorded: the synced version stays pinned to the last
+ *   durable write, which is exactly the copilot tool's final `edit_content` write that reconciles the file.
  *
- * Merges for a file run on a single serialized chain: each is chained after the current tail and
- * applies strictly after it, so ordering can never regress the doc — a DURABLE (versioned) write
- * always applies after any in-flight streaming merge AND after every earlier durable write, never
- * concurrently. The final durable write is therefore always the last merge applied and cannot be
- * clobbered by a late straggler. The copilot streaming caller uses {@link isLiveDocMergeInFlight} to
- * skip redundant snapshots while one is in flight, so a slow relay can't backlog stale snapshots.
+ * Pass one or the other, never both. Passing neither applies the merge without ordering it (legacy).
+ *
+ * Ordering is enforced at two scales. Within this process, merges for a file run on a single serialized
+ * chain — each chained after the current tail — so a durable write applies after any in-flight streaming
+ * merge and after every earlier durable write, never concurrently. Across processes, the per-process
+ * chain does not apply, so the relay orders merges by the monotonic version above (durable version /
+ * streaming `streamedAt`) under a cluster-wide lock. The copilot streaming caller uses
+ * {@link isLiveDocMergeInFlight} to skip redundant snapshots while one is in flight, so a slow relay
+ * can't backlog stale snapshots.
  */
+export interface LiveFileDocMergeOrder {
+  /** A durable write's `contentUpdatedAt` (epoch ms): orders the merge AND is recorded as the synced version. */
+  version?: number
+  /** A streaming snapshot's production time (epoch ms): orders the merge only — never recorded as a checkpoint. */
+  streamedAt?: number
+}
+
 export async function mergeEditIntoLiveFileDoc(
   fileId: string,
   markdown: string,
-  version?: number
+  order: LiveFileDocMergeOrder = {}
 ): Promise<void> {
   const tail = liveDocMergeChain.get(fileId) ?? Promise.resolve()
-  const run = tail.then(() => applyLiveFileDocMerge(fileId, markdown, version))
+  const run = tail.then(() => applyLiveFileDocMerge(fileId, markdown, order))
   liveDocMergeChain.set(fileId, run)
   try {
     await run
@@ -171,15 +185,21 @@ export function isLiveDocMergeInFlight(fileId: string): boolean {
 async function applyLiveFileDocMerge(
   fileId: string,
   markdown: string,
-  version?: number
+  order: LiveFileDocMergeOrder
 ): Promise<void> {
   try {
     const response = await fetch(`${getSocketServerUrl()}/api/file-doc/apply-edit`, {
       method: 'POST',
       headers: { 'Content-Type': 'application/json', 'x-api-key': env.INTERNAL_API_SECRET },
-      // A durable `version` (the durable `updatedAt` epoch ms) records the version the live doc now
-      // incorporates (the persist If-Match guard); omitted for a streaming intermediate merge.
-      body: JSON.stringify({ fileId, markdown, version }),
+      // `version` (durable `contentUpdatedAt`) records the synced version the live doc now incorporates
+      // (the persist If-Match guard); `streamedAt` orders a streaming snapshot without recording it.
+      // JSON.stringify drops whichever is undefined, so the wire shape is unchanged for durable writes.
+      body: JSON.stringify({
+        fileId,
+        markdown,
+        version: order.version,
+        streamedAt: order.streamedAt,
+      }),
       signal: AbortSignal.timeout(APPLY_EDIT_TIMEOUT_MS),
     })
     if (!response.ok) {

apps/sim/lib/uploads/contexts/workspace/workspace-file-manager.ts

@@ -1223,11 +1223,9 @@ export async function updateWorkspaceFileContent(
       // incorporates this durable version — the collab persist's optimistic-concurrency guard then won't
       // treat this (already-merged) write as an out-of-band conflict. Must be the SAME field the CAS
       // guards on (`contentUpdatedAt`), not `updatedAt`, or the relay's token wouldn't match the CAS.
-      await mergeEditIntoLiveFileDoc(
-        fileId,
-        content.toString('utf-8'),
-        finalized.file.contentUpdatedAt.getTime()
-      )
+      await mergeEditIntoLiveFileDoc(fileId, content.toString('utf-8'), {
+        version: finalized.file.contentUpdatedAt.getTime(),
+      })
     }
 
     const pathPrefix = getServePathPrefix()