fix(relay): stop the overflow marker from re-killing the link it protects

Round-1 review fixes on the P0 freeze work.

The control-lane overflow marker could reinstate the exact failure this P0
removes: dispatcher-client-writer closes the client when control-lane
admission fails, and admitControl is the only lane that returns an error, so
one marker per failing batch accumulated to the 256-frame/1MB bound and
dropped the link. Markers are now deduped to one outstanding per
(client, root), cleared on settle.

Chunking also defeated the renderer's per-payload directory dedupe -- events
are now stable-grouped by parent directory so one directory lands in one
chunk -- and the halving walk overshot the byte minimum ~1.7x while the fast
path paid three JSON encodes; both are fixed by publishing first and sizing
from a measured bytes-per-event estimate.

Agent-hook shedding now surrenders the blocking interactive prompt LAST
rather than first, so a degraded envelope cannot strand a pane at
state=waiting with no answerable question card.

The dropped-notification log now distinguishes over-capacity from producer
queue backpressure and no longer lets the first dropped method silence every
other producer for the life of the connection.

src/relay/agent-hook-envelope-publication.test.ts

@@ -131,40 +131,65 @@ describe('publishAgentHookEnvelope', () => {
     }
   })
 
-  it('sheds interactivePrompt next when dropping the assistant message is not enough', () => {
+  it('sheds subagents next when dropping the assistant message is not enough', () => {
     const primary = makeBoundedClient(16384)
     const dispatcher = new RelayDispatcher(primary.write, primary.options)
     try {
       publishAgentHookEnvelope(
         dispatcher,
-        makeEnvelope({ lastAssistantMessage: 8_000, interactivePrompt: 13_000, subagents: 2 })
+        makeEnvelope({ lastAssistantMessage: 8_000, interactivePrompt: 6_000, subagents: 40 })
       )
 
       const published = decodeEnvelopes(primary)
       expect(published).toHaveLength(1)
       expect(published[0].payload.lastAssistantMessage).toBeUndefined()
-      expect(published[0].payload.interactivePrompt).toBeUndefined()
-      expect(published[0].payload.subagents).toHaveLength(2)
+      expect(published[0].payload.subagents).toBeUndefined()
+      expect(published[0].payload.interactivePrompt).toBe('q'.repeat(6_000))
       expect(primary.closes).toBe(0)
     } finally {
       dispatcher.dispose()
     }
   })
 
-  it('sheds subagents last and still delivers the status', () => {
+  it('keeps the blocking question card when a waiting envelope has to shed', () => {
+    const primary = makeBoundedClient(16384)
+    const dispatcher = new RelayDispatcher(primary.write, primary.options)
+    try {
+      // A blocked pane is only answerable from web/mobile if interactivePrompt survives the ladder.
+      const envelope = makeEnvelope({
+        lastAssistantMessage: 8_000,
+        interactivePrompt: 4_000,
+        subagents: 40
+      })
+      envelope.payload.state = 'waiting'
+      publishAgentHookEnvelope(dispatcher, envelope)
+
+      const published = decodeEnvelopes(primary)
+      expect(published).toHaveLength(1)
+      expect(published[0].payload.state).toBe('waiting')
+      expect(published[0].payload.interactivePrompt).toBe('q'.repeat(4_000))
+      expect(published[0].payload.subagents).toBeUndefined()
+      expect(published[0].payload.lastAssistantMessage).toBeUndefined()
+      expect(primary.closes).toBe(0)
+    } finally {
+      dispatcher.dispose()
+    }
+  })
+
+  it('sheds interactivePrompt last and still delivers the status', () => {
     const primary = makeBoundedClient(16384)
     const dispatcher = new RelayDispatcher(primary.write, primary.options)
     try {
       publishAgentHookEnvelope(
         dispatcher,
-        makeEnvelope({ lastAssistantMessage: 6_000, interactivePrompt: 6_000, subagents: 40 })
+        makeEnvelope({ lastAssistantMessage: 6_000, interactivePrompt: 13_000, subagents: 2 })
       )
 
       const published = decodeEnvelopes(primary)
       expect(published).toHaveLength(1)
       expect(published[0].payload.lastAssistantMessage).toBeUndefined()
-      expect(published[0].payload.interactivePrompt).toBeUndefined()
       expect(published[0].payload.subagents).toBeUndefined()
+      expect(published[0].payload.interactivePrompt).toBeUndefined()
       expect(published[0].payload.state).toBe('working')
       expect(primary.closes).toBe(0)
     } finally {
@@ -172,6 +197,24 @@ describe('publishAgentHookEnvelope', () => {
     }
   })
 
+  it('measures the frame budget once up front and once per real shed', () => {
+    const primary = makeBoundedClient(16384)
+    const dispatcher = new RelayDispatcher(primary.write, primary.options)
+    const probe = vi.spyOn(dispatcher, 'producerEnvelopeBudget')
+    try {
+      publishAgentHookEnvelope(dispatcher, makeEnvelope({ lastAssistantMessage: 128 }))
+      expect(probe).toHaveBeenCalledTimes(1)
+
+      probe.mockClear()
+      // Two of the three ladder steps are absent here; skipping them must not cost an extra encode.
+      publishAgentHookEnvelope(dispatcher, makeEnvelope({ interactivePrompt: 20_000 }))
+      expect(probe).toHaveBeenCalledTimes(2)
+    } finally {
+      probe.mockRestore()
+      dispatcher.dispose()
+    }
+  })
+
   it('skips absent fields instead of consuming a shed step on them', () => {
     const primary = makeBoundedClient(16384)
     const dispatcher = new RelayDispatcher(primary.write, primary.options)

src/relay/agent-hook-envelope-publication.ts

@@ -6,7 +6,9 @@ import type { RelayDispatcher } from './dispatcher'
 
 // Why: shed the biggest, most reconstructible fields first — state/paneKey must survive or the pane
 // stays stuck on its last spinner, and an over-capacity frame is now dropped rather than sent.
-const SHED_ORDER = ['lastAssistantMessage', 'interactivePrompt', 'subagents'] as const
+// Why: interactivePrompt is last on purpose — a blocking question is load-bearing (without its card
+// the pane sits on `waiting` with no way to answer from web or mobile) while the roster is cosmetic.
+const SHED_ORDER = ['lastAssistantMessage', 'subagents', 'interactivePrompt'] as const
 
 function fitsProducerFrame(dispatcher: RelayDispatcher, envelope: AgentHookRelayEnvelope): boolean {
   return (
@@ -24,8 +26,10 @@ export function publishAgentHookEnvelope(
   envelope: AgentHookRelayEnvelope
 ): void {
   let candidate = envelope
+  // Why: re-encoding is the expensive part, so only re-measure after a field was actually dropped.
+  let fits = fitsProducerFrame(dispatcher, candidate)
   for (const field of SHED_ORDER) {
-    if (fitsProducerFrame(dispatcher, candidate)) {
+    if (fits) {
       break
     }
     if (candidate.payload[field] === undefined) {
@@ -35,6 +39,7 @@ export function publishAgentHookEnvelope(
     // would permanently strip the cached copy too.
     candidate = { ...candidate, payload: { ...candidate.payload } }
     delete candidate.payload[field]
+    fits = fitsProducerFrame(dispatcher, candidate)
   }
   dispatcher.notify(AGENT_HOOK_NOTIFICATION_METHOD, candidate as unknown as Record<string, unknown>)
 }

src/relay/dispatcher-capacity-degradation.test.ts

@@ -9,6 +9,17 @@ type BoundedClient = {
   write: (data: Buffer) => boolean
 }
 
+// Why: a sink that never accepts a write and never drains retains producer bytes, so the queue saturates
+// while every individual frame stays comfortably inside the per-frame capacity.
+function makeSaturatingClient(highWaterMark: number): BoundedClient {
+  const client = makeBoundedClient(highWaterMark)
+  client.write = (data: Buffer) => {
+    client.frames.push(Buffer.from(data))
+    return false
+  }
+  return client
+}
+
 function makeBoundedClient(highWaterMark: number): BoundedClient {
   const client: BoundedClient = {
     frames: [],
@@ -121,7 +132,7 @@ describe('RelayDispatcher bounded-capacity degradation', () => {
       expect(primary.frames).toHaveLength(0)
       expect(stderr).toHaveBeenCalledTimes(1)
       expect(stderr.mock.calls[0][0]).toMatch(
-        /^\[relay\] Dropped fs\.changed \(\d+B > producer capacity 12288B\)\n$/
+        /^\[relay\] Dropped fs\.changed \(\d+B > producer frame capacity 12288B\)\n$/
       )
 
       bounded.notify('pty.exit', { id: 'pty-1' })
@@ -153,6 +164,56 @@ describe('RelayDispatcher bounded-capacity degradation', () => {
     }
   })
 
+  it('distinguishes an over-capacity drop from a producer-queue-full drop on one client', () => {
+    const primary = makeSaturatingClient(4 * 1024 * 1024)
+    const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true)
+    const bounded = new RelayDispatcher(primary.write, primary.options)
+    try {
+      // The first 1.5MB frame is retained by the stalled sink; the second exceeds the 2MB producer queue.
+      bounded.notify('fs.changed', { events: ['x'.repeat(1_500_000)] })
+      expect(primary.frames).toHaveLength(1)
+      bounded.notify('fs.changed', { events: ['x'.repeat(1_500_000)] })
+      bounded.notify('fs.changed', { events: ['x'.repeat(5_000_000)] })
+
+      expect(primary.closes).toBe(0)
+      expect(stderr).toHaveBeenCalledTimes(2)
+      expect(stderr.mock.calls[0][0]).toMatch(
+        /^\[relay\] Dropped fs\.changed \(\d+B, producer queue full; frame capacity 4128768B\)\n$/
+      )
+      expect(stderr.mock.calls[1][0]).toMatch(
+        /^\[relay\] Dropped fs\.changed \(\d+B > producer frame capacity 4128768B\)\n$/
+      )
+
+      // Both classifications stay one-per-generation once logged.
+      bounded.notify('fs.changed', { events: ['x'.repeat(1_500_000)] })
+      bounded.notify('fs.changed', { events: ['x'.repeat(5_000_000)] })
+      expect(stderr).toHaveBeenCalledTimes(2)
+    } finally {
+      stderr.mockRestore()
+      bounded.dispose()
+    }
+  })
+
+  it('logs each dropped method once instead of letting the first method silence the rest', () => {
+    const primary = makeBoundedClient(16384)
+    const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true)
+    const bounded = new RelayDispatcher(primary.write, primary.options)
+    try {
+      const flood = { events: ['x'.repeat(20_000)] }
+      bounded.notify('fs.changed', flood)
+      bounded.notify('pty.data', flood)
+      bounded.notify('fs.changed', flood)
+      bounded.notify('pty.data', flood)
+
+      expect(stderr).toHaveBeenCalledTimes(2)
+      expect(stderr.mock.calls[0][0]).toContain('Dropped fs.changed')
+      expect(stderr.mock.calls[1][0]).toContain('Dropped pty.data')
+    } finally {
+      stderr.mockRestore()
+      bounded.dispose()
+    }
+  })
+
   it('publishProducerNotification publishes on the producer lane and never closes', () => {
     const primary = makeBoundedClient(65536)
     const secondary = makeBoundedClient(16384)

src/relay/dispatcher.ts

@@ -69,10 +69,19 @@ type RelayClient = {
   highestReceivedSeq: number
   generation: number
   closed: boolean
-  droppedNotificationLogGeneration: number | null
+  droppedNotificationLog: DroppedProducerNotificationLog | null
   sessionIdentity: RelayClientSessionIdentity
 }
 
+// Why: the log key set is rebuilt per generation, but a producer minting synthetic method names would still
+// grow it inside one generation — cap it well above the fixed relay method vocabulary.
+const DROPPED_NOTIFICATION_LOG_KEY_LIMIT = 64
+
+type DroppedProducerNotificationLog = {
+  generation: number
+  loggedKeys: Set<string>
+}
+
 type PendingRelayRequest = {
   resolve: (result: unknown) => void
   reject: (error: Error) => void
@@ -467,18 +476,30 @@ export class RelayDispatcher {
     )
   }
 
-  // Why: one line per connection generation — a flooding producer retries every batch and would spam stderr.
+  // Why: one line per generation, method and drop reason — a flooding producer retries every batch and would
+  // spam stderr, but a transient queue-full drop must not consume the slot a real over-capacity drop needs.
   private logDroppedProducerNotification(
     client: RelayClient,
     method: string,
     msg: JsonRpcNotification
   ): void {
-    if (client.droppedNotificationLogGeneration === client.generation) {
+    const bytes = this.estimateFrameBytes(msg)
+    const capacity = client.writer.producerFrameCapacity
+    const overCapacity = bytes > capacity
+    const key = `${method}:${overCapacity ? 'over-capacity' : 'queue-full'}`
+    let log = client.droppedNotificationLog
+    if (!log || log.generation !== client.generation) {
+      log = { generation: client.generation, loggedKeys: new Set() }
+      client.droppedNotificationLog = log
+    }
+    if (log.loggedKeys.has(key) || log.loggedKeys.size >= DROPPED_NOTIFICATION_LOG_KEY_LIMIT) {
       return
     }
-    client.droppedNotificationLogGeneration = client.generation
+    log.loggedKeys.add(key)
     process.stderr.write(
-      `[relay] Dropped ${method} (${this.estimateFrameBytes(msg)}B > producer capacity ${client.writer.producerFrameCapacity}B)\n`
+      overCapacity
+        ? `[relay] Dropped ${method} (${bytes}B > producer frame capacity ${capacity}B)\n`
+        : `[relay] Dropped ${method} (${bytes}B, producer queue full; frame capacity ${capacity}B)\n`
     )
   }
 
@@ -674,7 +695,7 @@ export class RelayDispatcher {
       highestReceivedSeq: 0,
       generation: 0,
       closed: false,
-      droppedNotificationLogGeneration: null,
+      droppedNotificationLog: null,
       sessionIdentity: sessionIdentity ?? {
         principal: `unproved:${id}`,
         authenticated: false,

src/relay/fs-handler-stream.test.ts

@@ -194,9 +194,12 @@ describe('FsHandler readFileStream', () => {
     expect(frames[lastChunkIndex].clientId).toBe(7)
   })
 
-  it('delivers fs.streamEnd on the control lane and releases the stream when the producer lane is saturated', async () => {
-    const filePath = path.join(tmpDir, 'saturated.png')
+  // Real saturation is not constructible: the fixed-bulk lane admits a chunk only while producerBytes is 0,
+  // so a real backlog parks the chunk pump long before the terminal frame — hence the lane-tagging mock.
+  it('orders fs.streamEnd after the final chunk on the control lane and releases the stream', async () => {
+    const filePath = path.join(tmpDir, 'stream-end-release.png')
     writeFileSync(filePath, Buffer.alloc(300 * 1024, 0x42))
+    // Producer-lane frames now divert to the drop log, so an empty log proves no stream frame took that lane.
     dispatcher.saturateProducerLane()
 
     const meta = (await dispatcher.callRequest(
@@ -209,6 +212,13 @@ describe('FsHandler readFileStream', () => {
     expect(dispatcher._droppedProducerFrames).toHaveLength(0)
     expect(collectStream(dispatcher).end).toEqual({ streamId: meta.streamId })
 
+    const frames = dispatcher._notifications
+    const endIndex = frames.findIndex((n) => n.method === 'fs.streamEnd')
+    const lastChunkIndex = frames.map((n) => n.method).lastIndexOf('fs.streamChunk')
+    expect(lastChunkIndex).toBeGreaterThanOrEqual(0)
+    expect(endIndex).toBe(lastChunkIndex + 1)
+    expect(frames[endIndex].lane).toBe('control')
+
     const registry = (handler as unknown as { streamRegistry: { size(): number } }).streamRegistry
     await waitFor(() => registry.size() === 0)
   })

src/relay/relay-filesystem-watch-registry.test.ts

@@ -72,6 +72,19 @@ function createDispatcher() {
     fsChanged,
     notify: vi.fn(),
     notifyClient: vi.fn(record),
+    tryNotifyClient: vi.fn(
+      (
+        clientId: number,
+        method: string,
+        params?: Record<string, unknown>,
+        onSettled?: (result: { ok: true } | { ok: false; error: Error }) => void
+      ) => {
+        record(clientId, method, params)
+        // A healthy sink settles as the frame is written, which releases the emitter's outstanding-marker slot.
+        onSettled?.({ ok: true })
+        return true
+      }
+    ),
     activeClientIds: vi.fn(() => [1]),
     producerEnvelopeBudget: vi.fn(() => Number.MAX_SAFE_INTEGER),
     publishProducerNotification: vi.fn(

src/relay/relay-oversized-notification-survival.test.ts

@@ -58,17 +58,6 @@ function buildMaximalHookEnvelope(): Record<string, unknown> {
   } as unknown as Record<string, unknown>
 }
 
-/** Pure measurement shape only: the real emitter never emits `watchId`. */
-function buildWatcherBatch(count: number): Record<string, unknown> {
-  return {
-    watchId: 'watch-1',
-    events: Array.from({ length: count }, (_, i) => ({
-      type: 'update',
-      path: `${WATCHER_ROOT}/node_modules/.vite/deps/chunk-${String(i).padStart(6, '0')}.js`
-    }))
-  }
-}
-
 function watcherProcessEvents(count: number): WatcherProcessEvent[] {
   return Array.from({ length: count }, (_, i) => ({
     type: 'update',
@@ -76,6 +65,16 @@ function watcherProcessEvents(count: number): WatcherProcessEvent[] {
   })) as unknown as WatcherProcessEvent[]
 }
 
+/** The exact `fs.changed` params emitRelayWatcherEvents publishes, so the pinned sizes measure reality. */
+function buildWatcherBatch(count: number): Record<string, unknown> {
+  return {
+    events: watcherProcessEvents(count).map((event) => ({
+      kind: event.type,
+      absolutePath: event.path
+    }))
+  }
+}
+
 function frameBytes(method: string, params: Record<string, unknown>): number {
   return encodeJsonRpcFrame({ jsonrpc: '2.0', method, params }, 1, 0).length
 }
@@ -102,6 +101,20 @@ function createDispatcher(hwm: number, onClose: () => void, sink: Buffer[]): Rel
 }
 
 describe('relay oversized notification survival', () => {
+  // Without this the pinned byte counts below could measure a payload the emitter never sends.
+  it('measures the params emitRelayWatcherEvents actually publishes', () => {
+    const sink: Buffer[] = []
+    const dispatcher = createDispatcher(NODE22_HWM, () => {}, sink)
+
+    try {
+      emitRelayWatcherEvents(dispatcher, WATCHER_ROOT, false, watcherProcessEvents(1))
+      expect(sink).toHaveLength(1)
+      expect(frameParams(sink[0])).toEqual(buildWatcherBatch(1))
+    } finally {
+      dispatcher.dispose()
+    }
+  })
+
   it('measures both producers against both Node tiers', () => {
     const hookBytes = frameBytes(AGENT_HOOK_NOTIFICATION_METHOD, buildMaximalHookEnvelope())
     const fullBatchBytes = frameBytes('fs.changed', buildWatcherBatch(MAX_BATCHED_WATCHER_EVENTS))
@@ -124,9 +137,9 @@ describe('relay oversized notification survival', () => {
     expect(capacityFor(NODE22_HWM)).toBe(49152)
     expect(capacityFor(NODE21_HWM)).toBe(12288)
     expect(hookBytes).toBe(29894)
-    expect(fullBatchBytes).toBe(425094)
-    expect(trip(capacityFor(NODE22_HWM))).toBe(577)
-    expect(trip(capacityFor(NODE21_HWM))).toBe(143)
+    expect(fullBatchBytes).toBe(465074)
+    expect(trip(capacityFor(NODE22_HWM))).toBe(527)
+    expect(trip(capacityFor(NODE21_HWM))).toBe(131)
 
     // agent.hook: fits a modern remote, over-cap on an older one.
     expect(hookBytes).toBeLessThan(capacityFor(NODE22_HWM))

src/relay/relay-watcher-event-emitter.test.ts

@@ -1,4 +1,5 @@
 import { describe, expect, it } from 'vitest'
+import type { WatcherProcessEvent } from '../main/ipc/parcel-watcher-process-protocol'
 import { RelayDispatcher } from './dispatcher'
 import type { RelayClientSinkOptions, RelayClientWrite } from './dispatcher-writer-sink'
 import { HEADER_LENGTH, parseJsonRpcMessage } from './protocol'
@@ -78,6 +79,25 @@ function watcherBatch(count: number): { type: 'create'; path: string; isDirector
   }))
 }
 
+function parentDir(absolutePath: string): string {
+  return absolutePath.slice(0, absolutePath.lastIndexOf('/'))
+}
+
+// Fills the 2 MB producer queue so every subsequent watcher batch is rejected on the producer lane.
+function saturateProducerQueue(dispatcher: RelayDispatcher): void {
+  let admitted = 0
+  for (const chunk of [40_000, 1_000, 1]) {
+    while (
+      admitted < 5_000 &&
+      dispatcher.tryNotifyPtyData({ paneId: 'pane', data: 'x'.repeat(chunk) })
+    ) {
+      admitted += 1
+    }
+  }
+  expect(admitted).toBeGreaterThan(0)
+  expect(admitted).toBeLessThan(5_000)
+}
+
 describe('relay watcher writer admission', () => {
   it('keeps watcher batches on the producer lane while overflow markers ride the control lane', () => {
     const sink = createRecordingSink()
@@ -158,17 +178,7 @@ describe('relay watcher batch chunking', () => {
 
     try {
       // Fill the 2 MB producer queue down to a residue smaller than a one-event fs.changed frame.
-      let admitted = 0
-      for (const chunk of [40_000, 1_000, 1]) {
-        while (
-          admitted < 5_000 &&
-          dispatcher.tryNotifyPtyData({ paneId: 'pane', data: 'x'.repeat(chunk) })
-        ) {
-          admitted += 1
-        }
-      }
-      expect(admitted).toBeGreaterThan(0)
-      expect(admitted).toBeLessThan(5_000)
+      saturateProducerQueue(dispatcher)
 
       emitRelayWatcherEvents(dispatcher, '/workspace', false, [
         { type: 'create', path: '/workspace/first', isDirectory: false }
@@ -186,6 +196,111 @@ describe('relay watcher batch chunking', () => {
     }
   })
 
+  it('keeps at most one outstanding overflow marker per root without closing the client', () => {
+    const sink = createRecordingSink(65536)
+    sink.blockNextWrite()
+    const dispatcher = new RelayDispatcher(sink.write, sink.options)
+
+    try {
+      saturateProducerQueue(dispatcher)
+
+      // More rejected batches than the 256-frame control queue can hold: one marker each would close the link.
+      for (let index = 0; index < 300; index += 1) {
+        emitRelayWatcherEvents(dispatcher, '/workspace', false, [
+          { type: 'create', path: `/workspace/file-${index}`, isDirectory: false }
+        ])
+      }
+      // A second root must still get its own marker, or that tree silently desyncs.
+      emitRelayWatcherEvents(dispatcher, '/other', false, [
+        { type: 'create', path: '/other/file', isDirectory: false }
+      ])
+      expect(sink.closes()).toBe(0)
+
+      sink.drainWaiters.shift()?.()
+      const markers = sink.frames
+        .filter((frame) => frameMethod(frame) === 'fs.changed')
+        .flatMap(frameEvents)
+      expect(markers).toEqual([
+        { kind: 'overflow', absolutePath: '/workspace' },
+        { kind: 'overflow', absolutePath: '/other' }
+      ])
+      expect(sink.closes()).toBe(0)
+
+      // The slot clears once the marker drains, so a later batch can still resync the same root.
+      emitRelayWatcherEvents(dispatcher, '/workspace', false, [
+        { type: 'create', path: '/workspace/later', isDirectory: false }
+      ])
+      expect(
+        sink.frames.filter((frame) => frameMethod(frame) === 'fs.changed').flatMap(frameEvents)
+      ).toHaveLength(3)
+    } finally {
+      dispatcher.dispose()
+    }
+  })
+
+  it('keeps one directory in one chunk when the batch interleaves directories', () => {
+    const sink = createRecordingSink(16384)
+    const dispatcher = new RelayDispatcher(sink.write, sink.options)
+    const directoryCount = 8
+    const replacedPath = '/workspace/project/src/module-7/component-0.tsx'
+    const events: WatcherProcessEvent[] = [
+      ...Array.from({ length: 200 }, (_unused, index) => ({
+        type: 'create' as const,
+        path: `/workspace/project/src/module-${index % directoryCount}/component-${Math.floor(index / directoryCount)}.tsx`,
+        isDirectory: false
+      })),
+      { type: 'delete' as const, path: replacedPath }
+    ]
+
+    try {
+      emitRelayWatcherEvents(dispatcher, '/workspace', false, events)
+
+      expect(sink.frames.length).toBeGreaterThan(1)
+      const frameDirs = sink.frames.map(
+        (frame) => new Set(frameEvents(frame).map((event) => parentDir(event.absolutePath)))
+      )
+      const dirs = Array.from(new Set(events.map((event) => parentDir(event.path))))
+      expect(dirs).toHaveLength(directoryCount)
+      // Each directory in one frame; the renderer dedupes refreshes per payload, so extra frames = extra readDir RPCs.
+      const refreshes = frameDirs.reduce((total, frameDir) => total + frameDir.size, 0)
+      expect(refreshes).toBeLessThanOrEqual(directoryCount + sink.frames.length - 1)
+
+      const delivered = sink.frames.flatMap(frameEvents)
+      for (const dir of dirs) {
+        const positions = delivered
+          .map((event, index) => ({ event, index }))
+          .filter((entry) => parentDir(entry.event.absolutePath) === dir)
+          .map((entry) => entry.index)
+        expect(positions.at(-1)! - positions[0]).toBe(positions.length - 1)
+      }
+      // Stable grouping keeps per-path order: the create still precedes the delete of the same path.
+      expect(
+        delivered.filter((event) => event.absolutePath === replacedPath).map((event) => event.kind)
+      ).toEqual(['create', 'delete'])
+      expect(delivered).toHaveLength(events.length)
+    } finally {
+      dispatcher.dispose()
+    }
+  })
+
+  it('sizes chunks near the byte minimum instead of overshooting it', () => {
+    const sink = createRecordingSink(16384)
+    const capacity = 12288
+    const dispatcher = new RelayDispatcher(sink.write, sink.options)
+
+    try {
+      emitRelayWatcherEvents(dispatcher, '/workspace', false, watcherBatch(5000))
+
+      const totalBytes = sink.frames.reduce((total, frame) => total + frame.length, 0)
+      expect(sink.frames.length).toBeLessThanOrEqual(Math.ceil(totalBytes / capacity) + 1)
+      for (const frame of sink.frames.slice(0, -1)) {
+        expect(frame.length).toBeGreaterThanOrEqual(capacity * 0.9)
+      }
+    } finally {
+      dispatcher.dispose()
+    }
+  })
+
   it('chunks per client so a small sink never degrades a healthy one', () => {
     const healthy = createRecordingSink(65536)
     const small = createRecordingSink(16384)