fix(files): fix new-file collab streaming latch and agent-edit duplication

- Latch collab readiness so a new file's post-seed `synced` flap can no longer
  re-gate agent streaming (the stream previously showed only the seed and the
  rest appeared only on reload)
- Relay defers the durable edit_content merge to an actively-streaming client:
  the client shadow stream and the server merge were both writing the same
  content into the live doc, duplicating it when the server ran ahead
- Render the collaborator caret bar out of flow so a peer caret never nudges
  the surrounding text by ~1px
- Remove dead code: unused FileDocMessageType alias, unnecessary
  LiveFileDocMergeOrder export

Covered by tests: readiness latch (flap/offline/latch cases), relay merge
deferral (single- and multi-replica), plus verified-failing guards.

Author: waleedlatif1

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

@@ -107,6 +107,11 @@ const SEED_LOCK_PREFIX = 'filedoc:seedlock:'
 const COMPACT_LOCK_PREFIX = 'filedoc:compactlock:'
 const PERSIST_LOCK_PREFIX = 'filedoc:persistlock:'
 const MERGE_LOCK_PREFIX = 'filedoc:mergelock:'
+/** Cluster-wide "a client is actively streaming an agent edit into this live doc" flag — set (refreshed)
+ *  on every agent frame so a durable {@link applyMarkdownToLiveFileDoc} merge defers to that client
+ *  (which is applying the same content) instead of double-writing it. Short-TTL'd so it self-clears the
+ *  moment streaming stops, after which the final durable merge lands as a near-noop. */
+const AGENT_STREAM_PREFIX = 'filedoc:agentstream:'
 
 /** The field each stream entry carries — a base64 Yjs update. */
 const UPDATE_FIELD = 'u'
@@ -555,6 +560,34 @@ export class FileDocStore {
     }
   }
 
+  /** Mark (or refresh) that a client is actively streaming an agent edit into this live doc — a plain
+   *  `SET key "1" PX ttl`, so it self-clears when streaming stops. Best-effort; no-op when disabled. */
+  async markAgentStreaming(name: string, ttlMs: number): Promise<void> {
+    if (!this.enabled || !this.write) return
+    try {
+      await this.write.set(`${AGENT_STREAM_PREFIX}${name}`, '1', { PX: ttlMs })
+    } catch (error) {
+      logger.warn(`FileDocStore markAgentStreaming failed for ${name}`, {
+        error: getErrorMessage(error),
+      })
+    }
+  }
+
+  /** Whether a client is currently streaming an agent edit into this live doc (see
+   *  {@link markAgentStreaming}). Best-effort; treats an error/disabled store as "not streaming" so a
+   *  merge never blocks on this check. */
+  async isAgentStreaming(name: string): Promise<boolean> {
+    if (!this.enabled || !this.write) return false
+    try {
+      return (await this.write.exists(`${AGENT_STREAM_PREFIX}${name}`)) === 1
+    } catch (error) {
+      logger.warn(`FileDocStore isAgentStreaming failed for ${name}`, {
+        error: getErrorMessage(error),
+      })
+      return false
+    }
+  }
+
   async releaseMergeSlot(name: string, token: string): Promise<void> {
     await this.releaseLock(`${MERGE_LOCK_PREFIX}${name}`, token)
   }

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

@@ -31,6 +31,8 @@ const fakeStore = {
   setSyncedVersion: vi.fn(async (name: string, version: number) => {
     fakeStore.versions.set(name, Math.max(fakeStore.versions.get(name) ?? 0, version))
   }),
+  markAgentStreaming: vi.fn(async () => {}),
+  isAgentStreaming: vi.fn(async () => false),
 }
 
 vi.mock('@sim/platform-authz/rooms', () => ({ authorizeRoom: vi.fn() }))
@@ -83,4 +85,26 @@ describe('applyMarkdownToLiveFileDoc — multi-replica (store-enabled) ordering'
     // setSyncedVersion fired only for the two applied durable writes, never for the stale one.
     expect(fakeStore.setSyncedVersion).toHaveBeenCalledTimes(2)
   })
