feat(copilot): stream file edits into the live collaborative Y.Doc (keep embedded view collaborative)

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

@@ -0,0 +1,90 @@
+/**
+ * @vitest-environment node
+ *
+ * Multi-replica (store-enabled) coverage for the copilot live-merge stale-check. The main
+ * `file-doc.test.ts` runs with the store DISABLED (single-replica fallback); this file mocks an ENABLED
+ * store so the cross-process branch of `mergeMarkdownIntoRoom` — staleness against the SHARED synced
+ * version under the merge lock, and `recordVersion` writing `setSyncedVersion` — is exercised directly.
+ * The enabled merge path reads its base from the shared store (not an in-memory room), so no JOIN/seed
+ * is needed: calling `applyMarkdownToLiveFileDoc` against the fake store drives the branch on its own.
+ */
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import * as Y from 'yjs'
+
+const { mockFetchFileDocMerge } = vi.hoisted(() => ({
+  mockFetchFileDocMerge: vi.fn(),
+}))
+
+/**
+ * A minimal ENABLED store: in-memory monotonic synced version (mirrors SET_VERSION_IF_NEWER_SCRIPT), a
+ * non-null stream state so the merge has a base, and no-op locks/publish. Only the surface the
+ * store-enabled merge path touches is implemented.
+ */
+const fakeStore = {
+  enabled: true,
+  versions: new Map<string, number>(),
+  acquireMergeSlot: vi.fn(async () => 'token'),
+  releaseMergeSlot: vi.fn(async () => {}),
+  getStreamState: vi.fn(async () => new Uint8Array([1])),
+  publishAndWait: vi.fn(async () => {}),
+  getSyncedVersion: vi.fn(async (name: string) => fakeStore.versions.get(name) ?? null),
+  setSyncedVersion: vi.fn(async (name: string, version: number) => {
+    fakeStore.versions.set(name, Math.max(fakeStore.versions.get(name) ?? 0, version))
+  }),
+}
+
+vi.mock('@sim/platform-authz/rooms', () => ({ authorizeRoom: vi.fn() }))
+
+vi.mock('@/handlers/file-doc-app', () => ({
+  fetchFileDocSeed: vi.fn(),
+  fetchFileDocMerge: mockFetchFileDocMerge,
+  fetchFileDocPersist: vi.fn(),
+}))
+
+vi.mock('@/handlers/file-doc-store', () => ({
+  getFileDocStore: () => fakeStore,
+  REDIS_ORIGIN: Symbol('redis'),
+  REDIS_SNAPSHOT_ORIGIN: Symbol('redis-snapshot'),
+}))
+
+import { applyMarkdownToLiveFileDoc } from '@/handlers/file-doc'
+
+const ROOM_NAME = 'workspace-file-doc:file-1'
+
+describe('applyMarkdownToLiveFileDoc — multi-replica (store-enabled) ordering', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    fakeStore.versions.clear()
+    fakeStore.acquireMergeSlot.mockResolvedValue('token')
+    fakeStore.getStreamState.mockResolvedValue(new Uint8Array([1]))
+    mockFetchFileDocMerge.mockResolvedValue(Y.encodeStateAsUpdate(new Y.Doc()))
+  })
+
+  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 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 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'
+    )
+    expect(fakeStore.setSyncedVersion).toHaveBeenCalledWith(ROOM_NAME, 150)
+    // setSyncedVersion fired only for the two durable writes, never for a streaming snapshot.
+    expect(fakeStore.setSyncedVersion).toHaveBeenCalledTimes(2)
+  })
+})

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

@@ -543,6 +543,65 @@ describe('setupWorkspaceFileDocHandlers', () => {
     expect(mockFetchFileDocMerge).not.toHaveBeenCalled()
   })
 
+  it('rejects a stale versioned merge (not newer than the synced version) without regressing the doc', 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', '# newer', { version: 100 })).toBe('applied')
+    mockFetchFileDocMerge.mockClear()
+
+    // An older durable version arriving out of order (e.g. a concurrent write on another process) is
+    // stale: skipped before any diff is computed, so the live doc never regresses to older content and
+    // no diff is published that a later persist could write back.
+    expect(await applyMarkdownToLiveFileDoc('file-1', '# older, stale', { version: 50 })).toBe(
+      'stale'
+    )
+    // The same version is idempotent — also skipped.
+    expect(await applyMarkdownToLiveFileDoc('file-1', '# same version', { version: 100 })).toBe(
+      'stale'
+    )
+    expect(mockFetchFileDocMerge).not.toHaveBeenCalled()
+  })
+
+  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)
+    await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 })
+    await flushMicrotasks()
+
+    mockFetchFileDocMerge.mockResolvedValue(Y.encodeStateAsUpdate(new Y.Doc()))
+
+    // 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 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 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'
+    )
+  })
+
   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

