fix(daemon): pause producers when stream backlogs grow

docs/audits/daemon-stream-retention/README.md

@@ -0,0 +1,34 @@
+# Daemon stream retention reproduction
+
+Run from the worktree root after installing dependencies:
+
+```sh
+ORCA_BACKGROUND_LAUNCH=1 node docs/audits/daemon-stream-retention/reproduce.mjs
+```
+
+The script bundles the current `DaemonStreamDataBatcher` with esbuild, then uses real loopback TCP sockets. It pauses the reader, feeds at most 8 MiB, honors the producer-pause callback, resumes the reader, and closes every socket in `finally`. It starts no PTY, agent, or application window. Output includes the bundle hash, queued payload sizes, RSS, and drain outcome.
+
+Visible 64 KiB and 1 KiB producers should pause within the backlog budget and resume after both queues drain. A hidden 1 KiB producer should keep running while the existing keep-tail policy bounds its held output. RSS includes allocator/GC timing; queue measurements are the direct evidence.
+
+`after.json` records the fixed run: the visible bulk producer paused after 2,686,976 characters with 131,300 socket bytes and 2,031,616 held characters; the visible small-write producer paused with 3,401,200 socket bytes. The hidden producer processed all 8 MiB with 131,300 socket bytes and 667,648 held characters. Every case resumed and drained both queues to zero.
+
+## Preserved measurements before the fix
+
+`before-1mib-chunks.json` and `before-64kib-chunks.json` were captured before adding producer backpressure, with a real paused reader and 96 MiB of input. The latter includes the original compiled bundle hash and a hidden-session comparison.
+
+| Input  | Socket buffered bytes | Held characters |
+| ------ | --------------------: | --------------: |
+| 32 MiB |               131,321 |      32,636,928 |
+| 64 MiB |            32,809,583 |      33,554,432 |
+| 96 MiB |            66,406,511 |      33,554,432 |
+
+The last 32 MiB retained another 33,596,928 queued payload bytes. The held queue plateaued at 32 MiB while the socket queue grew at 1.00127 bytes per produced ASCII byte. Resuming the reader drained both queues in 74 ms. Sampled RSS reached 1.38 GB with 64 KiB chunks; this includes string coalescing/slicing and GC effects.
+
+The original measurement's build command was:
+
+```sh
+ORCA_BACKGROUND_LAUNCH=1 node_modules/.bin/esbuild src/main/daemon/daemon-stream-data-batcher.ts --bundle --platform=node --format=esm --outfile=notes/daemon-stream-retention/daemon-stream-data-batcher.mjs
+ORCA_BACKGROUND_LAUNCH=1 node notes/daemon-stream-retention/compare.mjs
+```
+
+This proves a current stalled-consumer retaining path. It does not attribute issue #19831's whole-system memory total to this path; that report lacks per-process measurements.

docs/audits/daemon-stream-retention/after.json

@@ -0,0 +1,46 @@
+{
+  "node": "v26.6.0",
+  "platform": "darwin",
+  "bundleSha256": "7c63ca547aa62570878664f9e0504a8747f4daec2daf7c3cffcfce65ebede6bd",
+  "cases": [
+    {
+      "chunkSize": 65536,
+      "hidden": false,
+      "stalled": {
+        "producedChars": 2686976,
+        "producerPaused": true,
+        "socketBufferedBytes": 131300,
+        "batcherQueuedChars": 2031616,
+        "rssBytes": 62291968
+      },
+      "receivedBytes": 2690244,
+      "afterDrain": { "producerPaused": false, "socketBufferedBytes": 0, "batcherQueuedChars": 0 }
+    },
+    {
+      "chunkSize": 1024,
+      "hidden": false,
+      "stalled": {
+        "producedChars": 3674112,
+        "producerPaused": true,
+        "socketBufferedBytes": 3401200,
+        "batcherQueuedChars": 0,
+        "rssBytes": 92307456
+      },
+      "receivedBytes": 3946800,
+      "afterDrain": { "producerPaused": false, "socketBufferedBytes": 0, "batcherQueuedChars": 0 }
+    },
+    {
+      "chunkSize": 1024,
+      "hidden": true,
+      "stalled": {
+        "producedChars": 8388608,
+        "producerPaused": false,
+        "socketBufferedBytes": 131300,
+        "batcherQueuedChars": 667648,
+        "rssBytes": 105005056
+      },
+      "receivedBytes": 1607892,
+      "afterDrain": { "producerPaused": false, "socketBufferedBytes": 0, "batcherQueuedChars": 0 }
+    }
+  ]
+}

docs/audits/daemon-stream-retention/before-1mib-chunks.json

@@ -0,0 +1,37 @@
+{
+  "source": "src/main/daemon/daemon-stream-data-batcher.ts",
+  "node": "v26.6.0",
+  "platform": "darwin",
+  "stalledReader": true,
+  "samples": [
+    {
+      "fedMiB": 0,
+      "socketBufferedBytes": 0,
+      "batcherQueuedChars": 0,
+      "retainedAsciiPayloadBytes": 0,
+      "rssBytes": 51249152
+    },
+    {
+      "fedMiB": 32,
+      "socketBufferedBytes": 131333,
+      "batcherQueuedChars": 32636928,
+      "retainedAsciiPayloadBytes": 32768261,
+      "rssBytes": 57917440
+    },
+    {
+      "fedMiB": 64,
+      "socketBufferedBytes": 32811587,
+      "batcherQueuedChars": 33554432,
+      "retainedAsciiPayloadBytes": 66366019,
+      "rssBytes": 239484928
+    },
+    {
+      "fedMiB": 96,
+      "socketBufferedBytes": 66410563,
+      "batcherQueuedChars": 33554432,
+      "retainedAsciiPayloadBytes": 99964995,
+      "rssBytes": 311885824
+    }
+  ],
+  "retainedBytesPerProducedByteAfterValve": 1.0013275146484375
+}

docs/audits/daemon-stream-retention/before-64kib-chunks.json

@@ -0,0 +1,86 @@
+{
+  "source": "src/main/daemon/daemon-stream-data-batcher.ts",
+  "bundleSha256": "7555dbf7f6020022aca2a0d75f5d9eae2e2c3cf1586e0c1e66fbc357f3a3143e",
+  "node": "v26.6.0",
+  "platform": "darwin",
+  "cases": [
+    {
+      "droppable": false,
+      "samples": [
+        {
+          "fedMiB": 0,
+          "socketBufferedBytes": 0,
+          "batcherQueuedChars": 0,
+          "rssBytes": 53968896
+        },
+        {
+          "fedMiB": 32,
+          "socketBufferedBytes": 131321,
+          "batcherQueuedChars": 32636928,
+          "rssBytes": 59834368
+        },
+        {
+          "fedMiB": 64,
+          "socketBufferedBytes": 32809583,
+          "batcherQueuedChars": 33554432,
+          "rssBytes": 1241956352
+        },
+        {
+          "fedMiB": 96,
+          "socketBufferedBytes": 66406511,
+          "batcherQueuedChars": 33554432,
+          "rssBytes": 1381416960
+        }
+      ],
+      "finalRetainedQueuedPayloadBytes": 99960943,
+      "queuedBytesPerProducedByteAfter64MiB": 1.0012664794921875,
+      "afterDrain": {
+        "fedMiB": 96,
+        "socketBufferedBytes": 0,
+        "batcherQueuedChars": 0,
+        "rssBytes": 327680000
+      },
+      "drainMs": 74.25875000000042,
+      "receivedBytes": 100791448
+    },
+    {
+      "droppable": true,
+      "samples": [
+        {
+          "fedMiB": 0,
+          "socketBufferedBytes": 0,
+          "batcherQueuedChars": 0,
+          "rssBytes": 327974912
+        },
+        {
+          "fedMiB": 32,
+          "socketBufferedBytes": 137492,
+          "batcherQueuedChars": 786432,
+          "rssBytes": 390053888
+        },
+        {
+          "fedMiB": 64,
+          "socketBufferedBytes": 144389,
+          "batcherQueuedChars": 720896,
+          "rssBytes": 454852608
+        },
+        {
+          "fedMiB": 96,
+          "socketBufferedBytes": 151286,
+          "batcherQueuedChars": 655360,
+          "rssBytes": 519733248
+        }
+      ],
+      "finalRetainedQueuedPayloadBytes": 806646,
+      "queuedBytesPerProducedByteAfter64MiB": -0.0017475783824920654,
+      "afterDrain": {
+        "fedMiB": 96,
+        "socketBufferedBytes": 0,
+        "batcherQueuedChars": 0,
+        "rssBytes": 520437760
+      },
+      "drainMs": 5.139125000000149,
+      "receivedBytes": 1595433
+    }
+  ]
+}