+
+  it('defers the content merge cluster-wide when a client is streaming (records version, no diff)', async () => {
+    // The cluster-wide counterpart of the single-replica deferral: while the shared `isAgentStreaming`
+    // flag is set (a client on ANY replica is streaming this agent edit), the durable merge must record
+    // the version but skip the content diff — the streaming client owns the bytes, so a whole-document
+    // merge here would double-write them.
+    fakeStore.isAgentStreaming.mockResolvedValue(true)
+
+    expect(
+      await applyMarkdownToLiveFileDoc('file-1', '# streamed by a client', { version: 100 })
+    ).toBe('applied')
+    expect(mockFetchFileDocMerge).not.toHaveBeenCalled() // content deferred to the client
+    expect(fakeStore.publishAndWait).not.toHaveBeenCalled()
+    expect(fakeStore.setSyncedVersion).toHaveBeenCalledWith(ROOM_NAME, 100) // version still recorded
+
+    // Once streaming stops the flag clears and the (now near-noop) durable merge resumes normally.
+    fakeStore.isAgentStreaming.mockResolvedValue(false)
+    expect(await applyMarkdownToLiveFileDoc('file-1', '# final durable', { version: 150 })).toBe(
+      'applied'
+    )
+    expect(mockFetchFileDocMerge).toHaveBeenCalledTimes(1)
+  })
 })

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

@@ -568,6 +568,39 @@ describe('setupWorkspaceFileDocHandlers', () => {
     )
   })
 
