Add producer-side PTY flow control (watermarks + protocol v19)

Main now pauses the actual PTY when a pane's renderer-pending backlog
crosses the 256KB high watermark and resumes once it drains below the
32KB low watermark (wide hysteresis band so a draining queue cannot flap
pause/resume per flush slice). node-pty pause() stops the pty fd read, so
the kernel/ConPTY buffer fills and a flooding shell blocks on write —
flood-induced buffered lag becomes shell blocking instead of unbounded
main-process buffering (terminal-performance-initiative §5).

Transport: new fire-and-forget pausePty/resumePty daemon notifications
(protocol v19; 18 added to PREVIOUS_DAEMON_PROTOCOL_VERSIONS), routed
DaemonServer -> TerminalHost -> Session -> subprocess pause()/resume().
LocalPtyProvider pauses node-pty directly. Router/degraded providers
forward; IPtyProvider gains optional pauseProducer/resumeProducer.

Safety invariants:
- Lost-resume failsafe: daemon Session auto-resumes 5s after a pause with
  no matching resume; main re-asserts the pause at most once per 5s while
  still above the high watermark, so a lost resume can never wedge a shell
  and a sustained flood stays throttled.
- Resume on every teardown path: Session kill/exit/dispose/detach; main
  releases on pty exit and on window-destroyed bookkeeping wipes; the
  adapter owes paused sessions a resumePty on the next connect after a
  socket drop.
- Providers without support (SSH relay, legacy protocol <= v18) no-op
  silently, and the scrollback-scaled pending-output cap still bounds
  main memory when pause is unavailable.
- Kill switch: PRODUCER_FLOW_CONTROL_ENABLED in ipc/pty.ts flips the
  whole mechanism off in one line.

daemon-errors.ts is split out of types.ts to stay under the max-lines cap.

Tests: watermark transitions/hysteresis/re-assert (controller unit),
lost-resume failsafe + resume-on-kill/exit/dispose/detach (session),
notification routing + v18 gating + reconnect owed-resume (adapter),
direct pause/resume (local provider), and a flood test asserting pause
fires once, pending stays bounded at HIGH + one chunk, and resume fires
once after drain (ipc/pty).

Co-authored-by: Orca <help@stably.ai>

src/main/daemon/daemon-errors.ts

@@ -0,0 +1,23 @@
+// Error classes shared across the daemon protocol boundary (client, server,
+// host). Split from types.ts, which is capped for wire-shape declarations.
+
+export class TerminalAttachCanceledError extends Error {
+  constructor(sessionId: string) {
+    super(`Attach canceled for session ${sessionId}`)
+    this.name = 'TerminalAttachCanceledError'
+  }
+}
+
+export class DaemonProtocolError extends Error {
+  constructor(message: string) {
+    super(message)
+    this.name = 'DaemonProtocolError'
+  }
+}
+
+export class SessionNotFoundError extends Error {
+  constructor(sessionId: string) {
+    super(`Session not found: ${sessionId}`)
+    this.name = 'SessionNotFoundError'
+  }
+}

src/main/daemon/daemon-pty-adapter.test.ts

@@ -3,6 +3,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { mkdtempSync, mkdirSync, rmSync, existsSync, readFileSync, writeFileSync } from 'node:fs'
+import { DaemonClient } from './client'
 import { DaemonPtyAdapter } from './daemon-pty-adapter'
 import { DaemonServer } from './daemon-server'
 import { getHistorySessionDirName } from './history-paths'
@@ -29,6 +30,8 @@ function createTestDir(): string {
 }
 
 function createMockSubprocess(): SubprocessHandle & {
+  pause: ReturnType<typeof vi.fn<() => void>>
+  resume: ReturnType<typeof vi.fn<() => void>>
   _simulateData: (data: string) => void
   _simulateExit: (code: number) => void
 } {
@@ -41,6 +44,8 @@ function createMockSubprocess(): SubprocessHandle & {
     getForegroundProcess: vi.fn(() => null),
     write: vi.fn(),
     resize: vi.fn(),
+    pause: vi.fn<() => void>(),
+    resume: vi.fn<() => void>(),
     kill: vi.fn(() => setTimeout(() => onExitCb?.(0), 5)),
     forceKill: vi.fn(),
     signal: vi.fn(),
@@ -178,6 +183,75 @@ describe('DaemonPtyAdapter (IPtyProvider)', () => {
     })
   })
 