docs/audits/daemon-stream-retention/reproduce.mjs

@@ -0,0 +1,112 @@
+import { once } from 'node:events'
+import { createServer, Socket } from 'node:net'
+import { setImmediate as nextTurn, setTimeout as delay } from 'node:timers/promises'
+import { createHash } from 'node:crypto'
+import { fileURLToPath } from 'node:url'
+import { resolve } from 'node:path'
+import { build } from 'esbuild'
+
+if (process.env.ORCA_BACKGROUND_LAUNCH !== '1') {
+  throw new Error('Run with ORCA_BACKGROUND_LAUNCH=1')
+}
+const root = fileURLToPath(new URL('../../../', import.meta.url))
+const built = await build({
+  entryPoints: [resolve(root, 'src/main/daemon/daemon-stream-data-batcher.ts')],
+  bundle: true,
+  platform: 'node',
+  format: 'esm',
+  write: false
+})
+const bundle = built.outputFiles[0].text
+const { DaemonStreamDataBatcher } = await import(
+  `data:text/javascript;base64,${Buffer.from(bundle).toString('base64')}`
+)
+const MiB = 1024 * 1024
+
+async function run(chunkSize, hidden) {
+  const server = createServer()
+  const writer = new Socket()
+  let reader
+  let paused = false
+  let produced = 0
+  let receivedBytes = 0
+  const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: writer }), {
+    onProducerBackpressureChanged: (_sessionId, value) => {
+      paused = value
+    },
+    isSessionDroppable: () => hidden
+  })
+  writer.on('drain', () => batcher.flush('client'))
+  try {
+    const accepted = once(server, 'connection')
+    server.listen(0, '127.0.0.1')
+    await once(server, 'listening')
+    const address = server.address()
+    if (!address || typeof address === 'string') {
+      throw new Error('Expected TCP address')
+    }
+    writer.connect(address.port, '127.0.0.1')
+    await once(writer, 'connect')
+    ;[reader] = await accepted
+    reader.on('data', (data) => {
+      receivedBytes += data.length
+    })
+    reader.pause()
+    const chunk = 'x'.repeat(chunkSize)
+    while (!paused && produced < 8 * MiB) {
+      batcher.enqueue('client', 'session', chunk, { flushImmediately: true, flushMaxChars: 1024 })
+      batcher.flush('client')
+      produced += chunkSize
+      if (produced % MiB === 0) {
+        await nextTurn()
+      }
+    }
+    const stalled = {
+      producedChars: produced,
+      producerPaused: paused,
+      socketBufferedBytes: writer.writableLength,
+      batcherQueuedChars: batcher.queuedCharsForClient('client'),
+      rssBytes: process.memoryUsage().rss
+    }
+    reader.resume()
+    const start = performance.now()
+    while (paused || writer.writableLength || batcher.queuedCharsForClient('client')) {
+      if (performance.now() - start > 5000) {
+        throw new Error('Failed to drain')
+      }
+      await delay(5)
+    }
+    const ended = once(reader, 'end')
+    writer.end()
+    await ended
+    return {
+      chunkSize,
+      hidden,
+      stalled,
+      receivedBytes,
+      afterDrain: {
+        producerPaused: paused,
+        socketBufferedBytes: writer.writableLength,
+        batcherQueuedChars: batcher.queuedCharsForClient('client')
+      }
+    }
+  } finally {
+    batcher.clear()
+    writer.destroy()
+    reader?.destroy()
+    await new Promise((done) => server.close(done))
+  }
+}
+
+console.log(
+  JSON.stringify(
+    {
+      node: process.version,
+      platform: process.platform,
+      bundleSha256: createHash('sha256').update(bundle).digest('hex'),
+      cases: [await run(64 * 1024, false), await run(1024, false), await run(1024, true)]
+    },
+    null,
+    2
+  )
+)

src/main/daemon/daemon-client-connections.ts

@@ -244,6 +244,7 @@ export class DaemonClientConnections {
     socket.on('error', cleanup)
     if (previous && previous !== socket) {
       previous.destroy()
+      this.options.streamDataBatcher.flush(client.clientId)
     }
   }
 }

src/main/daemon/daemon-server.ts