+  it('defers the content merge to an actively-streaming client (records version, no double-write)', async () => {
+    // Two-writer duplication guard: while a client is streaming an agent edit into the shared doc
+    // (agent frames flowing), a durable apply-edit merge must NOT also publish the same content — the
+    // client's private shadow never observes this merge and would re-insert it. The merge still records
+    // the durable version; once streaming stops it resumes and lands as a near-noop.
+    mockFetchFileDocSeed.mockResolvedValue(seedResult('# Original'))
+    const { io } = createIo()
+    const { handlers } = setup('socket-1', io)
+    await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 })
+    await flushMicrotasks()
+
+    // A client streams an agent frame → the room is now "actively streaming".
+    const edit = new Y.Doc()
+    edit.getText(FILE_DOC_FIELD).insert(0, 'agent streaming this live')
+    handlers[FILE_DOC_EVENTS.MESSAGE](
+      frame(FILE_DOC_MESSAGE_TYPE.SYNC_NO_PERSIST, (e) =>
+        syncProtocol.writeUpdate(e, Y.encodeStateAsUpdate(edit))
+      )
+    )
+    await flushMicrotasks()
+
+    mockFetchFileDocMerge.mockResolvedValue(Y.encodeStateAsUpdate(new Y.Doc()))
+    // The durable merge lands mid-stream: it must defer (record version) and skip the content diff.
+    const result = await applyMarkdownToLiveFileDoc('file-1', '# Rewritten by copilot', {
+      version: 100,
+    })
+    expect(result).toBe('applied')
+    expect(mockFetchFileDocMerge).not.toHaveBeenCalled() // content deferred to the client
+
+    // The recorded version is honored: a later stale merge is still rejected on it.
+    expect(await applyMarkdownToLiveFileDoc('file-1', '# older', { version: 50 })).toBe('stale')
+  })
+
   it('reports no-live-room (and does not call the app) when the file has no seeded room', async () => {
     const result = await applyMarkdownToLiveFileDoc('file-1', '# anything')
     expect(result).toBe('no-live-room')

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

@@ -90,6 +90,15 @@ const MERGE_LOCK_RETRIES = Math.ceil(
   (MERGE_LOCK_TTL_MS + FILE_DOC_TIMEOUTS.mergeRequestMs) / MERGE_LOCK_RETRY_MS
 )
 
+/**
+ * How long after the last agent frame a client is still treated as "actively streaming" (so a durable
+ * merge defers to it). Comfortably longer than the gap between agent frames (per-rAF to the client's
+ * reparse throttle) so brief throttle pauses don't flip it off mid-stream, yet short enough that the
+ * final durable merge — which lands as a near-noop once the client has streamed everything — isn't
+ * deferred for long after the stream ends.
+ */
+const AGENT_STREAM_FLAG_TTL_MS = 10_000
+
 /** One presence ownership within a room: a (socket, clientID) pair. */
 interface FileDocOwner {
   /**
@@ -149,6 +158,15 @@ interface FileDocRoom {
    * edit the live doc hasn't incorporated. `null` before the first seed (persist writes unconditionally).
    */
   syncedVersion: number | null
+  /**
+   * Epoch ms until which a client is treated as actively streaming an agent edit into this doc —
+   * refreshed on every agent frame ({@link AGENT_SYNC_ORIGIN}/{@link REDIS_AGENT_ORIGIN}). While it is in
+   * the future, a durable {@link applyMarkdownToLiveFileDoc} merge DEFERS its content diff to that client
+   * (which applies the same content into the shared doc), preventing the two-writer duplication; the merge
+   * still records the durable version. The single-replica counterpart of the cluster-wide
+   * {@link FileDocStore.isAgentStreaming} flag. `0` when no agent stream is active.
+   */
+  agentStreamingUntil: number
 }
 
 /** Live documents keyed by Socket.IO room name. Module-global: one Y.Doc per file. */
@@ -645,6 +663,15 @@ async function mergeMarkdownIntoRoom(
       const shared = await store.getSyncedVersion(name)
       const current = Math.max(shared ?? 0, fileDocRooms.get(name)?.syncedVersion ?? 0)
       if (isStale(current)) return 'stale'
+      // Defer to an actively-streaming client: it is applying this SAME agent edit into the shared doc
+      // frame-by-frame, so also publishing a whole-document merge here would double-write the content (the
+      // client's private shadow never observes this merge, so it re-inserts what we added → duplication).
+      // Still record the durable version so the persist If-Match stays correct; the client owns the bytes,
+      // and once streaming stops the flag clears and the final durable merge lands as a near-noop.
+      if (await store.isAgentStreaming(name)) {
+        await recordVersion()
+        return 'applied'
+      }
       // 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
@@ -665,6 +692,12 @@ async function mergeMarkdownIntoRoom(
   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'
+  // Defer to an actively-streaming client (see the multi-replica branch above) — it applies this agent
+  // edit itself, so merging it here too would double-write. Record the version; skip the content merge.
+  if (room.agentStreamingUntil > Date.now()) {
+    await recordVersion()
+    return 'applied'
+  }
   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'
@@ -702,6 +735,7 @@ function getOrCreateRoom(io: Server, ref: RoomRef): FileDocRoom {
     persistTimer: null,
     persistDeadline: null,
     syncedVersion: null,
+    agentStreamingUntil: 0,
   }
   // Register synchronously BEFORE the async catch-up so a concurrent join sees this room, not a second.
   fileDocRooms.set(name, room)
@@ -735,6 +769,17 @@ function getOrCreateRoom(io: Server, ref: RoomRef): FileDocRoom {
       origin !== SEED_ORIGIN
     )
       getFileDocStore().publish(name, update, origin === AGENT_SYNC_ORIGIN)
+    // A locally-originated agent frame (this task's stream leader) means a client is applying this agent
+    // edit itself. Refresh the "actively streaming" markers so a durable `applyMarkdownToLiveFileDoc`
+    // merge defers to that client instead of double-writing the same content (the two-writer
+    // duplication). Both are TTL'd, so they self-clear once frames stop and the final durable merge then
+    // lands as a near-noop. Only AGENT_SYNC_ORIGIN is handled: the in-memory marker is read solely by the
+    // single-replica merge branch (where every agent frame is AGENT_SYNC_ORIGIN), and the cluster flag is
+    // read cluster-wide, so a peer task tailing REDIS_AGENT_ORIGIN never needs to set either.
+    if (origin === AGENT_SYNC_ORIGIN) {
+      room.agentStreamingUntil = Date.now() + AGENT_STREAM_FLAG_TTL_MS
+      void getFileDocStore().markAgentStreaming(name, AGENT_STREAM_FLAG_TTL_MS)
+    }
     // Edit tracking for persistence. Mark the doc dirty on any update applied AFTER it was seeded — a
     // local user edit (socket origin) OR a peer's edit relayed via the tailer (REDIS_ORIGIN) — so
     // whichever task is last to leave persists real edits, even one that only tailed them. A compaction

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

@@ -84,6 +84,13 @@ export function renderCaret(user: Record<string, unknown>): HTMLElement {
   // One inline var drives the caret bar, the dormant cap, and the name tag (all in CSS).
   caret.style.setProperty('--caret-color', color)
   if (clientId !== undefined) caret.dataset.caretClientId = String(clientId)
+  // The visible caret bar is a SEPARATE, absolutely-positioned child — never an inline border on the
+  // caret span. The caret is a ProseMirror inline widget inserted between characters; an in-flow bar
+  // (border + width) reflows the surrounding text by ~1px each time a peer's caret appears or moves.
+  // An out-of-flow bar has zero layout footprint, so peer carets never nudge the document.
+  const bar = document.createElement('span')
+  bar.className = 'collaboration-carets__bar'
+  caret.appendChild(bar)
   const label = document.createElement('div')
   label.className = 'collaboration-carets__label'
   label.textContent = name

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

@@ -0,0 +1,77 @@
+/**
+ * @vitest-environment node
+ */
+import { describe, expect, it } from 'vitest'
+import { type CollabReadinessInputs, nextCollabReadiness } from './readiness'
+
+/** Drive a sequence of observations through the latch, returning the readiness at each step. */
+function run(steps: CollabReadinessInputs[]): boolean[] {
+  let syncedOnce = false
+  return steps.map((input) => {
+    const next = nextCollabReadiness(syncedOnce, input)
+    syncedOnce = next.syncedOnce
+    return next.ready
+  })
+}
+
+describe('nextCollabReadiness', () => {
+  it('is not ready before syncing or seeding', () => {
+    const { syncedOnce, ready } = nextCollabReadiness(false, {
+      synced: false,
+      seeded: false,
+      offlineSeed: false,
+    })
+    expect(syncedOnce).toBe(false)
+    expect(ready).toBe(false)
+  })
+
+  it('is not ready when synced but not yet seeded', () => {
+    const { syncedOnce, ready } = nextCollabReadiness(false, {
+      synced: true,
+      seeded: false,
+      offlineSeed: false,
+    })
+    expect(syncedOnce).toBe(true) // latched
+    expect(ready).toBe(false) // waits for the seed
+  })
+
+  it('opens on the new-file flap sequence: synced true, then seed lands while synced flapped false', () => {
+    // The exact bug: `synced` and `seeded` are never true in the same observation. The latch must still
+    // open once BOTH have been seen across observations.
+    const readiness = run([
+      { synced: false, seeded: false, offlineSeed: false }, // joining
+      { synced: true, seeded: false, offlineSeed: false }, // initial (empty) sync
+      { synced: false, seeded: false, offlineSeed: false }, // synced flaps false on re-sync
+      { synced: false, seeded: true, offlineSeed: false }, // server seed lands (synced still false)
+    ])
+    expect(readiness).toEqual([false, false, false, true])
+  })
+
+  it('opens even if the seed lands before we ever observed synced (server seed proves a sync)', () => {
+    // If the flap beat our first observation, the seed flag alone (not the offline fallback) proves a
+    // completed sync happened.
+    const { syncedOnce, ready } = nextCollabReadiness(false, {
+      synced: false,
+      seeded: true,
+      offlineSeed: false,
+    })
+    expect(syncedOnce).toBe(true)
+    expect(ready).toBe(true)
+  })
+
+  it('stays read-only for an offline (local) seed that never reached the server', () => {
+    const readiness = run([
+      { synced: false, seeded: false, offlineSeed: false },
+      { synced: false, seeded: true, offlineSeed: true }, // offline fallback seeded locally
+    ])
+    expect(readiness).toEqual([false, false])
+  })
+
+  it('never reverts once ready, even if synced later flaps false', () => {
+    const readiness = run([
+      { synced: true, seeded: true, offlineSeed: false }, // ready
+      { synced: false, seeded: true, offlineSeed: false }, // synced flaps — must stay ready
+    ])
+    expect(readiness).toEqual([true, true])
+  })
+})

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

@@ -0,0 +1,37 @@
+/**
+ * Collaborative-readiness latch for the file editor.
+ *
+ * A file becomes "ready" (editable + agent streaming enabled) once its shared doc has both SYNCED and
+ * SEEDED. The subtlety this latch solves: a brand-new file's provider reports `synced: true` on the
+ * initial (empty) sync, then the SERVER pushes the seed and receiving that update flips `synced` back to
+ * `false` — so `synced` and `seeded` are never `true` in the same observation. An un-latched
+ * `synced && seeded` gate would therefore never open, and agent streaming would be dropped for the whole
+ * run (the file only fills in on reload).
+ *
+ * The latch: `syncedOnce` is sticky — set the first time a completed sync is observed, and it never
+ * reverts. A completed sync is proven by EITHER a live `synced`, OR the seed flag being present without
+ * the offline fallback having set it (`offlineSeed`) — because a SERVER seed can only arrive after a
+ * sync, whereas the offline fallback seeds locally without ever reaching the server and must stay
+ * read-only. Once `syncedOnce` is set, a later `synced` flap can no longer re-gate the doc.
+ */
+
+export interface CollabReadinessInputs {
+  /** The provider's current `synced` flag (may flap false after the seed update). */
+  synced: boolean
+  /** Whether the shared doc carries the seed flag. */
+  seeded: boolean
+  /** Whether the seed flag was set by the offline fallback (no server sync) rather than the server. */
+  offlineSeed: boolean
+}
+
+/**
+ * Pure transition for the readiness latch. `syncedOnce` is the sticky prior state — pass the returned
+ * `syncedOnce` back in on the next call. `ready` is whether the doc is synced-and-seeded.
+ */
+export function nextCollabReadiness(
+  syncedOnce: boolean,
+  input: CollabReadinessInputs
+): { syncedOnce: boolean; ready: boolean } {
+  const next = syncedOnce || input.synced || (input.seeded && !input.offlineSeed)
+  return { syncedOnce: next, ready: next && input.seeded }
+}

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

@@ -472,19 +472,26 @@
  * cell is not hidden behind adjacent cells.
  */
 .rich-markdown-prose .collaboration-carets__caret {
+  /* Zero inline footprint: a positioned anchor with NO width/border/margin, so inserting or moving this
+   * inline widget never reflows the surrounding text. The visible bar, dormant cap (::before), name label,
+   * and hover slop (::after) are all positioned relative to this anchor and are out of the inline flow. */
   position: relative;
-  margin-left: -1px;
-  margin-right: -1px;
-  border-left-width: 1px;
-  border-left-style: solid;
-  border-left-color: var(--caret-color);
-  border-right-width: 1px;
-  border-right-style: solid;
-  border-right-color: var(--caret-color);
   word-break: normal;
   z-index: 20;
 }
 
+/* The visible caret bar — absolutely positioned so it draws over the text without occupying inline
+ * width. `left: -1px` centers the 2px bar on the cursor position; top/bottom span the line box height. */
+.rich-markdown-prose .collaboration-carets__bar {
+  position: absolute;
+  top: -0.1em;
+  bottom: -0.1em;
+  left: -1px;
+  width: 2px;
+  background-color: var(--caret-color);
+  pointer-events: none;
+}
+
 /* Dormant affordance: a small cap at the top of the caret in the collaborator's color,
  * shaped like a collapsed presence name tag (same `rounded-xs` + notch corner as the
  * tables/canvas tags) so the whole presence system reads as one language. It signals
@@ -493,10 +500,9 @@
 .rich-markdown-prose .collaboration-carets__caret::before {
   content: "";
   position: absolute;
-  /* Seat the cap's square bottom-left corner on the pole's top-left, flush like a flag on its
-   * pole. `left: -1px` backs out the caret's 1px left border (the abs-positioning origin is the
-   * padding box, inside that border) so the cap's left edge lines up with the pole's left edge;
-   * `top` overlaps the pole's top a hair so the notch reads as continuous, no gap. */
+  /* Seat the cap's square bottom-left corner on the pole's top-left, flush like a flag on its pole.
+   * `left: -1px` matches the bar's `left: -1px` so the cap's left edge lines up with the bar's left edge;
+   * `top` overlaps the bar's top a hair so the notch reads as continuous, no gap. */
   top: -2px;
   left: -1px;
   width: 8px;