diff --git a/src/main/daemon/daemon-errors.ts b/src/main/daemon/daemon-errors.ts new file mode 100644 index 00000000000..99d19818a3c --- /dev/null +++ b/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' + } +} diff --git a/src/main/daemon/daemon-pty-adapter.test.ts b/src/main/daemon/daemon-pty-adapter.test.ts index 63b6bc4ab3e..d283e328093 100644 --- a/src/main/daemon/daemon-pty-adapter.test.ts +++ b/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 void>> + resume: ReturnType 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 }) diff --git a/src/main/daemon/daemon-pty-adapter.ts b/src/main/daemon/daemon-pty-adapter.ts index c30d385747d..950b13de4c2 100644 --- a/src/main/daemon/daemon-pty-adapter.ts +++ b/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() + // 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() 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 { // 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) } diff --git a/src/main/daemon/daemon-pty-router.ts b/src/main/daemon/daemon-pty-router.ts index 2212635d1fa..19bd6aa5adb 100644 --- a/src/main/daemon/daemon-pty-router.ts +++ b/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 { await this.adapterFor(id).shutdown(id, opts) // Why: sleep passes keepHistory=true and re-spawns against the same diff --git a/src/main/daemon/daemon-server.ts b/src/main/daemon/daemon-server.ts index 3dc1c648f0b..8b4e07b3baa 100644 --- a/src/main/daemon/daemon-server.ts +++ b/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 }) diff --git a/src/main/daemon/degraded-daemon-pty-provider.ts b/src/main/daemon/degraded-daemon-pty-provider.ts index acf95ffcdfe..a5a925d10a2 100644 --- a/src/main/daemon/degraded-daemon-pty-provider.ts +++ b/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 { await this.providerFor(id).shutdown(id, opts) if (!opts.keepHistory) { diff --git a/src/main/daemon/pty-subprocess.ts b/src/main/daemon/pty-subprocess.ts index a4bd1fca7e4..bfcf7e021bf 100644 --- a/src/main/daemon/pty-subprocess.ts +++ b/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 diff --git a/src/main/daemon/session.test.ts b/src/main/daemon/session.test.ts index c34b58743f8..c54f07e5afd 100644 --- a/src/main/daemon/session.test.ts +++ b/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) + }) + }) }) diff --git a/src/main/daemon/session.ts b/src/main/daemon/session.ts index 551b8ad276c..f6aff2519d1 100644 --- a/src/main/daemon/session.ts +++ b/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 | 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) { diff --git a/src/main/daemon/terminal-host.ts b/src/main/daemon/terminal-host.ts index 4731322e526..256c5d1e1be 100644 --- a/src/main/daemon/terminal-host.ts +++ b/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) diff --git a/src/main/daemon/types.ts b/src/main/daemon/types.ts index 0a882e9786a..0fb03617776 100644 --- a/src/main/daemon/types.ts +++ b/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' diff --git a/src/main/ipc/pty-producer-flow-control.test.ts b/src/main/ipc/pty-producer-flow-control.test.ts new file mode 100644 index 00000000000..ae7fcb6b655 --- /dev/null +++ b/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 void>> + let resumeProducer: ReturnType 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) + }) +}) diff --git a/src/main/ipc/pty-producer-flow-control.ts b/src/main/ipc/pty-producer-flow-control.ts new file mode 100644 index 00000000000..48a9fc0b678 --- /dev/null +++ b/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() + + 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 */ + } + } +} diff --git a/src/main/ipc/pty.test.ts b/src/main/ipc/pty.test.ts index 964fabf07d0..9e4f4a9b0d1 100644 --- a/src/main/ipc/pty.test.ts +++ b/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() diff --git a/src/main/ipc/pty.ts b/src/main/ipc/pty.ts index 22c3a09c390..41baf471afc 100644 --- a/src/main/ipc/pty.ts +++ b/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() 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 { 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) { diff --git a/src/main/providers/local-pty-provider.test.ts b/src/main/providers/local-pty-provider.test.ts index bc236e9f9bd..8ecb3e0e56d 100644 --- a/src/main/providers/local-pty-provider.test.ts +++ b/src/main/providers/local-pty-provider.test.ts @@ -91,6 +91,8 @@ describe('LocalPtyProvider', () => { onExit: ReturnType write: ReturnType resize: ReturnType + pause: ReturnType + resume: ReturnType kill: ReturnType 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 → diff --git a/src/main/providers/local-pty-provider.ts b/src/main/providers/local-pty-provider.ts index 01ea3aeea57..b0e8c2ef9c9 100644 --- a/src/main/providers/local-pty-provider.ts +++ b/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 diff --git a/src/main/providers/types.ts b/src/main/providers/types.ts index cd96765a62c..7d00e75f365 100644 --- a/src/main/providers/types.ts +++ b/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`),