fix(redis): address review findings on the byte bounds

- Measure ceilings in UTF-8 bytes on both the copilot and Tables paths, so the
  TypeScript checks bound a stream the same way the Lua's `string.len` does
  rather than under-reporting every non-ASCII frame.
- Split an oversized copilot batch on the per-write ceiling instead of refusing
  it. A flush carries whatever accumulated since the last one, so a run of large
  frames can exceed the ceiling collectively while each frame is individually
  writable; refusing that stopped replay for the rest of the stream over a
  batching artefact. A single frame past the ceiling is still refused.
- Re-check the copilot soft stop when an in-flight append resolves, not only at
  enqueue, so a batch queued behind a refusal cannot land and leave replay
  holding later events but not the refused ones.
- Deduct rather than zero the file-doc compaction counter, and only once the
  trim succeeds, so a failed fold leaves the trigger armed and a concurrent
  publish's bytes survive.

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

Author: waleedlatif1

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

@@ -470,6 +470,28 @@ describe('FileDocStore', () => {
     doc.destroy()
   })
 
+  it('keeps the byte trigger armed when compaction fails', async () => {
+    const a = await newStore()
+    const doc = new Y.Doc()
+    await a.attachRoom(NAME, doc)
+    const room = (a as any).rooms.get(NAME)
+    room.appendedBytes = 9 * 1024 * 1024
+    room.realEdited = true
+
+    const write = (a as any).write
+    const original = write.xTrim.bind(write)
+    write.xTrim = async () => {
+      throw new Error('redis blip')
+    }
+    await (a as any).maybeCompact(NAME, true)
+    write.xTrim = original
+
+    // 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)
+    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

@@ -787,12 +787,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')
-        // Counts deltas appended SINCE this fold, so it must not carry the snapshot's own size: a
-        // document whose snapshot already exceeds the ceiling would otherwise 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. Reset before the appends so a
-        // concurrent publish is counted against the new baseline rather than the one being retired.
-        room.appendedBytes = 0
+        // 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
         // 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
@@ -805,6 +803,10 @@ export class FileDocStore {
         // MINID keeps entries with id >= upTo: the snapshot, any un-integrated peer entries, and
         // `upTo` itself (redundant with the snapshot, harmless); it drops only the folded older deltas.
         await this.write.xTrim(streamKey(name), 'MINID', upTo)
+        // 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)
       } finally {
         await this.releaseLock(key, token)
       }

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

@@ -9,6 +9,7 @@ import {
   MothershipStreamV1TextChannel,
 } from '@/lib/copilot/generated/mothership-stream-v1'
 import { createEvent } from '@/lib/copilot/request/session/event'
+import { getRedisBudgetLimits } from '@/lib/core/redis/byte-budget.server'
 
 type StoredEnvelope = {
   score: number
@@ -136,6 +137,18 @@ import {
   scheduleBufferCleanup,
 } from '@/lib/copilot/request/session/buffer'
 
+async function makeEnvelope(text: string) {
+  const cursor = await allocateCursor('stream-1')
+  return createEvent({
+    streamId: 'stream-1',
+    cursor: cursor.cursor,
+    seq: cursor.seq,
+    requestId: 'req-1',
+    type: MothershipStreamV1EventType.text,
+    payload: { channel: MothershipStreamV1TextChannel.assistant, text },
+  })
+}
+
 describe('mothership-stream-outbox', () => {
   beforeEach(() => {
     mockRedis = createRedisStub()
@@ -346,4 +359,43 @@ describe('mothership-stream-outbox', () => {
     expect(replayed).toHaveLength(1)
     expect(replayed[0]?.payload.text).toBe('hello')
   })
+
+  it('splits an oversized batch instead of refusing it', async () => {
+    const limits = getRedisBudgetLimits('copilot_stream')
+    // Individually writable frames that collectively exceed the per-write ceiling. Refusing the
+    // whole batch would stop replay persistence for the rest of the stream over a batching artefact.
+    const envelopes = await Promise.all(
+      Array.from({ length: 3 }, () =>
+        makeEnvelope('x'.repeat(Math.floor(limits.maxSingleWriteBytes * 0.45)))
+      )
+    )
+
+    const result = await appendEvents(envelopes, { streamId: 'stream-1' })
+
+    expect(result.persisted).toBe(true)
+    expect(mockRedis.eval).toHaveBeenCalledTimes(2)
+  })
+
+  it('refuses a single frame that can never land, without splitting', async () => {
+    const limits = getRedisBudgetLimits('copilot_stream')
+    const oversized = await makeEnvelope('x'.repeat(limits.maxSingleWriteBytes + 10))
+    const result = await appendEvents([oversized], { streamId: 'stream-1' })
+
+    expect(result.persisted).toBe(false)
+    expect(mockRedis.eval).not.toHaveBeenCalled()
+  })
+
+  it('measures the ceiling in UTF-8 bytes, not UTF-16 units', async () => {
+    const limits = getRedisBudgetLimits('copilot_stream')
+    // Each astral char is 2 UTF-16 units but 4 UTF-8 bytes, so `String.length` under-reports by 2x
+    // and would call this batch writable when Redis will not.
+    const chars = Math.floor(limits.maxSingleWriteBytes / 3)
+    const astral = await makeEnvelope('\u{1D306}'.repeat(chars))
+    expect(JSON.stringify(astral).length).toBeLessThan(limits.maxSingleWriteBytes)
+
+    const result = await appendEvents([astral], { streamId: 'stream-1' })
+
+    expect(result.persisted).toBe(false)
+    expect(mockRedis.eval).not.toHaveBeenCalled()
+  })
 })

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

@@ -236,50 +236,77 @@ export async function appendEvents(
   }
   const budgetKeys = getRedisBudgetKeys(budgetScope)
 
-  const zaddArgs: Array<number | string> = []
-  let batchBytes = 0
-  for (const envelope of envelopes) {
+  /*
+    Redis measures a member in UTF-8 bytes, so the ceiling has to be measured the same
+    way — `String.length` counts UTF-16 units and under-reports every non-ASCII frame,
+    which would let a batch past a check the Lua then applies differently.
+  */
+  const members = envelopes.map((envelope) => {
     const member = JSON.stringify(envelope)
-    batchBytes += member.length
-    zaddArgs.push(envelope.seq, member)
-  }
+    return { seq: envelope.seq, member, bytes: Buffer.byteLength(member, 'utf8') }
+  })
 
   /*
-    A single batch past the per-write ceiling can never land, and retrying it would
-    stall every later batch behind it. Refuse it the same way the budget would.
+    Split on the per-write ceiling rather than refusing the whole batch: a flush carries
+    whatever accumulated since the last one, so an ordinary run of large frames can exceed
+    the ceiling collectively while every frame is individually writable. Refusing that
+    batch would stop replay persistence for the rest of the stream over a batching
+    artefact. Chunks are written in sequence order, so the stored cursor stays monotonic.
   */
-  if (batchBytes > limits.maxSingleWriteBytes) {
-    const refusal: RedisBudgetRefusal = {
-      resource: 'owner_redis_bytes',
-      currentBytes: 0,
-      limitBytes: limits.maxSingleWriteBytes,
-      attemptedBytes: batchBytes,
+  const chunks: Array<{ members: typeof members; bytes: number }> = []
+  for (const entry of members) {
+    const last = chunks[chunks.length - 1]
+    if (!last || last.bytes + entry.bytes > limits.maxSingleWriteBytes) {
+      chunks.push({ members: [entry], bytes: entry.bytes })
+    } else {
+      last.members.push(entry)
+      last.bytes += entry.bytes
     }
-    logRedisBudgetRefusal(refusal, { operation: 'append_event', scope: budgetScope, logger })
-    return { persisted: false, refusal }
   }
 
-  const result = await withRedisRetry({ operation: 'append_event', streamId }, async (redis) =>
-    redis.eval(
-      APPEND_EVENTS_SCRIPT,
-      2 + budgetKeys.length,
-      getEventsKey(streamId),
-      getSeqKey(streamId),
-      ...budgetKeys,
-      config.ttlSeconds,
-      config.eventLimit,
-      limits.maxOwnerBytes,
-      limits.maxUserBytes,
-      limits.ttlSeconds,
-      String(envelopes[envelopes.length - 1].seq),
-      ...zaddArgs
+  for (const chunk of chunks) {
+    /*
+      A single frame past the ceiling can never land, and retrying it would stall every
+      later batch behind it. Refuse it the same way the budget would.
+    */
+    if (chunk.bytes > limits.maxSingleWriteBytes) {
+      const refusal: RedisBudgetRefusal = {
+        resource: 'owner_redis_bytes',
+        currentBytes: 0,
+        limitBytes: limits.maxSingleWriteBytes,
+        attemptedBytes: chunk.bytes,
+      }
+      logRedisBudgetRefusal(refusal, { operation: 'append_event', scope: budgetScope, logger })
+      return { persisted: false, refusal }
+    }
+
+    const zaddArgs: Array<number | string> = []
+    for (const entry of chunk.members) {
+      zaddArgs.push(entry.seq, entry.member)
+    }
+
+    const result = await withRedisRetry({ operation: 'append_event', streamId }, async (redis) =>
+      redis.eval(
+        APPEND_EVENTS_SCRIPT,
+        2 + budgetKeys.length,
+        getEventsKey(streamId),
+        getSeqKey(streamId),
+        ...budgetKeys,
+        config.ttlSeconds,
+        config.eventLimit,
+        limits.maxOwnerBytes,
+        limits.maxUserBytes,
+        limits.ttlSeconds,
+        String(chunk.members[chunk.members.length - 1].seq),
+        ...zaddArgs
+      )
     )
-  )
 
-  const refusal = parseRedisBudgetRefusal(result, batchBytes, limits)
-  if (refusal) {
-    logRedisBudgetRefusal(refusal, { operation: 'append_event', scope: budgetScope, logger })
-    return { persisted: false, refusal }
+    const refusal = parseRedisBudgetRefusal(result, chunk.bytes, limits)
+    if (refusal) {
+      logRedisBudgetRefusal(refusal, { operation: 'append_event', scope: budgetScope, logger })
+      return { persisted: false, refusal }
+    }
   }
 
   return { persisted: true }

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

@@ -322,4 +322,58 @@ describe('StreamWriter', () => {
       userId: 'user-7',
     })
   })
+
+  it('does not persist a batch queued while an earlier append was already refusing', async () => {
+    vi.useFakeTimers()
+    let releaseFirst: () => void = () => {}
+    appendEvents
+      .mockImplementationOnce(
+        () =>
+          new Promise((resolve) => {
+            releaseFirst = () =>
+              resolve({
+                persisted: false,
+                refusal: {
+                  resource: 'owner_redis_bytes',
+                  currentBytes: 1,
+                  limitBytes: 1,
+                  attemptedBytes: 1,
+                },
+              })
+          })
+      )
+      .mockResolvedValue({ persisted: true })
+
+    const writer = new StreamWriter({
+      streamId: 'stream-1',
+      chatId: 'chat-1',
+      requestId: 'req-1',
+    })
+    const controller = {
+      enqueue: vi.fn(),
+      close: vi.fn(),
+    } as unknown as ReadableStreamDefaultController
+    writer.attach(controller)
+
+    await writer.publish({
+      type: MothershipStreamV1EventType.text,
+      payload: { channel: MothershipStreamV1TextChannel.assistant, text: 'one' },
+    })
+    await vi.advanceTimersByTimeAsync(15)
+
+    // Queued while the first append is still in flight, so the enqueue-time check cannot see the
+    // refusal about to latch. Persisting it would leave replay holding a later event but not the
+    // refused one — a hole a resuming client cannot detect.
+    await writer.publish({
+      type: MothershipStreamV1EventType.text,
+      payload: { channel: MothershipStreamV1TextChannel.assistant, text: 'two' },
+    })
+    await vi.advanceTimersByTimeAsync(15)
+
+    releaseFirst()
+    await writer.close()
+
+    expect(writer.persistenceStopped).toBe(true)
+    expect(appendEvents).toHaveBeenCalledTimes(1)
+  })
 })

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

@@ -22,6 +22,9 @@ export interface StreamWriterOptions {
   keepaliveMs?: number
 }
 
+/** Result used when the soft stop is already latched, so no further append is attempted. */
+const PERSISTENCE_ALREADY_STOPPED = { persisted: true } as const
+
 export class StreamWriter {
   private readonly streamId: string
   private readonly chatId: string | undefined
@@ -191,10 +194,18 @@ export class StreamWriter {
     this.persistenceTail = this.persistenceTail
       .catch(() => undefined)
       .then(() =>
-        appendEvents(batch, {
-          streamId: this.streamId,
-          ...(this.userId ? { userId: this.userId } : {}),
-        })
+        /*
+          Re-checked here, not only at enqueue: a batch queued while an earlier append was
+          in flight would otherwise land after that append had already stopped persistence,
+          leaving a replay that holds later events but not the refused ones — a hole a
+          resuming client cannot detect.
+        */
+        this._persistenceStopped
+          ? PERSISTENCE_ALREADY_STOPPED
+          : appendEvents(batch, {
+              streamId: this.streamId,
+              ...(this.userId ? { userId: this.userId } : {}),
+            })
       )
       .then((result) => {
         this.lastPersistenceError = null

apps/sim/lib/realtime/event-log.ts

@@ -198,12 +198,13 @@ export async function appendEvent<E extends EventLogEntry>(
           stream.events = stream.events.slice(-config.cap)
         }
         if (config.maxBytes > 0) {
-          let bytes = stream.events.reduce(
-            (total, event) => total + JSON.stringify(event).length,
-            0
-          )
+          // UTF-8 bytes, so this path bounds a stream identically to the Lua's `string.len`;
+          // `String.length` counts UTF-16 units and under-reports every non-ASCII event.
+          const entryBytes = (event: EventLogEntry) =>
+            Buffer.byteLength(JSON.stringify(event), 'utf8')
+          let bytes = stream.events.reduce((total, event) => total + entryBytes(event), 0)
           while (bytes > config.maxBytes && stream.events.length > 1) {
-            bytes -= JSON.stringify(stream.events[0]).length
+            bytes -= entryBytes(stream.events[0])
             stream.events = stream.events.slice(1)
           }
         }