fix(collab-doc): close review-round race/durability gaps

Address Greptile P1 + Cursor findings:

- Seed publishes to the shared stream AWAITED *before* seeding the local doc, so
  a publish failure leaves the doc unseeded and the stream empty for a clean
  retry rather than serving an unpublished local seed a peer would re-seed over
  (split-brain).
- flushPersist falls back to the synchronously-captured local snapshot when
  getStreamState throws (not only when it returns null), so a transient Redis
  read no longer drops the final durable write as the room is torn down.
- streamHasContent fails CLOSED (returns true on xLen error): a Redis blip can no
  longer let the seed fence pass and double-seed.
- Client collabReady initializes from the collaborative prop, so a collaborative
  editor never has a mount-window where client autosave could clobber the server
  write.

Author: waleedlatif1

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

@@ -268,15 +268,20 @@ export class FileDocStore {
 
   /**
    * Whether the file's stream already holds content — fences a seed apply against a peer that seeded
-   * while this task held a (possibly stale) lock, so we never write a second seed (split-brain). False
-   * when disabled or on error (the caller then proceeds under its lock, as before).
+   * while this task held a (possibly stale) lock, so we never write a second seed (split-brain). Both
+   * callers treat `true` as "do not seed", so this fails CLOSED: a Redis `xLen` error returns `true`
+   * (cannot confirm the stream is empty → do not risk a double-seed). `false` only when genuinely empty,
+   * or when disabled (single-replica, where seeding locally is always correct).
    */
   async streamHasContent(name: string): Promise<boolean> {
     if (!this.enabled || !this.write) return false
     try {
       return (await this.write.xLen(streamKey(name))) > 0
-    } catch {
-      return false
+    } catch (error) {
+      logger.warn(`FileDocStore streamHasContent failed for ${name}`, {
+        error: getErrorMessage(error),
+      })
+      return true
     }
   }
 

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

@@ -121,8 +121,8 @@ const FILE_DOC_FIELD = 'default'
 
 /** Let a fire-and-forget `void ensureServerSeed(...)` chain settle (mock resolves synchronously). */
 async function flushMicrotasks(): Promise<void> {
-  await Promise.resolve()
-  await Promise.resolve()
+  // Enough to drain the fire-and-forget seed chain (shouldSeed → fetch → fence → publish → apply).
+  for (let i = 0; i < 8; i++) await Promise.resolve()
 }
 
 /**

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

@@ -216,7 +216,19 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr
   try {
     if (!final && !(await store.acquirePersistSlot(name, FILE_DOC_TIMEOUTS.persistRequestMs)))
       return
-    const docState = (store.enabled ? await store.getStreamState(name) : null) ?? localState
+    let docState = localState
+    if (store.enabled) {
+      try {
+        docState = (await store.getStreamState(name)) ?? localState
+      } catch (streamError) {
+        // A transient Redis read must NOT drop the write when we already hold a valid local snapshot —
+        // else the last-disconnect flush loses the session's edits as the room is torn down.
+        if (!localState) throw streamError
+        logger.warn(`Stream state unavailable for file ${room.fileId}; persisting local snapshot`, {
+          error: getErrorMessage(streamError),
+        })
+      }
+    }
     if (!docState) return // nothing seeded/authoritative to persist yet
     await fetchFileDocPersist(room.workspaceId, room.fileId, room.lastEditorUserId, docState)
   } catch (error) {
@@ -332,18 +344,15 @@ async function ensureServerSeed(
     // Fence against a peer that seeded while we held a stale lock (a rare long stall): if the stream
     // already has content it is seeded, so never write a SECOND seed on top — that would split-brain.
     if (await store.streamHasContent(name)) return
-    // SEED_ORIGIN → `doc.on('update')` fans the seed to THIS task's clients but does NOT publish it
-    // (nor persist it — the seed is the file's current content). We publish it EXPLICITLY and AWAIT the
-    // stream write below, so the seed is durably in the stream BEFORE the lock releases — then any later
-    // seeder's `streamHasContent` fence is guaranteed to see it. Closes the fence's publish-after-release
-    // gap that a fire-and-forget publish left open.
-    if (update) Y.applyUpdate(room.doc, update, SEED_ORIGIN)
-    else
-      room.doc.transact(
-        () => room.doc.getMap(FILE_DOC_SEED.configMap).set(FILE_DOC_SEED.flag, true),
-        SEED_ORIGIN
-      )
-    await store.publishAndWait(name, Y.encodeStateAsUpdate(room.doc))
+    // Build the seed (file content + seed flag, or just the flag for an empty/missing file), then
+    // PUBLISH it to the shared stream AWAITED *before* seeding the local doc. The doc is marked seeded
+    // only once the seed is durably shared — so if the publish fails we leave the doc unseeded and the
+    // stream empty for a clean retry, rather than serving an unpublished local seed that a peer would
+    // re-seed on top of (split-brain). SEED_ORIGIN keeps `doc.on('update')` from re-publishing it.
+    const seedUpdate = update ?? emptySeedUpdate()
+    await store.publishAndWait(name, seedUpdate)
+    if (fileDocRooms.get(name) !== room || isDocSeeded(room.doc)) return
+    Y.applyUpdate(room.doc, seedUpdate, SEED_ORIGIN)
   } catch (error) {
     logger.warn(`Server seed failed for file ${room.fileId} (workspace ${workspaceId})`, error)
     room.serverSeedStarted = false
@@ -352,6 +361,18 @@ async function ensureServerSeed(
   }
 }
 
+/** The seed update for an empty/missing file: just the `initialContentLoaded` flag, so an empty doc
+ * still reaches readiness (and its emptiness is durably shared like any seed). */
+function emptySeedUpdate(): Uint8Array {
+  const doc = new Y.Doc()
+  doc.getMap(FILE_DOC_SEED.configMap).set(FILE_DOC_SEED.flag, true)
+  try {
+    return Y.encodeStateAsUpdate(doc)
+  } finally {
+    doc.destroy()
+  }
+}
+
 /** Serializes live merges per file so overlapping calls never race the same doc (see below). */
 const fileDocMergeChains = new Map<string, Promise<unknown>>()
 

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

@@ -126,8 +126,12 @@ export const RichMarkdownEditor = memo(function RichMarkdownEditor({
    * shared document to markdown server-side, so the client must never also autosave — a stale keystroke
    * saving over a server/copilot edit is exactly the clobber the server path closes. The child reports
    * the right value up via `onCollabReadyChange`.
+   *
+   * Initialize from the `collaborative` prop (NOT unconditionally `true`): a collaborative file must
+   * start with autosave OFF, or a save could fire in the window before the child mounts and reports —
+   * re-clobbering exactly what this closes. The child turns it on for the non-collaborative fallback.
    */
-  const [collabReady, setCollabReady] = useState(true)
+  const [collabReady, setCollabReady] = useState(!collaborative)
 
   const {
     content,