fix(copilot): address review — order merges, exclude update, gate throttle, unhide stream

Review round on #6108:
- Greptile P1 (durable merges lose ordering): serialize ALL merges per file on one chain in
  mergeEditIntoLiveFileDoc (each chains after the current tail), so concurrent durable writes
  can't resume-and-fire out of order. notify now exposes isLiveDocMergeInFlight.
- Cursor High (update stream blanks the doc): only append/patch stream — they build on the loaded
  base; update is a from-scratch rewrite whose partial snapshot would diff the full doc toward a
  fragment, so it applies atomically at the durable write.
- Cursor Medium (throttle advances on a dropped merge): the adapter gates on !isLiveDocMergeInFlight,
  so the send throttle advances only on an actual dispatch — no lag, no backlog behind a slow relay.
- Cursor Medium (placeholder hides a live stream): show the fast-render placeholder only when not
  streaming, so a stream that starts before the doc seeds shows through the editor.
- Soften merge.ts/notify.ts comments per the lifecycle audit: only UNTOUCHED regions are preserved;
  a region the merge rewrites reconciles toward copilot's content.

Tests updated: notify covers chain ordering + isLiveDocMergeInFlight; adapter covers append streaming,
throttle, non-markdown/base-less/update skips, and the in-flight skip.

Author: waleedlatif1

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

@@ -935,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}
@@ -959,7 +964,7 @@ export function LoadedRichMarkdownEditor({
           if (images.length > 0) void insertImagesRef.current(images, at)
         }}
       />