+  describe('producer flow control', () => {
+    it('routes pausePty/resumePty notifications to the daemon session subprocess', async () => {
+      const { id } = await adapter.spawn({ cols: 80, rows: 24 })
+
+      adapter.pauseProducer(id)
+      await waitFor(() => lastSubprocess.pause.mock.calls.length > 0)
+
+      adapter.resumeProducer(id)
+      await waitFor(() => lastSubprocess.resume.mock.calls.length > 0)
+    })
+
+    it('sends pause/resume as fire-and-forget notifications on the current protocol', async () => {
+      const { id } = await adapter.spawn({ cols: 80, rows: 24 })
+      const notifySpy = vi.spyOn(DaemonClient.prototype, 'notify')
+      try {
+        adapter.pauseProducer(id)
+        adapter.resumeProducer(id)
+        expect(notifySpy).toHaveBeenCalledWith('pausePty', { sessionId: id })
+        expect(notifySpy).toHaveBeenCalledWith('resumePty', { sessionId: id })
+      } finally {
+        notifySpy.mockRestore()
+      }
+    })
+
+    it('never sends pause/resume notifications on a legacy protocol version', () => {
+      const notifySpy = vi.spyOn(DaemonClient.prototype, 'notify')
+      const legacy = new DaemonPtyAdapter({ socketPath, tokenPath, protocolVersion: 18 })
+      try {
+        legacy.pauseProducer('legacy-session')
+        legacy.resumeProducer('legacy-session')
+        expect(notifySpy).not.toHaveBeenCalled()
+      } finally {
+        legacy.dispose()
+        notifySpy.mockRestore()
+      }
+    })
+
+    it('owes paused sessions a resumePty on the next connect after a socket drop', async () => {
+      const { id } = await adapter.spawn({ cols: 80, rows: 24 })
+      adapter.pauseProducer(id)
+      await waitFor(() => lastSubprocess.pause.mock.calls.length > 0)
+
+      // Drop the daemon out from under the adapter: the in-flight pause has no
+      // matching resume anymore.
+      await server.shutdown()
+      await waitFor(() => !(adapter as unknown as { client: DaemonClient }).client.isConnected())
+
+      server = new DaemonServer({
+        socketPath,
+        tokenPath,
+        spawnSubprocess: (opts) => {
+          lastSpawnOpts = opts
+          lastSubprocess = createMockSubprocess()
+          return lastSubprocess
+        }
+      })
+      await server.start()
+
+      const notifySpy = vi.spyOn(DaemonClient.prototype, 'notify')
+      try {
+        // Any reconnecting operation must flush the owed resume first.
+        await adapter.listProcesses()
+        expect(notifySpy).toHaveBeenCalledWith('resumePty', { sessionId: id })
+      } finally {
+        notifySpy.mockRestore()
+      }
+    })
+  })
+
   describe('getAppliedSize', () => {
     it('reports the spawn dims before any resize', async () => {
       const { id } = await adapter.spawn({ cols: 80, rows: 24 })

src/main/daemon/daemon-pty-adapter.ts

@@ -98,6 +98,14 @@ export class DaemonPtyAdapter implements IPtyProvider {
   // Why: incremental checkpoints require the takePendingOutput RPC (v13+).
   // Against older daemons the tick falls back to full-snapshot checkpoints.
   private supportsIncrementalCheckpoints: boolean
+  // Why: producer pause/resume notifications require v19+; legacy daemons
+  // must never see them, so gating makes them silent no-ops there.
+  private supportsProducerFlowControl: boolean
+  private pausedProducerSessionIds = new Set<string>()
+  // Why: a daemon that survives a socket drop can still hold a pause whose
+  // resume died with the connection. Owe those sessions a resume on the next
+  // connect; the daemon's 5s failsafe covers the window in between.
+  private producerResumesOwedOnReconnect = new Set<string>()
   private static CHECKPOINT_INTERVAL_MS = 5_000
   // Why: a streaming session (build logs, `yes`) re-triggers a full multi-MB
   // snapshot checkpoint on every 5s tick via pending-buffer overflow or the
@@ -122,6 +130,13 @@ export class DaemonPtyAdapter implements IPtyProvider {
     this.respawnFn = opts.respawn ?? null
     this.supportsCheckpoints = this.protocolVersion >= 4
     this.supportsIncrementalCheckpoints = this.protocolVersion >= 13
+    this.supportsProducerFlowControl = this.protocolVersion >= 19
+    this.client.onDisconnected(() => {
+      for (const id of this.pausedProducerSessionIds) {
+        this.producerResumesOwedOnReconnect.add(id)
+      }
+      this.pausedProducerSessionIds.clear()
+    })
   }
 
   getHistoryManager(): HistoryManager | null {
@@ -319,6 +334,23 @@ export class DaemonPtyAdapter implements IPtyProvider {
     this.client.notify('resize', { sessionId: id, cols, rows })
   }
 
+  pauseProducer(id: string): void {
+    if (!this.supportsProducerFlowControl) {
+      return
+    }
+    this.pausedProducerSessionIds.add(id)
+    this.client.notify('pausePty', { sessionId: id })
+  }
+
+  resumeProducer(id: string): void {
+    this.producerResumesOwedOnReconnect.delete(id)
+    if (!this.supportsProducerFlowControl) {
+      return
+    }
+    this.pausedProducerSessionIds.delete(id)
+    this.client.notify('resumePty', { sessionId: id })
+  }
+
   async shutdown(id: string, opts: { immediate?: boolean; keepHistory?: boolean }): Promise<void> {
     // Why: sleep/exact-stop kills the live PTY before the periodic checkpoint may run.
     // Force a final snapshot so wake can restore the pane users left.
@@ -566,6 +598,8 @@ export class DaemonPtyAdapter implements IPtyProvider {
     this.dirtySessionVersions.clear()
     this.lastFullCheckpointAt.clear()
     this.sessionsNeedingFullCheckpoint.clear()
+    this.pausedProducerSessionIds.clear()
+    this.producerResumesOwedOnReconnect.clear()
     this.stopCheckpointTimer()
     for (const id of ids) {
       this.coldRestoreCache.delete(id)
@@ -627,6 +661,8 @@ export class DaemonPtyAdapter implements IPtyProvider {
     this.dirtySessionVersions.clear()
     this.lastFullCheckpointAt.clear()
     this.coldRestoreCache.clear()
+    this.pausedProducerSessionIds.clear()
+    this.producerResumesOwedOnReconnect.clear()
     this.removeEventListener?.()
     this.removeEventListener = null
     // Why: final checkpoints are written daemon-side in TerminalHost.dispose()
@@ -663,6 +699,13 @@ export class DaemonPtyAdapter implements IPtyProvider {
     this.dirtySessionVersions.clear()
     this.lastFullCheckpointAt.clear()
     this.coldRestoreCache.clear()
+    // Why: the detached daemon keeps these PTYs alive for warm reattach; a
+    // pause left behind would block their shells for a failsafe window.
+    for (const id of this.pausedProducerSessionIds) {
+      this.client.notify('resumePty', { sessionId: id })
+    }
+    this.pausedProducerSessionIds.clear()
+    this.producerResumesOwedOnReconnect.clear()
     this.removeEventListener?.()
     this.removeEventListener = null
     this.client.disconnect()
@@ -672,6 +715,19 @@ export class DaemonPtyAdapter implements IPtyProvider {
     await this.client.ensureConnected()
     this.setupEventRouting()
     this.scheduleCheckpointTimer()
+    this.flushOwedProducerResumes()
+  }
+
+  private flushOwedProducerResumes(): void {
+    if (this.producerResumesOwedOnReconnect.size === 0) {
+      return
+    }
+    for (const id of this.producerResumesOwedOnReconnect) {
+      // Why: resuming a session the fresh daemon doesn't know is a harmless
+      // no-op; leaving a survivor paused would waste 5s of failsafe latency.
+      this.client.notify('resumePty', { sessionId: id })
+    }
+    this.producerResumesOwedOnReconnect.clear()
   }
 
   private stopCheckpointTimer(): void {
@@ -1013,6 +1069,10 @@ export class DaemonPtyAdapter implements IPtyProvider {
       } else if (event.event === 'exit') {
         this.activeSessionIds.delete(event.sessionId)
         this.dirtySessionVersions.delete(event.sessionId)
+        // Why: an exited session must not be owed a resume on reconnect — a
+        // reused sessionId would receive a stray resumePty.
+        this.pausedProducerSessionIds.delete(event.sessionId)
+        this.producerResumesOwedOnReconnect.delete(event.sessionId)
         if (!this.sleepRestoreSessionIds.has(event.sessionId)) {
           this.coldRestoreCache.delete(event.sessionId)
         }

src/main/daemon/daemon-pty-router.ts

@@ -71,6 +71,14 @@ export class DaemonPtyRouter implements IPtyProvider {
     this.adapterFor(id).resize(id, cols, rows)
   }
 
+  pauseProducer(id: string): void {
+    this.adapterFor(id).pauseProducer(id)
+  }
+
+  resumeProducer(id: string): void {
+    this.adapterFor(id).resumeProducer(id)
+  }
+
   async shutdown(id: string, opts: { immediate?: boolean; keepHistory?: boolean }): Promise<void> {
     await this.adapterFor(id).shutdown(id, opts)
     // Why: sleep passes keepHistory=true and re-spawns against the same

src/main/daemon/daemon-server.ts

@@ -347,6 +347,14 @@ export class DaemonServer {
         }
         return {}
 
+      case 'pausePty':
+        this.host.pauseProducer(request.payload.sessionId)
+        return {}
+
+      case 'resumePty':
+        this.host.resumeProducer(request.payload.sessionId)
+        return {}
+
       case 'kill':
         this.lastInputAtBySessionId.delete(request.payload.sessionId)
         this.host.kill(request.payload.sessionId, { immediate: request.payload.immediate })

src/main/daemon/degraded-daemon-pty-provider.ts

@@ -88,6 +88,14 @@ export class DegradedDaemonPtyProvider implements IPtyProvider {
     this.providerFor(id).resize(id, cols, rows)
   }
 
+  pauseProducer(id: string): void {
+    this.providerFor(id).pauseProducer?.(id)
+  }
+
+  resumeProducer(id: string): void {
+    this.providerFor(id).resumeProducer?.(id)
+  }
+
   async shutdown(id: string, opts: { immediate?: boolean; keepHistory?: boolean }): Promise<void> {
     await this.providerFor(id).shutdown(id, opts)
     if (!opts.keepHistory) {

src/main/daemon/pty-subprocess.ts

@@ -1023,6 +1023,30 @@ export function createPtySubprocess(opts: PtySubprocessOptions): SubprocessHandl
         dead = true
       }
     },
+    // Why pause/resume work on Windows too: node-pty's base Terminal
+    // implements both as socket pause/resume (lib/terminal.js), and
+    // WindowsTerminal wires _socket to the ConPTY conout pipe — pausing stops
+    // conout reads so ConPTY's bounded buffer backpressures the child.
+    pause: () => {
+      if (dead) {
+        return
+      }
+      try {
+        proc.pause()
+      } catch {
+        /* native handle already torn down — flow control is best-effort */
+      }
+    },
+    resume: () => {
+      if (dead) {
+        return
+      }
+      try {
+        proc.resume()
+      } catch {
+        /* native handle already torn down — flow control is best-effort */
+      }
+    },
     kill: () => {
       if (dead) {
         return

src/main/daemon/session.test.ts

@@ -1,5 +1,5 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
-import { Session } from './session'
+import { PRODUCER_PAUSE_FAILSAFE_MS, Session } from './session'
 import type { SessionState, ShellReadyState } from './types'
 
 // Stub the subprocess — Session talks to it via an interface, not child_process directly.
@@ -10,6 +10,8 @@ function createMockSubprocess() {
   let onExit: ((code: number) => void) | null = null
   let killed = false
   let pid = 12345
+  let pauseCalls = 0
+  let resumeCalls = 0
 
   return {
     written,
@@ -20,6 +22,12 @@ function createMockSubprocess() {
     get pid() {
       return pid
     },
+    get pauseCalls() {
+      return pauseCalls
+    },
+    get resumeCalls() {
+      return resumeCalls
+    },
     getForegroundProcess(): string | null {
       return null
     },
@@ -27,6 +35,12 @@ function createMockSubprocess() {
       written.push(data)
     },
     resize(_cols: number, _rows: number) {},
+    pause() {
+      pauseCalls++
+    },
+    resume() {
+      resumeCalls++
+    },
     kill() {
       killed = true
       // Simulate async exit
@@ -508,4 +522,103 @@ describe('Session', () => {
       expect(session.state).toBe('exited')
     })
   })
+
+  describe('producer flow control', () => {
+    it('pauses the subprocess and auto-resumes via the lost-resume failsafe', () => {
+      createSession()
+      session.pauseProducer()
+      expect(subprocess.pauseCalls).toBe(1)
+      expect(subprocess.resumeCalls).toBe(0)
+
+      vi.advanceTimersByTime(PRODUCER_PAUSE_FAILSAFE_MS - 1)
+      expect(subprocess.resumeCalls).toBe(0)
+      vi.advanceTimersByTime(1)
+      expect(subprocess.resumeCalls).toBe(1)
+    })
+
+    it('resumeProducer resumes once and cancels the failsafe timer', () => {
+      createSession()
+      session.pauseProducer()
+      session.resumeProducer()
+      expect(subprocess.resumeCalls).toBe(1)
+
+      vi.advanceTimersByTime(PRODUCER_PAUSE_FAILSAFE_MS * 2)
+      expect(subprocess.resumeCalls).toBe(1)
+    })
+
+    it('resumeProducer without a matching pause is a no-op', () => {
+      createSession()
+      session.resumeProducer()
+      expect(subprocess.resumeCalls).toBe(0)
+    })
+
+    it('re-pausing re-arms the failsafe window', () => {
+      createSession()
+      session.pauseProducer()
+      vi.advanceTimersByTime(PRODUCER_PAUSE_FAILSAFE_MS - 1_000)
+      session.pauseProducer()
+
+      vi.advanceTimersByTime(PRODUCER_PAUSE_FAILSAFE_MS - 1)
+      expect(subprocess.resumeCalls).toBe(0)
+      vi.advanceTimersByTime(1)
+      expect(subprocess.resumeCalls).toBe(1)
+    })
+
+    it('kill() resumes a paused producer before signalling the child', () => {
+      createSession()
+      session.pauseProducer()
+      session.kill()
+      expect(subprocess.resumeCalls).toBe(1)
+      expect(subprocess.killed).toBe(true)
+    })
+
+    it('dispose() resumes a paused producer and clears the failsafe', () => {
+      createSession()
+      session.pauseProducer()
+      session.dispose()
+      expect(subprocess.resumeCalls).toBe(1)
+      expect(vi.getTimerCount()).toBe(0)
+    })
+
+    it('subprocess exit clears the failsafe without resuming a reaped child', () => {
+      createSession()
+      session.pauseProducer()
+      subprocess.simulateExit(0)
+      vi.advanceTimersByTime(PRODUCER_PAUSE_FAILSAFE_MS * 2)
+      expect(subprocess.resumeCalls).toBe(0)
+    })
+
+    it('ignores pauseProducer on an exited session', () => {
+      createSession()
+      subprocess.simulateExit(0)
+      session.pauseProducer()
+      expect(subprocess.pauseCalls).toBe(0)
+      expect(vi.getTimerCount()).toBe(0)
+    })
+
+    it('detaching the last client resumes a paused producer', () => {
+      createSession()
+      const token = session.attachClient({ onData: () => {}, onExit: () => {} })
+      session.pauseProducer()
+      session.detachClient(token)
+      expect(subprocess.resumeCalls).toBe(1)
+    })
+
+    it('keeps the pause while another client is still attached', () => {
+      createSession()
+      const token = session.attachClient({ onData: () => {}, onExit: () => {} })
+      session.attachClient({ onData: () => {}, onExit: () => {} })
+      session.pauseProducer()
+      session.detachClient(token)
+      expect(subprocess.resumeCalls).toBe(0)
+    })
+
+    it('detachAllClients resumes a paused producer', () => {
+      createSession()
+      session.attachClient({ onData: () => {}, onExit: () => {} })
+      session.pauseProducer()
+      session.detachAllClients()
+      expect(subprocess.resumeCalls).toBe(1)
+    })
+  })
 })

src/main/daemon/session.ts

@@ -30,6 +30,11 @@ const KILL_TIMEOUT_MS = 5_000
 // Worst-case wire size for a full take is ~6x this (each control char
 // JSON-escapes to six bytes) and must stay under NDJSON_MAX_LINE_BYTES (16MB).
 const PENDING_OUTPUT_MAX_BYTES = 2 * 1024 * 1024
+// Why: producer pause is requested over a fire-and-forget notification, so the
+// matching resume can be lost (main crash, dropped socket). A lost resume must
+// never wedge a shell: auto-resume after this window; a still-flooded main
+// re-asserts the pause on its next watermark check.
+export const PRODUCER_PAUSE_FAILSAFE_MS = 5_000
 
 export type SubprocessHandle = {
   pid: number
@@ -41,6 +46,11 @@ export type SubprocessHandle = {
   startupCommandDeliveredInShellArgs?: boolean
   write(data: string): void
   resize(cols: number, rows: number): void
+  /** Stop reading the PTY fd (node-pty pause()) so the kernel/ConPTY buffer
+   *  fills and a flooding child blocks on write. Optional: handles that
+   *  cannot pause simply omit it and flow control degrades to a no-op. */
+  pause?(): void
+  resume?(): void
   kill(): void
   forceKill(): void
   signal(sig: string): void
@@ -94,6 +104,8 @@ export class Session {
   private pendingOutputBytes = 0
   private pendingOutputOverflowed = false
   private pendingOutputSeq = 0
+  private producerPaused = false
+  private producerPauseFailsafeTimer: ReturnType<typeof setTimeout> | null = null
 
   constructor(opts: SessionOptions) {
     this.sessionId = opts.sessionId
@@ -180,12 +192,52 @@ export class Session {
     this.subprocess.resize(cols, rows)
   }
 
+  /** Producer-side flow control: stop reading the PTY fd so the flooding
+   *  child blocks on write (kernel backpressure). Arms the lost-resume
+   *  failsafe; re-pausing re-arms it (main re-asserts during long floods). */
+  pauseProducer(): void {
+    if (this._state === 'exited' || this._disposed) {
+      return
+    }
+    this.producerPaused = true
+    this.subprocess.pause?.()
+    if (this.producerPauseFailsafeTimer) {
+      clearTimeout(this.producerPauseFailsafeTimer)
+    }
+    this.producerPauseFailsafeTimer = setTimeout(() => {
+      this.producerPauseFailsafeTimer = null
+      this.producerPaused = false
+      this.subprocess.resume?.()
+    }, PRODUCER_PAUSE_FAILSAFE_MS)
+  }
+
+  resumeProducer(): void {
+    this.releaseProducerPause({ resume: true })
+  }
+
+  private releaseProducerPause(opts: { resume: boolean }): void {
+    if (this.producerPauseFailsafeTimer) {
+      clearTimeout(this.producerPauseFailsafeTimer)
+      this.producerPauseFailsafeTimer = null
+    }
+    if (!this.producerPaused) {
+      return
+    }
+    this.producerPaused = false
+    if (opts.resume) {
+      this.subprocess.resume?.()
+    }
+  }
+
   kill(): void {
     if (this._state === 'exited' || this._isTerminating) {
       return
     }
     this._isTerminating = true
 
+    // Why: a paused child can be blocked inside write(); resume before
+    // signalling so it can run signal handlers and actually exit.
+    this.releaseProducerPause({ resume: true })
     this.subprocess.kill()
 
     this.killTimer = setTimeout(() => {
@@ -213,10 +265,16 @@ export class Session {
     if (idx !== -1) {
       this.attachedClients.splice(idx, 1)
     }
+    // Why: with no attached client, nobody will ever send resumePty — a
+    // paused shell would sit wedged until the failsafe. Resume eagerly.
+    if (this.attachedClients.length === 0) {
+      this.releaseProducerPause({ resume: true })
+    }
   }
 
   detachAllClients(): void {
     this.attachedClients.length = 0
+    this.releaseProducerPause({ resume: true })
   }
 
   getSnapshot(): TerminalSnapshot | null {
@@ -377,6 +435,9 @@ export class Session {
       return
     }
     this._disposed = true
+    // Why: never leave a paused fd behind on any teardown path — the handle's
+    // own dead-guard makes this a no-op when the child is already reaped.
+    this.releaseProducerPause({ resume: true })
     if (this.killTimer) {
       clearTimeout(this.killTimer)
       this.killTimer = null
@@ -460,6 +521,9 @@ export class Session {
 
     this._exitCode = code
     this._state = 'exited'
+    // Why resume:false — the child is reaped, so there is nothing to unblock;
+    // only the failsafe timer must not outlive the session.
+    this.releaseProducerPause({ resume: false })
     this.releaseHeldShellReadyBytes()
 
     if (this.killTimer) {

src/main/daemon/terminal-host.ts

@@ -183,6 +183,21 @@ export class TerminalHost {
     this.getAliveSession(sessionId).resize(cols, rows)
   }
 
+  // Why null-not-throw (unlike write/resize): pause/resume are best-effort
+  // flow-control hints; a session that exited while the notify was in flight
+  // must not surface an error or a synthetic exit.
+  pauseProducer(sessionId: string): void {
+    const session = this.sessions.get(sessionId)
+    if (!session || !session.isAlive) {
+      return
+    }
+    session.pauseProducer()
+  }
+
+  resumeProducer(sessionId: string): void {
+    this.sessions.get(sessionId)?.resumeProducer()
+  }
+
   kill(sessionId: string, opts: { immediate?: boolean } = {}): void {
     const session = this.getAliveSession(sessionId)
     this.recordTombstone(sessionId)

src/main/daemon/types.ts

@@ -7,9 +7,9 @@ import type { StartupCommandDelivery } from '../../shared/codex-startup-delivery
 // when daemon-baked behavior cannot be delivered by on-disk wrapper refresh.
 // Why: bump when adding daemon wire behavior so same-version old daemons do
 // not silently accept the handshake and then reject new RPCs.
-export const PROTOCOL_VERSION = 18
+export const PROTOCOL_VERSION = 19
 export const PREVIOUS_DAEMON_PROTOCOL_VERSIONS = [
-  1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17
+  1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18
 ] as const
 
 // ─── Session State Machine ──────────────────────────────────────────
@@ -133,6 +133,26 @@ export type ResizeRequest = {
   }
 }
 
+// ─── Producer flow control (v19+) ───────────────────────────────────
+// Why fire-and-forget notifications (like write/resize): pause/resume ride the
+// hot data path and are best-effort — the daemon-side 5s failsafe, not an RPC
+// reply, is what guarantees a paused shell can never stay wedged.
+export type PausePtyRequest = {
+  id: string
+  type: 'pausePty'
+  payload: {
+    sessionId: string
+  }
+}
+
+export type ResumePtyRequest = {
+  id: string
+  type: 'resumePty'
+  payload: {
+    sessionId: string
+  }
+}
+
 export type KillRequest = {
   id: string
   type: 'kill'
@@ -276,6 +296,8 @@ export type DaemonRequest =
   | CancelCreateOrAttachRequest
   | WriteRequest
   | ResizeRequest
+  | PausePtyRequest
+  | ResumePtyRequest
   | KillRequest
   | SignalRequest
   | ListSessionsRequest
@@ -396,23 +418,10 @@ export const FRAME_MAX_PAYLOAD = 1024 * 1024 // 1MB
 export const NOTIFY_PREFIX = 'notify_'
 
 // ─── Error types ────────────────────────────────────────────────────
-export class TerminalAttachCanceledError extends Error {
-  constructor(sessionId: string) {
-    super(`Attach canceled for session ${sessionId}`)
-    this.name = 'TerminalAttachCanceledError'
-  }
-}
-
-export class DaemonProtocolError extends Error {
-  constructor(message: string) {
-    super(message)
-    this.name = 'DaemonProtocolError'
-  }
-}
-
-export class SessionNotFoundError extends Error {
-  constructor(sessionId: string) {
-    super(`Session not found: ${sessionId}`)
-    this.name = 'SessionNotFoundError'
-  }
-}
+// Re-exported so existing importers of `./types` keep working; the classes
+// live in daemon-errors.ts (this file is capped for wire-shape declarations).
+export {
+  TerminalAttachCanceledError,
+  DaemonProtocolError,
+  SessionNotFoundError
+} from './daemon-errors'

src/main/ipc/pty-producer-flow-control.test.ts

@@ -0,0 +1,137 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import {
+  PRODUCER_FLOW_HIGH_WATERMARK_CHARS,
+  PRODUCER_FLOW_LOW_WATERMARK_CHARS,
+  PRODUCER_PAUSE_REASSERT_INTERVAL_MS,
+  PtyProducerFlowController
+} from './pty-producer-flow-control'
+
+const HIGH = PRODUCER_FLOW_HIGH_WATERMARK_CHARS
+const LOW = PRODUCER_FLOW_LOW_WATERMARK_CHARS
+
+describe('PtyProducerFlowController', () => {
+  let pauseProducer: ReturnType<typeof vi.fn<(id: string) => void>>
+  let resumeProducer: ReturnType<typeof vi.fn<(id: string) => void>>
+  let controller: PtyProducerFlowController
+
+  beforeEach(() => {
+    vi.useFakeTimers()
+    pauseProducer = vi.fn<(id: string) => void>()
+    resumeProducer = vi.fn<(id: string) => void>()
+    controller = new PtyProducerFlowController({
+      pauseProducer,
+      resumeProducer
+    })
+  })
+
+  afterEach(() => {
+    vi.useRealTimers()
+  })
+
+  it('does not pause at or below the high watermark', () => {
+    controller.update('pty-1', 0)
+    controller.update('pty-1', LOW)
+    controller.update('pty-1', HIGH)
+    expect(pauseProducer).not.toHaveBeenCalled()
+    expect(controller.isPaused('pty-1')).toBe(false)
+  })
+
+  it('pauses exactly once when pending crosses the high watermark, not per chunk', () => {
+    controller.update('pty-1', HIGH + 1)
+    controller.update('pty-1', HIGH + 64 * 1024)
+    controller.update('pty-1', HIGH + 128 * 1024)
+    expect(pauseProducer).toHaveBeenCalledTimes(1)
+    expect(pauseProducer).toHaveBeenCalledWith('pty-1')
+    expect(controller.isPaused('pty-1')).toBe(true)
+  })
+
+  it('resumes exactly once when pending drains below the low watermark', () => {
+    controller.update('pty-1', HIGH + 1)
+    controller.update('pty-1', LOW - 1)
+    expect(resumeProducer).toHaveBeenCalledTimes(1)
+    expect(resumeProducer).toHaveBeenCalledWith('pty-1')
+    expect(controller.isPaused('pty-1')).toBe(false)
+    // A second drain report on the now-unpaused pty must not resume again.
+    controller.update('pty-1', 0)
+    expect(resumeProducer).toHaveBeenCalledTimes(1)
+  })
+
+  it('holds hysteresis: no flapping while pending sits between the watermarks', () => {
+    controller.update('pty-1', HIGH + 1)
+    expect(pauseProducer).toHaveBeenCalledTimes(1)
+    // Draining but still above LOW: stay paused, no extra calls either way.
+    controller.update('pty-1', HIGH - 16 * 1024)
+    controller.update('pty-1', 128 * 1024)
+    controller.update('pty-1', LOW)
+    expect(pauseProducer).toHaveBeenCalledTimes(1)
+    expect(resumeProducer).not.toHaveBeenCalled()
+    expect(controller.isPaused('pty-1')).toBe(true)
+    // An unpaused pty hovering in the same band must not pause.
+    controller.update('pty-2', LOW + 1)
+    controller.update('pty-2', HIGH)
+    expect(pauseProducer).toHaveBeenCalledTimes(1)
+  })
+
+  it('re-asserts the pause after the failsafe interval while still flooded', () => {
+    controller.update('pty-1', HIGH + 1)
+    expect(pauseProducer).toHaveBeenCalledTimes(1)
+
+    // Within the failsafe window: no re-assert even far above HIGH.
+    vi.advanceTimersByTime(PRODUCER_PAUSE_REASSERT_INTERVAL_MS - 1)
+    controller.update('pty-1', HIGH * 4)
+    expect(pauseProducer).toHaveBeenCalledTimes(1)
+
+    // After the window (daemon failsafe has auto-resumed by now): re-pause.
+    vi.advanceTimersByTime(1)
+    controller.update('pty-1', HIGH * 4)
+    expect(pauseProducer).toHaveBeenCalledTimes(2)
+
+    // The re-assert re-stamps the clock — no immediate third pause.
+    controller.update('pty-1', HIGH * 4)
+    expect(pauseProducer).toHaveBeenCalledTimes(2)
+  })
+
+  it('release resumes only ptys that are actually paused', () => {
+    controller.update('paused-pty', HIGH + 1)
+    controller.release('paused-pty')
+    controller.release('never-paused-pty')
+    expect(resumeProducer).toHaveBeenCalledTimes(1)
+    expect(resumeProducer).toHaveBeenCalledWith('paused-pty')
+    expect(controller.isPaused('paused-pty')).toBe(false)
+  })
+
+  it('releaseAll resumes every paused pty', () => {
+    controller.update('pty-1', HIGH + 1)
+    controller.update('pty-2', HIGH + 1)
+    controller.update('pty-3', LOW)
+    controller.releaseAll()
+    expect(resumeProducer).toHaveBeenCalledTimes(2)
+    expect(resumeProducer).toHaveBeenCalledWith('pty-1')
+    expect(resumeProducer).toHaveBeenCalledWith('pty-2')
+    expect(controller.isPaused('pty-1')).toBe(false)
+    expect(controller.isPaused('pty-2')).toBe(false)
+  })
+
+  it('keeps bookkeeping consistent when the transport throws', () => {
+    pauseProducer.mockImplementation(() => {
+      throw new Error('provider gone')
+    })
+    resumeProducer.mockImplementation(() => {
+      throw new Error('provider gone')
+    })
+    expect(() => controller.update('pty-1', HIGH + 1)).not.toThrow()
+    expect(controller.isPaused('pty-1')).toBe(true)
+    expect(() => controller.update('pty-1', 0)).not.toThrow()
+    expect(controller.isPaused('pty-1')).toBe(false)
+  })
+
+  it('tracks watermark state per pty independently', () => {
+    controller.update('pty-1', HIGH + 1)
+    controller.update('pty-2', HIGH + 1)
+    controller.update('pty-1', 0)
+    expect(pauseProducer).toHaveBeenCalledTimes(2)
+    expect(resumeProducer).toHaveBeenCalledTimes(1)
+    expect(controller.isPaused('pty-1')).toBe(false)
+    expect(controller.isPaused('pty-2')).toBe(true)
+  })
+})

src/main/ipc/pty-producer-flow-control.ts

@@ -0,0 +1,105 @@
+// Producer-side PTY flow control (notes/terminal-performance-initiative.md §5).
+// Main tracks per-PTY renderer-pending chars; past HIGH it asks the provider to
+// pause the actual PTY read (node-pty pause() → kernel backpressure → the
+// flooding shell blocks on write), and below LOW it resumes. The wide
+// HIGH/LOW gap is deliberate hysteresis so a draining queue cannot flap
+// pause/resume once per flush slice.
+
+export const PRODUCER_FLOW_HIGH_WATERMARK_CHARS = 256 * 1024
+export const PRODUCER_FLOW_LOW_WATERMARK_CHARS = 32 * 1024
+// Why: the daemon auto-resumes a pause after its 5s lost-resume failsafe. If
+// pending is still above HIGH after that window, the pause must be re-asserted
+// or a sustained flood would run unthrottled after the first failsafe fires.
+export const PRODUCER_PAUSE_REASSERT_INTERVAL_MS = 5_000
+
+export type ProducerFlowControlTransport = {
+  pauseProducer: (id: string) => void
+  resumeProducer: (id: string) => void
+}
+
+export class PtyProducerFlowController {
+  private transport: ProducerFlowControlTransport
+  private highWatermarkChars: number
+  private lowWatermarkChars: number
+  private reassertIntervalMs: number
+  private pausedAtByPty = new Map<string, number>()
+
+  constructor(
+    transport: ProducerFlowControlTransport,
+    opts: {
+      highWatermarkChars?: number
+      lowWatermarkChars?: number
+      reassertIntervalMs?: number
+    } = {}
+  ) {
+    this.transport = transport
+    this.highWatermarkChars = opts.highWatermarkChars ?? PRODUCER_FLOW_HIGH_WATERMARK_CHARS
+    this.lowWatermarkChars = opts.lowWatermarkChars ?? PRODUCER_FLOW_LOW_WATERMARK_CHARS
+    this.reassertIntervalMs = opts.reassertIntervalMs ?? PRODUCER_PAUSE_REASSERT_INTERVAL_MS
+  }
+
+  /** Reports the current pending chars for a PTY. Fires pause exactly once at
+   *  the HIGH crossing (re-asserted only after the failsafe interval) and
+   *  resume exactly once when pending drains below LOW. */
+  update(id: string, pendingChars: number): void {
+    const pausedAt = this.pausedAtByPty.get(id)
+    if (pausedAt === undefined) {
+      if (pendingChars > this.highWatermarkChars) {
+        this.pausedAtByPty.set(id, Date.now())
+        this.safePause(id)
+      }
+      return
+    }
+    if (pendingChars < this.lowWatermarkChars) {
+      this.pausedAtByPty.delete(id)
+      this.safeResume(id)
+      return
+    }
+    if (
+      pendingChars > this.highWatermarkChars &&
+      Date.now() - pausedAt >= this.reassertIntervalMs
+    ) {
+      this.pausedAtByPty.set(id, Date.now())
+      this.safePause(id)
+    }
+  }
+
+  /** Resumes a PTY if it was paused. For teardown paths (exit, kill) where
+   *  the pending bookkeeping is being dropped rather than drained. */
+  release(id: string): void {
+    if (this.pausedAtByPty.delete(id)) {
+      this.safeResume(id)
+    }
+  }
+
+  /** Resumes every paused PTY. For wholesale bookkeeping wipes (window
+   *  destroyed) — a local PTY left paused here would stay wedged forever. */
+  releaseAll(): void {
+    // Deleting the visited entry during Map key iteration is spec-safe.
+    for (const id of this.pausedAtByPty.keys()) {
+      this.release(id)
+    }
+  }
+
+  isPaused(id: string): boolean {
+    return this.pausedAtByPty.has(id)
+  }
+
+  // Why swallow: pause/resume are optimizations riding the terminal data
+  // path — a provider throw must never break delivery or exit handling.
+  private safePause(id: string): void {
+    try {
+      this.transport.pauseProducer(id)
+    } catch {
+      /* best-effort */
+    }
+  }
+
+  private safeResume(id: string): void {
+    try {
+      this.transport.resumeProducer(id)
+    } catch {
+      /* best-effort */
+    }
+  }
+}

src/main/ipc/pty.test.ts

@@ -517,12 +517,16 @@ describe('registerPtyHandlers', () => {
       id: options.sessionId ?? 'daemon-pty'
     }))
     const write = vi.fn()
+    const pauseProducer = vi.fn()
+    const resumeProducer = vi.fn()
     let dataHandler: ((payload: { id: string; data: string }) => void) | null = null
     let exitHandler: ((payload: { id: string; code: number }) => void) | null = null
     setLocalPtyProvider({
       spawn,
       write,
       resize: vi.fn(),
+      pauseProducer,
+      resumeProducer,
       kill: vi.fn(),
       shutdown: vi.fn(),
       sendSignal: vi.fn(),
@@ -551,6 +555,8 @@ describe('registerPtyHandlers', () => {
     return {
       spawn,
       write,
+      pauseProducer,
+      resumeProducer,
       emitData: (id: string, data: string) => dataHandler?.({ id, data }),
       emitExit: (id: string, code = 0) => exitHandler?.({ id, code })
     }
@@ -7104,6 +7110,67 @@ describe('registerPtyHandlers', () => {
     }
   })
 
+  it('pauses the producer at the pending high watermark and resumes after drain', async () => {
+    vi.useFakeTimers()
+    try {
+      const provider = installObservableDaemonTestProvider()
+      registerPtyHandlers(mainWindow as never)
+      mainWindow.webContents.send.mockClear()
+
+      // Flood in 64KB chunks like a `yes`-style producer that honors pause —
+      // node-pty pause() stops the fd read, so a real producer stops emitting.
+      const chunk = 'x'.repeat(64 * 1024)
+      let chunks = 0
+      while (provider.pauseProducer.mock.calls.length === 0 && chunks < 100) {
+        provider.emitData('flood-pty', chunk)
+        chunks++
+      }
+
+      // Pause fires exactly once, on the first chunk past the 256KB high
+      // watermark (the 5th 64KB chunk), not once per chunk.
+      expect(provider.pauseProducer).toHaveBeenCalledTimes(1)
+      expect(provider.pauseProducer).toHaveBeenCalledWith('flood-pty')
+      expect(chunks).toBe(5)
+      // Bounded: main buffered at most HIGH + one chunk while paused.
+      expect(getPtyRendererDeliveryDebugSnapshot()).toMatchObject({
+        pendingPtyCount: 1,
+        pendingChars: 320 * 1024,
+        peakPendingChars: 320 * 1024
+      })
+
+      // Drain to the renderer. Resume must fire exactly once — when pending
+      // drops below the 32KB low watermark — with no pause/resume flapping
+      // while pending crosses the 32-256KB hysteresis band.
+      vi.runAllTimers()
+      expect(provider.resumeProducer).toHaveBeenCalledTimes(1)
+      expect(provider.resumeProducer).toHaveBeenCalledWith('flood-pty')
+      expect(provider.pauseProducer).toHaveBeenCalledTimes(1)
+      expect(getPtyRendererDeliveryDebugSnapshot()).toMatchObject({ pendingChars: 0 })
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
+  it('resumes a paused producer when the PTY exits before draining', async () => {
+    vi.useFakeTimers()
+    try {
+      const provider = installObservableDaemonTestProvider()
+      registerPtyHandlers(mainWindow as never)
+      mainWindow.webContents.send.mockClear()
+
+      provider.emitData('flood-pty', 'x'.repeat(320 * 1024))
+      expect(provider.pauseProducer).toHaveBeenCalledTimes(1)
+
+      // Exit while pending is still above the low watermark: the exit path
+      // must release the pause instead of leaving a stale mark behind.
+      provider.emitExit('flood-pty', 0)
+      expect(provider.resumeProducer).toHaveBeenCalledTimes(1)
+      expect(provider.resumeProducer).toHaveBeenCalledWith('flood-pty')
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
   it('forwards only actually in-flight bytes to provider ACK backpressure', async () => {
     vi.useFakeTimers()
     const acknowledgeDataEvent = vi.fn()

src/main/ipc/pty.ts

@@ -87,6 +87,7 @@ import {
 import { parseWslPath } from '../wsl'
 import { mergePersistedWindowsPath } from '../pty/windows-environment-path'
 import { addOrcaWslInteropEnv } from '../pty/wsl-orca-env'
+import { PtyProducerFlowController } from './pty-producer-flow-control'
 import {
   clearHiddenRendererPtyDeliveryState,
   getHiddenRendererPtyDeliveryDebug,
@@ -136,6 +137,10 @@ type FreshLocalFallbackProvider = IPtyProvider & {
 }
 const sshProviders = new Map<string, IPtyProvider>()
 const SYNTHETIC_KILL_EXIT_DUPLICATE_WINDOW_MS = 30_000
+// Why: producer flow control changes terminal physics — a flooding shell now
+// blocks on write instead of buffering in main. Kill switch: flip this one
+// line to disable pause/resume entirely without untangling the wiring.
+const PRODUCER_FLOW_CONTROL_ENABLED = true
 // Why: PTY IDs are assigned at spawn time with a connectionId, but subsequent
 // write/resize/kill calls only carry the PTY ID. This map lets us route
 // post-spawn operations to the correct provider without the renderer needing
@@ -1471,6 +1476,24 @@ export function registerPtyHandlers(
   let peakMaxRendererInFlightCharsByPty = 0
   let ackGatedFlushSkipCount = 0
 
+  // Why: watermark-driven producer pause/resume (terminal-performance
+  // initiative §5). Signal source is per-PTY pendingData only — renderer
+  // in-flight is already bounded by the ACK window above, while pendingData
+  // is what grows without bound when the renderer cannot keep up. Providers
+  // without support (SSH, legacy daemon protocol) surface no pauseProducer
+  // and the call chain no-ops; the pending cap still bounds memory then.
+  const producerFlowControl = new PtyProducerFlowController({
+    pauseProducer: (id) => tryGetProviderForPty(id)?.pauseProducer?.(id),
+    resumeProducer: (id) => tryGetProviderForPty(id)?.resumeProducer?.(id)
+  })
+
+  function updateProducerFlowControl(id: string): void {
+    if (!PRODUCER_FLOW_CONTROL_ENABLED) {
+      return
+    }
+    producerFlowControl.update(id, pendingData.get(id)?.data.length ?? 0)
+  }
+
   function getMaxMapValue(values: Iterable<number>): number {
     let max = 0
     for (const value of values) {
@@ -1752,6 +1775,9 @@ export function registerPtyHandlers(
   function flushPendingData(): void {
     flushTimer = null
     if (mainWindow.isDestroyed()) {
+      // Why: the bookkeeping is being wiped, so no future drain can ever
+      // resume these producers — release them now or local shells wedge.
+      producerFlowControl.releaseAll()
       pendingData.clear()
       pendingOverflowMarkedPtys.clear()
       rendererInFlightCharsByPty.clear()
@@ -1770,6 +1796,7 @@ export function registerPtyHandlers(
       if (shouldDropHiddenRendererPtyData(id, settings)) {
         pendingData.delete(id)
         pendingOverflowMarkedPtys.delete(id)
+        updateProducerFlowControl(id)
         const drop = recordHiddenRendererPtyDataDrop(id, pending.data.length)
         if (drop.shouldEmitRestoreMarker) {
           sendModelRestoreNeededMarker(id, 'hidden-drop', runtime?.getPtyOutputSequence(id))
@@ -1781,6 +1808,7 @@ export function registerPtyHandlers(
       }
       pendingData.delete(id)
       if (pending.droppedOutput === true) {
+        updateProducerFlowControl(id)
         // Why: the buffered bytes were dropped at the pending cap; tell the
         // renderer so the pane repaints from the main-owned buffer snapshot
         // instead of continuing a stream with a silent gap.
@@ -1803,6 +1831,7 @@ export function registerPtyHandlers(
       } else {
         pendingOverflowMarkedPtys.delete(id)
       }
+      updateProducerFlowControl(id)
       sendPtyDataToRenderer(
         id,
         makePtyDataPayload(id, chunk, pending.startSeq, pending.containsBackgroundOutput)
@@ -1874,6 +1903,9 @@ export function registerPtyHandlers(
       )
       pendingData.delete(payload.id)
     }
+    // Why: exit drops this PTY's bookkeeping; resume (no-op on a dead PTY)
+    // rather than leave a stale paused mark behind for a reused id.
+    producerFlowControl.release(payload.id)
     pendingOverflowMarkedPtys.delete(payload.id)
     lastInputAtByPty.delete(payload.id)
     interactiveOutputCharsByPty.delete(payload.id)
@@ -1936,6 +1968,7 @@ export function registerPtyHandlers(
           clearTimeout(flushTimer)
           flushTimer = null
         }
+        producerFlowControl.releaseAll()
         pendingData.clear()
         pendingOverflowMarkedPtys.clear()
         rendererInFlightCharsByPty.clear()
@@ -1985,10 +2018,12 @@ export function registerPtyHandlers(
         // bounded, and the per-PTY cap still prevents an active TUI runaway.
         if (!canSendPtyDataToRenderer(payload.id, { interactive: true })) {
           pendingData.set(payload.id, pending)
+          updateProducerFlowControl(payload.id)
           recordPtyRendererDeliveryPressure()
           return
         }
         pendingData.delete(payload.id)
+        updateProducerFlowControl(payload.id)
         pendingOverflowMarkedPtys.delete(payload.id)
         clearFlushTimerIfIdle()
         // Why: agent TUIs redraw small prompt regions after every keystroke.
@@ -2004,6 +2039,7 @@ export function registerPtyHandlers(
         return
       }
       pendingData.set(payload.id, pending)
+      updateProducerFlowControl(payload.id)
       recordPtyRendererDeliveryPressure()
       if (!flushTimer) {
         schedulePendingDataFlush(PTY_BATCH_INTERVAL_MS)
@@ -3802,6 +3838,7 @@ export function registerPtyHandlers(
       const pending = pendingData.get(args.id)
       if (pending && shouldDropHiddenRendererPtyData(args.id, getSettings?.())) {
         pendingData.delete(args.id)
+        updateProducerFlowControl(args.id)
         pendingOverflowMarkedPtys.delete(args.id)
         const drop = recordHiddenRendererPtyDataDrop(args.id, pending.data.length)
         if (drop.shouldEmitRestoreMarker) {

src/main/providers/local-pty-provider.test.ts

@@ -91,6 +91,8 @@ describe('LocalPtyProvider', () => {
     onExit: ReturnType<typeof vi.fn>
     write: ReturnType<typeof vi.fn>
     resize: ReturnType<typeof vi.fn>
+    pause: ReturnType<typeof vi.fn>
+    resume: ReturnType<typeof vi.fn>
     kill: ReturnType<typeof vi.fn>
     process: string
     pid: number
@@ -133,6 +135,8 @@ describe('LocalPtyProvider', () => {
       }),
       write: vi.fn(),
       resize: vi.fn(),
+      pause: vi.fn(),
+      resume: vi.fn(),
       kill: vi.fn(() => {
         exitCb?.({ exitCode: -1 })
       }),
@@ -840,6 +844,39 @@ describe('LocalPtyProvider', () => {
     })
   })
 
+  describe('producer flow control', () => {
+    it('pauses and resumes the node-pty process directly', async () => {
+      const { id } = await provider.spawn({ cols: 80, rows: 24 })
+      provider.pauseProducer(id)
+      expect(mockProc.pause).toHaveBeenCalledTimes(1)
+      provider.resumeProducer(id)
+      expect(mockProc.resume).toHaveBeenCalledTimes(1)
+    })
+
+    it('is a no-op for unknown PTY ids', () => {
+      expect(() => {
+        provider.pauseProducer('nonexistent')
+        provider.resumeProducer('nonexistent')
+      }).not.toThrow()
+      expect(mockProc.pause).not.toHaveBeenCalled()
+      expect(mockProc.resume).not.toHaveBeenCalled()
+    })
+
+    it('swallows node-pty throws from a torn-down PTY', async () => {
+      const { id } = await provider.spawn({ cols: 80, rows: 24 })
+      mockProc.pause.mockImplementation(() => {
+        throw new Error('read EIO')
+      })
+      mockProc.resume.mockImplementation(() => {
+        throw new Error('read EIO')
+      })
+      expect(() => {
+        provider.pauseProducer(id)
+        provider.resumeProducer(id)
+      }).not.toThrow()
+    })
+  })
+
   describe('shutdown', () => {
     it('kills the PTY process', async () => {
       // Why: capture the spy reference before shutdown triggers onExit →

src/main/providers/local-pty-provider.ts

@@ -828,6 +828,26 @@ export class LocalPtyProvider implements IPtyProvider {
     ptyProcesses.get(id)?.resize(cols, rows)
   }
 
+  // Why: node-pty pause() stops reading the pty master fd, so the kernel
+  // buffer fills and a flooding child blocks on write — true producer
+  // backpressure. Best-effort: a PTY torn down mid-call must never throw
+  // into the flow-control path.
+  pauseProducer(id: string): void {
+    try {
+      ptyProcesses.get(id)?.pause()
+    } catch {
+      /* PTY already destroyed */
+    }
+  }
+
+  resumeProducer(id: string): void {
+    try {
+      ptyProcesses.get(id)?.resume()
+    } catch {
+      /* PTY already destroyed */
+    }
+  }
+
   // Why: node-pty caches the last winsize it applied on the IPty handle, so its
   // cols/rows are the authoritative applied size (node-pty clamps invalid dims
   // and a resize on a dead handle is a no-op, neither of which the requested

src/main/providers/types.ts

@@ -110,6 +110,16 @@ export type IPtyProvider = {
   hasPty?: (id: string) => boolean
   write(id: string, data: string): void
   resize(id: string, cols: number, rows: number): void
+  /**
+   * Producer-side flow control: stop/restart reading the underlying PTY so a
+   * flooding child blocks on write (kernel backpressure) instead of growing
+   * main-process buffers. Best-effort and optional — providers that cannot
+   * pause (SSH relay, legacy daemon protocols) omit these or no-op silently,
+   * and callers must keep functioning without them (the pending-output cap
+   * still bounds memory when pause is unavailable).
+   */
+  pauseProducer?: (id: string) => void
+  resumeProducer?: (id: string) => void
   /**
    * The size the PTY has ACTUALLY applied, not the last size requested.
    * resize() is fire-and-forget for remote providers (daemon/SSH `notify`),