@@ -74,6 +74,14 @@ export class DaemonServer {
     this.streamDataBatcher = new DaemonStreamDataBatcher(
       (clientId) => this.connections.get(clientId),
       {
+        onProducerBackpressureChanged: (sessionId, paused) =>
+          paused
+            ? this.host.pauseProducer(sessionId, 'stream')
+            : this.host.resumeProducer(sessionId, 'stream'),
+        isSessionAttachedToClient: (clientId, sessionId) => {
+          const owner = this.attachments.clientIdForSession(sessionId)
+          return owner === undefined || owner === clientId
+        },
         isSessionDroppable: (sessionId) =>
           BACKGROUND_STREAM_DROP_ENABLED && this.transientFactRelay.isBackgrounded(sessionId),
         salvageDroppedData: (dropped) => {
@@ -128,6 +136,7 @@ export class DaemonServer {
       onControlReplaced: (clientId) => {
         this.preparations.cancelForClient(clientId)
         this.historySeedTransfers.clearOwner(clientId)
+        this.streamDataBatcher.clear(clientId)
       },
       onClientDisconnected: (clientId) => {
         this.preparations.cancelForClient(clientId)

src/main/daemon/daemon-stream-backpressure-socket.test.ts

@@ -0,0 +1,145 @@
+import { once } from 'node:events'
+import { createServer, Socket } from 'node:net'
+import { setImmediate as nextTurn, setTimeout as delay } from 'node:timers/promises'
+import { describe, expect, it } from 'vitest'
+import { DaemonStreamDataBatcher } from './daemon-stream-data-batcher'
+import { createNdjsonParser } from './ndjson'
+import { SessionProducerPause } from './session-producer-pause'
+
+const MiB = 1024 * 1024
+
+describe('daemon stream backpressure with a stalled Socket reader', () => {
+  it.each([
+    { chunkSize: 64 * 1024, hidden: false },
+    { chunkSize: 1024, hidden: false },
+    { chunkSize: 64 * 1024, hidden: true },
+    { chunkSize: 1024, hidden: true }
+  ])(
+    'bounds $chunkSize-character writes (hidden=$hidden) and resumes',
+    async ({ chunkSize, hidden }) => {
+      const server = createServer()
+      const writer = new Socket()
+      let reader: Socket | undefined
+      let paused = false
+      let produced = 0
+      let dropped = 0
+      const received = new Map<string, number>()
+      const producer = new SessionProducerPause({
+        pause: () => {
+          paused = true
+        },
+        resume: () => {
+          paused = false
+        }
+      })
+      const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: writer }), {
+        isSessionDroppable: (sessionId) => hidden && sessionId === 'flood',
+        onProducerBackpressureChanged: (sessionId, value) => {
+          expect(sessionId).toBe('flood')
+          producer.setStreamBackpressured(value)
+        }
+      })
+      writer.on('drain', () => batcher.flush('client'))
+      try {
+        const accepted = once(server, 'connection')
+        server.listen(0, '127.0.0.1')
+        await once(server, 'listening')
+        const address = server.address()
+        if (!address || typeof address === 'string') {
+          throw new Error('Expected TCP address')
+        }
+        writer.connect(address.port, '127.0.0.1')
+        await once(writer, 'connect')
+        const [acceptedSocket] = await accepted
+        if (!(acceptedSocket instanceof Socket)) {
+          throw new Error('Expected accepted Socket')
+        }
+        reader = acceptedSocket
+        const parser = createNdjsonParser(
+          (event) => {
+            if (
+              !event ||
+              typeof event !== 'object' ||
+              !('sessionId' in event) ||
+              typeof event.sessionId !== 'string' ||
+              !('payload' in event)
+            ) {
+              return
+            }
+            const payload = event.payload
+            if (
+              payload &&
+              typeof payload === 'object' &&
+              'droppedChars' in payload &&
+              typeof payload.droppedChars === 'number'
+            ) {
+              dropped += payload.droppedChars
+            }
+            if (
+              !payload ||
+              typeof payload !== 'object' ||
+              !('data' in payload) ||
+              typeof payload.data !== 'string'
+            ) {
+              return
+            }
+            received.set(
+              event.sessionId,
+              (received.get(event.sessionId) ?? 0) + payload.data.length
+            )
+          },
+          () => {
+            throw new Error('Invalid stream frame')
+          }
+        )
+        reader.on('data', (data) => parser.feed(data.toString('utf8')))
+        reader.pause()
+
+        const chunk = 'x'.repeat(chunkSize)
+        while (!paused && produced < 8 * MiB) {
+          batcher.enqueue('client', 'flood', chunk, {
+            flushImmediately: chunkSize <= 1024,
+            flushMaxChars: 1024
+          })
+          batcher.flush('client')
+          produced += chunkSize
+          if (produced % MiB === 0) {
+            await nextTurn()
+          }
+        }
+        expect(paused).toBe(!hidden)
+        expect(writer.writableLength + 2 * batcher.queuedCharsForClient('client')).toBeLessThan(
+          5 * MiB
+        )
+
+        if (!hidden) {
+          producer.pause()
+          producer.resumeClient()
+          expect(paused).toBe(true)
+        }
+        batcher.enqueue('client', 'typing', 'echo', { flushImmediately: true, flushMaxChars: 1024 })
+        reader.resume()
+        const startedAt = performance.now()
+        while (paused || writer.writableLength || batcher.queuedCharsForClient('client')) {
+          if (performance.now() - startedAt > 5_000) {
+            throw new Error('Stream failed to drain')
+          }
+          await delay(5)
+        }
+        const ended = once(reader, 'end')
+        writer.end()
+        await ended
+        expect((received.get('flood') ?? 0) + dropped).toBe(produced)
+        expect(dropped > 0).toBe(hidden)
+        expect(received.get('typing')).toBe(4)
+        expect(paused).toBe(false)
+      } finally {
+        batcher.clear()
+        producer.release({ resume: false })
+        writer.destroy()
+        reader?.destroy()
+        await new Promise<void>((resolve) => server.close(() => resolve()))
+      }
+    }
+  )
+})

src/main/daemon/daemon-stream-backpressure.test.ts

@@ -0,0 +1,140 @@
+import { describe, expect, it, vi } from 'vitest'
+import { DaemonStreamBackpressure } from './daemon-stream-backpressure'
+import { DaemonStreamHeldRefill } from './daemon-stream-held-refill'
+
+const MiB = 1024 * 1024
+
+function createBackpressure(isDroppable: (sessionId: string) => boolean = () => false) {
+  const paused = new Set<string>()
+  const setPaused = vi.fn((sessionId: string, value: boolean) => {
+    if (value) {
+      paused.add(sessionId)
+    } else {
+      paused.delete(sessionId)
+    }
+  })
+  return { pressure: new DaemonStreamBackpressure(setPaused, isDroppable), paused, setPaused }
+}
+
+describe('DaemonStreamBackpressure', () => {
+  it('aggregates held output across clients and spares low-output sessions', () => {
+    const { pressure, paused } = createBackpressure()
+    pressure.setQueued(
+      'a',
+      new Map([
+        ['flood-a', MiB],
+        ['typing', 4]
+      ])
+    )
+    expect(paused.size).toBe(0)
+    pressure.setQueued('b', new Map([['flood-b', MiB]]))
+    expect([...paused].sort()).toEqual(['flood-a', 'flood-b'])
+    pressure.clear()
+    expect(paused.size).toBe(0)
+  })
+
+  it('keeps a shared producer paused until every slow client releases its backlog', () => {
+    const { pressure, paused } = createBackpressure()
+    pressure.setQueued('a', new Map([['shared', MiB]]))
+    pressure.setQueued('b', new Map([['shared', MiB]]))
+    expect(paused.has('shared')).toBe(true)
+    pressure.clear('a')
+    expect(paused.has('shared')).toBe(true)
+    pressure.clear('b')
+    expect(paused.has('shared')).toBe(false)
+  })
+
+  it('releases a drained session while another client remains stalled', () => {
+    const { pressure, paused } = createBackpressure()
+    pressure.setQueued('a', new Map([['flood-a', 2 * MiB]]))
+    pressure.setQueued('b', new Map([['flood-b', MiB]]))
+    expect(paused.has('flood-b')).toBe(true)
+    pressure.setQueued('b', new Map())
+    expect([...paused]).toEqual(['flood-a'])
+  })
+
+  it('leaves background shedding in control when a session becomes hidden', () => {
+    let hidden = false
+    const { pressure, paused } = createBackpressure(() => hidden)
+    pressure.setQueued('a', new Map([['flood', 2 * MiB]]))
+    expect(paused.has('flood')).toBe(true)
+    hidden = true
+    pressure.refresh()
+    expect(paused.size).toBe(0)
+    pressure.setQueued('a', new Map([['flood', 3 * MiB]]))
+    expect(paused.size).toBe(0)
+  })
+
+  it('uses hysteresis and resumes after the aggregate falls below the low watermark', () => {
+    const { pressure, paused } = createBackpressure()
+    pressure.setQueued('a', new Map([['flood', 2 * MiB]]))
+    pressure.setQueued('a', new Map([['flood', MiB]]))
+    expect(paused.has('flood')).toBe(true)
+    pressure.setQueued('a', new Map([['flood', MiB / 2]]))
+    expect(paused.size).toBe(0)
+  })
+
+  it('does not let a superseded subscription pause the new attachment owner', () => {
+    let owner = 'old'
+    const setPaused = vi.fn()
+    const pressure = new DaemonStreamBackpressure(
+      setPaused,
+      () => false,
+      (clientId) => clientId === owner
+    )
+    pressure.setQueued('old', new Map([['session', 2 * MiB]]))
+    expect(setPaused).toHaveBeenLastCalledWith('session', true)
+    owner = 'new'
+    pressure.refresh()
+    expect(setPaused).toHaveBeenLastCalledWith('session', false)
+    setPaused.mockClear()
+    pressure.setQueued('new', new Map([['session', 1024]]))
+    expect(setPaused).not.toHaveBeenCalled()
+  })
+
+  it('bounds undroppable control output even when the producer is backgrounded', () => {
+    const { pressure, paused } = createBackpressure(() => true)
+    pressure.setQueued('a', new Map([['hidden', 2 * MiB]]), new Map([['hidden', 4 * MiB]]))
+    expect(paused.has('hidden')).toBe(true)
+    pressure.setQueued('a', new Map([['hidden', 2 * MiB]]))
+    expect(paused.size).toBe(0)
+  })
+
+  it('ignores old write completions after a client and session id are reused', () => {
+    const completions: (() => void)[] = []
+    const socket = {
+      write(...args: unknown[]): boolean {
+        const complete = args[1]
+        if (typeof complete === 'function') {
+          completions.push(() => complete())
+        }
+        return false
+      }
+    }
+    const { pressure, paused } = createBackpressure()
+    const line = 'x'.repeat(4 * MiB)
+    pressure.write('client', 'session', socket, line)
+    expect(paused.has('session')).toBe(true)
+    pressure.clear('client')
+    expect(paused.size).toBe(0)
+    pressure.write('client', 'session', socket, line)
+    completions.shift()?.()
+    expect(paused.has('session')).toBe(true)
+    completions.shift()?.()
+    expect(paused.size).toBe(0)
+  })
+
+  it('does not let an old refill callback flush a replacement connection', () => {
+    const flush = vi.fn()
+    const completions: (() => void)[] = []
+    const write = (_line: string, complete: () => void) => completions.push(complete)
+    const refill = new DaemonStreamHeldRefill(flush)
+    refill.arm('client', 'session', write)
+    refill.clear('client')
+    refill.arm('client', 'session', write)
+    completions.shift()?.()
+    expect(flush).not.toHaveBeenCalled()
+    completions.shift()?.()
+    expect(flush).toHaveBeenCalledOnce()
+  })
+})

src/main/daemon/daemon-stream-backpressure.ts

@@ -0,0 +1,145 @@
+import type { Socket } from 'node:net'
+import { STREAM_ENTRY_OVERHEAD_BYTES } from './daemon-stream-entry-accounting'
+
+const HIGH_WATER_BYTES = 4 * 1024 * 1024
+const LOW_WATER_BYTES = 1024 * 1024
+const SESSION_HIGH_WATER_BYTES = 64 * 1024
+const SESSION_LOW_WATER_BYTES = 32 * 1024
+
+type ClientBacklog = {
+  queuedChars: Map<string, number>
+  queuedMetadataBytes: Map<string, number>
+  pendingWriteBytes: Map<string, number>
+}
+
+/** Counts held strings and socket writes together, including small writes that bypass the hold. */
+export class DaemonStreamBackpressure {
+  private readonly clients = new Map<string, ClientBacklog>()
+  private readonly pausedSessions = new Set<string>()
+  private pressured = false
+
+  constructor(
+    private readonly setPaused: (sessionId: string, paused: boolean) => void,
+    private readonly isDroppable: (sessionId: string) => boolean,
+    private readonly ownsSession: (clientId: string, sessionId: string) => boolean = () => true
+  ) {}
+
+  setQueued(
+    clientId: string,
+    queuedChars: ReadonlyMap<string, number>,
+    queuedMetadataBytes: ReadonlyMap<string, number> = new Map()
+  ): void {
+    const client = this.getOrCreateClient(clientId)
+    client.queuedChars = new Map(queuedChars)
+    client.queuedMetadataBytes = new Map(queuedMetadataBytes)
+    this.pruneClient(clientId, client)
+    this.refresh()
+  }
+
+  write(
+    clientId: string,
+    sessionId: string,
+    socket: Pick<Socket, 'write'>,
+    line: string,
+    onComplete?: () => void
+  ): void {
+    const client = this.getOrCreateClient(clientId)
+    const bytes = Buffer.byteLength(line) + STREAM_ENTRY_OVERHEAD_BYTES
+    client.pendingWriteBytes.set(sessionId, (client.pendingWriteBytes.get(sessionId) ?? 0) + bytes)
+    this.refresh()
+    socket.write(line, () => {
+      // A disconnected client's callbacks must not debit a replacement connection's writes.
+      if (this.clients.get(clientId) === client) {
+        const remaining = (client.pendingWriteBytes.get(sessionId) ?? 0) - bytes
+        if (remaining > 0) {
+          client.pendingWriteBytes.set(sessionId, remaining)
+        } else {
+          client.pendingWriteBytes.delete(sessionId)
+        }
+        this.pruneClient(clientId, client)
+        this.refresh()
+      }
+      onComplete?.()
+    })
+  }
+
+  refresh(): void {
+    const bytesBySession = new Map<string, number>()
+    let total = 0
+    for (const [clientId, client] of this.clients) {
+      for (const [sessionId, chars] of client.queuedChars) {
+        // Background data has its own keep-tail budget; only its undroppable frames need pausing.
+        if (this.isDroppable(sessionId)) {
+          continue
+        }
+        const bytes = 2 * chars
+        total += bytes
+        if (this.ownsSession(clientId, sessionId)) {
+          bytesBySession.set(sessionId, (bytesBySession.get(sessionId) ?? 0) + bytes)
+        }
+      }
+      for (const counts of [client.queuedMetadataBytes, client.pendingWriteBytes]) {
+        for (const [sessionId, bytes] of counts) {
+          total += bytes
+          if (this.ownsSession(clientId, sessionId)) {
+            bytesBySession.set(sessionId, (bytesBySession.get(sessionId) ?? 0) + bytes)
+          }
+        }
+      }
+    }
+    if (total >= HIGH_WATER_BYTES) {
+      this.pressured = true
+    } else if (total <= LOW_WATER_BYTES) {
+      this.pressured = false
+    }
+
+    for (const sessionId of this.pausedSessions) {
+      if (!this.pressured || (bytesBySession.get(sessionId) ?? 0) <= SESSION_LOW_WATER_BYTES) {
+        this.pausedSessions.delete(sessionId)
+        this.setPaused(sessionId, false)
+      }
+    }
+    if (!this.pressured) {
+      return
+    }
+    for (const [sessionId, bytes] of bytesBySession) {
+      if (bytes >= SESSION_HIGH_WATER_BYTES) {
+        this.pausedSessions.add(sessionId)
+        // Detach/termination may have released the session's pause since the last update.
+        this.setPaused(sessionId, true)
+      }
+    }
+  }
+
+  clear(clientId?: string): void {
+    if (clientId === undefined) {
+      this.clients.clear()
+    } else {
+      this.clients.delete(clientId)
+    }
+    this.refresh()
+  }
+
+  private getOrCreateClient(clientId: string): ClientBacklog {
+    let client = this.clients.get(clientId)
+    if (!client) {
+      client = {
+        queuedChars: new Map(),
+        queuedMetadataBytes: new Map(),
+        pendingWriteBytes: new Map()
+      }
+      this.clients.set(clientId, client)
+    }
+    return client
+  }
+
+  private pruneClient(clientId: string, client: ClientBacklog): void {
+    if (
+      client.queuedChars.size === 0 &&
+      client.queuedMetadataBytes.size === 0 &&
+      client.pendingWriteBytes.size === 0
+    ) {
+      this.clients.delete(clientId)
+    }
+  }
+}

src/main/daemon/daemon-stream-data-batcher.test.ts

@@ -38,6 +38,30 @@ function nonSentinelWrites(streamSocket: { write: ReturnType<typeof vi.fn> }): P
 }
 
 describe('DaemonStreamDataBatcher', () => {
+  it.each(['', 'x'])('accounts held transformed entries with %i payload characters', (data) => {
+    vi.useFakeTimers()
+    let paused = false
+    const { batcher, streamSocket } = createBatcher({
+      onProducerBackpressureChanged: (_sessionId, value) => {
+        paused = value
+      }
+    })
+    try {
+      streamSocket.writableLength = 128 * 1024
+      batcher.enqueue('client-1', 'session-1', 'bulk'.repeat(16 * 1024))
+      for (let seq = 1; seq <= 20_000 && !paused; seq++) {
+        batcher.enqueue('client-1', 'session-1', data, { transformed: true, rawLength: 1, seq })
+      }
+      expect(paused).toBe(true)
+      expect(batcher.queuedCharsForClient('client-1')).toBeLessThan(100 * 1024)
+      batcher.clear()
+      expect(paused).toBe(false)
+    } finally {
+      batcher.clear()
+      vi.useRealTimers()
+    }
+  })
+
   it('coalesces background output before writing daemon stream events', () => {
     vi.useFakeTimers()
     try {

src/main/daemon/daemon-stream-data-batcher.ts

@@ -1,14 +1,20 @@
 import type { Socket } from 'node:net'
 import { encodeNdjson, NDJSON_MAX_LINE_BYTES } from './ndjson'
 import { recordDaemonStreamBacklogEvent } from './daemon-stream-backlog-probe'
+import { DaemonStreamBackpressure } from './daemon-stream-backpressure'
 import {
-  clampToSafeSplitIndex,
-  encodeStreamDataEvent,
-  writeStreamDataEvents
-} from './daemon-stream-data-split'
+  accountDaemonStreamEntry,
+  releaseDaemonStreamEntry
+} from './daemon-stream-entry-accounting'
+import { DaemonStreamHeldRefill } from './daemon-stream-held-refill'
+import { clampToSafeSplitIndex, writeStreamDataEvents } from './daemon-stream-data-split'
 import type { PendingStreamDataBatch } from './daemon-stream-keep-tail-drop'
 import type { DaemonEvent } from './types'
-import { appendDaemonStreamData, type DaemonStreamEnqueueOptions } from './daemon-stream-data-entry'
+import {
+  appendDaemonStreamData,
+  flushDaemonStreamSession,
+  type DaemonStreamEnqueueOptions
+} from './daemon-stream-data-entry'
 import {
   evaluateDroppableEnqueue,
   refreshDroppableSessionMembership
@@ -27,15 +33,15 @@ const SHALLOW_SOCKET_WRITE_GATE_BYTES =
   process.env.ORCA_DAEMON_SHALLOW_SOCKET_GATE === '0' ? Number.POSITIVE_INFINITY : 128 * 1024
 // Sliced writes: a coalesced entry can grow to megabytes; writing it whole would re-deepen the socket past the gate in one call.
 const BULK_WRITE_SLICE_CHARS = 64 * 1024
-// Safety valve: past this, write through — bounded daemon memory beats bounded echo latency in the extreme. Must sit FAR above the pacer's pause watermark + overshoot (~5MB) or an engaged valve buries interactive echo behind the whole backlog.
+// Last resort for handles without pause support; actual memory bounds come from producer backpressure.
 const HELD_WRITE_THROUGH_TOTAL_CHARS = 32 * 1024 * 1024
 // Small-session bypass: a few-KB session (echo, redraws, query replies) is never the flood, so it must not wait FIFO behind others' megabytes; backstops the 100ms interactive fast-path, which misses under event-loop load.
 const SMALL_SESSION_HOLD_BYPASS_CHARS = 4 * 1024
 
 type DaemonStreamDataBatcherOptions = {
   maxLineBytes?: number
-  /** Fires after each stream-socket write — the only place backlog grows, so the backlog pacer checks its watermark here. */
-  onAfterSocketWrite?: () => void
+  onProducerBackpressureChanged?: (sessionId: string, paused: boolean) => void
+  isSessionAttachedToClient?: (clientId: string, sessionId: string) => boolean
   /** True for sessions whose queued output may be keep-tail dropped (main-marked background sessions). */
   isSessionDroppable?: (sessionId: string) => boolean
   /** Carve reply-eliciting query bytes (DSR/DA/DECRQM/OSC probes) out of dropped data — the hidden program blocks on the reply, so they must still be delivered even when their flood is not. */
@@ -46,7 +52,8 @@ export class DaemonStreamDataBatcher {
   private pendingByClient = new Map<string, PendingStreamDataBatch>()
   private getClient: (clientId: string) => StreamDataClient | undefined
   private maxLineBytes: number
-  private onAfterSocketWrite: (() => void) | undefined
+  private readonly backpressure: DaemonStreamBackpressure | undefined
+  private readonly heldRefill = new DaemonStreamHeldRefill((clientId) => this.flush(clientId))
   private isSessionDroppable: (sessionId: string) => boolean
   private salvageDroppedData: (dropped: string) => string
 
@@ -56,9 +63,15 @@ export class DaemonStreamDataBatcher {
   ) {
     this.getClient = getClient
     this.maxLineBytes = Math.max(1, options.maxLineBytes ?? NDJSON_MAX_LINE_BYTES)
-    this.onAfterSocketWrite = options.onAfterSocketWrite
     this.isSessionDroppable = options.isSessionDroppable ?? (() => false)
     this.salvageDroppedData = options.salvageDroppedData ?? (() => '')
+    this.backpressure = options.onProducerBackpressureChanged
+      ? new DaemonStreamBackpressure(
+          options.onProducerBackpressureChanged,
+          this.isSessionDroppable,
+          options.isSessionAttachedToClient
+        )
+      : undefined
   }
 
   enqueue(
@@ -83,10 +96,13 @@ export class DaemonStreamDataBatcher {
       this.isSessionDroppable,
       this.salvageDroppedData
     )
+    this.updateBackpressure(clientId, batch)
 
     if (
       options.flushImmediately === true &&
-      this.queuedCharsForSession(batch, sessionId, options.flushMaxChars) <=
+      (!batch.droppableQueuedSessionIds.has(sessionId) ||
+        client.streamSocket.writableLength < SHALLOW_SOCKET_WRITE_GATE_BYTES) &&
+      (batch.queuedCharsBySession.get(sessionId) ?? 0) <=
         (options.flushMaxChars ?? Number.POSITIVE_INFINITY)
     ) {
       this.flushSession(clientId, sessionId)
@@ -104,7 +120,8 @@ export class DaemonStreamDataBatcher {
       return
     }
     const batch = this.getOrCreateBatch(clientId)
-    batch.queue.push({ sessionId, data: '', control })
+    batch.queue.push(accountDaemonStreamEntry(batch, { sessionId, data: '', control }))
+    this.updateBackpressure(clientId, batch)
     if (!batch.timer) {
       batch.timer = setTimeout(() => this.flush(clientId), STREAM_DATA_BATCH_INTERVAL_MS)
     }
@@ -113,6 +130,7 @@ export class DaemonStreamDataBatcher {
   refreshSessionDroppability(sessionId: string): void {
     const droppable = this.isSessionDroppable(sessionId)
     refreshDroppableSessionMembership(this.pendingByClient.values(), sessionId, droppable)
+    this.backpressure?.refresh()
   }
 
   private getOrCreateBatch(clientId: string): PendingStreamDataBatch {
@@ -123,6 +141,7 @@ export class DaemonStreamDataBatcher {
         queue: [],
         queuedChars: 0,
         queuedCharsBySession: new Map(),
+        queuedMetadataBytesBySession: new Map(),
         droppableQueuedSessionIds: new Set()
       }
       this.pendingByClient.set(clientId, batch)
@@ -148,7 +167,7 @@ export class DaemonStreamDataBatcher {
     const client = this.getClient(clientId)
     if (!client?.streamSocket || client.streamSocket.destroyed) {
       // A vanished stream socket drops the batch — the model owns the bytes and reconnect restores from a snapshot.
-      this.pendingByClient.delete(clientId)
+      this.clear(clientId)
       return
     }
 
@@ -158,22 +177,30 @@ export class DaemonStreamDataBatcher {
     const retained: PendingStreamDataBatch['queue'] = []
     while (batch.queue.length > 0) {
       const entry = batch.queue[0]
+      const socketDeep = (socket.writableLength ?? 0) >= SHALLOW_SOCKET_WRITE_GATE_BYTES
       if (entry.control) {
-        // Control entries only respect the held-session order latch; at ~100B, writing them onto a deep socket is as harmless as the small-session bypass.
-        if (heldSessions.has(entry.sessionId)) {
+        // Holding gaps lets repeated background drops coalesce without filling the socket with markers.
+        if (
+          heldSessions.has(entry.sessionId) ||
+          (socketDeep && entry.control.event === 'dataGap')
+        ) {
+          heldSessions.add(entry.sessionId)
           retained.push(entry)
           batch.queue.shift()
           continue
         }
         batch.queue.shift()
-        socket.write(encodeNdjson(entry.control))
-        this.onAfterSocketWrite?.()
+        releaseDaemonStreamEntry(batch, entry)
+        this.write(clientId, entry.sessionId, socket, encodeNdjson(entry.control))
         continue
       }
-      const socketDeep = (socket.writableLength ?? 0) >= SHALLOW_SOCKET_WRITE_GATE_BYTES
       if (socketDeep && batch.queuedChars <= HELD_WRITE_THROUGH_TOTAL_CHARS) {
         const sessionHeld = batch.queuedCharsBySession.get(entry.sessionId) ?? 0
-        if (heldSessions.has(entry.sessionId) || sessionHeld > SMALL_SESSION_HOLD_BYPASS_CHARS) {
+        if (
+          heldSessions.has(entry.sessionId) ||
+          batch.droppableQueuedSessionIds.has(entry.sessionId) ||
+          sessionHeld > SMALL_SESSION_HOLD_BYPASS_CHARS
+        ) {
           // Hold this flooding session's entry; small talkers keep flowing. No timer: a deep socket implies a prior false write(), so 'drain' (routed back to flush) is guaranteed to resume held bulk.
           heldSessions.add(entry.sessionId)
           retained.push(entry)
@@ -200,6 +227,7 @@ export class DaemonStreamDataBatcher {
           : slice.length
       if (end >= entry.data.length) {
         batch.queue.shift()
+        releaseDaemonStreamEntry(batch, entry)
       } else {
         entry.data = entry.data.slice(end)
         const remainingSequenceChars = entrySequenceChars - sliceSequenceChars
@@ -216,7 +244,7 @@ export class DaemonStreamDataBatcher {
         batch.queuedCharsBySession.set(entry.sessionId, sessionHeldAfter)
       }
       writeStreamDataEvents(
-        socket,
+        { write: (line) => this.write(clientId, entry.sessionId, socket, line) },
         entry.sessionId,
         slice,
         this.maxLineBytes,
@@ -224,46 +252,42 @@ export class DaemonStreamDataBatcher {
         entry.seq,
         entry.transformed
       )
-      this.onAfterSocketWrite?.()
     }
+    this.updateBackpressure(clientId, batch)
     if (retained.length > 0) {
       batch.queue = retained
       // 'drain' only fires when the buffer fully empties (one gate-depth/turn = seconds for multi-MB backlogs); arm a no-op data event whose flush callback re-flushes while bytes are still in flight.
-      this.armHeldQueueRefill(socket, clientId, retained[0].sessionId)
+      if (!socket.destroyed) {
+        const sessionId = retained[0].sessionId
+        this.heldRefill.arm(clientId, sessionId, (line, complete) =>
+          this.write(clientId, sessionId, socket, line, complete)
+        )
+      }
       return
     }
     this.pendingByClient.delete(clientId)
   }
 
-  private refillArmedClients = new Set<string>()
-
-  private armHeldQueueRefill(socket: Socket, clientId: string, sessionId: string): void {
-    if (this.refillArmedClients.has(clientId) || socket.destroyed) {
-      return
+  private write(
+    clientId: string,
+    sessionId: string,
+    socket: Socket,
+    line: string,
+    onComplete?: () => void
+  ): void {
+    if (this.backpressure) {
+      this.backpressure.write(clientId, sessionId, socket, line, onComplete)
+    } else {
+      socket.write(line, onComplete)
     }
-    this.refillArmedClients.add(clientId)
-    // Must be a real protocol no-op line, not an empty write: an empty write's callback fires immediately, defeating the in-flight re-flush.
-    socket.write(encodeStreamDataEvent(sessionId, ''), () => {
-      this.refillArmedClients.delete(clientId)
-      this.flush(clientId)
-    })
   }
 
-  private queuedCharsForSession(
-    batch: PendingStreamDataBatch,
-    sessionId: string,
-    stopAfter = Number.POSITIVE_INFINITY
-  ): number {
-    let chars = 0
-    for (const entry of batch.queue) {
-      if (entry.sessionId === sessionId) {
-        chars += entry.data.length
-        if (chars > stopAfter) {
-          return chars
-        }
-      }
-    }
-    return chars
+  private updateBackpressure(clientId: string, batch: PendingStreamDataBatch): void {
+    this.backpressure?.setQueued(
+      clientId,
+      batch.queuedCharsBySession,
+      batch.queuedMetadataBytesBySession
+    )
   }
 
   private flushSession(clientId: string, sessionId: string): void {
@@ -272,55 +296,20 @@ export class DaemonStreamDataBatcher {
       return
     }
 
-    const flushed: PendingStreamDataBatch['queue'] = []
-    const retained: PendingStreamDataBatch['queue'] = []
-    let flushedChars = 0
-    for (const entry of batch.queue) {
-      if (entry.sessionId === sessionId) {
-        flushed.push(entry)
-        flushedChars += entry.data.length
-      } else {
-        retained.push(entry)
-      }
-    }
-    if (flushed.length === 0) {
-      return
-    }
-
-    batch.queue = retained
-    batch.queuedChars -= flushedChars
-    batch.queuedCharsBySession.delete(sessionId)
-    batch.droppableQueuedSessionIds.delete(sessionId)
-    if (batch.queue.length === 0) {
-      if (batch.timer) {
-        clearTimeout(batch.timer)
-        batch.timer = null
-      }
-      this.pendingByClient.delete(clientId)
-    }
-
     const client = this.getClient(clientId)
     if (!client?.streamSocket || client.streamSocket.destroyed) {
+      this.clear(clientId)
       return
     }
 
-    for (const entry of flushed) {
-      if (entry.control) {
-        client.streamSocket.write(encodeNdjson(entry.control))
-        this.onAfterSocketWrite?.()
-      } else {
-        writeStreamDataEvents(
-          client.streamSocket,
-          entry.sessionId,
-          entry.data,
-          this.maxLineBytes,
-          entry.sequenceChars ?? entry.data.length,
-          entry.seq,
-          entry.transformed
-        )
-        this.onAfterSocketWrite?.()
-      }
+    const socket = client.streamSocket
+    flushDaemonStreamSession(batch, sessionId, this.maxLineBytes, (line) =>
+      this.write(clientId, sessionId, socket, line)
+    )
+    if (batch.queue.length === 0) {
+      this.pendingByClient.delete(clientId)
     }
+    this.updateBackpressure(clientId, batch)
   }
 
   clear(clientId?: string): void {
@@ -335,5 +324,7 @@ export class DaemonStreamDataBatcher {
       }
       this.pendingByClient.delete(id)
     }
+    this.heldRefill.clear(clientId)
+    this.backpressure?.clear(clientId)
   }
 }

src/main/daemon/daemon-stream-data-entry.ts

@@ -1,4 +1,7 @@
 import type { PendingStreamDataBatch } from './daemon-stream-keep-tail-drop'
+import { writeStreamDataEvents } from './daemon-stream-data-split'
+import { encodeNdjson } from './ndjson'
+import { accountDaemonStreamEntry } from './daemon-stream-entry-accounting'
 
 export type DaemonStreamEnqueueOptions = {
   flushImmediately?: boolean
@@ -28,18 +31,68 @@ export function appendDaemonStreamData(
     last.sequenceChars = combinedRawLength === last.data.length ? undefined : combinedRawLength
     last.seq = options.seq
   } else {
-    batch.queue.push({
-      sessionId,
-      data,
-      ...(options.rawLength === undefined || options.rawLength === data.length
-        ? {}
-        : { sequenceChars: options.rawLength }),
-      ...(options.transformed ? { transformed: true } : {}),
-      ...(options.seq === undefined ? {} : { seq: options.seq })
-    })
+    batch.queue.push(
+      accountDaemonStreamEntry(batch, {
+        sessionId,
+        data,
+        ...(options.rawLength === undefined || options.rawLength === data.length
+          ? {}
+          : { sequenceChars: options.rawLength }),
+        ...(options.transformed ? { transformed: true } : {}),
+        ...(options.seq === undefined ? {} : { seq: options.seq })
+      })
+    )
   }
   batch.queuedChars += data.length
   const queuedAfter = (batch.queuedCharsBySession.get(sessionId) ?? 0) + data.length
   batch.queuedCharsBySession.set(sessionId, queuedAfter)
   return queuedAfter
 }
+
+function takeDaemonStreamSession(
+  batch: PendingStreamDataBatch,
+  sessionId: string
+): PendingStreamDataBatch['queue'] {
+  const flushed: PendingStreamDataBatch['queue'] = []
+  const retained: PendingStreamDataBatch['queue'] = []
+  for (const entry of batch.queue) {
+    if (entry.sessionId === sessionId) {
+      flushed.push(entry)
+      batch.queuedChars -= entry.data.length
+    } else {
+      retained.push(entry)
+    }
+  }
+  batch.queue = retained
+  batch.queuedCharsBySession.delete(sessionId)
+  batch.queuedMetadataBytesBySession.delete(sessionId)
+  batch.droppableQueuedSessionIds.delete(sessionId)
+  if (batch.queue.length === 0 && batch.timer) {
+    clearTimeout(batch.timer)
+    batch.timer = null
+  }
+  return flushed
+}
+
+export function flushDaemonStreamSession(
+  batch: PendingStreamDataBatch,
+  sessionId: string,
+  maxLineBytes: number,
+  write: (line: string) => void
+): void {
+  for (const entry of takeDaemonStreamSession(batch, sessionId)) {
+    if (entry.control) {
+      write(encodeNdjson(entry.control))
+    } else {
+      writeStreamDataEvents(
+        { write },
+        entry.sessionId,
+        entry.data,
+        maxLineBytes,
+        entry.sequenceChars ?? entry.data.length,
+        entry.seq,
+        entry.transformed
+      )
+    }
+  }
+}

src/main/daemon/daemon-stream-data-split.ts

@@ -4,7 +4,6 @@
  * clamp shared by the batcher's bulk write slicing and keep-tail dropping.
  */
 import { encodeNdjson } from './ndjson'
-import type { Socket } from 'node:net'
 
 export function encodeStreamDataEvent(
   sessionId: string,
@@ -113,7 +112,7 @@ function splitOversizedStreamDataForNdjson(
 }
 
 export function writeStreamDataEvents(
-  streamSocket: Pick<Socket, 'write'>,
+  streamSocket: { write(data: string): void },
   sessionId: string,
   data: string,
   maxLineBytes: number,

src/main/daemon/daemon-stream-entry-accounting.ts

@@ -0,0 +1,32 @@
+import { encodeNdjson } from './ndjson'
+import type { PendingStreamDataBatch, StreamQueueEntry } from './daemon-stream-keep-tail-drop'
+
+// Budget callbacks, stream requests and queue objects even when their data payload is empty.
+export const STREAM_ENTRY_OVERHEAD_BYTES = 256
+
+export function accountDaemonStreamEntry(
+  batch: PendingStreamDataBatch,
+  entry: StreamQueueEntry
+): StreamQueueEntry {
+  entry.retainedBytes =
+    STREAM_ENTRY_OVERHEAD_BYTES +
+    (entry.control ? Buffer.byteLength(encodeNdjson(entry.control)) : 0)
+  batch.queuedMetadataBytesBySession.set(
+    entry.sessionId,
+    (batch.queuedMetadataBytesBySession.get(entry.sessionId) ?? 0) + entry.retainedBytes
+  )
+  return entry
+}
+
+export function releaseDaemonStreamEntry(
+  batch: PendingStreamDataBatch,
+  entry: StreamQueueEntry
+): void {
+  const remaining =
+    (batch.queuedMetadataBytesBySession.get(entry.sessionId) ?? 0) - (entry.retainedBytes ?? 0)
+  if (remaining > 0) {
+    batch.queuedMetadataBytesBySession.set(entry.sessionId, remaining)
+  } else {
+    batch.queuedMetadataBytesBySession.delete(entry.sessionId)
+  }
+}

src/main/daemon/daemon-stream-held-refill.ts

@@ -0,0 +1,35 @@
+import { encodeStreamDataEvent } from './daemon-stream-data-split'
+
+export class DaemonStreamHeldRefill {
+  private readonly armed = new Map<string, symbol>()
+
+  constructor(private readonly flush: (clientId: string) => void) {}
+
+  arm(
+    clientId: string,
+    sessionId: string,
+    write: (line: string, complete: () => void) => void
+  ): void {
+    if (this.armed.has(clientId)) {
+      return
+    }
+    const refill = Symbol()
+    this.armed.set(clientId, refill)
+    // A real no-op frame waits for preceding writes; an empty write can complete immediately.
+    write(encodeStreamDataEvent(sessionId, ''), () => {
+      if (this.armed.get(clientId) !== refill) {
+        return
+      }
+      this.armed.delete(clientId)
+      this.flush(clientId)
+    })
+  }
+
+  clear(clientId?: string): void {
+    if (clientId === undefined) {
+      this.armed.clear()
+    } else {
+      this.armed.delete(clientId)
+    }
+  }
+}

src/main/daemon/daemon-stream-keep-tail-drop.ts

@@ -10,6 +10,10 @@
 import { clampToSafeSplitIndex } from './daemon-stream-data-split'
 import { recordDaemonStreamBacklogEvent } from './daemon-stream-backlog-probe'
 import type { DaemonEvent, DataGapEvent } from './types'
+import {
+  accountDaemonStreamEntry,
+  releaseDaemonStreamEntry
+} from './daemon-stream-entry-accounting'
 
 // A control entry carries a whole pre-shaped stream event (background marker,
 // data gap, transient fact) that must ride at its exact position in the
@@ -24,6 +28,7 @@ export type StreamQueueEntry = {
   seq?: number
   transformed?: boolean
   control?: DaemonEvent
+  retainedBytes?: number
 }
 
 export type PendingStreamDataBatch = {
@@ -33,6 +38,7 @@ export type PendingStreamDataBatch = {
   // Per-session held totals so the flush hold can spare small talkers
   // (echo/replies) from waiting behind other sessions' floods.
   queuedCharsBySession: Map<string, number>
+  queuedMetadataBytesBySession: Map<string, number>
   // Membership is reconciled when queued data first appears and on rare
   // background lifecycle changes, keeping steady-state enqueue constant-time.
   droppableQueuedSessionIds: Set<string>
@@ -121,6 +127,7 @@ export function dropOldestQueuedForSession(
       if (insertGapAt === -1) {
         insertGapAt = i
       }
+      releaseDaemonStreamEntry(batch, entry)
       batch.queue.splice(i, 1)
       i--
     } else {
@@ -159,16 +166,20 @@ export function dropOldestQueuedForSession(
       sessionIdSuffix: sessionId.slice(-10),
       droppedChars: dropped
     })
-    batch.queue.splice(Math.max(0, insertGapAt), 0, {
-      sessionId,
-      data: '',
-      control: {
-        type: 'event',
-        event: 'dataGap',
+    batch.queue.splice(
+      Math.max(0, insertGapAt),
+      0,
+      accountDaemonStreamEntry(batch, {
         sessionId,
-        payload: { droppedChars: dropped, sequenceChars: droppedSequenceChars }
-      }
-    })
+        data: '',
+        control: {
+          type: 'event',
+          event: 'dataGap',
+          sessionId,
+          payload: { droppedChars: dropped, sequenceChars: droppedSequenceChars }
+        }
+      })
+    )
     insertGapAt = Math.max(0, insertGapAt) + 1
   }
   if (salvaged.length > 0) {
@@ -177,7 +188,11 @@ export function dropOldestQueuedForSession(
     const at = existingGap
       ? batch.queue.findIndex((e) => e.control === existingGap) + 1
       : insertGapAt
-    batch.queue.splice(at, 0, { sessionId, data: salvaged, sequenceChars: 0 })
+    batch.queue.splice(
+      at,
+      0,
+      accountDaemonStreamEntry(batch, { sessionId, data: salvaged, sequenceChars: 0 })
+    )
     batch.queuedChars += salvaged.length
     batch.queuedCharsBySession.set(
       sessionId,

src/main/daemon/session-producer-pause.ts

@@ -7,33 +7,73 @@ export const PRODUCER_PAUSE_FAILSAFE_MS = 5_000
 /** Producer-side flow control for one session's PTY fd, with the lost-resume failsafe. */
 export class SessionProducerPause {
   private paused = false
+  private streamBackpressured = false
   private failsafeTimer: ReturnType<typeof setTimeout> | null = null
 
   constructor(private readonly subprocess: Pick<SubprocessHandle, 'pause' | 'resume'>) {}
 
   /** Stop reading the PTY fd so a flooding child blocks on write. Arms the failsafe; re-pausing re-arms it. */
-  pause(): void {
+  pause(source?: 'stream', canPauseStream = true): void {
+    if (source === 'stream') {
+      if (canPauseStream) {
+        this.setStreamBackpressured(true)
+      }
+      return
+    }
+    const wasPaused = this.paused || this.streamBackpressured
     this.paused = true
-    this.subprocess.pause?.()
+    if (!wasPaused) {
+      this.subprocess.pause?.()
+    }
     if (this.failsafeTimer) {
       clearTimeout(this.failsafeTimer)
     }
     this.failsafeTimer = setTimeout(() => {
       this.failsafeTimer = null
       this.paused = false
-      this.subprocess.resume?.()
+      if (!this.streamBackpressured) {
+        this.subprocess.resume?.()
+      }
     }, PRODUCER_PAUSE_FAILSAFE_MS)
   }
 
+  setStreamBackpressured(paused: boolean): void {
+    const wasPaused = this.paused || this.streamBackpressured
+    this.streamBackpressured = paused
+    const nowPaused = this.paused || this.streamBackpressured
+    if (nowPaused && !wasPaused) {
+      this.subprocess.pause?.()
+    } else if (wasPaused && !nowPaused) {
+      this.subprocess.resume?.()
+    }
+  }
+
+  resumeClient(source?: 'stream'): void {
+    if (source === 'stream') {
+      this.setStreamBackpressured(false)
+      return
+    }
+    if (this.failsafeTimer) {
+      clearTimeout(this.failsafeTimer)
+      this.failsafeTimer = null
+    }
+    const wasPaused = this.paused
+    this.paused = false
+    if (wasPaused && !this.streamBackpressured) {
+      this.subprocess.resume?.()
+    }
+  }
+
   release(opts: { resume: boolean }): void {
     if (this.failsafeTimer) {
       clearTimeout(this.failsafeTimer)
       this.failsafeTimer = null
     }
-    if (!this.paused) {
+    if (!this.paused && !this.streamBackpressured) {
       return
     }
     this.paused = false
+    this.streamBackpressured = false
     if (opts.resume) {
       this.subprocess.resume?.()
     }

src/main/daemon/session-terminal-control.test.ts

@@ -175,6 +175,48 @@ describe('Session terminal control', () => {
   })
 
   describe('producer flow control', () => {
+    it('client resume and its failsafe cannot release daemon stream backpressure', () => {
+      createSession()
+      session.attachClient({ onData: () => {}, onExit: () => {} })
+      session.pauseProducer('stream')
+      session.pauseProducer()
+      session.resumeProducer()
+      expect(subprocess.resumeCalls).toBe(0)
+      session.pauseProducer()
+      vi.advanceTimersByTime(PRODUCER_PAUSE_FAILSAFE_MS * 2)
+      expect(subprocess.resumeCalls).toBe(0)
+      session.resumeProducer('stream')
+      expect(subprocess.resumeCalls).toBe(1)
+    })
+
+    it('draining the stream cannot release an outstanding client pause', () => {
+      createSession()
+      session.attachClient({ onData: () => {}, onExit: () => {} })
+      session.pauseProducer()
+      session.pauseProducer('stream')
+      session.resumeProducer('stream')
+      expect(subprocess.resumeCalls).toBe(0)
+      session.resumeProducer()
+      expect(subprocess.resumeCalls).toBe(1)
+    })
+
+    it('detach and termination release daemon stream backpressure', () => {
+      createSession()
+      const client = { onData: () => {}, onExit: () => {} }
+      const token = session.attachClient(client)
+      session.pauseProducer('stream')
+      session.detachClient(token)
+      expect(subprocess.resumeCalls).toBe(1)
+      session.pauseProducer('stream')
+      expect(subprocess.pauseCalls).toBe(1)
+      session.attachClient(client)
+      session.pauseProducer('stream')
+      session.kill()
+      expect(subprocess.resumeCalls).toBe(2)
+      session.pauseProducer('stream')
+      expect(subprocess.pauseCalls).toBe(2)
+    })
+
     it('auto-resumes when the owner loses the resume signal', () => {
       createSession()
       session.pauseProducer()

src/main/daemon/session.ts

@@ -169,15 +169,15 @@ export class Session {
 
   /** Producer-side flow control: stop reading the PTY fd so a flooding child blocks on write.
    *  Arms the lost-resume failsafe; re-pausing re-arms it. */
-  pauseProducer(): void {
+  pauseProducer(source?: 'stream'): void {
     if (this._state === 'exited' || this._disposed) {
       return
     }
-    this.producerPause.pause()
+    this.producerPause.pause(source, this.hasAttachedClients && !this.isTerminating)
   }
 
-  resumeProducer(): void {
-    this.producerPause.release({ resume: true })
+  resumeProducer(source?: 'stream'): void {
+    this.producerPause.resumeClient(source)
   }
 
   kill(): void {

src/main/daemon/terminal-host.ts

@@ -168,16 +168,16 @@ export class TerminalHost {
   }
 
   // Why null-not-throw (unlike write/resize): pause/resume are best-effort hints against a session that may have exited.
-  pauseProducer(sessionId: string): void {
+  pauseProducer(sessionId: string, source?: 'stream'): void {
     const session = this.sessions.get(sessionId)
     if (!session || !session.isAlive) {
       return
     }
-    session.pauseProducer()
+    session.pauseProducer(source)
   }
 
-  resumeProducer(sessionId: string): void {
-    this.sessions.get(sessionId)?.resumeProducer()
+  resumeProducer(sessionId: string, source?: 'stream'): void {
+    this.sessions.get(sessionId)?.resumeProducer(source)
   }
 
   kill(sessionId: string, opts: { immediate?: boolean } = {}): Promise<void> {