fix(collab-doc): order streaming merges by causal base version, not wall-clock

A streaming snapshot now carries baseVersion (the durable contentUpdatedAt it was
built from) instead of a wall-clock streamedAt. The relay drops the snapshot when a
newer durable write landed since that base, so a concurrent human save can no longer
be clobbered in the live doc and then persisted over the durable file. Skew-immune:
both keys are DB-monotonic contentUpdatedAt values.

Author: waleedlatif1

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

@@ -60,27 +60,26 @@ describe('applyMarkdownToLiveFileDoc — multi-replica (store-enabled) ordering'
     mockFetchFileDocMerge.mockResolvedValue(Y.encodeStateAsUpdate(new Y.Doc()))
   })
 
-  it('stale-checks against the SHARED synced version and never records a streaming merge', async () => {
-    // A durable write records the shared synced version cluster-wide.
+  it('drops a stale-base streaming snapshot against the SHARED synced version and never records it', async () => {
+    // A durable write (e.g. a concurrent human save on another process) records the shared synced version.
     expect(await applyMarkdownToLiveFileDoc('file-1', '# durable', { version: 100 })).toBe(
       'applied'
     )
     expect(fakeStore.setSyncedVersion).toHaveBeenCalledWith(ROOM_NAME, 100)
     mockFetchFileDocMerge.mockClear()
 
-    // A delayed streaming snapshot older than the SHARED version (e.g. from another process) is stale —
-    // rejected under the lock before any diff is built, so the live doc never regresses.
-    expect(await applyMarkdownToLiveFileDoc('file-1', '# older stream', { streamedAt: 50 })).toBe(
-      'stale'
-    )
+    // A streaming snapshot built from an older base (50) than the SHARED synced version is stale —
+    // rejected under the lock before any diff is built, so it can't clobber the durable write.
+    expect(
+      await applyMarkdownToLiveFileDoc('file-1', '# stale-base stream', { baseVersion: 50 })
+    ).toBe('stale')
     expect(mockFetchFileDocMerge).not.toHaveBeenCalled()
 
-    // A streaming snapshot newer than the shared version applies (advances the live view)...
-    expect(await applyMarkdownToLiveFileDoc('file-1', '# newer stream', { streamedAt: 200 })).toBe(
-      'applied'
-    )
-    // ...but it is never recorded: a durable write between the two (150) still applies. Had 200 been
-    // recorded to the shared store, 150 would be rejected as stale.
+    // A streaming snapshot whose base is the current shared version applies (nothing newer to clobber)...
+    expect(
+      await applyMarkdownToLiveFileDoc('file-1', '# current-base stream', { baseVersion: 100 })
+    ).toBe('applied')
+    // ...but is never recorded: a later durable write at 150 still applies.
     expect(await applyMarkdownToLiveFileDoc('file-1', '# durable again', { version: 150 })).toBe(
       'applied'
     )

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

@@ -569,7 +569,7 @@ describe('setupWorkspaceFileDocHandlers', () => {
     expect(mockFetchFileDocMerge).not.toHaveBeenCalled()
   })
 
-  it('orders a streaming merge by streamedAt but never records it (stays behind the durable version)', async () => {
+  it('drops a streaming snapshot whose base predates a newer durable write, but never records it', async () => {
     mockFetchFileDocSeed.mockResolvedValue(seedResult('# Original')) // seed version 1
     const { io } = createIo()
     const { handlers } = setup('socket-1', io)
@@ -578,23 +578,25 @@ describe('setupWorkspaceFileDocHandlers', () => {
 
     mockFetchFileDocMerge.mockResolvedValue(Y.encodeStateAsUpdate(new Y.Doc()))
 
-    // A newer durable version lands and is recorded as the synced version.
+    // A durable write (e.g. a concurrent human save) lands and is recorded as the synced version.
     expect(await applyMarkdownToLiveFileDoc('file-1', '# durable', { version: 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.
-    expect(await applyMarkdownToLiveFileDoc('file-1', '# older stream', { streamedAt: 50 })).toBe(
-      'stale'
-    )
+    // A streaming snapshot built from an OLDER base (50) — copilot loaded the file before that durable
+    // write — is stale: applying it would diff the live doc back toward the copilot content and clobber
+    // the durable write, which a later persist would then write over the file.
+    expect(
+      await applyMarkdownToLiveFileDoc('file-1', '# stale-base stream', { baseVersion: 50 })
+    ).toBe('stale')
 
-    // A streaming snapshot NEWER than the durable version applies (it advances the live view)...
-    expect(await applyMarkdownToLiveFileDoc('file-1', '# newer stream', { streamedAt: 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.
+    // A streaming snapshot whose base IS the current durable version applies — nothing newer to clobber.
+    expect(
+      await applyMarkdownToLiveFileDoc('file-1', '# current-base stream', { baseVersion: 100 })
+    ).toBe('applied')
+
+    // ...and a streaming merge is never recorded as the synced version: a later durable write at 150 still
+    // applies (only durable writes move the synced version; the final edit_content write reconciles).
     expect(await applyMarkdownToLiveFileDoc('file-1', '# durable again', { version: 150 })).toBe(
       'applied'
     )

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

@@ -517,11 +517,12 @@ const fileDocMergeChains = new Map<string, Promise<unknown>>()
 
 /**
  * How a merge is positioned on the file's version line — mirrors the sim-side `LiveFileDocMergeOrder`
- * wire fields. A durable `version` is checked AND recorded; a streaming `streamedAt` is checked only.
+ * wire fields. A durable `version` is checked AND recorded; a streaming `baseVersion` (the durable version
+ * the snapshot was built from) is checked only — dropped if a newer durable write has since landed.
  */
 interface MergeOrder {
   version?: number
-  streamedAt?: number
+  baseVersion?: number
 }
 
 /**
@@ -568,15 +569,15 @@ async function mergeMarkdownIntoRoom(
   name: string,
   fileId: string,
   markdown: string,
-  { version, streamedAt }: MergeOrder
+  { version, baseVersion }: MergeOrder
 ): 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. Only a durable
-  // `version` is recorded — a streaming `streamedAt` orders the merge (below) but is never a checkpoint,
+  // `version` is recorded — a streaming `baseVersion` 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
@@ -588,21 +589,24 @@ async function mergeMarkdownIntoRoom(
     await store.setSyncedVersion(name, version)
   }
 
-  // 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.
-  //
-  // Only durable `version` (DB-monotonic `contentUpdatedAt`) is ever recorded, so the durable line is
-  // immune to clock skew. `streamedAt` is best-effort wall-clock used ONLY to order transient streaming
-  // snapshots; correctness never depends on it — a skewed clock at worst drops or briefly regresses the
-  // live preview, which the next durable write reconciles.
-  const orderingVersion = version ?? streamedAt
-  const isStale = (current: number): boolean =>
-    orderingVersion !== undefined && orderingVersion <= current
+  // Order this merge on the file's version line, where `current` is the durable version the doc already
+  // incorporates. Both keys are DB-monotonic `contentUpdatedAt` values (no wall-clock), so ordering is
+  // immune to clock skew:
+  //   - A durable `version` is stale if it is NOT strictly newer than `current` — a newer durable write
+  //     already landed (possibly on another process, out of dispatch order); applying its older markdown
+  //     would regress the doc while the monotonic token stays high.
+  //   - A streaming `baseVersion` (the durable version the snapshot was built from) is stale if `current`
+  //     has moved PAST it — a newer durable write landed since the snapshot's base, so diffing the live
+  //     doc back toward the snapshot would clobber that write's content (which a later persist, still
+  //     holding the current If-Match token, would then write over the durable file). This is what stops a
+  //     concurrent human edit from being silently lost; the per-process caller chain cannot see it.
+  // A merge with neither key is never stale (legacy, unordered). Only a durable `version` is recorded,
+  // so a streaming snapshot never advances the synced version — the final `edit_content` write does.
+  const isStale = (current: number): boolean => {
+    if (version !== undefined) return version <= current
+    if (baseVersion !== undefined) return current > baseVersion
+    return false
+  }
 
   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,16 @@ 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, streamedAt } = JSON.parse(body)
+        const { fileId, markdown, version, baseVersion } = 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.
+        // `baseVersion` is a streaming snapshot's causal base: dropped if a newer durable write landed.
         const result = await applyMarkdownToLiveFileDoc(fileId, markdown, {
           version: typeof version === 'number' ? version : undefined,
-          streamedAt: typeof streamedAt === 'number' ? streamedAt : undefined,
+          baseVersion: typeof baseVersion === 'number' ? baseVersion : 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

@@ -15,7 +15,7 @@ const { mergeEditIntoLiveFileDocMock, isLiveDocMergeInFlightMock } = vi.hoisted(
       (
         fileId: string,
         markdown: string,
-        order?: { version?: number; streamedAt?: number }
+        order?: { version?: number; baseVersion?: number }
       ) => Promise<void>
     >(),
   isLiveDocMergeInFlightMock: vi.fn<(fileId: string) => boolean>(),
@@ -46,6 +46,8 @@ import type { ActiveFileIntent, ExecutionContext, StreamEvent } from '@/lib/copi
 const STREAM_ID = 'stream-1'
 const EDIT_TOOL_CALL_ID = 'edit-content-1'
 const WORKSPACE_FILE_TOOL_CALL_ID = 'workspace-file-1'
+/** The durable version (`contentUpdatedAt`, epoch ms) the streamed base content is at. */
+const BASE_VERSION_MS = 900_000
 
 /** One args_delta chunk of the streamed `edit_content` JSON, as a driveable StreamEvent. */
 function editContentDelta(argumentsDelta: string): StreamEvent {
@@ -104,8 +106,12 @@ describe('processFilePreviewStreamEvent — live-doc streaming merge', () => {
     vi.clearAllMocks()
     mergeEditIntoLiveFileDocMock.mockResolvedValue(undefined)
     isLiveDocMergeInFlightMock.mockReturnValue(false)
-    // Default: an append/patch base is available (a non-empty file), so the base-present gate passes.
-    peekFileIntentMock.mockResolvedValue({ existingContent: 'Base.' })
+    // Default: an append/patch base is available (a non-empty file) at durable version BASE_VERSION_MS,
+    // so the base-present gate passes and the streaming merge carries that base version.
+    peekFileIntentMock.mockResolvedValue({
+      existingContent: 'Base.',
+      fileRecord: { contentUpdatedAt: new Date(BASE_VERSION_MS) },
+    })
     state = createFilePreviewAdapterState()
     nowMs = 1_000_000
     vi.spyOn(Date, 'now').mockImplementation(() => nowMs)
@@ -129,7 +135,7 @@ describe('processFilePreviewStreamEvent — live-doc streaming merge', () => {
     })
   }
 
-  it('merges the growing full content (no version arg) into the live doc as it streams', async () => {
+  it('merges the growing full content (base version, no durable version) into the live doc as it streams', async () => {
     const intent = makeIntent({ operation: 'append', fileId: 'file-grow', fileName: 'notes.md' })
 
     await drive(editContentDelta('{"content":"Hello'), intent)
@@ -143,16 +149,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. 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.
+    // carries `baseVersion` (the durable version it was built from) to order it — never `version`, which
+    // rides the final edit_content write; so the relay drops it if a newer durable write has since landed
+    // but never records it as a durable checkpoint.
     expect(first[0]).toBe('file-grow')
     expect(first[1]).toContain('Base.')
     expect(first[1]).toContain('Hello')
-    expect(typeof first[2]?.streamedAt).toBe('number')
+    expect(first[2]?.baseVersion).toBe(BASE_VERSION_MS)
     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')
+    expect(second[2]?.baseVersion).toBe(BASE_VERSION_MS)
   })
 
   it('throttles merges: two deltas within LIVE_DOC_MERGE_THROTTLE_MS yield one merge', async () => {