fix(copilot): stream append previews as deltas instead of a snapshot per token

An `append` preview is `existingContent + streamed`, so forcing a full snapshot on
every emission re-sent the entire file once per streamed token. Cost is
`O(file x tokens)` into the stream buffer, whose only trim is a 100,000-member
rank cap — a count cap does not bound a member whose size is the file. One 250 KB
file cost gigabytes.

The forced snapshot was correct when it was written: the preview consumer could
not merge a delta. It gained that ability two months later and the producer was
never relaxed, so the branch has been dead weight since — `update` has shipped
deltas through the same reducer ever since, and `deriveFilePreviewSession` is the
only thing in the app that reads `contentMode`.

Removing it changes no rendered text. The two conditions that make a snapshot
necessary still force one: a base that diverges mid-stream fails
`startsWith`, and the checkpoint interval still emits a recoverable full snapshot
so no delta chain runs longer than a second.

`file-preview-append-roundtrip.test.ts` drives the real producer into the real
consumer and asserts the reconstructed text is byte-identical, including at
token-scale chunking over a 250 KB base, across a diverging base, and under
duplicate delivery. Both halves were verified to fail when the behaviour they pin
is reverted.

Measured against the session that prompted this: 2.29 GiB -> 55 MB.

Author: waleedlatif1

.claude/rules/sim-redis-payloads.md

@@ -0,0 +1,102 @@
+# Redis Payloads
+
+Rules for anything written to the shared Redis. `sim-caching.md` governs *in-process*
+caches; this governs the cluster every process shares.
+
+Redis is one fixed-size box with `maxmemory-policy allkeys-lru` and no AOF. There is no
+per-feature quota and no isolation: a key that grows without a bound does not fail its
+own feature, it evicts *somebody else's* keys. Idempotency records, usage reservations,
+cancellation flags and distributed locks all live there, so an unbounded write is not a
+performance problem — it is a correctness problem in a feature you did not touch.
+
+## Cap the bytes, never the count
+
+A count cap is not a memory bound. `ZREMRANGEBYRANK(key, 0, -limit-1)` bounds how many
+members a key holds; it says nothing about how large each member is. A key holding 400
+members of 250 KB is 100 MB and passes a 100,000-member cap untouched.
+
+Cap the dimension you are actually short of:
+
+```ts
+// ✗ Bad — bounds cardinality, not memory
+pipeline.zadd(key, seq, JSON.stringify(envelope))
+pipeline.zremrangebyrank(key, 0, -EVENT_LIMIT - 1)
+
+// ✓ Good — refuse the write that would exceed the byte budget
+const bytes = Buffer.byteLength(payload)
+const allowed = await reserveRedisBudget({ ownerId, category, bytes, operation })
+if (!allowed) return  // drop, offload, or degrade — but do not write
+```
+
+A count cap on top of a byte cap is fine and often useful. A count cap *instead of* one
+is the bug this rule exists to prevent — it reads like a guard in review and holds
+nothing.
+
+## Use the budget that already exists
+
+`lib/execution/redis-budget.server.ts` is the sanctioned implementation. Three tiers,
+enforced atomically in the same Lua as the write:
+
+| Limit | Value | Stops |
+|---|---|---|
+| `maxSingleWriteBytes` | 8 MB | one pathological value |
+| `maxExecutionBytes` | 64 MB | one runaway session |
+| `maxUserBytes` | 256 MB | one runaway tenant |
+
+Pair it with `lib/execution/payloads/large-value-ref.ts`: above the threshold the value
+goes to blob storage and Redis holds a typed reference instead. Do not write a second
+budget module — extend this one to your owner kind.
+
+Any new multi-MB Redis writer must go through it. Today only `event-buffer.ts` and
+`user-file-base64.server.ts` do, which is why the copilot stream buffer, the table event
+log and the file-doc stream each independently grew a key with no byte bound.
+
+## Never store a value whose size scales with something you do not control
+
+Before writing, name what bounds the value. If the answer is "the size of a user's file",
+"the number of tokens the model emits", or "how long the session runs", it is unbounded
+and belongs in blob storage behind a reference.
+
+The failure mode to watch for is **re-serialization per event**: emitting a full snapshot
+of a growing document on every streamed chunk makes total bytes `O(document × chunks)` —
+quadratic in the document, and the client already has every prior byte. Send the delta the
+producer usually already computed. A snapshot is a recovery affordance; send it on a
+timer, not per chunk.
+
+```ts
+// ✗ Bad — every chunk carries the whole document
+emit({ content: nextText, contentMode: 'snapshot' })
+
+// ✓ Good — snapshot on an interval, delta in between
+const isCheckpoint = now - lastSnapshotAt >= CHECKPOINT_INTERVAL_MS
+emit(
+  isCheckpoint || !nextText.startsWith(previousText)
+    ? { content: nextText, contentMode: 'snapshot' }
+    : { content: nextText.slice(previousText.length), contentMode: 'delta' }
+)
+```
+
+Throttles must cover every branch that can emit. A throttle written for one operation and
+keyed on that operation's name silently exempts every sibling added later — bound the
+emission, not the operation.
+
+## Sliding TTLs do not expire
+
+`EXPIRE` on every write means the key dies a TTL after the *last* write, not after the
+first. For an append-only key under continuous traffic that is never. A sliding TTL is
+correct for a session that should outlive its own idle gaps; it is wrong as the only
+bound on a key that grows. Pair it with a byte budget, or set the TTL once on creation.
+
+Cleanup that runs only on a clean close is not a bound either — the process that dies
+mid-stream is exactly the one holding the largest key.
+
+## Reviewer checklist
+
+For any new or changed Redis write:
+
+- What is the largest this value can be? If you cannot state it in bytes, it is unbounded.
+- Is the cap on bytes, or only on entries?
+- Does the TTL slide? If so, what stops the key growing between writes?
+- Does the write go through `redis-budget.server.ts`? If not, why is this one exempt?
+- If it appends per streamed event, does it send deltas, or re-send the whole state?
+- If this key grew 100× tomorrow, which *other* feature breaks first?

