fix(files): elect a single agent-stream writer across tabs

Cursor round 4 (High): with the stream applied client-side, two tabs/windows on
the same chat could each derive streamingContent (the reconnect/resume path
re-consumes preview events) and each independently insert the stream under a
different Yjs clientID, duplicating content until the durable reconcile.

Fix — single-writer election via the file-doc awareness (new agent-stream-leader):
- a client applying an agent stream announces `agentApplying` on its own awareness
- only the leader (min clientID among announcers) applies mid-stream AND at settle;
  a non-leader renders the leader's ops via Yjs and does not apply (a non-leader
  applying the final body would re-insert the whole doc as a duplicate)
- re-checked each frame, so it converges to one writer the moment awareness
  propagates; the sub-frame startup race is reconciled by the durable write
- single-client (the common case) is unaffected: it is the only announcer, so it
  always leads

Author: waleedlatif1

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

@@ -0,0 +1,55 @@
+/**
+ * @vitest-environment node
+ */
+import { describe, expect, it } from 'vitest'
+import { Awareness } from 'y-protocols/awareness'
+import * as Y from 'yjs'
+import {
+  announceAgentApplying,
+  clearAgentApplying,
+  isAgentStreamLeader,
+} from './agent-stream-leader'
+
+/** An Awareness with explicit peer states injected (self, if present, carries no `agentApplying`). */
+function awarenessWith(entries: Array<[number, Record<string, unknown>]>): Awareness {
+  const aw = new Awareness(new Y.Doc())
+  const states = aw.getStates() as Map<number, Record<string, unknown>>
+  for (const [clientId, state] of entries) states.set(clientId, state)
+  return aw
+}
+
+describe('agent-stream leader election', () => {
+  it('a sole announcer is the leader', () => {
+    expect(isAgentStreamLeader(awarenessWith([[5, { agentApplying: true }]]), 5)).toBe(true)
+  })
+
+  it('the lowest clientID among announcers leads; higher announcers do not', () => {
+    const aw = awarenessWith([
+      [7, { agentApplying: true }],
+      [3, { agentApplying: true }],
+      [9, { user: { name: 'someone else, not applying' } }],
+    ])
+    expect(isAgentStreamLeader(aw, 3)).toBe(true)
+    expect(isAgentStreamLeader(aw, 7)).toBe(false)
+  })
+
+  it('a client that is not announcing is never the leader', () => {
+    expect(isAgentStreamLeader(awarenessWith([[3, { agentApplying: true }]]), 8)).toBe(false)
+  })
+
+  it('with no announcers, nobody leads', () => {
+    expect(isAgentStreamLeader(awarenessWith([[3, { user: {} }]]), 3)).toBe(false)
+  })
+
+  it('announce makes self the leader; clear relinquishes it', () => {
+    const doc = new Y.Doc()
+    const aw = new Awareness(doc)
+    announceAgentApplying(aw)
+    expect(aw.getLocalState()?.agentApplying).toBe(true)
+    expect(isAgentStreamLeader(aw, doc.clientID)).toBe(true)
+
+    clearAgentApplying(aw)
+    expect(aw.getLocalState()?.agentApplying ?? null).toBeNull()
+    expect(isAgentStreamLeader(aw, doc.clientID)).toBe(false)
+  })
+})

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

@@ -0,0 +1,37 @@
+import type { Awareness } from 'y-protocols/awareness'
+
+/**
+ * Awareness field a collaborative client sets on its OWN state while it is applying an agent stream into
+ * the shared doc. Read by every peer to run the single-writer election below. Coexists with the caret
+ * `user` field (`setLocalStateField` writes one field without clobbering others).
+ */
+const AGENT_APPLYING_FIELD = 'agentApplying'
+
+/** Announce that this client is applying an agent stream (candidate in the leader election). */
+export function announceAgentApplying(awareness: Awareness): void {
+  awareness.setLocalStateField(AGENT_APPLYING_FIELD, true)
+}
+
+/** Stop announcing (this client is no longer applying an agent stream). */
+export function clearAgentApplying(awareness: Awareness): void {
+  awareness.setLocalStateField(AGENT_APPLYING_FIELD, null)
+}
+
+/**
+ * Single-writer election: exactly one collaborative client applies a given agent stream into the shared
+ * doc, so N tabs/windows watching the same live copilot stream don't each insert it under a different
+ * Yjs clientID and duplicate the content. The leader is the MINIMUM clientID among all clients currently
+ * announcing (via {@link announceAgentApplying}) that they are applying — a deterministic tie-break that
+ * needs no coordinator. A brief startup race (before an announcement propagates to peers) is bounded to a
+ * frame or two — self-corrected the moment awareness converges, and reconciled anyway by the durable
+ * server write. In the common single-client case the caller is the only announcer, so it always leads.
+ */
+export function isAgentStreamLeader(awareness: Awareness, selfClientId: number): boolean {
+  let leader = Number.POSITIVE_INFINITY
+  awareness.getStates().forEach((state, clientId) => {
+    if ((state as Record<string, unknown> | undefined)?.[AGENT_APPLYING_FIELD] === true) {
+      leader = Math.min(leader, clientId)
+    }
+  })
+  return leader === selfClientId
+}

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

