fix(files): close realEdited data-loss race + elect a settle writer

Independent audit surfaced two real gaps:

- file-doc-store: realEdited was latched AFTER appendUpdate's awaits, but the
  edit already sits in room.doc synchronously. A concurrent agent-frame
  compaction could read realEdited=false, snapshot that real content, and stamp
  it a no-persist agent frame — a lost edit. Latch it synchronously (same tick
  as the doc mutation) before any await. Deterministic falsifiable test added.

- rich-markdown-editor: at settle every tab applied the final body, and a
  non-leader's local microtask runs before the leader's final propagates, so
  both insert the tail (Yjs keeps both) -> duplicated tail. Elect a single
  settle writer (reliable — awareness is long converged by settle), reading
  leadership before clearing the announcement. Corrects the overclaiming
  idempotency comment and the handoff pick-up comment.

Adds a y-tiptap internals upgrade-guardrail test.

Author: waleedlatif1

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

@@ -270,6 +270,23 @@ describe('FileDocStore', () => {
     bDoc.destroy()
   })
 
+  it('latches realEdited synchronously so a concurrent compaction can never mislabel a real edit', async () => {
+    // The data-loss race: a real edit sits in room.doc synchronously, but if realEdited were set only
+    // AFTER appendUpdate's awaits, a concurrent agent-triggered compaction could snapshot that content and
+    // stamp it an agent (no-persist) frame — losing the edit. The latch must be set in the same tick.
+    const a = await newStore()
+    const doc = new Y.Doc()
+    await a.attachRoom(NAME, doc)
+    const room = (a as any).rooms.get(NAME)
+    expect(room.realEdited).toBe(false)
+    // Kick off a real (non-agent) append but do NOT await it: realEdited must already be true before the
+    // xAdd/expire awaits resolve, so any compaction racing on the awaits sees the real edit.
+    const pending = (a as any).appendUpdate(NAME, updateFor('real user edit'))
+    expect(room.realEdited).toBe(true)
+    await pending
+    doc.destroy()
+  })
+
   it('stamps a compaction snapshot of an agent-ONLY stream as an agent frame (never persisted)', async () => {
     const streamKey = `filedoc:stream:${NAME}`
     const noop = Buffer.from(Y.encodeStateAsUpdate(new Y.Doc())).toString('base64')

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

@@ -299,6 +299,17 @@ export class FileDocStore {
    */
   private async appendUpdate(name: string, update: Uint8Array, agent = false): Promise<void> {
     if (!this.write) return
+    // Latch realEdited SYNCHRONOUSLY — before the first await — for a real (non-agent) publish. The edit
+    // already sits in room.doc (applied in doc.on('update') before publish was called), so if this set
+    // were deferred past the xAdd/expire awaits a CONCURRENT agent-frame-triggered maybeCompact could read
+    // realEdited=false, snapshot the doc (which already holds this real edit), and stamp it an agent
+    // (no-persist) snapshot — a lost edit. Setting it in the same synchronous tick as the doc mutation
+    // makes "room.doc holds a real edit ⇒ realEdited" hold before any compaction (always async) can run.
+    // Monotonic latch, so an eager set is safe; the seed never flows through here (it uses seedIfEmpty).
+    if (!agent) {
+      const editedRoom = this.rooms.get(name)
+      if (editedRoom) editedRoom.realEdited = true
+    }
     const encoded = Buffer.from(update).toString('base64')
     const fields: Record<string, string> = { [UPDATE_FIELD]: encoded }
     if (agent) fields[AGENT_FIELD] = '1'
@@ -318,13 +329,7 @@ export class FileDocStore {
     }
     await this.write.expire(streamKey(name), STREAM_TTL_SEC).catch(() => {})
     const room = this.rooms.get(name)
-    if (room) {
-      // A local non-agent publish (a user edit or the awaited copilot durable merge) is a real edit; the
-      // seed never flows through here (it uses seedIfEmpty). Set it BEFORE the compaction check below so a
-      // real edit can never be folded into an agent (no-persist) snapshot due to a tail-back race.
-      if (!agent) room.realEdited = true
-      if (++room.publishes % COMPACT_CHECK_EVERY === 0) void this.maybeCompact(name)
-    }
+    if (room && ++room.publishes % COMPACT_CHECK_EVERY === 0) void this.maybeCompact(name)
   }
 
   /**

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

@@ -2,6 +2,7 @@
  * @vitest-environment jsdom
  */
 import { Editor } from '@tiptap/core'
+import { initProseMirrorDoc, updateYFragment, ySyncPluginKey } from '@tiptap/y-tiptap'
 import { afterEach, beforeAll, describe, expect, it } from 'vitest'
 import { Awareness } from 'y-protocols/awareness'
 import * as Y from 'yjs'
@@ -49,6 +50,18 @@ function track(t: { editor: Editor; doc: Y.Doc; awareness: Awareness }) {
 }
 
 describe('agent-stream applier', () => {
+  it('relies on y-tiptap internals that still exist (upgrade guardrail)', () => {
+    // beginAgentStream/applyAgentStreamFrame reach into y-tiptap internals (not public TipTap API):
+    // `ySyncPluginKey`, `updateYFragment`, `initProseMirrorDoc`. A y-tiptap bump that renames or drops
+    // any of them can pass typecheck yet break at runtime — assert their runtime shape here so an upgrade
+    // fails loudly at test time instead of in production. Pinned to an exact y-tiptap version in
+    // package.json; bump that pin and this guard together.
+    expect(typeof updateYFragment).toBe('function')
+    expect(typeof initProseMirrorDoc).toBe('function')
+    expect(ySyncPluginKey).toBeDefined()
+    expect(typeof ySyncPluginKey.getState).toBe('function')
+  })
+
   it('streams agent content into the live collaborative doc and broadcasts it as Yjs ops', () => {
     const { editor, doc } = track(makeCollabEditor())
 

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

@@ -908,7 +908,10 @@ export function LoadedRichMarkdownEditor({
           // Single-writer election: only the leader (min clientID among clients announcing they apply 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.
+          // Yjs; re-checked each frame, so a co-leader stops the moment awareness propagates. (The pick-up
+          // direction — a successor beginning to write after the leader tab closes — waits for the next
+          // content frame to run a tick; a stream that already delivered its last frame is covered by
+          // settle and the durable write, so at worst a brief end-of-stream display lag, never a loss.)
           // Bounded residual (accepted): if two tabs start the SAME stream within the awareness-propagation
           // window they briefly both lead and duplicate a frame or two — a rare, transient, never-persisted
           // glitch (SYNC_NO_PERSIST keeps it out of storage; the durable edit_content write reconciles the
@@ -962,27 +965,41 @@ export function LoadedRichMarkdownEditor({
         cancelAnimationFrame(streamRafRef.current)
         streamRafRef.current = null
       }
-      // Settle: apply the FINAL body so the Y.Doc exactly equals the streamed result. The mid-stream
-      // leader REUSES its up-to-date shadow (just catching a throttled last frame); a client that never
-      // applied mid-stream (a non-leader, a held `update`, or a pre-seed stream) opens a FRESH shadow
-      // seeded from the CURRENT doc. Reconciling current→final is idempotent — a client that settles after
-      // another already wrote the final reconciles to a noop — so there is NO settle-time election and no
-      // base-shadow duplication, and a lone client (incl. an `update`) still applies rather than waiting on
-      // the durable merge. That durable `edit_content` write then lands as a noop diff too.
+      // Settle: apply the FINAL body so the Y.Doc exactly equals the streamed result — but ONLY the
+      // elected writer applies it (the same min-clientID election the streaming tick uses). Without this,
+      // N tabs watching one run each open a fresh shadow and reconcile current→final; a non-leader's local
+      // settle microtask runs BEFORE the leader's final propagates (a server round-trip), so both insert
+      // the same tail and Yjs keeps both (it does not dedupe identical text from two clients) → a
+      // duplicated tail. The election is reliable here (unlike the bounded startup window): the stream ran
+      // for seconds, so awareness is long converged. Each tab reads leadership BEFORE clearing its own
+      // announcement — a remote clear is a network round-trip, always slower than these local microtasks,
+      // so every tab sees the same announcer set and agrees on one leader. The leader reuses its
+      // up-to-date shadow (catching a throttled last frame) or, if it never applied mid-stream (a held
+      // `update`, or a pre-seed stream), opens a FRESH shadow from the current doc; a non-leader applies
+      // nothing (the leader's final broadcasts to it) and frees any shadow it still held. The durable
+      // `edit_content` write then lands as a noop diff for everyone.
       if (wasStreamingRef.current && collabReady) {
         wasStreamingRef.current = false
         agentAnnouncedRef.current = false
+        const isSettleWriter =
+          !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 heldSession = agentStreamSessionRef.current
         agentStreamSessionRef.current = null
-        if (session) {
-          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.
-          queueMicrotask(() => endAgentStream(session))
+        if (isSettleWriter) {
+          const finalBody = splitFrontmatter(content).body
+          const session = heldSession ?? beginAgentStream(editor)
+          if (session) {
+            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.
+            queueMicrotask(() => endAgentStream(session))
+          }
+        } else if (heldSession) {
+          // Non-leader: it never writes the final (the leader does + broadcasts it); free any shadow it held.
+          queueMicrotask(() => endAgentStream(heldSession))
         }
       }
       return