apps/sim/lib/copilot/request/go/file-preview-adapter.ts

@@ -326,7 +326,6 @@ export function buildPreviewContentUpdate(
     previousText.length === 0 ||
     !nextText.startsWith(previousText) ||
     operation === 'patch' ||
-    operation === 'append' ||
     now - lastSnapshotAt >= DELTA_PREVIEW_CHECKPOINT_INTERVAL_MS
 
   if (shouldForceSnapshot) {

apps/sim/lib/copilot/request/go/file-preview-append-roundtrip.test.ts

@@ -0,0 +1,174 @@
+/**
+ * @vitest-environment node
+ */
+import { describe, expect, it } from 'vitest'
+import { buildPreviewContentUpdate } from '@/lib/copilot/request/go/file-preview-adapter'
+import type { FilePreviewSession } from '@/lib/copilot/request/session/file-preview-session-contract'
+import { deriveFilePreviewSession } from '@/app/workspace/[workspaceId]/home/hooks/preview/apply-file-preview-phase'
+
+const CHECKPOINT_MS = 1_000
+
+/**
+ * Producer -> consumer round trip for an `append` preview.
+ *
+ * `buildPreviewContentUpdate` decides snapshot vs delta; `deriveFilePreviewSession` is
+ * the only thing in the app that reads `contentMode`. Emitting deltas instead of a full
+ * snapshot per chunk is only safe if replaying what the producer emits reconstructs the
+ * text exactly — this drives the real functions against each other and checks that,
+ * rather than reasoning about it.
+ */
+function roundTrip(
+  chunks: string[],
+  base: string,
+  msPerChunk: number
+): { rendered: string; expected: string; snapshots: number; deltas: number } {
+  let lastEmitted = ''
+  let lastSnapshotAt = 0
+  let now = 0
+  let streamed = ''
+  let session: FilePreviewSession | undefined
+  let version = 0
+  let snapshots = 0
+  let deltas = 0
+
+  for (const chunk of chunks) {
+    streamed += chunk
+    now += msPerChunk
+    const nextText = base.length > 0 ? `${base}\n${streamed}` : streamed
+    const update = buildPreviewContentUpdate(lastEmitted, nextText, lastSnapshotAt, now, 'append')
+    lastEmitted = nextText
+    lastSnapshotAt = update.lastSnapshotAt
+    version += 1
+    if (update.contentMode === 'snapshot') snapshots++
+    else deltas++
+
+    session = deriveFilePreviewSession(
+      session,
+      {
+        previewPhase: 'file_preview_content',
+        content: update.content,
+        contentMode: update.contentMode,
+        previewVersion: version,
+        toolCallId: 'tc_1',
+        toolName: 'prepare_file_edit',
+        fileName: 'notes.md',
+        operation: 'append',
+      } as never,
+      'stream_1',
+      new Date(now).toISOString()
+    )
+  }
+
+  return {
+    rendered: session?.previewText ?? '',
+    expected: base.length > 0 ? `${base}\n${streamed}` : streamed,
+    snapshots,
+    deltas,
+  }
+}
+
+function chunksOf(text: string, size: number): string[] {
+  const out: string[] = []
+  for (let i = 0; i < text.length; i += size) out.push(text.slice(i, i + size))
+  return out
+}
+
+describe('append preview round trip', () => {
+  it('reconstructs the exact text the user should see, and does it with deltas', () => {
+    const base = 'Existing file body.\nSecond line.'
+    const r = roundTrip(chunksOf('The appended paragraph goes here.', 4), base, 20)
+
+    expect(r.rendered).toBe(r.expected)
+    expect(r.deltas).toBeGreaterThan(0)
+  })
+
+  it('holds for realistic token-scale chunking on a large base file', () => {
+    const base = 'x'.repeat(250 * 1024)
+    const r = roundTrip(chunksOf('y'.repeat(4096), 10), base, 20)
+
+    expect(r.rendered).toBe(r.expected)
+    expect(r.rendered.length).toBe(250 * 1024 + 1 + 4096)
+  })
+
+  it('still emits a recoverable full snapshot on the checkpoint interval', () => {
+    // One chunk per 400ms crosses the 1s checkpoint repeatedly.
+    const r = roundTrip(chunksOf('abcdefghij', 1), 'base', 400)
+
+    expect(r.rendered).toBe(r.expected)
+    expect(r.snapshots).toBeGreaterThan(1)
+    expect(r.snapshots * CHECKPOINT_MS).toBeGreaterThan(0)
+  })
+
+  it('recovers exactly when the base file changes underneath the stream', () => {
+    // A divergent base must fall back to a snapshot, not a delta on stale text.
+    const first = buildPreviewContentUpdate('Old base\nabc', 'New base\nabcd', 100, 200, 'append')
+    expect(first.contentMode).toBe('snapshot')
+    expect(first.content).toBe('New base\nabcd')
+
+    const session = deriveFilePreviewSession(
+      undefined,
+      {
+        previewPhase: 'file_preview_content',
+        content: first.content,
+        contentMode: first.contentMode,
+        previewVersion: 1,
+        toolCallId: 'tc_1',
+        toolName: 'prepare_file_edit',
+        fileName: 'notes.md',
+        operation: 'append',
+      } as never,
+      'stream_1',
+      new Date().toISOString()
+    )
+    expect(session.previewText).toBe('New base\nabcd')
+  })
+
+  it('ignores a replayed event rather than double-appending its delta', () => {
+    const base = 'Base.'
+    const chunks = chunksOf('hello world', 3)
+    let lastEmitted = ''
+    let lastSnapshotAt = 0
+    let now = 0
+    let streamed = ''
+    let session: FilePreviewSession | undefined
+    let version = 0
+    const emitted: Array<{ content: string; contentMode: string; version: number }> = []
+
+    for (const chunk of chunks) {
+      streamed += chunk
+      now += 20
+      const u = buildPreviewContentUpdate(
+        lastEmitted,
+        `${base}\n${streamed}`,
+        lastSnapshotAt,
+        now,
+        'append'
+      )
+      lastEmitted = `${base}\n${streamed}`
+      lastSnapshotAt = u.lastSnapshotAt
+      version += 1
+      emitted.push({ content: u.content, contentMode: u.contentMode, version })
+    }
+
+    // Deliver every event twice, out of order for the duplicates.
+    for (const e of [...emitted, ...emitted]) {
+      session = deriveFilePreviewSession(
+        session,
+        {
+          previewPhase: 'file_preview_content',
+          content: e.content,
+          contentMode: e.contentMode,
+          previewVersion: e.version,
+          toolCallId: 'tc_1',
+          toolName: 'prepare_file_edit',
+          fileName: 'notes.md',
+          operation: 'append',
+        } as never,
+        'stream_1',
+        new Date().toISOString()
+      )
+    }
+
+    expect(session?.previewText).toBe(`${base}\n${streamed}`)
+  })
+})

apps/sim/lib/copilot/request/go/stream.test.ts

@@ -193,14 +193,39 @@ describe('copilot go stream helpers', () => {
     expect(extractEditContent('{"content":"tab\\tvalue"}')).toBe('tab\tvalue')
   })
 
-  it('emits full snapshots for append (sidebar viewer uses replace mode; no delta merge)', () => {
+  /**
+   * Append extends its own text, so it deltas like `update` does.
+   *
+   * It forced a snapshot per emission until the only consumer that could not merge a
+   * delta was gone — `apply-file-preview-phase.ts` has accumulated them since #4923.
+   * Because an append preview is `existingContent + streamed`, a snapshot per chunk
+   * re-sent the whole file on every streamed token, which is `O(file x tokens)` into
+   * the stream buffer: one 250 KB file cost gigabytes of Redis.
+   */
+  it('emits deltas for append when the preview extends the previous text', () => {
     expect(buildPreviewContentUpdate('hello', 'hello world', 100, 200, 'append')).toEqual({
-      content: 'hello world',
+      content: ' world',
+      contentMode: 'delta',
+      lastSnapshotAt: 100,
+    })
+  })
+
+  it('still snapshots an append whose base changed underneath it', () => {
+    expect(buildPreviewContentUpdate('hello', 'HELLO world', 100, 200, 'append')).toEqual({
+      content: 'HELLO world',
       contentMode: 'snapshot',
       lastSnapshotAt: 200,
     })
   })
 
+  it('still checkpoints an append with a full snapshot on the interval', () => {
+    expect(buildPreviewContentUpdate('hello', 'hello world', 0, 1_000, 'append')).toEqual({
+      content: 'hello world',
+      contentMode: 'snapshot',
+      lastSnapshotAt: 1_000,
+    })
+  })
+
   it('emits deltas for update when the preview extends the previous text', () => {
     expect(buildPreviewContentUpdate('hello', 'hello world', 100, 200, 'update')).toEqual({
       content: ' world',