@@ -20,6 +20,11 @@ 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 {
+  announceAgentApplying,
+  clearAgentApplying,
+  isAgentStreamLeader,
+} from './collaboration/agent-stream-leader'
 import {
   type AgentStreamSession,
   applyAgentStreamFrame,
@@ -874,8 +879,12 @@ export function LoadedRichMarkdownEditor({
         if (!collabReady) return
         // Open the stream's shadow on the FIRST ready frame so it captures the pre-stream base (immune to
         // later peer edits) — including for an `update`, whose frames are all held until settle, so settle
-        // still has a shadow through which to apply the final rewrite.
-        agentStreamSessionRef.current ??= beginAgentStream(editor)
+        // still has a shadow through which to apply the final rewrite. Announce candidacy in the
+        // single-writer election so only one tab/window actually applies this stream (see the tick).
+        if (agentStreamSessionRef.current === null) {
+          agentStreamSessionRef.current = beginAgentStream(editor)
+          if (collaboration) announceAgentApplying(collaboration.awareness)
+        }
         const session = agentStreamSessionRef.current
         const body = splitFrontmatter(content).body
         if (body === lastStreamedBodyRef.current) return
@@ -895,6 +904,17 @@ export function LoadedRichMarkdownEditor({
             streamRafRef.current = null
             return
           }
+          // Single-writer election: only the leader (min clientID among clients applying this stream)
+          // writes it into the shared doc, so multiple tabs/windows watching the same live copilot stream
+          // don't each insert it and duplicate content. A non-leader renders the leader's ops via Yjs;
+          // re-checked each frame, so it converges to one writer the moment awareness propagates.
+          if (
+            collaboration &&
+            !isAgentStreamLeader(collaboration.awareness, collaboration.doc.clientID)
+          ) {
+            streamRafRef.current = null
+            return
+          }
           if (
             pending.length > STREAM_REPARSE_THROTTLE_THRESHOLD &&
             performance.now() - lastStreamParseAtRef.current < STREAM_REPARSE_THROTTLE_MS
@@ -928,14 +948,21 @@ export function LoadedRichMarkdownEditor({
       // survive); otherwise open one on demand. The durable server write then lands as a noop diff.
       if (wasStreamingRef.current && collabReady) {
         wasStreamingRef.current = false
+        // Only the elected leader applies the final body — a non-leader never applied mid-stream, so
+        // reconciling its base-seeded shadow to the final body would re-insert the whole doc as a
+        // duplicate; it converges to the final state via Yjs + the durable server write instead. Compute
+        // leadership BEFORE clearing our announcement, then stop announcing.
+        const wasLeader =
+          !collaboration || isAgentStreamLeader(collaboration.awareness, collaboration.doc.clientID)
+        if (collaboration) clearAgentApplying(collaboration.awareness)
+        lastStreamedBodyRef.current = null
         const finalBody = splitFrontmatter(content).body
-        const session = agentStreamSessionRef.current ?? beginAgentStream(editor)
+        const session = wasLeader
+          ? (agentStreamSessionRef.current ?? beginAgentStream(editor))
+          : agentStreamSessionRef.current
         agentStreamSessionRef.current = null
-        lastStreamedBodyRef.current = null
         if (session) {
-          runOffRender(() => {
-            applyAgentStreamFrame(editor, session, finalBody)
-          })
+          if (wasLeader) runOffRender(() => applyAgentStreamFrame(editor, session, finalBody))
           // Free the shadow with an UNGUARDED microtask (not `runOffRender`): a rapid follow-up stream
           // can supersede the run token and drop the apply above, but the shadow must always be
           // destroyed. Queued after the apply, so it frees the shadow only once that has had its chance.