fix(files): apply agent stream as a true CRDT peer + guard base-less snapshots

Review round 1 (Greptile P1s):
- apply the stream against a private shadow replica (seeded from the live doc at
  stream start) and relay only the agent's own delta into the shared doc, so a
  concurrent peer edit to a region the agent snapshot didn't include is no longer
  reverted (previously the whole-body reconcile deleted it)
- gate append snapshots on "must extend the base": a base-less append fragment
  (emitted before the base loads) can no longer reconcile the seeded doc to a wipe;
  patch still legitimately replaces a mid-region
- gate the apply on collabReady so diffs never land on an unseeded doc; keep the
  placeholder visible until the seed swaps in
- plumb streamOperation through the preview surfaces to drive the append gate
- add a peer-edit-preservation test (fails under whole-body reconcile) and refresh
  the undo-isolation + broadcast tests for the session API

Author: waleedlatif1

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

@@ -108,6 +108,7 @@ interface FileViewerProps {
   streamingContent?: string
   isAgentEditing?: boolean
   streamIsIncremental?: boolean
+  streamOperation?: string
   disableStreamingAutoScroll?: boolean
   previewContextKey?: string
   /**
@@ -150,6 +151,7 @@ function FileViewerContent({
   streamingContent,
   isAgentEditing,
   streamIsIncremental,
+  streamOperation,
   disableStreamingAutoScroll = false,
   previewContextKey,
   collaborative,
@@ -196,6 +198,7 @@ function FileViewerContent({
           streamingContent={streamingContent}
           isAgentEditing={isAgentEditing}
           streamIsIncremental={streamIsIncremental}
+          streamOperation={streamOperation}
           disableStreamingAutoScroll={disableStreamingAutoScroll}
           previewContextKey={previewContextKey}
           collaborative={collaborative}

apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/apply-streamed-markdown.test.ts

@@ -6,7 +6,7 @@ import { afterEach, beforeAll, describe, expect, it } from 'vitest'
 import { Awareness } from 'y-protocols/awareness'
 import * as Y from 'yjs'
 import { createMarkdownEditorExtensions } from '../editor-extensions'
-import { applyStreamedMarkdownToLiveDoc } from './apply-streamed-markdown'
+import { applyAgentStreamFrame, beginAgentStream, endAgentStream } from './apply-streamed-markdown'
 
 beforeAll(() => {
   // jsdom does not implement elementFromPoint; the Placeholder extension's viewport tracking calls it
@@ -48,11 +48,13 @@ function track(t: { editor: Editor; doc: Y.Doc; awareness: Awareness }) {
   return t
 }
 
-describe('applyStreamedMarkdownToLiveDoc', () => {
+describe('agent-stream applier', () => {
   it('streams agent content into the live collaborative doc and broadcasts it as Yjs ops', () => {
     const { editor, doc } = track(makeCollabEditor())
 
-    expect(applyStreamedMarkdownToLiveDoc(editor, '# Title\n\nHello world.')).toBe(true)
+    const session = beginAgentStream(editor)
+    expect(session).not.toBeNull()
+    expect(applyAgentStreamFrame(editor, session!, '# Title\n\nHello world.')).toBe(true)
     expect(editor.getText()).toContain('Hello world')
 
     // The write lands as ops on the shared doc, so any peer receives it (this is what makes a
@@ -61,25 +63,27 @@ describe('applyStreamedMarkdownToLiveDoc', () => {
     Y.applyUpdate(peer, Y.encodeStateAsUpdate(doc))
     expect(peer.getXmlFragment('default').toString()).toContain('Hello world')
     peer.destroy()
+    endAgentStream(session!)
   })
 
-  it('returns false when the doc is not yet bound (no ySync binding)', () => {
-    // A plain editor with no collaboration has no ySync binding, so the applier reports "not ready"
-    // rather than throwing — the streaming tick re-arms until the doc is seeded.
+  it('beginAgentStream returns null when the editor has no ySync binding', () => {
+    // A plain editor with no collaboration has no ySync binding, so a stream cannot start against it.
     const editor = new Editor({
       extensions: createMarkdownEditorExtensions({ placeholder: '' }),
       content: '',
     })
     teardown.push(() => editor.destroy())
-    expect(applyStreamedMarkdownToLiveDoc(editor, '# Nope')).toBe(false)
+    expect(beginAgentStream(editor)).toBeNull()
   })
 
   it('keeps agent-streamed ops out of the undo stack while user edits stay undoable', () => {
     const { editor } = track(makeCollabEditor())
 
-    applyStreamedMarkdownToLiveDoc(editor, '# Streamed\n\nAgent wrote this.')
-    // The streamed op used AGENT_STREAM_ORIGIN, which the Collaboration UndoManager does not track —
-    // so there is nothing to undo, and an undo must not revert the agent's content.
+    const session = beginAgentStream(editor)!
+    applyAgentStreamFrame(editor, session, '# Streamed\n\nAgent wrote this.')
+    endAgentStream(session)
+    // The streamed op relayed under a non-`ySyncPluginKey` origin, which the Collaboration UndoManager
+    // does not track — so there is nothing to undo, and an undo must not revert the agent's content.
     expect(editor.can().undo()).toBe(false)
     editor.commands.undo()
     expect(editor.getText()).toContain('Agent wrote this')
@@ -95,31 +99,38 @@ describe('applyStreamedMarkdownToLiveDoc', () => {
     expect(editor.getText()).toContain('Agent wrote this')
   })
 
-  it('merges an agent write with a concurrent peer edit (minimal diff, no clobber)', () => {
+  it('preserves a concurrent peer edit to a region the agent snapshot does not include', () => {
+    // This is the core "AI as a CRDT peer" guarantee: the agent relays only its OWN delta (computed
+    // against a private shadow), never a whole-document reconcile that would revert a peer's edit.
     const { editor, doc } = track(makeCollabEditor())
-    applyStreamedMarkdownToLiveDoc(editor, 'Alpha paragraph.\n\nBeta paragraph.')
-
-    // A peer forks the current state and edits the FIRST paragraph directly on the shared type…
-    const remote = new Y.Doc()
-    Y.applyUpdate(remote, Y.encodeStateAsUpdate(doc))
-    const remoteFrag = remote.getXmlFragment('default')
-    remote.transact(() => {
-      const firstPara = remoteFrag.get(0) as Y.XmlElement
+
+    const session = beginAgentStream(editor)!
+    applyAgentStreamFrame(editor, session, 'Alpha paragraph.\n\nBeta paragraph.')
+
+    // A peer edits the FIRST paragraph directly on the shared doc — the agent's later snapshot still
+    // carries the ORIGINAL first paragraph (it was built from the base, before this edit).
+    const peer = new Y.Doc()
+    Y.applyUpdate(peer, Y.encodeStateAsUpdate(doc))
+    const peerFrag = peer.getXmlFragment('default')
+    peer.transact(() => {
+      const firstPara = peerFrag.get(0) as Y.XmlElement
       const textNode = firstPara.get(0) as Y.XmlText
       textNode.insert(textNode.toString().length, ' EDITED')
     })
+    Y.applyUpdate(doc, Y.encodeStateAsUpdate(peer, Y.encodeStateVector(doc)))
+    peer.destroy()
 
-    // …while the agent rewrites the SECOND paragraph through the live editor binding.
-    applyStreamedMarkdownToLiveDoc(editor, 'Alpha paragraph.\n\nBeta paragraph, expanded.')
-
-    // Exchange updates both ways (as the relay would); a full-document replace would have clobbered
-    // the peer's concurrent edit — a minimal `updateYFragment` diff preserves both.
-    Y.applyUpdate(doc, Y.encodeStateAsUpdate(remote, Y.encodeStateVector(doc)))
-    Y.applyUpdate(remote, Y.encodeStateAsUpdate(doc, Y.encodeStateVector(remote)))
-
-    const merged = doc.getXmlFragment('default').toString()
-    expect(merged).toContain('EDITED')
-    expect(merged).toContain('expanded')
-    remote.destroy()
+    // The agent appends a third paragraph. Its snapshot's first paragraph is the stale original, but the
+    // shadow-relayed delta only inserts the new paragraph — so the peer's " EDITED" must survive.
+    applyAgentStreamFrame(
+      editor,
+      session,
+      'Alpha paragraph.\n\nBeta paragraph.\n\nGamma paragraph.'
+    )
+    endAgentStream(session)
+
+    const live = doc.getXmlFragment('default').toString()
+    expect(live).toContain('EDITED')
+    expect(live).toContain('Gamma paragraph')
   })
 })

apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/apply-streamed-markdown.ts

@@ -1,8 +1,12 @@
 import type { Editor } from '@tiptap/core'
 import { Node as PMNode } from '@tiptap/pm/model'
-import { updateYFragment, ySyncPluginKey } from '@tiptap/y-tiptap'
+import { initProseMirrorDoc, updateYFragment, ySyncPluginKey } from '@tiptap/y-tiptap'
+import * as Y from 'yjs'
 import { parseMarkdownToDoc } from '../markdown-parse'
 
+/** The Yjs fragment name TipTap's Collaboration extension binds to (its default `field`). */
+const COLLAB_DOC_FIELD = 'default'
+
 /**
  * Transaction origin for agent-streamed writes into a live collaborative doc. It is deliberately NOT
  * the `ySyncPluginKey` origin that local user edits use, so the Collaboration UndoManager — which
@@ -11,22 +15,64 @@ import { parseMarkdownToDoc } from '../markdown-parse'
 const AGENT_STREAM_ORIGIN = Symbol('agent-stream')
 
 /**
- * Apply a streamed markdown body into the editor's live collaborative Y.Doc as a minimal CRDT diff.
- *
- * Uses the running `ySyncPlugin` binding's {@link updateYFragment} — the same primitive TipTap runs on
- * every keystroke — so only the delta between the doc's current content and `body` is written, never a
- * full-document replace that would wipe collaborators. Each diff is a small Yjs op that renders locally
- * (via the binding's observer, the remote-edit render path) AND broadcasts to every peer, so the stream
- * is smooth here and on other clients alike. Runs under {@link AGENT_STREAM_ORIGIN} so the streamed ops
- * stay out of the user's undo stack. Returns `false` when the editor has no live ySync binding (e.g. a
- * non-collaborative editor); the caller gates seed-readiness separately via `collabReady`.
+ * A private Yjs replica the agent stream reconciles against, so a stream writes into the live doc as a
+ * TRUE peer: only the agent's own delta reaches the shared doc, never a whole-document reconcile that
+ * would revert a collaborator's concurrent edit. Seeded from the live doc at stream start; it receives
+ * ONLY agent reconciles (never peer updates), so `shadow → nextTarget` yields exactly the agent's change.
+ */
+export interface AgentStreamSession {
+  shadow: Y.Doc
+  fragment: Y.XmlFragment
+}
+
+/**
+ * Begin an agent stream by snapshotting the live doc into a private shadow replica. Returns `null` when
+ * the editor has no live ySync binding (e.g. a non-collaborative editor).
+ */
+export function beginAgentStream(editor: Editor): AgentStreamSession | null {
+  const binding = ySyncPluginKey.getState(editor.state)?.binding
+  if (!binding) return null
+  const shadow = new Y.Doc()
+  Y.applyUpdate(shadow, Y.encodeStateAsUpdate(binding.doc))
+  return { shadow, fragment: shadow.getXmlFragment(COLLAB_DOC_FIELD) }
+}
+
+/**
+ * Apply one streamed markdown body. Reconciles the shadow toward `body` with `updateYFragment` (the same
+ * minimal-diff primitive TipTap runs per keystroke), captures ONLY the resulting agent delta, and relays
+ * it into the live doc under {@link AGENT_STREAM_ORIGIN}. Because the shadow never sees peer updates, the
+ * delta touches only what the agent changed — so concurrent peer edits elsewhere in the live doc survive,
+ * the change renders locally (via the binding's observer, the remote-edit path), broadcasts to every
+ * peer, and stays out of the user's undo stack. Returns `false` when the editor has no live ySync binding.
  */
-export function applyStreamedMarkdownToLiveDoc(editor: Editor, body: string): boolean {
+export function applyAgentStreamFrame(
+  editor: Editor,
+  session: AgentStreamSession,
+  body: string
+): boolean {
   const binding = ySyncPluginKey.getState(editor.state)?.binding
   if (!binding) return false
   const target = PMNode.fromJSON(editor.schema, parseMarkdownToDoc(body))
-  binding.doc.transact(() => {
-    updateYFragment(binding.doc, binding.type, target, binding)
-  }, AGENT_STREAM_ORIGIN)
+  let delta: Uint8Array | null = null
+  const capture = (update: Uint8Array, origin: unknown) => {
+    if (origin === AGENT_STREAM_ORIGIN) delta = update
+  }
+  session.shadow.on('update', capture)
+  try {
+    session.shadow.transact(() => {
+      // `updateYFragment` diffs against the fragment's CURRENT content, so it needs the fragment↔PM
+      // binding metadata; `initProseMirrorDoc` reconstructs it from the fragment's present state.
+      const { meta } = initProseMirrorDoc(session.fragment, editor.schema)
+      updateYFragment(session.shadow, session.fragment, target, meta)
+    }, AGENT_STREAM_ORIGIN)
+  } finally {
+    session.shadow.off('update', capture)
+  }
+  if (delta) Y.applyUpdate(binding.doc, delta, AGENT_STREAM_ORIGIN)
   return true
 }
+
+/** End an agent stream and free its shadow replica. */
+export function endAgentStream(session: AgentStreamSession): void {
+  session.shadow.destroy()
+}

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

@@ -20,7 +20,12 @@ import type { SaveStatus } from '@/hooks/use-autosave'
 import { useFileContentSource } from '@/hooks/use-file-content-source'
 import { PreviewLoadingFrame } from '../preview-shared'
 import { useEditableFileContent } from '../use-editable-file-content'
-import { applyStreamedMarkdownToLiveDoc } from './collaboration/apply-streamed-markdown'
+import {
+  type AgentStreamSession,
+  applyAgentStreamFrame,
+  beginAgentStream,
+  endAgentStream,
+} from './collaboration/apply-streamed-markdown'
 import { useFileDocCollaboration } from './collaboration/use-file-doc-collaboration'
 import { createMarkdownEditorExtensions } from './editor-extensions'
 import { findHeadingPos } from './heading-anchors'
@@ -79,14 +84,21 @@ interface RichMarkdownEditorProps {
    * applied live; a rebuild is only revealed while it extends what's shown (see the streaming tick).
    */
   streamIsIncremental?: boolean
+  /**
+   * The agent edit operation driving the stream, when known (`create`/`append`/`update`/`patch`). Used
+   * only to relax the "must extend" gate for `patch` (which legitimately replaces a mid-document region):
+   * every other operation's snapshot must extend what's shown, so a base-less `append` fragment can't
+   * reconcile the live doc to a wipe.
+   */
+  streamOperation?: string
   disableStreamingAutoScroll?: boolean
   previewContextKey?: string
   /** Disable the `@` tag-insertion menu (existing tags still render). Defaults off — the file editor keeps tagging. */
   disableTagging?: boolean
   /**
    * Opt this surface into live collaborative editing (Files page + the embedded chat file preview).
    * Collaboration can coexist with agent streaming: while streaming, the growing content is applied to
-   * the shared Y.Doc as minimal CRDT diffs (see {@link applyStreamedMarkdownToLiveDoc}) rather than a
+   * the shared Y.Doc as minimal CRDT diffs (see {@link applyAgentStreamFrame}) rather than a
    * full-document `setContent`, so the stream stays smooth and every peer sees it live.
    */
   collaborative?: boolean
@@ -111,6 +123,7 @@ export const RichMarkdownEditor = memo(function RichMarkdownEditor({
   streamingContent,
   isAgentEditing,
   streamIsIncremental,
+  streamOperation,
   disableStreamingAutoScroll = false,
   previewContextKey,
   disableTagging,
@@ -181,6 +194,7 @@ export const RichMarkdownEditor = memo(function RichMarkdownEditor({
       userName={userName}
       autoFocus={autoFocus}
       streamIsIncremental={streamIsIncremental}
+      streamOperation={streamOperation}
       disableStreamingAutoScroll={disableStreamingAutoScroll}
       disableTagging={disableTagging}
       collaborative={collaborative}
@@ -206,6 +220,8 @@ interface LoadedRichMarkdownEditorProps {
   autoFocus?: boolean
   /** See {@link RichMarkdownEditorProps.streamIsIncremental}. */
   streamIsIncremental?: boolean
+  /** See {@link RichMarkdownEditorProps.streamOperation}. */
+  streamOperation?: string
   disableStreamingAutoScroll?: boolean
   disableTagging?: boolean
   /** See {@link RichMarkdownEditorProps.collaborative}. */
@@ -239,6 +255,7 @@ export function LoadedRichMarkdownEditor({
   userName,
   autoFocus,
   streamIsIncremental,
+  streamOperation,
   disableStreamingAutoScroll,
   disableTagging,
   collaborative = false,
@@ -356,6 +373,10 @@ export function LoadedRichMarkdownEditor({
    */
   const streamIsIncrementalRef = useRef(streamIsIncremental)
   streamIsIncrementalRef.current = streamIsIncremental
+  const streamOperationRef = useRef(streamOperation)
+  streamOperationRef.current = streamOperation
+  /** The live agent-stream shadow replica, held for the current stream and freed on settle/unmount. */
+  const agentStreamSessionRef = useRef<AgentStreamSession | null>(null)
   const router = useRouter()
   const routerRef = useRef(router)
   routerRef.current = router
@@ -855,7 +876,11 @@ export function LoadedRichMarkdownEditor({
           }
           const shownBody = lastSyncedBodyRef.current
           const extendsShown = shownBody === null || pending.startsWith(shownBody)
-          if (!streamIsIncrementalRef.current && !extendsShown) {
+          // Every snapshot except a mid-document `patch` must EXTEND what's shown: a from-scratch rebuild
+          // (`create`/`update`) is only revealed as it grows, and an `append` snapshot that doesn't extend
+          // the base is a base-less fragment (the server emits one before the base loads) which would
+          // reconcile the seeded doc down to a wipe. Only `patch` legitimately replaces a mid-region.
+          if (!extendsShown && streamOperationRef.current !== 'patch') {
             streamRafRef.current = null
             return
           }
@@ -868,9 +893,11 @@ export function LoadedRichMarkdownEditor({
           }
           const el = containerRef.current
           const pinnedToBottom = el ? el.scrollHeight - el.scrollTop - el.clientHeight < 80 : false
+          agentStreamSessionRef.current ??= beginAgentStream(editor)
+          const session = agentStreamSessionRef.current
           // Defensive: a ready collab editor always has a ySync binding, so this applies; if one is
           // somehow absent, bail this frame without advancing rather than looping.
-          if (!applyStreamedMarkdownToLiveDoc(editor, pending)) {
+          if (!session || !applyAgentStreamFrame(editor, session, pending)) {
             streamRafRef.current = null
             return
           }
@@ -892,13 +919,15 @@ export function LoadedRichMarkdownEditor({
       if (wasStreamingRef.current && collabReady) {
         wasStreamingRef.current = false
         const finalBody = splitFrontmatter(content).body
-        if (finalBody !== lastSyncedBodyRef.current) {
-          runOffRender(() => {
-            if (applyStreamedMarkdownToLiveDoc(editor, finalBody)) {
+        const session = agentStreamSessionRef.current
+        agentStreamSessionRef.current = null
+        runOffRender(() => {
+          if (session && finalBody !== lastSyncedBodyRef.current) {
+            if (applyAgentStreamFrame(editor, session, finalBody))
               lastSyncedBodyRef.current = finalBody
-            }
-          })
-        }
+          }
+          if (session) endAgentStream(session)
+        })
       }
       return
     }
@@ -1002,6 +1031,10 @@ export function LoadedRichMarkdownEditor({
   useEffect(
     () => () => {
       if (streamRafRef.current !== null) cancelAnimationFrame(streamRafRef.current)
+      if (agentStreamSessionRef.current) {
+        endAgentStream(agentStreamSessionRef.current)
+        agentStreamSessionRef.current = null
+      }
     },
     []
   )