-      {collaborationEnabled && !collabReady && placeholderHtml && (
+      {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
@@ -971,7 +976,7 @@ export function LoadedRichMarkdownEditor({
         editor={editor}
         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',
-          collaborationEnabled && !collabReady && 'hidden'
+          showPlaceholder && placeholderHtml && 'hidden'
         )}
       />
     </div>

apps/sim/lib/collab-doc/merge.ts

@@ -10,10 +10,12 @@ import { applyMarkdownToYDoc } from './converter'
  * and applies the returned diff, which Yjs merges with any concurrent user edits before relaying it to
  * every connected editor.
  *
- * `applyMarkdownToYDoc` performs a real `updateYFragment` diff (not a replace), so unrelated
- * paragraphs the user is editing are preserved. The returned update is relative to the document's
- * state at call time (`Y.encodeStateAsUpdate(doc, before)`), so it is exactly the change to apply — and
- * is empty (a no-op update) when `markdown` already matches the document.
+ * `applyMarkdownToYDoc` performs a real `updateYFragment` diff (not a replace), so paragraphs the diff
+ * does not touch are preserved even while the user edits them. A region the incoming `markdown` DOES
+ * change is reconciled toward that markdown — a concurrent user edit inside such a region is diffed
+ * away, since `markdown` is built from a base snapshot, not the user's in-flight text. The returned
+ * update is relative to the document's state at call time (`Y.encodeStateAsUpdate(doc, before)`), so it
+ * is exactly the change to apply — and is empty (a no-op update) when `markdown` already matches.
  */
 export function buildFileDocMergeUpdate(docState: Uint8Array, markdown: string): Uint8Array {
   const doc = new Y.Doc()

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

@@ -9,9 +9,10 @@ import {
   MothershipStreamV1ToolPhase,
 } from '@/lib/copilot/generated/mothership-stream-v1'
 
-const { mergeEditIntoLiveFileDocMock } = vi.hoisted(() => ({
+const { mergeEditIntoLiveFileDocMock, isLiveDocMergeInFlightMock } = vi.hoisted(() => ({
   mergeEditIntoLiveFileDocMock:
     vi.fn<(fileId: string, markdown: string, version?: number) => Promise<void>>(),
+  isLiveDocMergeInFlightMock: vi.fn<(fileId: string) => boolean>(),
 }))
 
 const { peekFileIntentMock } = vi.hoisted(() => ({
@@ -20,6 +21,7 @@ const { peekFileIntentMock } = vi.hoisted(() => ({
 
 vi.mock('@/lib/realtime/notify', () => ({
   mergeEditIntoLiveFileDoc: mergeEditIntoLiveFileDocMock,
+  isLiveDocMergeInFlight: isLiveDocMergeInFlightMock,
 }))
 
 vi.mock('@/lib/copilot/tools/server/files/file-intent-store', () => ({
@@ -95,7 +97,9 @@ describe('processFilePreviewStreamEvent — live-doc streaming merge', () => {
   beforeEach(() => {
     vi.clearAllMocks()
     mergeEditIntoLiveFileDocMock.mockResolvedValue(undefined)
-    peekFileIntentMock.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.' })
     state = createFilePreviewAdapterState()
     nowMs = 1_000_000
     vi.spyOn(Date, 'now').mockImplementation(() => nowMs)
@@ -120,7 +124,7 @@ describe('processFilePreviewStreamEvent — live-doc streaming merge', () => {
   }
 
   it('merges the growing full content (no version arg) into the live doc as it streams', async () => {
-    const intent = makeIntent({ operation: 'update', fileId: 'file-grow', fileName: 'notes.md' })
+    const intent = makeIntent({ operation: 'append', fileId: 'file-grow', fileName: 'notes.md' })
 
     await drive(editContentDelta('{"content":"Hello'), intent)
     await flushMicrotasks()
@@ -131,17 +135,21 @@ describe('processFilePreviewStreamEvent — live-doc streaming merge', () => {
     await flushMicrotasks()
 
     expect(mergeEditIntoLiveFileDocMock).toHaveBeenCalledTimes(2)
-    expect(mergeEditIntoLiveFileDocMock.mock.calls[0]).toEqual(['file-grow', 'Hello'])
-    // The second merge carries the GROWN full text, never a diff.
-    expect(mergeEditIntoLiveFileDocMock.mock.calls[1]).toEqual(['file-grow', 'Hello world'])
-    // No version arg on either streaming merge — the durable version rides the final edit_content write.
-    expect(mergeEditIntoLiveFileDocMock.mock.calls[0]).toHaveLength(2)
-    expect(mergeEditIntoLiveFileDocMock.mock.calls[1]).toHaveLength(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.
+    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(second[1]).toContain('Hello world')
+    expect(second[1].length).toBeGreaterThan(first[1].length)
   })
 
   it('throttles merges: two deltas within LIVE_DOC_MERGE_THROTTLE_MS yield one merge', async () => {
     const intent = makeIntent({
-      operation: 'update',
+      operation: 'append',
       fileId: 'file-throttle',
       fileName: 'notes.md',
     })
@@ -155,11 +163,10 @@ describe('processFilePreviewStreamEvent — live-doc streaming merge', () => {
     await flushMicrotasks()
 
     expect(mergeEditIntoLiveFileDocMock).toHaveBeenCalledTimes(1)
-    expect(mergeEditIntoLiveFileDocMock).toHaveBeenCalledWith('file-throttle', 'Hel')
   })
 
   it('does not merge for a non-markdown file (no collaborative room)', async () => {
-    const intent = makeIntent({ operation: 'update', fileId: 'file-txt', fileName: 'notes.txt' })
+    const intent = makeIntent({ operation: 'append', fileId: 'file-txt', fileName: 'notes.txt' })
 
     await drive(editContentDelta('{"content":"plain text body'), intent)
     await flushMicrotasks()
@@ -178,4 +185,25 @@ describe('processFilePreviewStreamEvent — live-doc streaming merge', () => {
     // A base-less snapshot would diff to a delete-everything wipe of the seeded doc, so it must be skipped.
     expect(mergeEditIntoLiveFileDocMock).not.toHaveBeenCalled()
   })
+
+  it('does not stream an update (from-scratch rewrite) — it would blank the doc mid-stream', async () => {
+    const intent = makeIntent({ operation: 'update', fileId: 'file-update', fileName: 'notes.md' })
+
+    await drive(editContentDelta('{"content":"Rewritten intro'), intent)
+    await flushMicrotasks()
+
+    // Update streams a partial rewrite; diffing the full doc toward it would delete most of the file
+    // until it grows back, so update applies atomically at the final durable write instead.
+    expect(mergeEditIntoLiveFileDocMock).not.toHaveBeenCalled()
+  })
+
+  it('skips the merge while one is already in flight for the file (does not backlog / advance throttle)', async () => {
+    isLiveDocMergeInFlightMock.mockReturnValue(true)
+    const intent = makeIntent({ operation: 'append', fileId: 'file-busy', fileName: 'notes.md' })
+
+    await drive(editContentDelta('{"content":"Hello'), intent)
+    await flushMicrotasks()
+
+    expect(mergeEditIntoLiveFileDocMock).not.toHaveBeenCalled()
+  })
 })

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

@@ -24,7 +24,7 @@ import {
   buildFilePreviewText,
   loadWorkspaceFileTextForPreview,
 } from '@/lib/copilot/tools/server/files/file-preview'
-import { mergeEditIntoLiveFileDoc } from '@/lib/realtime/notify'
+import { isLiveDocMergeInFlight, mergeEditIntoLiveFileDoc } from '@/lib/realtime/notify'
 import { resolveWorkspaceFileReference } from '@/lib/uploads/contexts/workspace/workspace-file-manager'
 import { isMarkdownFile } from '@/lib/uploads/utils/file-utils'
 
@@ -653,19 +653,25 @@ 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. Throttled per file;
-          // fire-and-forget so a slow relay never stalls the stream (`mergeEditIntoLiveFileDoc` never
-          // throws, coordinates ordering per file, and treats a versionless call as a non-durable
-          // preview merge). 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; and for `append`/`patch`, only once the base file content has loaded —
-          // a base-less snapshot would diff to a delete-everything wipe of the seeded doc. `update`
-          // streams a full rewrite from scratch, so it needs no base.
+          // 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`).
+          //
+          // 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
+          // would diff to a delete-everything wipe of the seeded doc). `update` is a from-scratch
+          // rewrite: streaming its partial content would diff the full doc toward a fragment and blank
+          // it mid-stream, so it applies atomically at the final durable write instead. Skip while a
+          // merge is in flight for this file — one at a time, and don't advance the throttle on a
+          // no-op — so a slow relay can't backlog stale snapshots or make the doc lag the stream.
           const dueForLiveMerge =
             nextSession.fileId !== undefined &&
             isMarkdownFile({ name: nextSession.fileName ?? '' }) &&
-            (editIntent.operation === 'update' ||
-              currentPreview.session.baseContent !== undefined) &&
+            (editIntent.operation === 'append' || editIntent.operation === 'patch') &&
+            currentPreview.session.baseContent !== undefined &&
+            !isLiveDocMergeInFlight(nextSession.fileId) &&
             now - currentPreview.lastLiveMergeAt >= LIVE_DOC_MERGE_THROTTLE_MS
           const nextLiveMergeAt = dueForLiveMerge ? now : currentPreview.lastLiveMergeAt
           if (dueForLiveMerge && nextSession.fileId) {

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

@@ -6,7 +6,7 @@ import { afterEach, describe, expect, it, vi } from 'vitest'
 vi.mock('@/lib/core/utils/urls', () => ({ getSocketServerUrl: () => 'http://realtime' }))
 vi.mock('@/lib/core/config/env', () => ({ env: { INTERNAL_API_SECRET: 'secret' } }))
 
-import { mergeEditIntoLiveFileDoc } from './notify'
+import { isLiveDocMergeInFlight, mergeEditIntoLiveFileDoc } from './notify'
 
 describe('mergeEditIntoLiveFileDoc', () => {
   afterEach(() => {
@@ -39,23 +39,23 @@ describe('mergeEditIntoLiveFileDoc', () => {
     await expect(mergeEditIntoLiveFileDoc('file-1', '# hello', 42)).resolves.toBeUndefined()
   })
 
-  it('drops a streaming (versionless) merge while one is already in flight for the same file', async () => {
-    let resolveFirst: (value: { ok: boolean }) => void = () => {}
-    const fetchMock = vi
-      .fn()
-      .mockImplementationOnce(() => new Promise((resolve) => (resolveFirst = resolve)))
-      .mockResolvedValue({ ok: true })
-    vi.stubGlobal('fetch', fetchMock)
+  it('reports isLiveDocMergeInFlight while a merge runs and clears when it settles', async () => {
+    let resolveFetch: (value: { ok: boolean }) => void = () => {}
+    vi.stubGlobal(
+      'fetch',
+      vi.fn(() => new Promise((resolve) => (resolveFetch = resolve)))
+    )
 
-    const first = mergeEditIntoLiveFileDoc('file-inflight', 'v1') // versionless → in flight (pending)
+    expect(isLiveDocMergeInFlight('file-flight')).toBe(false)
+    const run = mergeEditIntoLiveFileDoc('file-flight', 'v1')
     await Promise.resolve()
-    // A second versionless merge while the first is pending is dropped, not queued — so a stale
-    // snapshot can never land after a newer one and regress the doc.
-    await mergeEditIntoLiveFileDoc('file-inflight', 'v2')
-    expect(fetchMock).toHaveBeenCalledTimes(1)
+    // The streaming caller checks this to skip a redundant merge (and not advance its throttle) while
+    // one is in flight, so a slow relay can't backlog stale snapshots.
+    expect(isLiveDocMergeInFlight('file-flight')).toBe(true)
 
-    resolveFirst({ ok: true })
-    await first
+    resolveFetch({ ok: true })
+    await run
+    expect(isLiveDocMergeInFlight('file-flight')).toBe(false)
   })
 
   it('a durable (versioned) merge waits for an in-flight streaming merge, then applies last', async () => {
@@ -86,4 +86,41 @@ describe('mergeEditIntoLiveFileDoc', () => {
       JSON.stringify({ fileId: 'file-durable', markdown: 'final content', version: 100 })
     )
   })
+
+  it('serializes concurrent durable writes behind a streaming merge, strictly in order', async () => {
+    const applied: Array<number | 'stream'> = []
+    const resolvers: Array<() => void> = []
+    vi.stubGlobal(
+      'fetch',
+      vi.fn((_url: string, init: { body: string }) => {
+        applied.push(JSON.parse(init.body).version ?? 'stream')
+        return new Promise<{ ok: boolean }>((resolve) =>
+          resolvers.push(() => resolve({ ok: true }))
+        )
+      })
+    )
+    const flush = async () => {
+      for (let i = 0; i < 6; i++) await Promise.resolve()
+    }
+
+    const s = mergeEditIntoLiveFileDoc('file-order', 's') // streaming, in flight
+    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)
+    await flush()
+    expect(applied).toEqual(['stream']) // A and B queued behind streaming
+
+    resolvers[0]() // finish streaming → A applies next (not B)
+    await flush()
+    expect(applied).toEqual(['stream', 1])
+
+    resolvers[1]() // finish A → B applies after A
+    await flush()
+    expect(applied).toEqual(['stream', 1, 2])
+
+    resolvers[2]()
+    await Promise.all([s, a, b])
+  })
 })

apps/sim/lib/realtime/notify.ts

@@ -121,41 +121,51 @@ export async function notifyFolderResourceChanged(
  * server-side), and the relay applies this merge THROUGH the shared Redis stream, so it reaches the
  * live doc on whichever task holds it and can't go stale relative to this direct write.
  *
- * Awaited (not fire-and-forget) so the fetch dispatches before the route handler returns; bounded to
- * {@link APPLY_EDIT_TIMEOUT_MS}, so it adds latency only when the socket pod is unreachable.
+ * A durable caller awaits this (so the fetch dispatches before the route handler returns); the copilot
+ * 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.
  *
- * Concurrency is coordinated per file so that ordering can never regress the doc: a STREAMING
- * (versionless) merge is DROPPED while another is already in flight for the file — the next throttled
- * caller sends the latest snapshot, and a stale snapshot can never land after a newer one; a DURABLE
- * (versioned) write instead WAITS for the in-flight streaming merge, so the final content is always the
- * last merge applied and cannot be clobbered by a late straggler.
+ * 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.
  */
 export async function mergeEditIntoLiveFileDoc(
   fileId: string,
   markdown: string,
   version?: number
 ): Promise<void> {
-  const pending = liveDocMergeInFlight.get(fileId)
-  if (version === undefined && pending) return
-  if (pending) await pending
-
-  const run = applyLiveFileDocMerge(fileId, markdown, version)
-  liveDocMergeInFlight.set(fileId, run)
+  const tail = liveDocMergeChain.get(fileId) ?? Promise.resolve()
+  const run = tail.then(() => applyLiveFileDocMerge(fileId, markdown, version))
+  liveDocMergeChain.set(fileId, run)
   try {
     await run
   } finally {
-    if (liveDocMergeInFlight.get(fileId) === run) liveDocMergeInFlight.delete(fileId)
+    if (liveDocMergeChain.get(fileId) === run) liveDocMergeChain.delete(fileId)
   }
 }
 
-/** Files with a live-doc merge in flight → the running merge promise (never rejects). */
-const liveDocMergeInFlight = new Map<string, Promise<void>>()
+/** Per file, the tail of the serialized merge chain (each merge applies after it); never rejects
+ *  because {@link applyLiveFileDocMerge} never throws. Absent when the file's chain is idle. */
+const liveDocMergeChain = new Map<string, Promise<void>>()
+
+/**
+ * Whether a live-doc merge is currently running or queued for the file. The copilot streaming caller
+ * checks this to skip a redundant snapshot (and to not advance its send throttle) while a merge is in
+ * flight — bounding the stream to one live merge per file at a time without backlogging stale
+ * snapshots behind a slow relay.
+ */
+export function isLiveDocMergeInFlight(fileId: string): boolean {
+  return liveDocMergeChain.has(fileId)
+}
 
 /** POST the merge to the relay. Never throws (a live-doc merge is best-effort). */
 async function applyLiveFileDocMerge(