diff --git a/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/apply-streamed-markdown.concurrent.test.ts b/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/apply-streamed-markdown.concurrent.test.ts
new file mode 100644
index 00000000000..19b02ea7182
--- /dev/null
+++ b/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/apply-streamed-markdown.concurrent.test.ts
@@ -0,0 +1,214 @@
+/**
+ * @vitest-environment jsdom
+ *
+ * Two-writer evaluation: does a PEER editing the shared doc WHILE the agent streams cause corruption,
+ * clobbering, duplication, or stray empty paragraphs? The agent applies via the real
+ * `beginAgentStream`/`applyAgentStreamFrame` path (a shadow doc diffed with `updateYFragment`, seeded
+ * once and never shown the peer's edits). A second editor is wired as a genuine Yjs peer (bidirectional
+ * update forwarding), so this reproduces the production two-client scenario, not a mock.
+ *
+ * Convergence is a hard invariant everywhere (CRDT MUST converge). Peer-edit survival is hard-asserted
+ * only for the NON-overlapping case (an agent that appends must not clobber an unrelated peer edit); for
+ * the overlapping case it is diagnostic (CRDT last-writer semantics are acceptable there), so those are
+ * logged for judgement. Run: bunx vitest run <thisfile> --disable-console-intercept
+ */
+import { Editor } from '@tiptap/core'
+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 { parseMarkdownToDoc } from '../markdown-parse'
+import { applyAgentStreamFrame, beginAgentStream, endAgentStream } from './apply-streamed-markdown'
+
+beforeAll(() => {
+  if (!document.elementFromPoint) document.elementFromPoint = () => null
+})
+
+function makeCollabEditor() {
+  const doc = new Y.Doc()
+  const awareness = new Awareness(doc)
+  const editor = new Editor({
+    extensions: createMarkdownEditorExtensions({
+      placeholder: '',
+      collaboration: { doc, awareness, user: { name: 'U', color: '#fff', clientId: doc.clientID } },
+    }),
+    content: '',
+  })
+  return { editor, doc, awareness }
+}
+
+const teardown: Array<() => void> = []
+afterEach(() => {
+  for (const fn of teardown.splice(0)) fn()
+})
+function track(t: { editor: Editor; doc: Y.Doc; awareness: Awareness }) {
+  teardown.push(() => {
+    t.editor.destroy()
+    t.awareness.destroy()
+    t.doc.destroy()
+  })
+  return t
+}
+
+/** Wire two Y.Docs as real peers: forward each update to the other, origin-guarded to avoid echo. */
+function wirePeers(a: Y.Doc, b: Y.Doc) {
+  const A2B = Symbol('a->b')
+  const B2A = Symbol('b->a')
+  a.on('update', (u: Uint8Array, origin: unknown) => {
+    if (origin !== B2A) Y.applyUpdate(b, u, A2B)
+  })
+  b.on('update', (u: Uint8Array, origin: unknown) => {
+    if (origin !== A2B) Y.applyUpdate(a, u, B2A)
+  })
+}
+
+/** Seed editor A with markdown (through the real parse), then bring up B as a synced peer. */
+function seededPair(markdown: string) {
+  const A = track(makeCollabEditor())
+  A.editor.commands.setContent(parseMarkdownToDoc(markdown), { contentType: 'json' })
+  const B = track(makeCollabEditor())
+  Y.applyUpdate(B.doc, Y.encodeStateAsUpdate(A.doc))
+  wirePeers(A.doc, B.doc)
+  return { A, B }
+}
+
+/** A peer edit: insert `text` at the start of the first text node containing `needle`. */
+function peerInsertNear(editor: Editor, needle: string, text: string): boolean {
+  let pos: number | null = null
+  editor.state.doc.descendants((node, p) => {
+    if (pos !== null) return false
+    if (node.isText && node.text?.includes(needle)) pos = p + node.text.indexOf(needle)
+  })
+  if (pos === null) return false
+  return editor.commands.insertContentAt(pos, text)
+}
+
+function fragStr(doc: Y.Doc): string {
+  return doc.getXmlFragment('default').toString()
+}
+function count(hay: string, needle: string): number {
+  return hay.split(needle).length - 1
+}
+function emptyParas(editor: Editor): number {
+  let n = 0
+  editor.state.doc.descendants((node) => {
+    if (node.type.name === 'paragraph' && node.childCount === 0) n++
+  })
+  return n
+}
+
+describe('two-writer: peer edits while the agent streams', () => {
+  it('SANITY: peers converge on seed and a plain peer edit with no agent activity', () => {
+    const { A, B } = seededPair('# Title\n\nAlpha\n\nBeta')
+    expect(peerInsertNear(B.editor, 'Alpha', 'PEER ')).toBe(true)
+    expect(fragStr(A.doc)).toBe(fragStr(B.doc))
+    expect(A.editor.state.doc.textContent).toContain('PEER Alpha')
+  })
+
+  it('NON-OVERLAPPING: agent appends at the bottom while the peer edits the top — peer edit MUST survive', () => {
+    const { A, B } = seededPair('# Title\n\nAlpha\n\nBeta')
+    const session = beginAgentStream(A.editor)!
+
+    // Frame 1: agent appends Gamma (region far from the peer's target).
+    applyAgentStreamFrame(A.editor, session, '# Title\n\nAlpha\n\nBeta\n\nGamma')
+    // Peer edits the TOP paragraph mid-stream (the agent never touches or knows about this).
+    expect(peerInsertNear(B.editor, 'Alpha', 'PEER ')).toBe(true)
+    // Frames 2-3: agent keeps appending. Its bodies say "Alpha" (no PEER) — the test is whether the
+    // (aggressive) updateYFragment re-emits/clobbers the unchanged Alpha paragraph.
+    applyAgentStreamFrame(A.editor, session, '# Title\n\nAlpha\n\nBeta\n\nGamma\n\nDelta')
+    applyAgentStreamFrame(
+      A.editor,
+      session,
+      '# Title\n\nAlpha\n\nBeta\n\nGamma\n\nDelta\n\nEpsilon'
+    )
+    endAgentStream(session)
+
+    const textA = A.editor.state.doc.textContent
+    console.log(`\n[NON-OVERLAP] A: ${JSON.stringify(textA)}`)
+    console.log(
+      `[NON-OVERLAP] converged=${fragStr(A.doc) === fragStr(B.doc)} peerCount=${count(textA, 'PEER ')} emptyParas=${emptyParas(A.editor)}`
+    )
+
+    expect(fragStr(A.doc)).toBe(fragStr(B.doc)) // CRDT convergence
+    expect(count(textA, 'PEER ')).toBe(1) // peer edit survives, exactly once (no clobber, no dup)
+    expect(textA).toContain('Epsilon') // agent's stream landed
+    expect(textA).toContain('Beta') // untouched content intact
+    expect(emptyParas(A.editor)).toBe(0) // no stray empties from the merge
+  })
+
+  it('POSITION DRIFT: agent inserts a paragraph ABOVE while the peer edits the paragraph BELOW', () => {
+    // The exact scenario relative-position anchoring is meant to protect: the agent shifts positions by
+    // inserting content above the region the peer is editing. Without anchoring, an offset-based writer
+    // would misplace the edit; a whole-doc CRDT diff should not.
+    const { A, B } = seededPair('# Title\n\nAlpha\n\nBeta')
+    const session = beginAgentStream(A.editor)!
+
+    applyAgentStreamFrame(A.editor, session, '# Title\n\nAlpha\n\nMIDDLE\n\nBeta')
+    // Peer edits Beta, which just shifted down by the agent's inserted MIDDLE paragraph.
+    expect(peerInsertNear(B.editor, 'Beta', 'PEER ')).toBe(true)
+    applyAgentStreamFrame(A.editor, session, '# Title\n\nAlpha\n\nMIDDLE\n\nMIDDLE2\n\nBeta')
+    endAgentStream(session)
+
+    const textA = A.editor.state.doc.textContent
+    console.log(`\n[POS-DRIFT] A: ${JSON.stringify(textA)}`)
+    console.log(
+      `[POS-DRIFT] converged=${fragStr(A.doc) === fragStr(B.doc)} peerCount=${count(textA, 'PEER ')} peerOnBeta=${textA.includes('PEER Beta')} emptyParas=${emptyParas(A.editor)}`
+    )
+
+    expect(fragStr(A.doc)).toBe(fragStr(B.doc)) // convergence
+    expect(count(textA, 'PEER ')).toBe(1) // no duplication
+    expect(textA).toContain('PEER Beta') // peer edit stayed attached to Beta despite the insert above
+    expect(textA).toContain('MIDDLE2') // agent's inserts landed
+    expect(emptyParas(A.editor)).toBe(0)
+  })
+
+  it('OVERLAPPING: agent rewrites the exact paragraph the peer is editing (diagnostic + must converge)', () => {
+    const { A, B } = seededPair('# Title\n\noriginal body text')
+    const session = beginAgentStream(A.editor)!
+
+    applyAgentStreamFrame(A.editor, session, '# Title\n\noriginal body text extended')
+    // Peer edits the SAME paragraph the agent is rewriting.
+    expect(peerInsertNear(B.editor, 'original', 'PEER ')).toBe(true)
+    applyAgentStreamFrame(A.editor, session, '# Title\n\nagent fully rewrote this paragraph')
+    endAgentStream(session)
+
+    const textA = A.editor.state.doc.textContent
+    console.log(`\n[OVERLAP] A: ${JSON.stringify(textA)}`)
+    console.log(
+      `[OVERLAP] converged=${fragStr(A.doc) === fragStr(B.doc)} peerSurvived=${textA.includes('PEER')} emptyParas=${emptyParas(A.editor)}`
+    )
+
+    expect(fragStr(A.doc)).toBe(fragStr(B.doc)) // convergence is non-negotiable even in conflict
+    expect(emptyParas(A.editor)).toBe(0) // conflict must not leave stray empty paragraphs
+    // peer survival here is CRDT-dependent — reported above, not hard-asserted.
+  })
+
+  it('FULL REWRITE: peer edits original content that the agent then deletes in a full rewrite', () => {
+    const { A, B } = seededPair('# Title\n\nAlpha\n\nBeta\n\nGamma')
+    const session = beginAgentStream(A.editor)!
+
+    // Peer edits Beta WHILE it still exists — genuinely concurrent with the impending rewrite.
+    // (Asserting the insert landed guards against a false-green where the target was already gone.)
+    expect(peerInsertNear(B.editor, 'Beta', 'PEER ')).toBe(true)
+    // Agent replaces the WHOLE doc across two frames, deleting Alpha/Beta/Gamma.
+    applyAgentStreamFrame(A.editor, session, '# Report\n\nOne\n\nTwo')
+    applyAgentStreamFrame(A.editor, session, '# Report\n\nOne\n\nTwo\n\nThree')
+    endAgentStream(session)
+
+    const textA = A.editor.state.doc.textContent
+    console.log(`\n[FULL-REWRITE] A: ${JSON.stringify(textA)}`)
+    console.log(
+      `[FULL-REWRITE] converged=${fragStr(A.doc) === fragStr(B.doc)} peerCount=${count(textA, 'PEER ')} oneCount=${count(textA, 'One')} threeCount=${count(textA, 'Three')} emptyParas=${emptyParas(A.editor)}`
+    )
+
+    expect(fragStr(A.doc)).toBe(fragStr(B.doc)) // convergence
+    expect(count(textA, 'One')).toBe(1) // agent content not duplicated by the concurrent merge
+    expect(count(textA, 'Three')).toBe(1)
+    expect(emptyParas(A.editor)).toBe(0) // no stray empties from a delete/insert conflict
+    // The peer's insert is NOT lost when the rewrite deletes its surrounding paragraph: Yjs preserves
+    // the inserted text and reattaches it to the nearest surviving anchor (it relocates into the
+    // rewritten content rather than vanishing). What matters is that it survives exactly once — never
+    // duplicated, never silently dropped.
+    expect(count(textA, 'PEER ')).toBe(1)
+  })
+})
diff --git a/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/collab-streaming-integration.test.ts b/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/collab-streaming-integration.test.ts
new file mode 100644
index 00000000000..b8996800923
--- /dev/null
+++ b/apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/collab-streaming-integration.test.ts
@@ -0,0 +1,200 @@
+/**
+ * @vitest-environment jsdom
+ *
+ * Integration coverage for the collaborative agent-streaming surface with all the moving pieces:
+ * multiple peers, undo isolation, the durable persist→reopen round-trip, empty-collapse on the live
+ * streaming path, and a late joiner. Editors are wired as genuine Yjs peers (mesh update forwarding).
+ * This exercises the CRDT/merge/convert LOGIC deterministically; it does NOT cover the realtime socket
+ * transport, RAF-paced stream loop, or real browser timing (those need a live 2-browser E2E harness).
+ * Run: bunx vitest run <thisfile> --disable-console-intercept
+ */
+import { Editor } from '@tiptap/core'
+import { afterEach, beforeAll, describe, expect, it } from 'vitest'
+import { Awareness } from 'y-protocols/awareness'
+import * as Y from 'yjs'
+import { markdownToYDoc, yDocToMarkdown } from '@/lib/collab-doc/converter'
+import { createMarkdownEditorExtensions } from '../editor-extensions'
+import { parseMarkdownToDoc } from '../markdown-parse'
+import { applyAgentStreamFrame, beginAgentStream, endAgentStream } from './apply-streamed-markdown'
+
+beforeAll(() => {
+  if (!document.elementFromPoint) document.elementFromPoint = () => null
+})
+
+const teardown: Array<() => void> = []
+afterEach(() => {
+  for (const fn of teardown.splice(0)) fn()
+})
+
+function makeCollabEditor() {
+  const doc = new Y.Doc()
+  const awareness = new Awareness(doc)
+  const editor = new Editor({
+    extensions: createMarkdownEditorExtensions({
+      placeholder: '',
+      collaboration: { doc, awareness, user: { name: 'U', color: '#fff', clientId: doc.clientID } },
+    }),
+    content: '',
+  })
+  const t = { editor, doc, awareness }
+  teardown.push(() => {
+    editor.destroy()
+    awareness.destroy()
+    doc.destroy()
+  })
+  return t
+}
+
+/** Forward every local/agent update from each doc to all others (origin-guarded), a full mesh. */
+function wireMesh(docs: Y.Doc[]) {
+  const MESH = Symbol('mesh')
+  for (const d of docs) {
+    d.on('update', (u: Uint8Array, origin: unknown) => {
+      if (origin === MESH) return
+      for (const other of docs) if (other !== d) Y.applyUpdate(other, u, MESH)
+    })
+  }
+}
+
+function peerInsertNear(editor: Editor, needle: string, text: string): boolean {
+  let pos: number | null = null
+  editor.state.doc.descendants((node, p) => {
+    if (pos !== null) return false
+    if (node.isText && node.text?.includes(needle)) pos = p + node.text.indexOf(needle)
+  })
+  if (pos === null) return false
+  return editor.commands.insertContentAt(pos, text)
+}
+
+const fragStr = (doc: Y.Doc) => doc.getXmlFragment('default').toString()
+const countText = (hay: string, needle: string) => hay.split(needle).length - 1
+function emptyParas(editor: Editor): number {
+  let n = 0
+  editor.state.doc.descendants((node) => {
+    if (node.type.name === 'paragraph' && node.childCount === 0) n++
+  })
+  return n
+}
+
+describe('collab streaming integration — moving pieces', () => {
+  it('THREE-WAY: agent + two peers editing different regions all converge, both peer edits survive', () => {
+    const A = makeCollabEditor()
+    A.editor.commands.setContent(parseMarkdownToDoc('# Title\n\nAlpha\n\nBeta\n\nGamma'), {
+      contentType: 'json',
+    })
+    const B = makeCollabEditor()
+    const C = makeCollabEditor()
+    Y.applyUpdate(B.doc, Y.encodeStateAsUpdate(A.doc))
+    Y.applyUpdate(C.doc, Y.encodeStateAsUpdate(A.doc))
+    wireMesh([A.doc, B.doc, C.doc])
+
+    const session = beginAgentStream(A.editor)!
+    applyAgentStreamFrame(A.editor, session, '# Title\n\nAlpha\n\nBeta\n\nGamma\n\nDelta')
+    expect(peerInsertNear(B.editor, 'Alpha', 'B_EDIT ')).toBe(true) // peer B edits the top
+    expect(peerInsertNear(C.editor, 'Gamma', 'C_EDIT ')).toBe(true) // peer C edits the bottom
+    applyAgentStreamFrame(
+      A.editor,
+      session,
+      '# Title\n\nAlpha\n\nBeta\n\nGamma\n\nDelta\n\nEpsilon'
+    )
+    endAgentStream(session)
+
+    const textA = A.editor.state.doc.textContent
+    console.log(`\n[3-WAY] A: ${JSON.stringify(textA)}`)
+    console.log(
+      `[3-WAY] converged=${fragStr(A.doc) === fragStr(B.doc) && fragStr(B.doc) === fragStr(C.doc)} B_EDIT=${countText(textA, 'B_EDIT ')} C_EDIT=${countText(textA, 'C_EDIT ')} empty=${emptyParas(A.editor)}`
+    )
+
+    expect(fragStr(A.doc)).toBe(fragStr(B.doc))
+    expect(fragStr(B.doc)).toBe(fragStr(C.doc))
+    expect(countText(textA, 'B_EDIT ')).toBe(1)
+    expect(countText(textA, 'C_EDIT ')).toBe(1)
+    expect(textA).toContain('Epsilon')
+    expect(emptyParas(A.editor)).toBe(0)
+  })
+
+  it('UNDO ISOLATION: a peer undo reverts only the peer’s own edit, never the agent’s stream', () => {
+    const A = makeCollabEditor()
+    A.editor.commands.setContent(parseMarkdownToDoc('# Title\n\nbase'), { contentType: 'json' })
+    const B = makeCollabEditor()
+    Y.applyUpdate(B.doc, Y.encodeStateAsUpdate(A.doc))
+    wireMesh([A.doc, B.doc])
+
+    expect(peerInsertNear(B.editor, 'base', 'PEER_UNDOABLE ')).toBe(true) // peer's own edit (undo stack)
+    const session = beginAgentStream(A.editor)!
+    applyAgentStreamFrame(
+      A.editor,
+      session,
+      '# Title\n\nPEER_UNDOABLE base\n\nagent added this line'
+    )
+    endAgentStream(session)
+
+    const undid = B.editor.commands.undo()
+    const textB = B.editor.state.doc.textContent
+    console.log(`\n[UNDO] undoRan=${undid} afterUndo=${JSON.stringify(textB)}`)
+
+    expect(fragStr(A.doc)).toBe(fragStr(B.doc)) // still converged after undo
+    expect(textB).toContain('agent added this line') // agent content NOT undone by the peer
+    expect(textB).not.toContain('PEER_UNDOABLE') // peer's own edit was undone
+  })
+
+  it('PERSIST ROUND-TRIP: stream → serialize to durable markdown → reopen yields the same content, no empties', () => {
+    const A = makeCollabEditor()
+    const session = beginAgentStream(A.editor)!
+    applyAgentStreamFrame(A.editor, session, '# Report\n\n## Section 1\n\nbody one')
+    applyAgentStreamFrame(
+      A.editor,
+      session,
+      '# Report\n\n## Section 1\n\nbody one\n\n## Section 2\n\nbody two'
+    )
+    endAgentStream(session)
+
+    const durable = yDocToMarkdown(A.doc) // server-side projection to durable markdown
+    const reopened = markdownToYDoc(durable) // cold reopen from durable
+    const reopenedMd = yDocToMarkdown(reopened)
+    const blankRuns = (durable.match(/\n{3,}/g) ?? []).length
+    console.log(`\n[ROUND-TRIP] durable=${JSON.stringify(durable)}`)
+    console.log(`[ROUND-TRIP] reopenStable=${reopenedMd === durable} blankRuns=${blankRuns}`)
+
+    expect(durable).toContain('Section 1')
+    expect(durable).toContain('Section 2')
+    expect(durable).toContain('body two')
+    expect(blankRuns).toBe(0) // no pathological blank runs in the persisted markdown
+    expect(reopenedMd).toBe(durable) // reopen is a fixed point (stable)
+    reopened.destroy()
+  })
+
+  it('EMPTY-COLLAPSE ON THE STREAM PATH: an agent body with a huge blank run does not strand empties', () => {
+    const A = makeCollabEditor()
+    A.editor.commands.setContent(parseMarkdownToDoc('# Title\n\nintro'), { contentType: 'json' })
+    const session = beginAgentStream(A.editor)!
+    // The agent emits a pathological blank run between two blocks (the original incident's shape).
+    applyAgentStreamFrame(A.editor, session, `# Title\n\nintro${'\n'.repeat(400)}tail paragraph`)
+    endAgentStream(session)
+
+    console.log(
+      `\n[STREAM-COLLAPSE] text=${JSON.stringify(A.editor.state.doc.textContent)} empty=${emptyParas(A.editor)}`
+    )
+    expect(emptyParas(A.editor)).toBe(0) // collapse protects the live streaming path, not just static open
+    expect(A.editor.state.doc.textContent).toContain('tail paragraph')
+  })
+
+  it('LATE JOINER: a peer that syncs AFTER the stream sees the full, clean document', () => {
+    const A = makeCollabEditor()
+    A.editor.commands.setContent(parseMarkdownToDoc('# Doc\n\nstart'), { contentType: 'json' })
+    const session = beginAgentStream(A.editor)!
+    applyAgentStreamFrame(A.editor, session, '# Doc\n\nstart\n\nstreamed body')
+    endAgentStream(session)
+
+    // A brand-new client joins now and syncs from the current state.
+    const D = makeCollabEditor()
+    Y.applyUpdate(D.doc, Y.encodeStateAsUpdate(A.doc))
+
+    console.log(
+      `\n[LATE-JOIN] D: ${JSON.stringify(D.editor.state.doc.textContent)} converged=${fragStr(A.doc) === fragStr(D.doc)}`
+    )
+    expect(fragStr(A.doc)).toBe(fragStr(D.doc))
+    expect(D.editor.state.doc.textContent).toContain('streamed body')
+    expect(emptyParas(D.editor)).toBe(0)
+  })
+})