fix(redis): bound the three unbudgeted stream writers by bytes

The copilot stream buffer, the Tables event log and the realtime file-doc
streams were each bounded by entry count and nothing else. An entry cap bounds
how many entries a key holds and says nothing about how large each one is, so a
single writer emitting large entries reaches gigabytes well inside its cap —
which is how one file-edit stream filled a shared Redis under a 100,000-entry
cap and evicted the whole keyspace.

Each writer gets the bound its read semantics allow:

- Copilot's replay buffer is read from a cursor and must stay contiguous, so it
  now reserves bytes against per-stream and per-user ceilings inside the same
  Lua that appends, and the writer soft-stops persistence on refusal rather
  than failing the live stream.
- The Tables event log is a live feed whose readers already refetch on a prune,
  so it drops oldest-first once past a byte ceiling — the existing `pruned`
  path carries it, with the running total kept in meta under the same TTL as
  the bytes it counts.
- The file-doc streams are Yjs deltas replayed in full by a task attaching
  later, so dropping the oldest would lose edits and a native MAXLEN bound is
  unsafe. Compaction, which folds deltas into a snapshot first, is lossless —
  it now triggers on appended bytes as well as entry count.

Also folds `lib/execution/redis-budget.server.ts` into the shared module rather
than leaving two definitions of the same prefix and ceilings writing the same
Redis keys. Key layout and every execution limit are unchanged, and pinned by
test.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: waleedlatif1

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

@@ -128,6 +128,7 @@ vi.mock('redis', () => ({ createClient: () => makeClient() }))
 import { FileDocStore, REDIS_AGENT_ORIGIN, REDIS_ORIGIN } from '@/handlers/file-doc-store'
 
 const REDIS_URL = 'redis://fake'
