fix(redis): account for deltas a fold retains, and release cleared counters

The compaction counter was a single total, so a fold deducted bytes for entries
`XTRIM MINID` had retained — anything published past the fold boundary the
tailer had not yet integrated. Those bytes are still in Redis, so the trigger
disarmed while the stream kept growing. Deltas are now tracked as `{id, bytes}`
and dropped only once a trim provably removed them.

Arming the trigger on retained bytes would be the opposite fault: a fold that
reclaims nothing would re-arm immediately and force a full snapshot append per
publish. Only bytes at or before the fold boundary arm it, and because that
boundary moves in the tailer rather than on publish, the tailer re-checks it —
otherwise a burst of large edits followed by silence would sit unfolded until
the next keystroke.

Also releases the copilot owner counter when the buffer is cleared, crediting
the user counter by exactly what the owner held. Those keys are deleted rather
than expired, so the counter otherwise outlived its data and a retry reusing the
streamId would be refused against bytes that no longer exist.

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

Author: waleedlatif1

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

@@ -475,7 +475,7 @@ describe('FileDocStore', () => {
     const doc = new Y.Doc()
     await a.attachRoom(NAME, doc)
     const room = (a as any).rooms.get(NAME)
-    room.appendedBytes = 9 * 1024 * 1024
+    room.pendingDeltas = [{ id: '1-0', bytes: 9 * 1024 * 1024 }]
     room.realEdited = true
 
     const write = (a as any).write
@@ -488,7 +488,27 @@ describe('FileDocStore', () => {
 
     // A failed fold must not disarm the trigger — otherwise the stream stays oversized until
     // this task happens to append another full threshold's worth of deltas.
-    expect(room.appendedBytes).toBe(9 * 1024 * 1024)
+    expect(room.pendingDeltas).toEqual([{ id: '1-0', bytes: 9 * 1024 * 1024 }])
+    doc.destroy()
+  })
+
+  it('keeps counting deltas the trim retained because they sit past the fold boundary', async () => {
+    const a = await newStore()
+    const doc = new Y.Doc()
+    await a.attachRoom(NAME, doc)
+    const room = (a as any).rooms.get(NAME)
+    room.realEdited = true
+    // The tailer has integrated up to 5-0, so MINID retains 9-0. Its bytes are still in Redis,
+    // and dropping them would disarm the byte trigger while the stream kept growing.
+    room.lastId = '5-0'
+    room.pendingDeltas = [
+      { id: '3-0', bytes: 4 * 1024 * 1024 },
+      { id: '9-0', bytes: 7 * 1024 * 1024 },
+    ]
+
+    await (a as any).maybeCompact(NAME, true)
+
+    expect(room.pendingDeltas).toEqual([{ id: '9-0', bytes: 7 * 1024 * 1024 }])
     doc.destroy()
   })
 

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

@@ -186,6 +186,22 @@ const READER_ERROR_LOG_EVERY = 20
 
 const streamKey = (name: string) => `${STREAM_PREFIX}${name}`
 
+/**
+ * Unfolded delta bytes a compaction could actually reclaim right now.
+ *
+ * Only entries at or before `room.lastId` count: a fold trims to that boundary, so bytes past it
+ * would survive the trim and re-arm the trigger immediately, forcing a full snapshot append per
+ * publish that reclaims nothing. They stay in `pendingDeltas` and start counting once the tailer
+ * has integrated them.
+ */
+function foldableDeltaBytes(room: StoreRoom): number {
+  let bytes = 0
+  for (const delta of room.pendingDeltas) {
+    if (!isAfterStreamId(delta.id, room.lastId)) bytes += delta.bytes
+  }
+  return bytes
+}
+
 /**
  * Decode one stream entry's base64 Yjs update and apply it to `doc`. A malformed entry is logged and
  * SKIPPED — never thrown — so one bad frame can neither wedge the tailer nor abort a headless
@@ -236,14 +252,20 @@ interface StoreRoom {
   /** Local publish count, to pace compaction checks. */
   publishes: number
   /**
-   * Delta bytes this task has appended since the last compaction it performed, so the byte
-   * threshold costs no extra round-trip. Counts deltas only — never the snapshot a compaction
-   * writes, which is a function of document size rather than of edit volume and would make a
-   * large document breach the threshold permanently. Locally tracked, so it under-counts a peer
-   * task's appends: it is a trigger, not an accounting, and {@link COMPACT_THRESHOLD} still
-   * covers many small edits arriving from elsewhere.
+   * Deltas this task has appended and not yet folded, as `{id, bytes}` pairs in append order.
+   *
+   * Keyed by stream id rather than summed, because a fold trims to `room.lastId` and RETAINS
+   * anything published past it — those bytes are still in Redis, so deducting them would
+   * disarm the trigger while the stream keeps growing. Entries are dropped only once an
+   * `XTRIM` has provably removed them.
+   *
+   * Counts deltas only — never the snapshot a compaction writes, which is a function of
+   * document size rather than of edit volume and would make a large document breach the
+   * threshold permanently. Locally tracked, so it under-counts a peer task's appends: it is a
+   * trigger, not an accounting, and {@link COMPACT_THRESHOLD} still covers many small edits
+   * arriving from elsewhere.
    */
-  appendedBytes: number
+  pendingDeltas: Array<{ id: string; bytes: number }>
   /** Set once the doc has been observed seeded, so the seed transition itself is never mistaken for an
    * edit (mirrors the relay's `seededObserved`). */
   seededObserved: boolean
@@ -325,7 +347,7 @@ export class FileDocStore {
       doc,
       lastId: '0',
       publishes: 0,
-      appendedBytes: 0,
+      pendingDeltas: [],
       seededObserved: false,
       realEdited: false,
     }
@@ -390,9 +412,10 @@ export class FileDocStore {
     const encoded = Buffer.from(update).toString('base64')
     const fields: Record<string, string> = { [UPDATE_FIELD]: encoded }
     if (agent) fields[AGENT_FIELD] = '1'
+    let appendedId: string | null = null
     for (let attempt = 0; attempt <= PUBLISH_MAX_RETRIES; attempt++) {
       try {
-        await this.write.xAdd(streamKey(name), '*', fields)
+        appendedId = await this.write.xAdd(streamKey(name), '*', fields)
         break
       } catch (error) {
         if (attempt === PUBLISH_MAX_RETRIES) {
@@ -407,11 +430,11 @@ export class FileDocStore {
     await this.write.expire(streamKey(name), STREAM_TTL_SEC).catch(() => {})
     const room = this.rooms.get(name)
     if (!room) return
-    room.appendedBytes += encoded.length
+    if (appendedId) room.pendingDeltas.push({ id: appendedId, bytes: encoded.length })
     // Bytes are checked every publish: one entry can cross the ceiling on its own, so pacing this
     // check the way the entry count is paced would let a stream sit far over the ceiling for up to
-    // COMPACT_CHECK_EVERY more appends. The check itself is a local comparison.
-    const overBytes = room.appendedBytes >= COMPACT_BYTES_THRESHOLD
+    // COMPACT_CHECK_EVERY more appends. The check itself is a local sum over unfolded entries.
+    const overBytes = foldableDeltaBytes(room) >= COMPACT_BYTES_THRESHOLD
     if (overBytes || ++room.publishes % COMPACT_CHECK_EVERY === 0) {
       void this.maybeCompact(name, overBytes)
     }
@@ -726,6 +749,11 @@ export class FileDocStore {
           // but wasteful re-delivery). The new room caught itself up via xRange already.
           if (!room || room !== snapshot.get(name)) continue
           for (const entry of stream.messages) this.applyEntry(room, entry.id, entry.message)
+          // Foldability is decided by `lastId`, which only the tailer advances — so a burst of
+          // large edits followed by silence would otherwise sit unfolded until the next publish
+          // happened to re-evaluate the trigger. Re-check it where the boundary actually moved.
+          if (foldableDeltaBytes(room) >= COMPACT_BYTES_THRESHOLD)
+            void this.maybeCompact(name, true)
         }
       } catch (error) {
         if (!this.running) break
@@ -787,10 +815,7 @@ export class FileDocStore {
         // appended snapshot id instead would silently drop those un-integrated peer entries.
         const upTo = room.lastId
         const snapshot = Buffer.from(Y.encodeStateAsUpdate(room.doc)).toString('base64')
-        // Bytes this fold is accountable for. Deducted only once the trim succeeds, so a failed
-        // compaction leaves the trigger armed instead of silently disarming it — and deducting
-        // rather than zeroing preserves whatever a concurrent publish added while it ran.
-        const foldedBytes = room.appendedBytes
+        // Captured with `upTo` so the two agree: exactly the entries this fold will trim.
         // Stamp the snapshot by what it folds: a real edit → SNAPSHOT_FIELD (a fresh catch-up treats it
         // as edited content, not a bare seed). An agent-ONLY stream (no real edit yet) → AGENT_FIELD, so a
         // peer catching up applies it as REDIS_AGENT_ORIGIN and never marks the doc edited — preserving
@@ -806,7 +831,11 @@ export class FileDocStore {
         // Never the snapshot's own size: a document whose snapshot already exceeds the ceiling
         // would re-breach it the instant compaction finished and force a full snapshot append on
         // every subsequent keystroke — the write amplification this threshold exists to prevent.
-        room.appendedBytes = Math.max(0, room.appendedBytes - foldedBytes)
+        // Drop only what the trim provably removed. An entry published past `upTo` is retained by
+        // MINID and its bytes are still in Redis, so it stays counted; dropping it would disarm the
+        // trigger while the stream kept growing. Done after the trim, so a failed fold changes
+        // nothing and leaves the trigger armed.
+        room.pendingDeltas = room.pendingDeltas.filter((delta) => isAfterStreamId(delta.id, upTo))
       } finally {
         await this.releaseLock(key, token)
       }

apps/sim/lib/copilot/request/lifecycle/start.ts

@@ -206,7 +206,10 @@ export function createSSEStream(params: StreamingOrchestrationParams): ReadableS
               }
             | undefined
 
-          await Promise.all([resetBuffer(streamId), clearFilePreviewSessions(streamId)])
+          await Promise.all([
+            resetBuffer(streamId, { streamId, ...(userId ? { userId } : {}) }),
+            clearFilePreviewSessions(streamId),
+          ])
 
           if (chatId) {
             createRunSegment({

apps/sim/lib/copilot/request/session/buffer.test.ts

@@ -398,4 +398,24 @@ describe('mothership-stream-outbox', () => {
     expect(result.persisted).toBe(false)
     expect(mockRedis.eval).not.toHaveBeenCalled()
   })
+
+  it('releases the owner counter and credits the user when the buffer is cleared', async () => {
+    // The buffer keys are deleted rather than expired, so a counter left behind would refuse a
+    // retry that reuses the same streamId against bytes that no longer exist anywhere.
+    await clearBuffer('stream-1', 'clear_outbox', { streamId: 'stream-1', userId: 'user-1' })
+
+    expect(mockRedis.del).toHaveBeenCalled()
+    const evalCall = mockRedis.eval.mock.calls.at(-1)
+    expect(evalCall?.[1]).toBe(2)
+    expect(evalCall?.[2]).toBe('execution:redis-budget:copilot_stream:stream-1')
+    expect(evalCall?.[3]).toBe('execution:redis-budget:user:user-1')
+  })
+
+  it('releases only the owner counter when no user is in scope', async () => {
+    await clearBuffer('stream-1')
+
+    const evalCall = mockRedis.eval.mock.calls.at(-1)
+    expect(evalCall?.[1]).toBe(1)
+    expect(evalCall?.[2]).toBe('execution:redis-budget:copilot_stream:stream-1')
+  })
 })

apps/sim/lib/copilot/request/session/buffer.ts

@@ -8,6 +8,7 @@ import {
   getRedisBudgetLimits,
   logRedisBudgetRefusal,
   parseRedisBudgetRefusal,
+  REDIS_BUDGET_RELEASE_SCRIPT,
   type RedisBudgetRefusal,
   renderRedisBudgetLua,
 } from '@/lib/core/redis/byte-budget.server'
@@ -102,13 +103,28 @@ export async function allocateCursor(streamId: string): Promise<{
   return { seq, cursor: String(seq) }
 }
 
-export async function resetBuffer(streamId: string): Promise<void> {
-  await clearBuffer(streamId, 'reset_outbox')
+export async function resetBuffer(streamId: string, scope?: StreamBudgetScope): Promise<void> {
+  await clearBuffer(streamId, 'reset_outbox', scope)
 }
 
-export async function clearBuffer(streamId: string, operation = 'clear_outbox'): Promise<void> {
+export async function clearBuffer(
+  streamId: string,
+  operation = 'clear_outbox',
+  scope?: StreamBudgetScope
+): Promise<void> {
   await withRedisRetry({ operation, streamId }, async (redis) => {
     await redis.del(getEventsKey(streamId), getSeqKey(streamId), getAbortKey(streamId))
+    /*
+      The counter outlives the data it accounts for unless it is released here: the keys
+      above are deleted rather than expired, so without this a retry reusing the same
+      streamId would be refused against bytes that no longer exist anywhere.
+    */
+    const budgetKeys = getRedisBudgetKeys({
+      kind: 'copilot_stream',
+      id: streamId,
+      ...(scope?.userId ? { userId: scope.userId } : {}),
+    })
+    await redis.eval(REDIS_BUDGET_RELEASE_SCRIPT, budgetKeys.length, ...budgetKeys)
   })
 }
 

apps/sim/lib/core/redis/byte-budget.server.ts

@@ -201,6 +201,27 @@ end
 `
 }
 
+/**
+ * Releases an owner's whole reservation when its data is deleted rather than expired.
+ *
+ * The owner counter is dropped and the user counter credited by exactly what the owner
+ * held, in one script — crediting the user from a separately read value would let a
+ * concurrent write land in between and be released twice.
+ *
+ * KEYS: [ownerKey] or [ownerKey, userKey], as {@link getRedisBudgetKeys} returns them.
+ */
+export const REDIS_BUDGET_RELEASE_SCRIPT = `
+local owner_bytes = tonumber(redis.call('GET', KEYS[1]) or '0')
+redis.call('DEL', KEYS[1])
+if #KEYS >= 2 and owner_bytes > 0 then
+  local user_next = redis.call('DECRBY', KEYS[2], owner_bytes)
+  if user_next <= 0 then
+    redis.call('DEL', KEYS[2])
+  end
+end
+return owner_bytes
+`
+
 /** Parses the `{0, resource, current}` refusal a guarded script returns. */
 export function parseRedisBudgetRefusal(
   result: unknown,