@@ -515,6 +515,16 @@ function emptySeedUpdate(): Uint8Array {
 /** Serializes live merges per file so overlapping calls never race the same doc (see below). */
 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 `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
+  baseVersion?: number
+}
+
 /**
  * Apply new markdown into a file's LIVE collaborative document (Stage C — copilot writing into an open
  * doc). Ships the document's current state to the app to build a minimal Yjs diff, applies it — which
@@ -540,14 +550,12 @@ const fileDocMergeChains = new Map<string, Promise<unknown>>()
 export function applyMarkdownToLiveFileDoc(
   fileId: string,
   markdown: string,
-  version?: number
-): Promise<'applied' | 'no-live-room' | 'merge-unavailable'> {
+  order: MergeOrder = {}
+): 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))
+  const run = prior.catch(() => {}).then(() => mergeMarkdownIntoRoom(name, fileId, markdown, order))
   fileDocMergeChains.set(
     name,
     run.finally(() => {
@@ -561,21 +569,51 @@ async function mergeMarkdownIntoRoom(
   name: string,
   fileId: string,
   markdown: string,
-  version?: number
-): Promise<'applied' | 'no-live-room' | 'merge-unavailable'> {
+  { 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. Set on success below.
-  const recordVersion = () => {
+  // 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 `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
     const room = fileDocRooms.get(name)
-    // Never regress the token: merges/seeds/persists all write it (locally and via fire-and-forget
-    // Redis), so a lower value arriving out of order must not shadow a higher one the doc already
-    // incorporates (the Redis side is guarded identically by SET_VERSION_IF_NEWER_SCRIPT).
+    // Never regress the token: merges/seeds/persists all write it, so a lower value arriving out of
+    // order must not shadow a higher one the doc already incorporates (the Redis side is guarded
+    // identically by SET_VERSION_IF_NEWER_SCRIPT).
     if (room) room.syncedVersion = Math.max(room.syncedVersion ?? 0, version)
-    void store.setSyncedVersion(name, version)
+    await store.setSyncedVersion(name, version)
+  }
+
+  // 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.
+  //
+  // Known, accepted limitation: two INDEPENDENT copilot streams editing the SAME file at once share one
+  // base version, so neither is stale relative to the other and their snapshots can interleave in the live
+  // doc. This is transient only — each stream's final durable write is version-ordered and reconciles the
+  // doc, so the steady state is deterministic (last durable wins) and the durable file is never corrupted.
+  // Ordering two independent snapshot streams would need a shared sequence they don't have; the fully
+  // robust form (a per-file streaming lease, or embedding the version in each stream entry) is a scoped
+  // follow-up, not a durability fix owed here.
+  const isStale = (current: number): boolean => {
+    if (version !== undefined) return version <= current
+    if (baseVersion !== undefined) return current > baseVersion
+    return false
   }
 
   if (store.enabled) {
@@ -595,6 +633,11 @@ async function mergeMarkdownIntoRoom(
       return 'merge-unavailable'
     }
     try {
+      // Staleness is checked under the lock against the cluster-wide synced version, so a durable merge
+      // that lost the race to a newer one (on any process) is dropped rather than regressing the doc.
+      const shared = await store.getSyncedVersion(name)
+      const current = Math.max(shared ?? 0, fileDocRooms.get(name)?.syncedVersion ?? 0)
+      if (isStale(current)) return 'stale'
       // Compute the diff against the committed SHARED state and PUBLISH it — every task with the doc
       // live (including this one, via its own tailer) applies it and fans it out to its clients, so the
       // merge reaches the live doc no matter which task the apply-edit call landed on. An empty stream
@@ -604,7 +647,7 @@ async function mergeMarkdownIntoRoom(
       if (!base) return 'no-live-room'
       const diff = await fetchFileDocMerge(fileId, base, markdown)
       await store.publishAndWait(name, diff)
-      recordVersion()
+      await recordVersion()
       return 'applied'
     } finally {
       await store.releaseMergeSlot(name, token)
@@ -614,12 +657,13 @@ async function mergeMarkdownIntoRoom(
   // Single-replica fallback: apply straight to the local authoritative doc.
   const room = fileDocRooms.get(name)
   if (!room || room.owners.size === 0 || !isDocSeeded(room.doc)) return 'no-live-room'
+  if (isStale(room.syncedVersion ?? 0)) return 'stale'
   const update = await fetchFileDocMerge(fileId, Y.encodeStateAsUpdate(room.doc), markdown)
   // The room may have been dropped while the diff was being built; never touch a destroyed doc.
   if (fileDocRooms.get(name) !== room) return 'no-live-room'
   // No transaction origin → `doc.on('update')` relays to the WHOLE room (every editor sees copilot).
   Y.applyUpdate(room.doc, update)
-  recordVersion()
+  await recordVersion()
   return 'applied'
 }
 

apps/realtime/src/routes/http.ts

@@ -209,17 +209,17 @@ 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, 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.
-        const result = await applyMarkdownToLiveFileDoc(
-          fileId,
-          markdown,
-          typeof version === 'number' ? version : undefined
-        )
+        // `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,
+          baseVersion: typeof baseVersion === 'number' ? baseVersion : undefined,
+        })
         res.writeHead(200, { 'Content-Type': 'application/json' })
         res.end(JSON.stringify({ applied: result === 'applied' }))
       } catch (error) {

apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/rich-markdown-editor.tsx

@@ -3,7 +3,7 @@
 import { memo, useCallback, useEffect, useRef, useState } from 'react'
 import { cn, toast } from '@sim/emcn'
 import { FILE_DOC_SEED, type JoinFileDocError } from '@sim/realtime-protocol/file-doc'
-import type { Extensions, JSONContent } from '@tiptap/core'
+import { type Extensions, generateHTML, type JSONContent } from '@tiptap/core'
 import { isChangeOrigin } from '@tiptap/extension-collaboration'
 import { Fragment, Slice } from '@tiptap/pm/model'
 import { NodeSelection } from '@tiptap/pm/state'
@@ -297,6 +297,18 @@ export function LoadedRichMarkdownEditor({
       ? ''
       : parseMarkdownToDoc(splitFrontmatter(content).body)
   )
+  /**
+   * A read-only placeholder rendered from the already-fetched markdown while a collaborative doc waits
+   * for its server seed, so the pane shows content instantly instead of blocking blank on the socket
+   * round-trip (the seed IS the same markdown, so the swap on {@link collabReady} is seamless). Static
+   * HTML — it holds no editor, doc, or awareness, so it structurally cannot write to the Y.Doc, which
+   * is the invariant that keeps seeding out of the client (a client seed duplicates the doc).
+   */
+  const [placeholderHtml] = useState<string | null>(() =>
+    collaborationEnabled
+      ? generateHTML(parseMarkdownToDoc(splitFrontmatter(content).body), EXTENSIONS)
+      : null
+  )
   /**
    * The body currently shown in the editor: seeded from a settled mount, updated on local edits (via
    * onUpdate) and on each streamed sync. Incremental edits (append/patch) stream complete snapshots and
@@ -923,6 +935,11 @@ export function LoadedRichMarkdownEditor({
     []
   )
 
+  // Show the read-only placeholder only for a plain cold open — never during an agent stream. A stream
+  // that begins before the doc has seeded fills the (hidden) editor via Yjs, so gating the placeholder
+  // off while streaming lets that live content show through instead of hiding it behind stale markdown.
+  const showPlaceholder = collaborationEnabled && !collabReady && !isStreaming
+
   return (
     <div
       ref={containerRef}
@@ -947,9 +964,20 @@ export function LoadedRichMarkdownEditor({
           if (images.length > 0) void insertImagesRef.current(images, at)
         }}
       />
+      {showPlaceholder && placeholderHtml && (
+        // Instant read-only content while the collaborative doc seeds; the editor stays mounted-but-
+        // hidden below so it renders the seeded doc before the swap. Same layout box → no reflow.
+        <div
+          className='rich-markdown-prose mx-auto w-full max-w-[48rem] px-8 py-6'
+          dangerouslySetInnerHTML={{ __html: placeholderHtml }}
+        />
+      )}
       <EditorContent
         editor={editor}
-        className='mx-auto flex w-full max-w-[48rem] flex-1 flex-col px-8 py-6 selection:bg-[var(--selection-bg)] selection:text-[var(--text-primary)] dark:selection:bg-[var(--selection-dark)] dark:selection:text-white'
+        className={cn(
+          'mx-auto flex w-full max-w-[48rem] flex-1 flex-col px-8 py-6 selection:bg-[var(--selection-bg)] selection:text-[var(--text-primary)] dark:selection:bg-[var(--selection-dark)] dark:selection:text-white',
+          showPlaceholder && placeholderHtml && 'hidden'
+        )}
       />
     </div>
   )