+const COMPACT_THRESHOLD_ENTRIES = 400
 const NAME = 'workspace-file-doc:file-1'
 
 function docWithText(text: string): Y.Doc {
@@ -401,6 +402,40 @@ describe('FileDocStore', () => {
     doc.destroy()
   })
 
+  it('compacts on appended bytes, before the entry threshold is anywhere near reached', async () => {
+    const streamKey = `filedoc:stream:${NAME}`
+    const a = await newStore()
+    const doc = new Y.Doc()
+    await a.attachRoom(NAME, doc)
+
+    // A handful of large pastes: far below COMPACT_THRESHOLD entries, far above the byte ceiling.
+    // Before bytes were counted this stream held tens of megabytes and never compacted.
+    const updates: Uint8Array[] = []
+    doc.on('update', (u: Uint8Array) => updates.push(u))
+    for (let i = 0; i < 4; i++) {
+      doc.getText('body').insert(0, 'x'.repeat(3 * 1024 * 1024))
+    }
+    for (const update of updates) {
+      await a.publishAndWait(NAME, update)
+    }
+
+    await vi.waitFor(
+      () => {
+        const stream = state.backing!.streams.get(streamKey)!
+        expect(stream.length).toBeLessThan(COMPACT_THRESHOLD_ENTRIES)
+        expect(stream.some((entry) => entry.message.s === '1')).toBe(true)
+      },
+      { timeout: 5000 }
+    )
+
+    // Compaction must be lossless: the whole document is still reconstructable from what remains.
+    const rebuilt = new Y.Doc()
+    Y.applyUpdate(rebuilt, (await a.getStreamState(NAME))!)
+    expect(rebuilt.getText('body').length).toBe(4 * 3 * 1024 * 1024)
+    rebuilt.destroy()
+    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

@@ -140,6 +140,20 @@ const IDLE_POLL_MS = 250
 const READ_COUNT = 200
 /** Compact a stream once it exceeds this many entries (snapshot + trim). */
 const COMPACT_THRESHOLD = 400
+/**
+ * Compact a stream once its appended deltas exceed this many bytes, whichever comes first.
+ *
+ * The entry threshold alone bounds how many entries a stream holds and says nothing about
+ * how large each one is: one pasted block is a single entry carrying megabytes, so a stream
+ * can sit at a few dozen entries and hundreds of megabytes and never reach
+ * {@link COMPACT_THRESHOLD} before its TTL. Folding by bytes as well keeps a stream's cost
+ * proportional to its document rather than to the size of the edits that produced it.
+ *
+ * Compaction is the only safe way to shrink one of these streams: a task attaching later
+ * replays every entry to rebuild the doc, so dropping the oldest entries — what a native
+ * `MAXLEN` retention bound would do — loses edits outright. A snapshot folds them first.
+ */
+const COMPACT_BYTES_THRESHOLD = 8 * 1024 * 1024
 /** Check whether compaction is due only every Nth local publish, to avoid an XLEN per keystroke. */
 const COMPACT_CHECK_EVERY = 64
 /** Compaction critical section (snapshot + xAdd + xTrim) is fast; a generous TTL covers a slow Redis
@@ -218,6 +232,13 @@ interface StoreRoom {
   lastId: string
   /** Local publish count, to pace compaction checks. */
   publishes: number
+  /**
+   * Bytes this task has appended since the last compaction it observed, so the byte threshold
+   * costs no extra round-trip. Locally tracked, so it under-counts a peer task's appends — it
+   * is a trigger, not an accounting, and {@link COMPACT_THRESHOLD} still covers the case where
+   * many small edits arrive from elsewhere.
+   */
+  appendedBytes: 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
@@ -299,6 +320,7 @@ export class FileDocStore {
       doc,
       lastId: '0',
       publishes: 0,
+      appendedBytes: 0,
       seededObserved: false,
       realEdited: false,
     }
@@ -379,7 +401,15 @@ export class FileDocStore {
     }
     await this.write.expire(streamKey(name), STREAM_TTL_SEC).catch(() => {})
     const room = this.rooms.get(name)
-    if (room && ++room.publishes % COMPACT_CHECK_EVERY === 0) void this.maybeCompact(name)
+    if (!room) return
+    room.appendedBytes += 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
+    if (overBytes || ++room.publishes % COMPACT_CHECK_EVERY === 0) {
+      void this.maybeCompact(name, overBytes)
+    }
   }
 
   /**
@@ -734,12 +764,12 @@ export class FileDocStore {
    * only one task compacts a given stream at a time (concurrent snapshot+trim would race). Trims only up
    * to what the snapshot provably contains — never un-integrated peer entries (see below).
    */
-  private async maybeCompact(name: string): Promise<void> {
+  private async maybeCompact(name: string, force = false): Promise<void> {
     if (!this.write) return
     const room = this.rooms.get(name)
     if (!room) return
     try {
-      if ((await this.write.xLen(streamKey(name))) < COMPACT_THRESHOLD) return
+      if (!force && (await this.write.xLen(streamKey(name))) < COMPACT_THRESHOLD) return
       const key = `${COMPACT_LOCK_PREFIX}${name}`
       const token = await this.acquireLock(key, COMPACT_LOCK_TTL_MS)
       if (!token) return
@@ -752,6 +782,10 @@ 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')
+        // The folded deltas are about to be trimmed; what remains of this task's contribution is the
+        // snapshot. Reset before the appends so a concurrent publish's bytes are counted against the
+        // new baseline rather than the one being retired.
+        room.appendedBytes = snapshot.length
         // 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

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

@@ -116,7 +116,7 @@ export function createSSEStream(params: StreamingOrchestrationParams): ReadableS
   const abortController = new AbortController()
   registerActiveStream(streamId, abortController)
 
-  const publisher = new StreamWriter({ streamId, chatId, requestId })
+  const publisher = new StreamWriter({ streamId, chatId, requestId, userId })
 
   // Declared at function scope (same rationale as `cancelReason` below) so the
   // leak backstop in the orchestration's outer finally can always reach them:

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

@@ -64,6 +64,33 @@ const createRedisStub = () => {
       return Promise.resolve('OK')
     }),
     get: vi.fn().mockImplementation((key: string) => Promise.resolve(values.get(key) ?? null)),
+    /**
+     * Stands in for `APPEND_EVENTS_SCRIPT`. It reproduces the script's observable
+     * effects — dedupe, zadd, rank-trim, seq — so the read-path tests still exercise
+     * real data, and exposes `budgetRefusal` so the refusal branch can be driven
+     * without reimplementing the budget arithmetic here.
+     */
+    budgetRefusal: null as null | [number, string, number],
+    eval: vi.fn().mockImplementation((...args: unknown[]) => {
+      const numKeys = Number(args[1])
+      const keys = args.slice(2, 2 + numKeys) as string[]
+      const argv = args.slice(2 + numKeys) as Array<string | number>
+      if (api.budgetRefusal) return Promise.resolve(api.budgetRefusal)
+
+      const [eventsKey, seqKey] = keys
+      const eventLimit = Number(argv[1])
+      const lastSeq = String(argv[5])
+      const entries = sortedSets.get(eventsKey) ?? []
+      for (let i = 6; i < argv.length; i += 2) {
+        const score = Number(argv[i])
+        const value = String(argv[i + 1])
+        if (!entries.some((entry) => entry.value === value)) entries.push({ score, value })
+      }
+      entries.sort((a, b) => a.score - b.score)
+      sortedSets.set(eventsKey, entries.slice(Math.max(0, entries.length - eventLimit)))
+      values.set(seqKey, lastSeq)
+      return Promise.resolve([1])
+    }),
     pipeline: vi.fn().mockImplementation(() => {
       const operations: Array<() => Promise<unknown>> = []
       const pipeline = {
@@ -103,6 +130,7 @@ let mockRedis: ReturnType<typeof createRedisStub>
 import {
   allocateCursor,
   appendEvent,
+  appendEvents,
   clearBuffer,
   readEvents,
   scheduleBufferCleanup,
@@ -161,11 +189,84 @@ describe('mothership-stream-outbox', () => {
       })
     )
 
-    expect(mockRedis.zremrangebyrank).toHaveBeenCalledWith(
-      'mothership_stream:stream-1:events',
-      0,
-      -100_001
-    )
+    // KEYS: [events, seq, budgetOwner]; ARGV follows.
+    const [, numKeys, eventsKey, seqKey, ownerKey, ...argv] = mockRedis.eval.mock.calls[0]
+    expect(numKeys).toBe(3)
+    expect(eventsKey).toBe('mothership_stream:stream-1:events')
+    expect(seqKey).toBe('mothership_stream:stream-1:seq')
+    expect(ownerKey).toBe('execution:redis-budget:copilot_stream:stream-1')
+    // ARGV: [ttl, eventLimit, ownerLimit, userLimit, budgetTtl, lastSeq, ...zaddArgs]
+    expect(argv[1]).toBe(100_000)
+  })
+
+  /**
+   * The stream's replay copy is charged to a budget, and a refusal is reported rather
+   * than thrown: `flush()` rethrows what it is handed, and that throw reaches the
+   * error-path finalize, which would reject a response stream whose bytes the user
+   * already received.
+   */
+  it('reports a budget refusal instead of throwing', async () => {
+    const cursor = await allocateCursor('stream-1')
+    mockRedis.budgetRefusal = [0, 'owner_redis_bytes', 40_000_000]
+
+    const result = await appendEvents([
+      createEvent({
+        streamId: 'stream-1',
+        cursor: cursor.cursor,
+        seq: cursor.seq,
+        requestId: 'req-1',
+        type: MothershipStreamV1EventType.text,
+        payload: { channel: MothershipStreamV1TextChannel.assistant, text: 'hello' },
+      }),
+    ])
+
+    expect(result.persisted).toBe(false)
+    if (!result.persisted) {
+      expect(result.refusal.resource).toBe('owner_redis_bytes')
+      expect(result.refusal.currentBytes).toBe(40_000_000)
+    }
+  })
+
+  it('refuses a batch past the single-write ceiling without reaching Redis', async () => {
+    const cursor = await allocateCursor('stream-1')
+
+    const result = await appendEvents([
+      createEvent({
+        streamId: 'stream-1',
+        cursor: cursor.cursor,
+        seq: cursor.seq,
+        requestId: 'req-1',
+        type: MothershipStreamV1EventType.text,
+        payload: {
+          channel: MothershipStreamV1TextChannel.assistant,
+          text: 'x'.repeat(2 * 1024 * 1024),
+        },
+      }),
+    ])
+
+    expect(result.persisted).toBe(false)
+    expect(mockRedis.eval).not.toHaveBeenCalled()
+  })
+
+  it('charges the user ceiling only when a user is in scope', async () => {
+    const cursor = await allocateCursor('stream-1')
+    const envelope = createEvent({
+      streamId: 'stream-1',
+      cursor: cursor.cursor,
+      seq: cursor.seq,
+      requestId: 'req-1',
+      type: MothershipStreamV1EventType.text,
+      payload: { channel: MothershipStreamV1TextChannel.assistant, text: 'hello' },
+    })
+
+    await appendEvents([envelope], { streamId: 'stream-1' })
+    expect(mockRedis.eval.mock.calls[0][1]).toBe(3)
+    expect(mockRedis.eval.mock.calls[0][4]).toBe('execution:redis-budget:copilot_stream:stream-1')
+
+    mockRedis.eval.mockClear()
+    await appendEvents([envelope], { streamId: 'stream-1', userId: 'user-1' })
+    expect(mockRedis.eval.mock.calls[0][1]).toBe(4)
+    expect(mockRedis.eval.mock.calls[0][5]).toBe('execution:redis-budget:user:user-1')
   })
 
   it('clears persisted stream state during teardown cleanup', async () => {