fix(realtime): adopt byte accounting for a stream taken over

A room attaching to an existing stream started from an empty ledger, so a
multi-megabyte stream under the entry threshold stayed unfolded while that
room's own heartbeat kept refreshing its TTL — a restart or handoff could hold
one open indefinitely.

`catchUp` already reads every entry to rebuild the doc, so adopting their bytes
costs no extra work. Plain deltas only: a compaction snapshot is the result of a
fold rather than something a fold can reclaim, so counting one would arm the
trigger against itself. The trigger is re-checked once, after catch-up, since
nothing else re-checks until the next local publish and a read-only participant
never makes one.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: waleedlatif1

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

@@ -552,6 +552,45 @@ describe('FileDocStore', () => {
     doc.destroy()
   })
 
+  it('adopts accounting for a stream it takes over, and folds it if already over the ceiling', async () => {
+    const streamKey = `filedoc:stream:${NAME}`
+    // A stream left behind by a previous task: two entries, so far under the entry threshold, and
+    // far over the byte ceiling. A fresh room starting from an empty ledger would never fold it,
+    // while its own heartbeat kept refreshing the TTL.
+    const seedDoc = new Y.Doc()
+    const updates: Uint8Array[] = []
+    seedDoc.on('update', (u: Uint8Array) => updates.push(u))
+    seedDoc.getText('body').insert(0, 'x'.repeat(9 * 1024 * 1024))
+    seedDoc.getText('body').insert(0, 'tail')
+    state.backing!.streams.set(
+      streamKey,
+      updates.map((update, index) => ({
+        id: `${index + 1}-0`,
+        message: { u: Buffer.from(update).toString('base64') },
+      }))
+    )
+    state.backing!.seq = updates.length
+
+    const a = await newStore()
+    const doc = new Y.Doc()
+    await a.attachRoom(NAME, doc)
+
+    // Either marker counts as a fold: this room only ever replayed entries, so it never observed
+    // a real edit and its snapshot is stamped as an agent frame (the no-persist guarantee).
+    await vi.waitFor(() => {
+      const stream = state.backing!.streams.get(streamKey)!
+      expect(stream.some((entry) => entry.message.s === '1' || entry.message.a === '1')).toBe(true)
+    })
+
+    // Lossless: the adopted content survives the fold it triggered.
+    const rebuilt = new Y.Doc()
+    Y.applyUpdate(rebuilt, (await a.getStreamState(NAME))!)
+    expect(rebuilt.getText('body').length).toBe(9 * 1024 * 1024 + 4)
+    rebuilt.destroy()
+    doc.destroy()
+    seedDoc.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

@@ -391,9 +391,26 @@ export class FileDocStore {
         // the SEED after `seededObserved` latched would count it as a post-seed edit and let a
         // compaction snapshot claim content no user ever typed. Skip what this room already holds.
         if (!isAfterStreamId(entry.id, room.lastId)) continue
+        // Adopt the accounting for what is already in the stream. A task taking one over would
+        // otherwise start from an empty ledger, so a multi-megabyte stream under the entry
+        // threshold would stay unfolded while this room's heartbeat keeps refreshing its TTL.
+        // These entries are already being read to rebuild the doc, so this costs no extra work —
+        // unlike seeding from a scan we would not otherwise do.
+        //
+        // Plain deltas only: a compaction snapshot is the RESULT of a fold, not something a fold
+        // can reclaim, so counting one would arm the trigger against itself.
+        if (!entry.message[SNAPSHOT_FIELD] && !entry.message[AGENT_FIELD]) {
+          room.pendingDeltas.push({
+            id: entry.id,
+            bytes: entry.message[UPDATE_FIELD]?.length ?? 0,
+          })
+        }
         this.applyEntry(room, entry.id, entry.message)
       }
       await this.write.expire(streamKey(name), STREAM_TTL_SEC)
+      // The adopted entries may already be past the ceiling, and nothing else re-checks until the
+      // next local publish — which a read-only participant never makes.
+      if (foldableDeltaBytes(room) >= COMPACT_BYTES_THRESHOLD) void this.maybeCompact(name, true)
     } catch (error) {
       logger.warn(`FileDocStore catch-up failed for ${name}`, { error: getErrorMessage(error) })
     }