Revert terminal stream backpressure changes (#2380)

src/main/daemon/client.test.ts

@@ -50,14 +50,10 @@ describe('DaemonClient', () => {
   function startMockDaemon(opts?: {
     onControlMessage?: (msg: unknown) => string | null
     onStreamHello?: (msg: HelloMessage) => void
-    streamHelloGate?: Promise<void>
     rejectVersion?: boolean
   }): Promise<void> {
     return new Promise((resolve) => {
       server = createServer((socket) => {
-        socket.on('error', () => {
-          /* tests intentionally destroy sockets mid-handshake */
-        })
         let buffer = ''
         socket.on('data', (chunk) => {
           buffer += chunk.toString()
@@ -77,22 +73,10 @@ describe('DaemonClient', () => {
                 socket.write(encodeNdjson({ type: 'hello', ok: false, error: 'Version mismatch' }))
                 return
               }
+              socket.write(encodeNdjson({ type: 'hello', ok: true }))
               if (hello.role === 'stream') {
                 opts?.onStreamHello?.(hello)
-                if (opts?.streamHelloGate) {
-                  void opts.streamHelloGate.then(() => {
-                    if (!socket.destroyed) {
-                      try {
-                        socket.write(encodeNdjson({ type: 'hello', ok: true }))
-                      } catch {
-                        /* client intentionally disconnected during handshake */
-                      }
-                    }
-                  })
-                  return
-                }
               }
-              socket.write(encodeNdjson({ type: 'hello', ok: true }))
             } else if (opts?.onControlMessage) {
               const response = opts.onControlMessage(msg)
               if (response) {
@@ -268,30 +252,6 @@ describe('DaemonClient', () => {
       client = new DaemonClient({ socketPath, tokenPath })
       expect(() => client.disconnect()).not.toThrow()
     })
-
-    it('disconnect() cancels an in-flight connection attempt', async () => {
-      let releaseStreamHello!: () => void
-      const streamHelloGate = new Promise<void>((resolve) => {
-        releaseStreamHello = resolve
-      })
-      let sawStreamHello = false
-      await startMockDaemon({
-        streamHelloGate,
-        onStreamHello: () => {
-          sawStreamHello = true
-        }
-      })
-
-      client = new DaemonClient({ socketPath, tokenPath })
-      const connectPromise = client.ensureConnected()
-      await waitFor(() => sawStreamHello)
-
-      client.disconnect()
-      releaseStreamHello()
-
-      await expect(connectPromise).rejects.toThrow()
-      expect(client.isConnected()).toBe(false)
-    })
   })
 
   describe('notify (fire-and-forget)', () => {

src/main/daemon/client.ts

@@ -31,7 +31,6 @@ export class DaemonClient {
   private streamSocket: Socket | null = null
   private connected = false
   private disconnectArmed = false
-  private disconnectGeneration = 0
   // Why: after a disconnect + reconnect (daemon respawn), a stale 'close'
   // event from the old sockets can fire. Without a generation check, that
   // event would tear down the fresh connection. Each doConnect() increments
@@ -75,20 +74,15 @@ export class DaemonClient {
 
   private async doConnect(): Promise<void> {
     const token = readFileSync(this.tokenPath, 'utf-8').trim()
-    const disconnectGeneration = this.disconnectGeneration
 
     try {
       // Sequential: control first, then stream
       this.controlSocket = await this.connectSocket()
-      this.assertConnectNotCancelled(disconnectGeneration)
       await this.sendHello(this.controlSocket, token, 'control')
-      this.assertConnectNotCancelled(disconnectGeneration)
       this.setupControlParser()
 
       this.streamSocket = await this.connectSocket()
-      this.assertConnectNotCancelled(disconnectGeneration)
       await this.sendHello(this.streamSocket, token, 'stream')
-      this.assertConnectNotCancelled(disconnectGeneration)
       this.setupStreamParser()
 
       this.connected = true
@@ -96,18 +90,11 @@ export class DaemonClient {
       this.connectionGeneration++
 
       const gen = this.connectionGeneration
-      this.controlSocket.on('close', (hadError) =>
-        this.handleDisconnect(gen, `control socket closed${hadError ? ' after error' : ''}`)
-      )
-      this.controlSocket.on('error', (error) =>
-        this.handleDisconnect(gen, `control socket error: ${error.message}`)
-      )
-      this.streamSocket.on('close', (hadError) =>
-        this.handleDisconnect(gen, `stream socket closed${hadError ? ' after error' : ''}`)
-      )
-      this.streamSocket.on('error', (error) =>
-        this.handleDisconnect(gen, `stream socket error: ${error.message}`)
-      )
+      const handleClose = () => this.handleDisconnect(gen)
+      this.controlSocket.on('close', handleClose)
+      this.controlSocket.on('error', handleClose)
+      this.streamSocket.on('close', handleClose)
+      this.streamSocket.on('error', handleClose)
     } catch (error) {
       this.controlSocket?.destroy()
       this.streamSocket?.destroy()
@@ -143,19 +130,14 @@ export class DaemonClient {
     })
   }
 
-  notify(type: string, payload: unknown): boolean {
+  notify(type: string, payload: unknown): void {
     if (!this.connected || !this.controlSocket) {
-      return false
+      return
     }
 
     const id = `${NOTIFY_PREFIX}${++this.requestCounter}`
     const msg = { id, type, ...(payload !== undefined ? { payload } : {}) }
-    try {
-      this.controlSocket.write(encodeNdjson(msg))
-      return true
-    } catch {
-      return false
-    }
+    this.controlSocket.write(encodeNdjson(msg))
   }
 
   onEvent(listener: (event: unknown) => void): () => void {
@@ -179,10 +161,8 @@ export class DaemonClient {
   }
 
   disconnect(): void {
-    this.disconnectGeneration++
     this.connected = false
     this.disconnectArmed = false
-    this.connectingPromise = null
 
     for (const [id, pending] of this.pendingRequests) {
       clearTimeout(pending.timer)
@@ -196,12 +176,6 @@ export class DaemonClient {
     this.streamSocket = null
   }
 
-  private assertConnectNotCancelled(disconnectGeneration: number): void {
-    if (disconnectGeneration !== this.disconnectGeneration) {
-      throw new DaemonProtocolError('Connection cancelled')
-    }
-  }
-
   private connectSocket(): Promise<Socket> {
     return new Promise((resolve, reject) => {
       const socket = connect(this.socketPath)
@@ -233,11 +207,6 @@ export class DaemonClient {
       }
 
       let buffer = ''
-      const cleanup = (): void => {
-        socket.removeListener('data', onData)
-        socket.removeListener('error', onError)
-        socket.removeListener('close', onClose)
-      }
       const onData = (chunk: Buffer): void => {
         buffer += chunk.toString()
         const newlineIdx = buffer.indexOf('\n')
@@ -245,7 +214,7 @@ export class DaemonClient {
           return
         }
 
-        cleanup()
+        socket.removeListener('data', onData)
         const line = buffer.slice(0, newlineIdx)
         try {
           const response = JSON.parse(line) as HelloResponse
@@ -260,18 +229,8 @@ export class DaemonClient {
           reject(new DaemonProtocolError('Invalid hello response'))
         }
       }
-      const onError = (error: Error): void => {
-        cleanup()
-        reject(error)
-      }
-      const onClose = (): void => {
-        cleanup()
-        reject(new DaemonProtocolError('Connection closed during hello'))
-      }
 
       socket.on('data', onData)
-      socket.once('error', onError)
-      socket.once('close', onClose)
       socket.write(encodeNdjson(hello))
     })
   }
@@ -323,18 +282,13 @@ export class DaemonClient {
     this.streamSocket.on('data', (chunk) => parser.feed(chunk.toString()))
   }
 
-  private handleDisconnect(generation: number, reason: string): void {
+  private handleDisconnect(generation: number): void {
     if (!this.disconnectArmed || generation !== this.connectionGeneration) {
       return
     }
     this.disconnectArmed = false
     this.connected = false
 
-    console.warn('[daemon-client] disconnected', {
-      reason,
-      pendingRequests: this.pendingRequests.size
-    })
-
     for (const [id, pending] of this.pendingRequests) {
       clearTimeout(pending.timer)
       pending.reject(new DaemonProtocolError('Connection lost'))

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

@@ -114,22 +114,6 @@ describe('DaemonPtyAdapter (IPtyProvider)', () => {
       await new Promise((r) => setTimeout(r, 50))
       expect(lastSubprocess.write).toHaveBeenCalledWith('ls\n')
     })
-
-    it('reattaches active sessions and flushes writes after daemon stream failure', async () => {
-      const warn = vi.spyOn(console, 'warn').mockImplementation(() => {})
-      try {
-        const { id } = await adapter.spawn({ cols: 80, rows: 24 })
-
-        lastSubprocess._simulateData('x'.repeat(8 * 1024 * 1024 + 1))
-        await new Promise((resolve) => setTimeout(resolve, 100))
-        adapter.write(id, 'after-reconnect\n')
-
-        await waitFor(() => vi.mocked(lastSubprocess.write).mock.calls.length > 0, 3000)
-        expect(lastSubprocess.write).toHaveBeenCalledWith('after-reconnect\n')
-      } finally {
-        warn.mockRestore()
-      }
-    })
   })
 
   describe('resize', () => {

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

@@ -31,12 +31,6 @@ export type DaemonPtyAdapterOptions = {
 }
 
 const MAX_TOMBSTONES = 1000
-const MAX_PENDING_DAEMON_NOTIFICATIONS = 512
-
-type PendingDaemonNotification = {
-  type: 'write' | 'resize'
-  payload: unknown
-}
 
 export class TerminalKilledError extends Error {
   constructor(sessionId: string) {
@@ -59,10 +53,6 @@ export class DaemonPtyAdapter implements IPtyProvider {
   private dataListeners: ((payload: { id: string; data: string }) => void)[] = []
   private exitListeners: ((payload: { id: string; code: number }) => void)[] = []
   private removeEventListener: (() => void) | null = null
-  private removeDisconnectedListener: (() => void) | null = null
-  private recoveryPromise: Promise<void> | null = null
-  private pendingNotifications: PendingDaemonNotification[] = []
-  private disposed = false
   private initialCwds = new Map<string, string>()
   // Why: React re-renders and StrictMode double-mounts can call createOrAttach
   // for a session the user just killed. Without tombstones, the daemon would
@@ -70,7 +60,6 @@ export class DaemonPtyAdapter implements IPtyProvider {
   // Uses a Map<id, timestamp> so eviction removes the oldest by insertion order,
   // matching terminal-host.ts tombstone semantics.
   private killedSessionTombstones = new Map<string, number>()
-  private sessionSizes = new Map<string, { cols: number; rows: number }>()
   // Why: React StrictMode double-mounts: mount → cold restore → unmount →
   // mount → ??? The sticky cache returns the same cold restore data on the
   // second mount until the renderer explicitly acknowledges it.
@@ -95,11 +84,6 @@ export class DaemonPtyAdapter implements IPtyProvider {
     this.historyReader = opts.historyPath ? new HistoryReader(opts.historyPath) : null
     this.respawnFn = opts.respawn ?? null
     this.supportsCheckpoints = this.protocolVersion >= 4
-    this.removeDisconnectedListener = this.client.onDisconnected(() => {
-      void this.recoverActiveSessionsAfterDisconnect().catch((err) =>
-        console.warn('[daemon] reconnect after stream failure failed:', err)
-      )
-    })
   }
 
   getHistoryManager(): HistoryManager | null {
@@ -167,7 +151,6 @@ export class DaemonPtyAdapter implements IPtyProvider {
     }
 
     this.activeSessionIds.add(sessionId)
-    this.sessionSizes.set(sessionId, { cols: effectiveCols, rows: effectiveRows })
 
     // Cold restore: daemon created a new session but disk history shows
     // an unclean shutdown → return saved scrollback so the renderer can
@@ -251,13 +234,12 @@ export class DaemonPtyAdapter implements IPtyProvider {
 
   write(id: string, data: string): void {
     this.markSessionDirty(id)
-    this.sendNotification('write', { sessionId: id, data })
+    this.client.notify('write', { sessionId: id, data })
   }
 
   resize(id: string, cols: number, rows: number): void {
     this.markSessionDirty(id)
-    this.sessionSizes.set(id, { cols, rows })
-    this.sendNotification('resize', { sessionId: id, cols, rows })
+    this.client.notify('resize', { sessionId: id, cols, rows })
   }
 
   async shutdown(id: string, opts: { immediate?: boolean; keepHistory?: boolean }): Promise<void> {
@@ -265,7 +247,6 @@ export class DaemonPtyAdapter implements IPtyProvider {
     this.activeSessionIds.delete(id)
     this.dirtySessionVersions.delete(id)
     this.initialCwds.delete(id)
-    this.sessionSizes.delete(id)
     // Why: history removal is for the "user explicitly closed this terminal"
     // path. Sleep also calls shutdown but expects scrollback to survive — wake
     // re-spawns and the cold-restore reader needs the dir intact. Caller
@@ -382,7 +363,6 @@ export class DaemonPtyAdapter implements IPtyProvider {
         // set, disconnectOnly()'s final checkpoint would skip them, leaving
         // stale recovery data if the daemon later crashes.
         this.activeSessionIds.add(session.sessionId)
-        this.sessionSizes.set(session.sessionId, { cols: session.cols, rows: session.rows })
         this.historyManager?.registerWriter(session.sessionId)
       }
     }
@@ -427,7 +407,6 @@ export class DaemonPtyAdapter implements IPtyProvider {
     const ids = [...this.activeSessionIds]
     this.activeSessionIds.clear()
     this.dirtySessionVersions.clear()
-    this.sessionSizes.clear()
     for (const id of ids) {
       // Why: listener throws are intentionally *not* caught — matches the
       // natural onExit fanout in setupEventRouting, so synthetic exits don't
@@ -483,13 +462,11 @@ export class DaemonPtyAdapter implements IPtyProvider {
   }
 
   dispose(): void {
-    this.disposed = true
     if (this.checkpointInterval) {
       clearInterval(this.checkpointInterval)
       this.checkpointInterval = null
     }
     this.dirtySessionVersions.clear()
-    this.sessionSizes.clear()
     this.removeEventListener?.()
     this.removeEventListener = null
     // Why: final checkpoints are written daemon-side in TerminalHost.dispose()
@@ -501,9 +478,6 @@ export class DaemonPtyAdapter implements IPtyProvider {
         .catch((err) => console.warn('[history] dispose failed:', err))
     }
     this.client.disconnect()
-    this.removeDisconnectedListener?.()
-    this.removeDisconnectedListener = null
-    this.pendingNotifications = []
   }
 
   // Why: for in-process daemon mode, disconnect without flushing history.
@@ -513,7 +487,6 @@ export class DaemonPtyAdapter implements IPtyProvider {
   // We write a final checkpoint before disconnecting so that if the daemon
   // later crashes while Orca is closed, checkpoint.json has recovery data.
   async disconnectOnly(): Promise<void> {
-    this.disposed = true
     if (this.checkpointInterval) {
       clearInterval(this.checkpointInterval)
       this.checkpointInterval = null
@@ -534,9 +507,6 @@ export class DaemonPtyAdapter implements IPtyProvider {
     this.removeEventListener?.()
     this.removeEventListener = null
     this.client.disconnect()
-    this.removeDisconnectedListener?.()
-    this.removeDisconnectedListener = null
-    this.pendingNotifications = []
   }
 
   private async ensureConnected(): Promise<void> {
@@ -563,72 +533,6 @@ export class DaemonPtyAdapter implements IPtyProvider {
     }, DaemonPtyAdapter.CHECKPOINT_INTERVAL_MS)
   }
 
-  private sendNotification(type: PendingDaemonNotification['type'], payload: unknown): void {
-    if (this.recoveryPromise) {
-      this.queueNotification(type, payload)
-      return
-    }
-    if (this.client.notify(type, payload)) {
-      return
-    }
-    this.queueNotification(type, payload)
-    void this.recoverActiveSessionsAfterDisconnect().catch((err) =>
-      console.warn('[daemon] reconnect after notification failure failed:', err)
-    )
-  }
-
-  private queueNotification(type: PendingDaemonNotification['type'], payload: unknown): void {
-    this.pendingNotifications.push({ type, payload })
-    if (this.pendingNotifications.length > MAX_PENDING_DAEMON_NOTIFICATIONS) {
-      this.pendingNotifications.splice(
-        0,
-        this.pendingNotifications.length - MAX_PENDING_DAEMON_NOTIFICATIONS
-      )
-    }
-  }
-
-  private async recoverActiveSessionsAfterDisconnect(): Promise<void> {
-    if (this.disposed || this.activeSessionIds.size === 0) {
-      return
-    }
-    if (!this.recoveryPromise) {
-      this.recoveryPromise = this.reattachActiveSessions().finally(() => {
-        this.recoveryPromise = null
-      })
-    }
-    await this.recoveryPromise
-  }
-
-  private async reattachActiveSessions(): Promise<void> {
-    await this.ensureConnected()
-    // Why: daemon stream failure only breaks the renderer socket pair; the
-    // backing PTYs stay alive in TerminalHost. Reattach active sessions so
-    // stream events resume instead of letting panes black-hole input.
-    for (const sessionId of this.activeSessionIds) {
-      const size = this.sessionSizes.get(sessionId) ?? { cols: 80, rows: 24 }
-      await this.client.request<CreateOrAttachResult>('createOrAttach', {
-        sessionId,
-        cols: size.cols,
-        rows: size.rows
-      })
-    }
-    this.flushPendingNotifications()
-  }
-
-  private flushPendingNotifications(): void {
-    const pending = this.pendingNotifications
-    this.pendingNotifications = []
-    for (const notification of pending) {
-      if (!this.client.notify(notification.type, notification.payload)) {
-        this.queueNotification(notification.type, notification.payload)
-        void this.recoverActiveSessionsAfterDisconnect().catch((err) =>
-          console.warn('[daemon] reconnect after pending notification failed:', err)
-        )
-        return
-      }
-    }
-  }
-
   private markSessionDirty(sessionId: string): void {
     if (!this.activeSessionIds.has(sessionId)) {
       return
@@ -746,7 +650,6 @@ export class DaemonPtyAdapter implements IPtyProvider {
       } else if (event.event === 'exit') {
         this.activeSessionIds.delete(event.sessionId)
         this.dirtySessionVersions.delete(event.sessionId)
-        this.sessionSizes.delete(event.sessionId)
         if (this.historyManager) {
           void this.historyManager
             .closeSession(event.sessionId, event.payload.code)

src/main/daemon/daemon-server.test.ts

@@ -1,5 +1,5 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
-import { connect, type Socket } from 'net'
+import { connect } from 'net'
 import { tmpdir } from 'os'
 import { join } from 'path'
 import { mkdtempSync, rmSync, readFileSync } from 'fs'
@@ -66,39 +66,6 @@ describe('DaemonServer', () => {
     return client
   }
 
-  async function connectRawSocket(role: 'control' | 'stream', clientId: string): Promise<Socket> {
-    const socket = connect(socketPath)
-    await new Promise<void>((resolve) => socket.on('connect', resolve))
-    socket.write(
-      encodeNdjson({
-        type: 'hello',
-        version: PROTOCOL_VERSION,
-        token: readFileSync(tokenPath, 'utf-8').trim(),
-        clientId,
-        role
-      })
-    )
-    const response = await readSocketLine(socket)
-    expect(JSON.parse(response)).toMatchObject({ ok: true })
-    return socket
-  }
-
-  function readSocketLine(socket: Socket): Promise<string> {
-    return new Promise((resolve) => {
-      let buffer = ''
-      const onData = (chunk: Buffer): void => {
-        buffer += chunk.toString()
-        const newlineIdx = buffer.indexOf('\n')
-        if (newlineIdx === -1) {
-          return
-        }
-        socket.removeListener('data', onData)
-        resolve(buffer.slice(0, newlineIdx))
-      }
-      socket.on('data', onData)
-    })
-  }
-
   describe('startup', () => {
     it('creates token file and starts listening', async () => {
       await startServer()
@@ -155,22 +122,6 @@ describe('DaemonServer', () => {
       expect(result).toEqual({ pong: true })
     })
 
-    it('keeps a replacement control socket alive when the old socket closes later', async () => {
-      await startServer()
-      const firstControl = await connectRawSocket('control', 'same-client')
-      const secondControl = await connectRawSocket('control', 'same-client')
-
-      await new Promise((resolve) => setTimeout(resolve, 20))
-      secondControl.write(encodeNdjson({ id: 'req-1', type: 'ping' }))
-
-      await expect(readSocketLine(secondControl)).resolves.toMatch(
-        /"id":"req-1".*"payload":\{"pong":true\}/
-      )
-
-      firstControl.destroy()
-      secondControl.destroy()
-    })
-
     it('handles write (fire-and-forget)', async () => {
       await startServer()
       const c = await connectClient()

src/main/daemon/daemon-server.ts

@@ -41,12 +41,7 @@ export class DaemonServer {
   private tokenPath: string
 
   private clients = new Map<string, ConnectedClient>()
-  private streamDataBatcher = new DaemonStreamDataBatcher(
-    (clientId) => this.clients.get(clientId),
-    {
-      onStreamFailure: (clientId) => this.disconnectClient(clientId)
-    }
-  )
+  private streamDataBatcher = new DaemonStreamDataBatcher((clientId) => this.clients.get(clientId))
 
   constructor(opts: DaemonServerOptions) {
     this.socketPath = opts.socketPath
@@ -139,7 +134,6 @@ export class DaemonServer {
     socket.write(encodeNdjson({ type: 'hello', ok: true }))
 
     if (hello.role === 'control') {
-      this.disconnectClient(hello.clientId)
       const client: ConnectedClient = {
         clientId: hello.clientId,
         controlSocket: socket,
@@ -150,8 +144,6 @@ export class DaemonServer {
     } else if (hello.role === 'stream') {
       const client = this.clients.get(hello.clientId)
       if (client) {
-        this.streamDataBatcher.clear(hello.clientId)
-        client.streamSocket?.destroy()
         client.streamSocket = socket
       }
       // Stream socket is receive-only from daemon's perspective (for events)
@@ -169,10 +161,8 @@ export class DaemonServer {
     socket.on('data', (chunk) => parser.feed(chunk.toString()))
 
     socket.on('close', () => {
-      const client = this.clients.get(clientId)
-      if (client?.controlSocket === socket) {
-        this.disconnectClient(clientId, client)
-      }
+      this.streamDataBatcher.clear(clientId)
+      this.clients.delete(clientId)
     })
   }
 
@@ -222,11 +212,19 @@ export class DaemonServer {
               this.streamDataBatcher.enqueue(clientId, p.sessionId, data)
             },
             onExit: (code) => {
-              // Why: exit tears down renderer handlers; queue it behind any
-              // pending data so the final PTY bytes cannot be overtaken under
-              // stream backpressure.
-              this.streamDataBatcher.enqueueExit(clientId, p.sessionId, code)
-              this.streamDataBatcher.clearSessionInput(clientId, p.sessionId)
+              // Why: exit tears down renderer handlers; flush final output first
+              // so the last few milliseconds of PTY data are not stranded.
+              this.streamDataBatcher.flush(clientId)
+              if (client?.streamSocket) {
+                client.streamSocket.write(
+                  encodeNdjson({
+                    type: 'event',
+                    event: 'exit',
+                    sessionId: p.sessionId,
+                    payload: { code }
+                  })
+                )
+              }
             }
           }
         })
@@ -240,10 +238,8 @@ export class DaemonServer {
 
       case 'write':
         try {
-          this.streamDataBatcher.markInput(clientId, request.payload.sessionId)
           this.host.write(request.payload.sessionId, request.payload.data)
         } catch (err) {
-          this.streamDataBatcher.clearSessionInput(clientId, request.payload.sessionId)
           if (err instanceof SessionNotFoundError) {
             this.sendExitEvent(client, request.payload.sessionId, -1)
           }
@@ -308,23 +304,19 @@ export class DaemonServer {
     sessionId: string,
     code: number
   ): void {
-    if (!client) {
+    if (!client?.streamSocket) {
       return
     }
     // Why: write/resize are notification-heavy and intentionally do not wait
     // for replies. If their target session is gone, this synthetic exit is the
     // only signal the renderer gets to clear stale terminal pane bindings.
-    this.streamDataBatcher.enqueueExit(client.clientId, sessionId, code)
-  }
-
-  private disconnectClient(clientId: string, expectedClient?: ConnectedClient): void {
-    const client = this.clients.get(clientId)
-    if (expectedClient && client !== expectedClient) {
-      return
-    }
-    this.streamDataBatcher.clear(clientId)
-    this.clients.delete(clientId)
-    client?.streamSocket?.destroy()
-    client?.controlSocket.destroy()
+    client.streamSocket.write(
+      encodeNdjson({
+        type: 'event',
+        event: 'exit',
+        sessionId,
+        payload: { code }
+      })
+    )
   }
 }

src/main/daemon/daemon-stream-data-batch-state.ts

@@ -1,100 +0,0 @@
-import type { Socket } from 'net'
-import { encodeNdjson } from './ndjson'
-
-export type StreamDataClient = {
-  streamSocket: Socket | null
-}
-
-export type PendingStreamEvent =
-  | { kind: 'data'; sessionId: string; data: string }
-  | { kind: 'exit'; sessionId: string; code: number }
-
-export type PendingStreamDataBatch = {
-  timer: ReturnType<typeof setTimeout> | null
-  drainTimer: ReturnType<typeof setTimeout> | null
-  cleanupWait: (() => void) | null
-  queue: PendingStreamEvent[]
-  queueHead: number
-  queuedDataBytes: number
-  waitingForDrain: boolean
-  warnedBackpressure: boolean
-}
-
-// Why: match main-process PTY IPC batching to avoid adding latency while
-// removing daemon socket writes and JSON framing during bursty output.
-export const STREAM_DATA_BATCH_INTERVAL_MS = 8
-export const STREAM_DATA_BACKPRESSURE_WARN_BYTES = 512 * 1024
-export const STREAM_DATA_DRAIN_TIMEOUT_MS = 30_000
-export const STREAM_DATA_MAX_QUEUED_BYTES = 8 * 1024 * 1024
-export const STREAM_DATA_MAX_PAYLOAD_CHARS = 64 * 1024
-export const STREAM_DATA_MAX_EVENTS_PER_FLUSH = 1024
-export const INTERACTIVE_OUTPUT_WINDOW_MS = 100
-export const INTERACTIVE_OUTPUT_MAX_CHARS = 1024
-
-export function createPendingStreamDataBatch(): PendingStreamDataBatch {
-  return {
-    timer: null,
-    drainTimer: null,
-    cleanupWait: null,
-    queue: [],
-    queueHead: 0,
-    queuedDataBytes: 0,
-    waitingForDrain: false,
-    warnedBackpressure: false
-  }
-}
-
-export function compactStreamDataBatch(batch: PendingStreamDataBatch): void {
-  if (batch.queueHead === 0) {
-    return
-  }
-  batch.queue = batch.queue.slice(batch.queueHead)
-  batch.queueHead = 0
-}
-
-export function streamInputKey(clientId: string, sessionId: string): string {
-  return `${clientId}\0${sessionId}`
-}
-
-export function getQueuedDataForSession(batch: PendingStreamDataBatch, sessionId: string): string {
-  let data = ''
-  for (let index = batch.queueHead; index < batch.queue.length; index++) {
-    const entry = batch.queue[index]!
-    if (entry.kind === 'data' && entry.sessionId === sessionId) {
-      data += entry.data
-    }
-  }
-  return data
-}
-
-export function removeQueuedDataForSession(batch: PendingStreamDataBatch, sessionId: string): void {
-  const remaining: PendingStreamEvent[] = []
-  for (let index = batch.queueHead; index < batch.queue.length; index++) {
-    const entry = batch.queue[index]!
-    if (entry.kind === 'data' && entry.sessionId === sessionId) {
-      batch.queuedDataBytes -= Buffer.byteLength(entry.data, 'utf8')
-    } else {
-      remaining.push(entry)
-    }
-  }
-  batch.queue = remaining
-  batch.queueHead = 0
-}
-
-export function writeStreamEvent(streamSocket: Socket, entry: PendingStreamEvent): boolean {
-  const payload =
-    entry.kind === 'data'
-      ? {
-          type: 'event',
-          event: 'data',
-          sessionId: entry.sessionId,
-          payload: { data: entry.data }
-        }
-      : {
-          type: 'event',
-          event: 'exit',
-          sessionId: entry.sessionId,
-          payload: { code: entry.code }
-        }
-  return streamSocket.write(encodeNdjson(payload))
-}

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

@@ -1,174 +0,0 @@
-import { describe, expect, it, vi } from 'vitest'
-import { createHarness, parseWrite } from './daemon-stream-data-batcher-test-harness'
-
-describe('DaemonStreamDataBatcher interactive output', () => {
-  it('coalesces non-interactive output before writing to the stream socket', () => {
-    vi.useFakeTimers()
-    try {
-      const { batcher, fake } = createHarness()
-
-      batcher.enqueue('client-1', 'session-1', 'a')
-      batcher.enqueue('client-1', 'session-1', 'b')
-
-      expect(fake.write).not.toHaveBeenCalled()
-      vi.advanceTimersByTime(7)
-      expect(fake.write).not.toHaveBeenCalled()
-      vi.advanceTimersByTime(1)
-
-      expect(fake.write).toHaveBeenCalledTimes(1)
-      expect(parseWrite(fake.write.mock.calls[0])).toMatchObject({
-        type: 'event',
-        event: 'data',
-        sessionId: 'session-1',
-        payload: { data: 'ab' }
-      })
-    } finally {
-      vi.useRealTimers()
-    }
-  })
-
-  it('sends small redraws immediately after terminal input', () => {
-    vi.useFakeTimers()
-    try {
-      const { batcher, fake, setNow } = createHarness()
-
-      setNow(10)
-      batcher.markInput('client-1', 'session-1')
-      setNow(15)
-      batcher.enqueue('client-1', 'session-1', '\x1b[20;2Hredraw')
-
-      expect(fake.write).toHaveBeenCalledTimes(1)
-      expect(parseWrite(fake.write.mock.calls[0])).toMatchObject({
-        type: 'event',
-        event: 'data',
-        sessionId: 'session-1',
-        payload: { data: '\x1b[20;2Hredraw' }
-      })
-      vi.advanceTimersByTime(8)
-      expect(fake.write).toHaveBeenCalledTimes(1)
-    } finally {
-      vi.useRealTimers()
-    }
-  })
-
-  it('flushes only the interactive session when another session has pending output', () => {
-    vi.useFakeTimers()
-    try {
-      const { batcher, fake, setNow } = createHarness()
-
-      batcher.enqueue('client-1', 'background-session', 'background')
-      setNow(20)
-      batcher.markInput('client-1', 'interactive-session')
-      setNow(21)
-      batcher.enqueue('client-1', 'interactive-session', 'redraw')
-
-      expect(fake.write).toHaveBeenCalledTimes(1)
-      expect(parseWrite(fake.write.mock.calls[0])).toMatchObject({
-        sessionId: 'interactive-session',
-        payload: { data: 'redraw' }
-      })
-
-      vi.advanceTimersByTime(8)
-      expect(fake.write).toHaveBeenCalledTimes(2)
-      expect(parseWrite(fake.write.mock.calls[1])).toMatchObject({
-        sessionId: 'background-session',
-        payload: { data: 'background' }
-      })
-    } finally {
-      vi.useRealTimers()
-    }
-  })
-
-  it('waits for drain after an immediate interactive write backpressures', () => {
-    vi.useFakeTimers()
-    try {
-      const { batcher, fake, setNow } = createHarness([false, true])
-
-      setNow(10)
-      batcher.markInput('client-1', 'session-1')
-      setNow(11)
-      batcher.enqueue('client-1', 'session-1', 'redraw')
-      batcher.enqueue('client-1', 'session-2', 'queued')
-      batcher.flush('client-1')
-
-      expect(fake.write).toHaveBeenCalledTimes(1)
-      fake.drain()
-
-      expect(fake.write).toHaveBeenCalledTimes(2)
-      expect(parseWrite(fake.write.mock.calls[1])).toMatchObject({
-        sessionId: 'session-2',
-        payload: { data: 'queued' }
-      })
-    } finally {
-      vi.useRealTimers()
-    }
-  })
-
-  it('batches large output even after recent terminal input', () => {
-    vi.useFakeTimers()
-    try {
-      const { batcher, fake, setNow } = createHarness()
-      const largeOutput = 'x'.repeat(1025)
-
-      setNow(10)
-      batcher.markInput('client-1', 'session-1')
-      setNow(11)
-      batcher.enqueue('client-1', 'session-1', largeOutput)
-
-      expect(fake.write).not.toHaveBeenCalled()
-      vi.advanceTimersByTime(8)
-      expect(fake.write).toHaveBeenCalledTimes(1)
-      expect(parseWrite(fake.write.mock.calls[0])).toMatchObject({
-        sessionId: 'session-1',
-        payload: { data: largeOutput }
-      })
-    } finally {
-      vi.useRealTimers()
-    }
-  })
-
-  it('batches stale output after the interactive window expires', () => {
-    vi.useFakeTimers()
-    try {
-      const { batcher, fake, setNow } = createHarness()
-
-      setNow(10)
-      batcher.markInput('client-1', 'session-1')
-      setNow(111)
-      batcher.enqueue('client-1', 'session-1', 'stale redraw')
-
-      expect(fake.write).not.toHaveBeenCalled()
-      vi.advanceTimersByTime(8)
-      expect(fake.write).toHaveBeenCalledTimes(1)
-      expect(parseWrite(fake.write.mock.calls[0])).toMatchObject({
-        sessionId: 'session-1',
-        payload: { data: 'stale redraw' }
-      })
-    } finally {
-      vi.useRealTimers()
-    }
-  })
-
-  it('forgets recent input when a session is cleared', () => {
-    vi.useFakeTimers()
-    try {
-      const { batcher, fake, setNow } = createHarness()
-
-      setNow(10)
-      batcher.markInput('client-1', 'session-1')
-      batcher.clearSessionInput('client-1', 'session-1')
-      setNow(11)
-      batcher.enqueue('client-1', 'session-1', 'redraw')
-
-      expect(fake.write).not.toHaveBeenCalled()
-      vi.advanceTimersByTime(8)
-      expect(fake.write).toHaveBeenCalledTimes(1)
-      expect(parseWrite(fake.write.mock.calls[0])).toMatchObject({
-        sessionId: 'session-1',
-        payload: { data: 'redraw' }
-      })
-    } finally {
-      vi.useRealTimers()
-    }
-  })
-})

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

@@ -1,83 +0,0 @@
-import type { Socket } from 'net'
-import { vi } from 'vitest'
-import { DaemonStreamDataBatcher } from './daemon-stream-data-batcher'
-
-export function parseWrite(call: unknown[]): unknown {
-  return JSON.parse(String(call[0]).trim())
-}
-
-export function createFakeSocket(writeResults: boolean[] = [true]): {
-  socket: Socket
-  write: ReturnType<typeof vi.fn>
-  removeListener: ReturnType<typeof vi.fn>
-  drain: () => void
-  close: () => void
-  error: () => void
-  staleDrain: () => void
-  staleClose: () => void
-  staleError: () => void
-} {
-  let drainHandler: (() => void) | null = null
-  let closeHandler: (() => void) | null = null
-  let errorHandler: (() => void) | null = null
-  let removedDrainHandler: (() => void) | null = null
-  let removedCloseHandler: (() => void) | null = null
-  let removedErrorHandler: (() => void) | null = null
-  const write = vi.fn(() => writeResults.shift() ?? true)
-  const removeListener = vi.fn((event: string, handler: () => void) => {
-    if (event === 'drain' && drainHandler === handler) {
-      removedDrainHandler = handler
-      drainHandler = null
-    } else if (event === 'close' && closeHandler === handler) {
-      removedCloseHandler = handler
-      closeHandler = null
-    } else if (event === 'error' && errorHandler === handler) {
-      removedErrorHandler = handler
-      errorHandler = null
-    }
-    return socket
-  })
-  const socket = {
-    destroyed: false,
-    write,
-    removeListener,
-    once: vi.fn((event: string, handler: () => void) => {
-      if (event === 'drain') {
-        drainHandler = handler
-      } else if (event === 'close') {
-        closeHandler = handler
-      } else if (event === 'error') {
-        errorHandler = handler
-      }
-      return socket
-    })
-  } as unknown as Socket
-
-  return {
-    socket,
-    write,
-    removeListener,
-    drain: () => drainHandler?.(),
-    close: () => closeHandler?.(),
-    error: () => errorHandler?.(),
-    staleDrain: () => removedDrainHandler?.(),
-    staleClose: () => removedCloseHandler?.(),
-    staleError: () => removedErrorHandler?.()
-  }
-}
-
-export function createHarness(writeResults: boolean[] = [true]) {
-  let now = 0
-  const fake = createFakeSocket(writeResults)
-  const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: fake.socket }), {
-    now: () => now
-  })
-
-  return {
-    batcher,
-    fake,
-    setNow(value: number) {
-      now = value
-    }
-  }
-}

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

@@ -1,249 +0,0 @@
-import { describe, expect, it, vi } from 'vitest'
-import { DaemonStreamDataBatcher } from './daemon-stream-data-batcher'
-import { createFakeSocket } from './daemon-stream-data-batcher-test-harness'
-
-describe('DaemonStreamDataBatcher', () => {
-  it('drops queued output when the backpressured stream errors before drain', () => {
-    const fake = createFakeSocket([false, true])
-    const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: fake.socket }))
-
-    batcher.enqueue('client-1', 'session-a', 'first')
-    batcher.enqueue('client-1', 'session-b', 'second')
-
-    batcher.flush('client-1')
-    fake.error()
-    fake.drain()
-
-    expect(fake.write).toHaveBeenCalledTimes(1)
-  })
-
-  it('cleans up unused backpressure listeners after drain', () => {
-    const fake = createFakeSocket([false, true])
-    const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: fake.socket }))
-
-    batcher.enqueue('client-1', 'session-a', 'first')
-    batcher.enqueue('client-1', 'session-b', 'second')
-
-    batcher.flush('client-1')
-    fake.drain()
-
-    expect(fake.removeListener).toHaveBeenCalledWith('close', expect.any(Function))
-    expect(fake.removeListener).toHaveBeenCalledWith('error', expect.any(Function))
-    fake.close()
-
-    expect(fake.write).toHaveBeenCalledTimes(2)
-  })
-
-  it('drops queued output when the backpressured stream closes before drain', () => {
-    const fake = createFakeSocket([false, true])
-    const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: fake.socket }))
-
-    batcher.enqueue('client-1', 'session-a', 'first')
-    batcher.enqueue('client-1', 'session-b', 'second')
-
-    batcher.flush('client-1')
-    fake.close()
-    fake.drain()
-
-    expect(fake.write).toHaveBeenCalledTimes(1)
-  })
-
-  it('does not keep queued output forever when drain never arrives', () => {
-    vi.useFakeTimers()
-    const warn = vi.spyOn(console, 'warn').mockImplementation(() => {})
-    try {
-      const fake = createFakeSocket([false, true])
-      const onStreamFailure = vi.fn()
-      const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: fake.socket }), {
-        onStreamFailure
-      })
-
-      batcher.enqueue('client-1', 'session-a', 'first')
-      batcher.enqueue('client-1', 'session-b', 'second')
-      batcher.flush('client-1')
-
-      vi.advanceTimersByTime(30_000)
-      fake.drain()
-
-      expect(fake.write).toHaveBeenCalledTimes(1)
-      expect(warn).toHaveBeenCalledWith(
-        '[daemon] PTY stream socket drain timed out',
-        expect.objectContaining({ clientId: 'client-1' })
-      )
-      expect(onStreamFailure).toHaveBeenCalledWith('client-1')
-    } finally {
-      warn.mockRestore()
-      vi.useRealTimers()
-    }
-  })
-
-  it('orders exit behind queued data while waiting for drain', () => {
-    const fake = createFakeSocket([false, true, true])
-    const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: fake.socket }))
-
-    batcher.enqueue('client-1', 'session-a', 'first')
-    batcher.flush('client-1')
-    batcher.enqueue('client-1', 'session-b', 'second')
-    batcher.enqueueExit('client-1', 'session-a', 0)
-
-    expect(fake.write).toHaveBeenCalledTimes(1)
-    fake.drain()
-
-    expect(fake.write).toHaveBeenCalledTimes(3)
-    expect(fake.write.mock.calls[1]?.[0]).toContain('"data"')
-    expect(fake.write.mock.calls[1]?.[0]).toContain('second')
-    expect(fake.write.mock.calls[2]?.[0]).toContain('"exit"')
-  })
-
-  it('waits for socket drain before continuing after stream backpressure', () => {
-    const fake = createFakeSocket([false, true])
-    const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: fake.socket }))
-
-    batcher.enqueue('client-1', 'session-a', 'first')
-    batcher.enqueue('client-1', 'session-b', 'second')
-
-    batcher.flush('client-1')
-    expect(fake.write).toHaveBeenCalledTimes(1)
-    expect(fake.write.mock.calls[0]?.[0]).toContain('first')
-
-    fake.drain()
-
-    expect(fake.write).toHaveBeenCalledTimes(2)
-    expect(fake.write.mock.calls[1]?.[0]).toContain('second')
-  })
-
-  it('queues additional output while waiting for drain', () => {
-    const fake = createFakeSocket([false, true])
-    const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: fake.socket }))
-
-    batcher.enqueue('client-1', 'session-a', 'first')
-    batcher.flush('client-1')
-
-    batcher.enqueue('client-1', 'session-b', 'second')
-    batcher.flush('client-1')
-    expect(fake.write).toHaveBeenCalledTimes(1)
-
-    fake.drain()
-
-    expect(fake.write).toHaveBeenCalledTimes(2)
-    expect(fake.write.mock.calls[1]?.[0]).toContain('second')
-  })
-
-  it('ignores stale drain callbacks after clearing an old client stream', () => {
-    const oldStream = createFakeSocket([false, true])
-    const newStream = createFakeSocket([true])
-    let streamSocket = oldStream.socket
-    const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket }))
-
-    batcher.enqueue('client-1', 'session-a', 'first')
-    batcher.enqueue('client-1', 'session-b', 'second')
-    batcher.flush('client-1')
-
-    batcher.clear('client-1')
-    streamSocket = newStream.socket
-    batcher.enqueue('client-1', 'session-c', 'third')
-    batcher.flush('client-1')
-    oldStream.staleDrain()
-
-    expect(oldStream.write).toHaveBeenCalledTimes(1)
-    expect(newStream.write).toHaveBeenCalledTimes(1)
-    expect(newStream.write.mock.calls[0]?.[0]).toContain('third')
-  })
-
-  it('ignores stale close and error callbacks after clearing an old client stream', () => {
-    const oldStream = createFakeSocket([false, true])
-    const newStream = createFakeSocket([true])
-    let streamSocket = oldStream.socket
-    const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket }))
-
-    batcher.enqueue('client-1', 'session-a', 'first')
-    batcher.enqueue('client-1', 'session-b', 'second')
-    batcher.flush('client-1')
-
-    batcher.clear('client-1')
-    streamSocket = newStream.socket
-    batcher.enqueue('client-1', 'session-c', 'third')
-    batcher.flush('client-1')
-    oldStream.staleClose()
-    oldStream.staleError()
-
-    expect(oldStream.write).toHaveBeenCalledTimes(1)
-    expect(newStream.write).toHaveBeenCalledTimes(1)
-    expect(newStream.write.mock.calls[0]?.[0]).toContain('third')
-  })
-
-  it('fails the stream when queued output exceeds the hard cap', () => {
-    const warn = vi.spyOn(console, 'warn').mockImplementation(() => {})
-    try {
-      const fake = createFakeSocket([true])
-      const onStreamFailure = vi.fn()
-      const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: fake.socket }), {
-        onStreamFailure
-      })
-
-      batcher.enqueue('client-1', 'session-a', 'x'.repeat(8 * 1024 * 1024 + 1))
-
-      expect(fake.write).not.toHaveBeenCalled()
-      expect(onStreamFailure).toHaveBeenCalledWith('client-1')
-      expect(warn).toHaveBeenCalledWith(
-        '[daemon] PTY stream socket queue exceeded limit',
-        expect.objectContaining({ clientId: 'client-1' })
-      )
-    } finally {
-      warn.mockRestore()
-    }
-  })
-
-  it('fails the stream when queued output cumulatively exceeds the hard cap while backpressured', () => {
-    const warn = vi.spyOn(console, 'warn').mockImplementation(() => {})
-    try {
-      const fake = createFakeSocket([false])
-      const onStreamFailure = vi.fn()
-      const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: fake.socket }), {
-        onStreamFailure
-      })
-
-      batcher.enqueue('client-1', 'session-a', 'first')
-      batcher.flush('client-1')
-      batcher.enqueue('client-1', 'session-a', 'x'.repeat(8 * 1024 * 1024 + 1))
-
-      expect(fake.write).toHaveBeenCalledTimes(1)
-      expect(onStreamFailure).toHaveBeenCalledWith('client-1')
-      expect(warn).toHaveBeenCalledWith(
-        '[daemon] PTY stream socket queue exceeded limit',
-        expect.objectContaining({ clientId: 'client-1' })
-      )
-    } finally {
-      warn.mockRestore()
-    }
-  })
-
-  it('splits very large output into bounded stream frames', () => {
-    const fake = createFakeSocket([true, true])
-    const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: fake.socket }))
-
-    batcher.enqueue('client-1', 'session-a', 'x'.repeat(65 * 1024))
-    batcher.flush('client-1')
-
-    expect(fake.write).toHaveBeenCalledTimes(2)
-  })
-
-  it('yields between large queued flushes', () => {
-    vi.useFakeTimers()
-    try {
-      const fake = createFakeSocket(Array.from({ length: 1025 }, () => true))
-      const batcher = new DaemonStreamDataBatcher(() => ({ streamSocket: fake.socket }))
-
-      for (let i = 0; i < 1025; i++) {
-        batcher.enqueue('client-1', `session-${i}`, `${i}`)
-      }
-      batcher.flush('client-1')
-
-      expect(fake.write).toHaveBeenCalledTimes(1024)
-      vi.advanceTimersByTime(0)
-      expect(fake.write).toHaveBeenCalledTimes(1025)
-    } finally {
-      vi.useRealTimers()
-    }
-  })
-})

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

@@ -1,169 +1,49 @@
 import type { Socket } from 'net'
-import { performance } from 'node:perf_hooks'
-import {
-  compactStreamDataBatch,
-  createPendingStreamDataBatch,
-  getQueuedDataForSession,
-  INTERACTIVE_OUTPUT_MAX_CHARS,
-  INTERACTIVE_OUTPUT_WINDOW_MS,
-  removeQueuedDataForSession,
-  STREAM_DATA_BACKPRESSURE_WARN_BYTES,
-  STREAM_DATA_BATCH_INTERVAL_MS,
-  STREAM_DATA_DRAIN_TIMEOUT_MS,
-  STREAM_DATA_MAX_EVENTS_PER_FLUSH,
-  STREAM_DATA_MAX_PAYLOAD_CHARS,
-  STREAM_DATA_MAX_QUEUED_BYTES,
-  streamInputKey,
-  type PendingStreamDataBatch,
-  type PendingStreamEvent,
-  type StreamDataClient,
-  writeStreamEvent
-} from './daemon-stream-data-batch-state'
+import { encodeNdjson } from './ndjson'
+
+type StreamDataClient = {
+  streamSocket: Socket | null
+}
+
+type PendingStreamDataBatch = {
+  timer: ReturnType<typeof setTimeout> | null
+  queue: { sessionId: string; data: string }[]
+}
+
+// Why: match main-process PTY IPC batching to avoid adding latency while
+// removing daemon socket writes and JSON framing during bursty output.
+const STREAM_DATA_BATCH_INTERVAL_MS = 8
 
 export class DaemonStreamDataBatcher {
   private pendingByClient = new Map<string, PendingStreamDataBatch>()
-  private lastInputAtBySession = new Map<string, number>()
   private getClient: (clientId: string) => StreamDataClient | undefined
-  private onStreamFailure: (clientId: string) => void
-  private now: () => number
 
-  constructor(
-    getClient: (clientId: string) => StreamDataClient | undefined,
-    opts: {
-      onStreamFailure?: (clientId: string) => void
-      now?: () => number
-    } = {}
-  ) {
+  constructor(getClient: (clientId: string) => StreamDataClient | undefined) {
     this.getClient = getClient
-    this.onStreamFailure = opts.onStreamFailure ?? (() => {})
-    this.now = opts.now ?? (() => performance.now())
-  }
-
-  markInput(clientId: string, sessionId: string): void {
-    this.lastInputAtBySession.set(streamInputKey(clientId, sessionId), this.now())
-  }
-
-  clearSessionInput(clientId: string, sessionId: string): void {
-    this.lastInputAtBySession.delete(streamInputKey(clientId, sessionId))
   }
 
   enqueue(clientId: string, sessionId: string, data: string): void {
-    for (let offset = 0; offset < data.length; offset += STREAM_DATA_MAX_PAYLOAD_CHARS) {
-      const shouldContinue = this.enqueueEvent(clientId, {
-        kind: 'data',
-        sessionId,
-        data: data.slice(offset, offset + STREAM_DATA_MAX_PAYLOAD_CHARS)
-      })
-      if (!shouldContinue) {
-        return
-      }
-    }
-  }
-
-  enqueueExit(clientId: string, sessionId: string, code: number): void {
-    this.clearSessionInput(clientId, sessionId)
-    this.enqueueEvent(clientId, { kind: 'exit', sessionId, code })
-    this.flush(clientId)
-  }
-
-  private enqueueEvent(clientId: string, event: PendingStreamEvent): boolean {
     const client = this.getClient(clientId)
     if (!client?.streamSocket || client.streamSocket.destroyed) {
-      return false
-    }
-
-    const batch = this.getOrCreateBatch(clientId)
-    compactStreamDataBatch(batch)
-
-    if (event.kind === 'data' && !batch.waitingForDrain) {
-      const queuedSessionData = getQueuedDataForSession(batch, event.sessionId)
-      const nextSessionData = queuedSessionData + event.data
-      if (this.shouldSendImmediately(clientId, event.sessionId, nextSessionData)) {
-        removeQueuedDataForSession(batch, event.sessionId)
-        const ok = writeStreamEvent(client.streamSocket, {
-          kind: 'data',
-          sessionId: event.sessionId,
-          data: nextSessionData
-        })
-        if (!ok) {
-          this.handleBackpressure(clientId, batch, client.streamSocket)
-          return true
-        }
-        this.deleteBatchIfIdle(clientId, batch)
-        return true
-      }
-    }
-
-    return this.enqueueForBatch(clientId, batch, event)
-  }
-
-  private getOrCreateBatch(clientId: string): PendingStreamDataBatch {
-    let batch = this.pendingByClient.get(clientId)
-    if (!batch) {
-      batch = createPendingStreamDataBatch()
-      this.pendingByClient.set(clientId, batch)
-    }
-    return batch
-  }
-
-  private enqueueForBatch(
-    clientId: string,
-    batch: PendingStreamDataBatch,
-    event: PendingStreamEvent
-  ): boolean {
-    const last = batch.queue.at(-1)
-    if (
-      event.kind === 'data' &&
-      last?.kind === 'data' &&
-      last.sessionId === event.sessionId &&
-      last.data.length + event.data.length <= STREAM_DATA_MAX_PAYLOAD_CHARS
-    ) {
-      last.data += event.data
-      batch.queuedDataBytes += Buffer.byteLength(event.data, 'utf8')
-    } else {
-      batch.queue.push(event)
-      if (event.kind === 'data') {
-        batch.queuedDataBytes += Buffer.byteLength(event.data, 'utf8')
-      }
-    }
-
-    if (batch.queuedDataBytes > STREAM_DATA_MAX_QUEUED_BYTES) {
-      console.warn('[daemon] PTY stream socket queue exceeded limit', {
-        clientId,
-        queuedEvents: batch.queue.length - batch.queueHead,
-        queuedBytes: batch.queuedDataBytes
-      })
-      // Why: backpressure is on the renderer's single stream socket. Once that
-      // socket is unhealthy, per-PTY recovery cannot make progress until the
-      // client reconnects and reattaches its active sessions.
-      this.failStream(clientId)
-      return false
-    }
-
-    if (!batch.timer && !batch.waitingForDrain) {
-      batch.timer = setTimeout(() => this.flush(clientId), STREAM_DATA_BATCH_INTERVAL_MS)
-    }
-    return true
-  }
-
-  private shouldSendImmediately(clientId: string, sessionId: string, data: string): boolean {
-    const lastInputAt = this.lastInputAtBySession.get(streamInputKey(clientId, sessionId))
-    return (
-      data.length <= INTERACTIVE_OUTPUT_MAX_CHARS &&
-      lastInputAt !== undefined &&
-      this.now() - lastInputAt <= INTERACTIVE_OUTPUT_WINDOW_MS
-    )
-  }
-
-  private deleteBatchIfIdle(clientId: string, batch: PendingStreamDataBatch): void {
-    if (batch.waitingForDrain || batch.queueHead < batch.queue.length) {
       return
     }
-    if (batch.timer) {
-      clearTimeout(batch.timer)
-      batch.timer = null
+
+    let batch = this.pendingByClient.get(clientId)
+    if (!batch) {
+      batch = { timer: null, queue: [] }
+      this.pendingByClient.set(clientId, batch)
+    }
+
+    const last = batch.queue.at(-1)
+    if (last?.sessionId === sessionId) {
+      last.data += data
+    } else {
+      batch.queue.push({ sessionId, data })
+    }
+
+    if (!batch.timer) {
+      batch.timer = setTimeout(() => this.flush(clientId), STREAM_DATA_BATCH_INTERVAL_MS)
     }
-    this.pendingByClient.delete(clientId)
   }
 
   flush(clientId: string): void {
@@ -176,41 +56,23 @@ export class DaemonStreamDataBatcher {
       clearTimeout(batch.timer)
       batch.timer = null
     }
-    if (batch.waitingForDrain) {
-      return
-    }
+    this.pendingByClient.delete(clientId)
 
     const client = this.getClient(clientId)
     if (!client?.streamSocket || client.streamSocket.destroyed) {
-      this.pendingByClient.delete(clientId)
       return
     }
-    const streamSocket = client.streamSocket
 
-    let flushedEvents = 0
-    while (batch.queueHead < batch.queue.length) {
-      const entry = batch.queue[batch.queueHead]!
-      batch.queueHead += 1
-      flushedEvents += 1
-      if (entry.kind === 'data') {
-        batch.queuedDataBytes -= Buffer.byteLength(entry.data, 'utf8')
-      }
-      const ok = writeStreamEvent(streamSocket, entry)
-      if (!ok) {
-        this.handleBackpressure(clientId, batch, streamSocket)
-        return
-      }
-      if (
-        flushedEvents >= STREAM_DATA_MAX_EVENTS_PER_FLUSH &&
-        batch.queueHead < batch.queue.length
-      ) {
-        compactStreamDataBatch(batch)
-        batch.timer = setTimeout(() => this.flush(clientId), 0)
-        batch.timer.unref?.()
-        return
-      }
+    for (const entry of batch.queue) {
+      client.streamSocket.write(
+        encodeNdjson({
+          type: 'event',
+          event: 'data',
+          sessionId: entry.sessionId,
+          payload: { data: entry.data }
+        })
+      )
     }
-    this.pendingByClient.delete(clientId)
   }
 
   clear(clientId?: string): void {
@@ -220,99 +82,10 @@ export class DaemonStreamDataBatcher {
         : [[clientId, this.pendingByClient.get(clientId)] as const]
 
     for (const [id, batch] of batches) {
-      batch?.cleanupWait?.()
       if (batch?.timer) {
         clearTimeout(batch.timer)
       }
-      if (batch?.drainTimer) {
-        clearTimeout(batch.drainTimer)
-      }
       this.pendingByClient.delete(id)
     }
-    if (clientId === undefined) {
-      this.lastInputAtBySession.clear()
-    } else {
-      for (const key of this.lastInputAtBySession.keys()) {
-        if (key.startsWith(`${clientId}\0`)) {
-          this.lastInputAtBySession.delete(key)
-        }
-      }
-    }
-  }
-
-  private handleBackpressure(
-    clientId: string,
-    batch: PendingStreamDataBatch,
-    streamSocket: Socket
-  ): void {
-    batch.waitingForDrain = true
-    if (batch.timer) {
-      clearTimeout(batch.timer)
-      batch.timer = null
-    }
-    compactStreamDataBatch(batch)
-    if (batch.queuedDataBytes >= STREAM_DATA_BACKPRESSURE_WARN_BYTES && !batch.warnedBackpressure) {
-      batch.warnedBackpressure = true
-      console.warn('[daemon] PTY stream socket backpressure', {
-        clientId,
-        queuedEvents: batch.queue.length - batch.queueHead,
-        queuedBytes: batch.queuedDataBytes
-      })
-    }
-    let settled = false
-    const handleDrain = (): void => {
-      if (settled) {
-        return
-      }
-      cleanupWait()
-      const current = this.pendingByClient.get(clientId)
-      if (current !== batch) {
-        return
-      }
-      current.waitingForDrain = false
-      this.flush(clientId)
-    }
-    const handleTerminal = (): void => {
-      if (settled) {
-        return
-      }
-      cleanupWait()
-      if (this.pendingByClient.get(clientId) === batch) {
-        this.pendingByClient.delete(clientId)
-      }
-    }
-    const cleanupWait = (): void => {
-      settled = true
-      if (batch.drainTimer) {
-        clearTimeout(batch.drainTimer)
-        batch.drainTimer = null
-      }
-      batch.cleanupWait = null
-      streamSocket.removeListener('drain', handleDrain)
-      streamSocket.removeListener('close', handleTerminal)
-      streamSocket.removeListener('error', handleTerminal)
-    }
-    batch.cleanupWait = cleanupWait
-    batch.drainTimer = setTimeout(() => {
-      if (settled) {
-        return
-      }
-      console.warn('[daemon] PTY stream socket drain timed out', {
-        clientId,
-        queuedEvents: batch.queue.length - batch.queueHead,
-        queuedBytes: batch.queuedDataBytes
-      })
-      cleanupWait()
-      this.failStream(clientId)
-    }, STREAM_DATA_DRAIN_TIMEOUT_MS)
-    batch.drainTimer.unref?.()
-    streamSocket.once('close', handleTerminal)
-    streamSocket.once('error', handleTerminal)
-    streamSocket.once('drain', handleDrain)
-  }
-
-  private failStream(clientId: string): void {
-    this.clear(clientId)
-    this.onStreamFailure(clientId)
   }
 }

src/main/ipc/pty.test.ts

@@ -2452,41 +2452,6 @@ describe('registerPtyHandlers', () => {
     }
   })
 
-  it('splits very large batched PTY output across IPC flush turns', async () => {
-    vi.useFakeTimers()
-    const mockProc = createMockProc()
-    spawnMock.mockReturnValue(mockProc.proc)
-
-    try {
-      registerPtyHandlers(mainWindow as never)
-      const spawnResult = (await handlers.get('pty:spawn')!(null, {
-        cols: 80,
-        rows: 24,
-        cwd: '/tmp'
-      })) as { id: string }
-      mainWindow.webContents.send.mockClear()
-
-      const largeOutput = 'x'.repeat(9 * 64 * 1024)
-      mockProc.emitData(largeOutput)
-
-      vi.advanceTimersByTime(8)
-      expect(mainWindow.webContents.send).toHaveBeenCalledTimes(8)
-      expect(mainWindow.webContents.send).toHaveBeenLastCalledWith('pty:data', {
-        id: spawnResult.id,
-        data: 'x'.repeat(64 * 1024)
-      })
-
-      vi.advanceTimersByTime(8)
-      expect(mainWindow.webContents.send).toHaveBeenCalledTimes(9)
-      expect(mainWindow.webContents.send).toHaveBeenLastCalledWith('pty:data', {
-        id: spawnResult.id,
-        data: 'x'.repeat(64 * 1024)
-      })
-    } finally {
-      vi.useRealTimers()
-    }
-  })
-
   it('batches combined pending output that exceeds the interactive size limit', async () => {
     vi.useFakeTimers()
     const mockProc = createMockProc()

src/main/ipc/pty.ts

@@ -650,12 +650,6 @@ export function registerPtyHandlers(
   // large output and non-interactive output must still use the batcher.
   const INTERACTIVE_OUTPUT_WINDOW_MS = 100
   const INTERACTIVE_OUTPUT_MAX_CHARS = 1024
-  const PTY_IPC_CHUNK_CHARS = 64 * 1024
-  const PTY_MAX_IPC_CHUNKS_PER_FLUSH = 8
-
-  const sendPtyData = (id: string, data: string): void => {
-    mainWindow.webContents.send('pty:data', { id, data })
-  }
 
   const flushPendingData = (): void => {
     flushTimer = null
@@ -663,30 +657,10 @@ export function registerPtyHandlers(
       pendingData.clear()
       return
     }
-
-    let sentChunks = 0
-    while (pendingData.size > 0 && sentChunks < PTY_MAX_IPC_CHUNKS_PER_FLUSH) {
-      const entry = pendingData.entries().next().value
-      if (!entry) {
-        break
-      }
-      const [id, data] = entry
-      pendingData.delete(id)
-      if (data.length <= PTY_IPC_CHUNK_CHARS) {
-        sendPtyData(id, data)
-      } else {
-        sendPtyData(id, data.slice(0, PTY_IPC_CHUNK_CHARS))
-        // Why: a single noisy PTY can otherwise serialize megabytes of IPC in
-        // one main-process turn. Requeue the remainder behind other PTYs so
-        // active terminal redraws and control IPC keep getting turns.
-        pendingData.set(id, data.slice(PTY_IPC_CHUNK_CHARS))
-      }
-      sentChunks++
-    }
-
-    if (pendingData.size > 0) {
-      flushTimer = setTimeout(flushPendingData, PTY_BATCH_INTERVAL_MS)
+    for (const [id, data] of pendingData) {
+      mainWindow.webContents.send('pty:data', { id, data })
     }
+    pendingData.clear()
   }
 
   const clearFlushTimerIfIdle = (): void => {
@@ -740,7 +714,10 @@ export function registerPtyHandlers(
         clearFlushTimerIfIdle()
         // Why: agent TUIs redraw small prompt regions after every keystroke.
         // Waiting for the throughput batch timer adds visible input latency.
-        sendPtyData(payload.id, nextData)
+        mainWindow.webContents.send('pty:data', {
+          id: payload.id,
+          data: nextData
+        })
         return
       }
       pendingData.set(payload.id, nextData)
@@ -761,7 +738,7 @@ export function registerPtyHandlers(
         // tears down the terminal on pty:exit before the batch timer fires.
         const remaining = pendingData.get(payload.id)
         if (remaining) {
-          sendPtyData(payload.id, remaining)
+          mainWindow.webContents.send('pty:data', { id: payload.id, data: remaining })
           pendingData.delete(payload.id)
         }
         lastInputAtByPty.delete(payload.id)

src/renderer/src/components/terminal-pane/pty-connection.test.ts

@@ -2009,51 +2009,7 @@ describe('connectPanePty', () => {
     }
   })
 
-  it('queues visible inactive split-pane PTY bytes so active pane input stays responsive', async () => {
-    const pendingTimeouts: (() => void)[] = []
-    const originalSetTimeout = globalThis.setTimeout
-    globalThis.setTimeout = vi.fn((fn: () => void) => {
-      pendingTimeouts.push(fn)
-      return 999 as unknown as ReturnType<typeof setTimeout>
-    }) as unknown as typeof setTimeout
-
-    try {
-      const { connectPanePty } = await import('./pty-connection')
-      const transport = createMockTransport()
-      const capturedDataCallback: { current: ((data: string) => void) | null } = { current: null }
-      transport.connect.mockImplementation(
-        async ({ callbacks }: { callbacks: ConnectCallbacks }) => {
-          capturedDataCallback.current = callbacks.onData ?? null
-          return 'pty-id'
-        }
-      )
-      transportFactoryQueue.push(transport)
-
-      const pane = createPane(1)
-      const manager = createManager(2)
-      manager.getActivePane.mockReturnValue({ id: 2 })
-      const deps = createDeps({
-        isVisibleRef: { current: true }
-      })
-
-      connectPanePty(pane as never, manager as never, deps as never)
-      await flushAsyncTicks(6)
-
-      expect(capturedDataCallback.current).not.toBeNull()
-      capturedDataCallback.current?.('visible split output\r\n')
-      expect(pane.terminal.write).not.toHaveBeenCalledWith('visible split output\r\n')
-
-      for (const fn of pendingTimeouts) {
-        fn()
-      }
-
-      expect(pane.terminal.write).toHaveBeenCalledWith('visible split output\r\n')
-    } finally {
-      globalThis.setTimeout = originalSetTimeout
-    }
-  })
-
-  it('writes active visible split-pane PTY bytes immediately', async () => {
+  it('writes visible split-pane PTY bytes immediately even when the tab is not active', async () => {
     const { connectPanePty } = await import('./pty-connection')
     const transport = createMockTransport()
     const capturedDataCallback: { current: ((data: string) => void) | null } = { current: null }
@@ -2064,9 +2020,9 @@ describe('connectPanePty', () => {
     transportFactoryQueue.push(transport)
 
     const pane = createPane(1)
-    const manager = createManager(2)
-    manager.getActivePane.mockReturnValue({ id: 1 })
+    const manager = createManager(1)
     const deps = createDeps({
+      isActiveRef: { current: false },
       isVisibleRef: { current: true }
     })
 
@@ -2074,9 +2030,9 @@ describe('connectPanePty', () => {
     await flushAsyncTicks(6)
 
     expect(capturedDataCallback.current).not.toBeNull()
-    capturedDataCallback.current?.('active split output\r\n')
+    capturedDataCallback.current?.('visible split output\r\n')
 
-    expect(pane.terminal.write).toHaveBeenCalledWith('active split output\r\n')
+    expect(pane.terminal.write).toHaveBeenCalledWith('visible split output\r\n')
   })
 
   it('marks panes that receive Arabic output for DOM rendering', async () => {

src/renderer/src/components/terminal-pane/pty-connection.ts

@@ -1139,12 +1139,11 @@ export function connectPanePty(
         manager.markPaneHasComplexScriptOutput(pane.id)
       }
       recordTerminalOutput(pane.terminal)
-      // Why: the active split pane owns keyboard latency. Visible inactive
-      // panes still drain, but through the shared scheduler so a build log in
-      // another split cannot monopolize xterm writes while the user types.
-      const activePaneId = manager.getActivePane()?.id ?? pane.id
+      // Why: visibility is the right gate — split-pane layouts have multiple
+      // visible-but-inactive panes whose output the user is watching. Only
+      // hidden panes (background tabs) should be throttled.
       writeTerminalOutput(pane.terminal, data, {
-        foreground: deps.isVisibleRef.current && activePaneId === pane.id
+        foreground: deps.isVisibleRef.current
       })
 
       if (pendingStartupCommand) {

src/renderer/src/store/slices/agent-status.test.ts

@@ -192,46 +192,6 @@ describe('agent status tool + assistant fields', () => {
     expect(store.getState().sortEpoch).toBe(firstSortEpoch + 1)
   })
 
-  it('throttles unchanged fresh same-state heartbeats to avoid status-map churn', () => {
-    vi.useFakeTimers()
-    const store = createTestStore()
-    store
-      .getState()
-      .setAgentStatus(
-        'tab-1:1',
-        { state: 'working', prompt: 'p1', agentType: 'claude', toolName: 'Read' },
-        'claude',
-        { updatedAt: 1_000, stateStartedAt: 1_000 }
-      )
-    const firstMap = store.getState().agentStatusByPaneKey
-    const firstEntry = firstMap['tab-1:1']
-
-    store
-      .getState()
-      .setAgentStatus(
-        'tab-1:1',
-        { state: 'working', prompt: 'p1', agentType: 'claude', toolName: 'Read' },
-        'claude',
-        { updatedAt: 1_500, stateStartedAt: 1_000 }
-      )
-
-    expect(store.getState().agentStatusByPaneKey).toBe(firstMap)
-    expect(store.getState().agentStatusByPaneKey['tab-1:1']).toBe(firstEntry)
-    expect(store.getState().agentStatusByPaneKey['tab-1:1'].updatedAt).toBe(1_000)
-
-    store
-      .getState()
-      .setAgentStatus(
-        'tab-1:1',
-        { state: 'working', prompt: 'p1', agentType: 'claude', toolName: 'Read' },
-        'claude',
-        { updatedAt: 2_000, stateStartedAt: 1_000 }
-      )
-
-    expect(store.getState().agentStatusByPaneKey).not.toBe(firstMap)
-    expect(store.getState().agentStatusByPaneKey['tab-1:1'].updatedAt).toBe(2_000)
-  })
-
   it('bumps global epochs when a stale same-state entry refreshes', () => {
     vi.useFakeTimers()
     const store = createTestStore()

src/renderer/src/store/slices/agent-status.ts

@@ -123,27 +123,6 @@ function paneKeyMatchesAnyTabPrefix(paneKey: string, tabPrefixes: string[]): boo
   return false
 }
 
-const UNCHANGED_AGENT_STATUS_UPDATE_MIN_INTERVAL_MS = 1_000
-
-function isUnchangedAgentStatusHeartbeat(
-  previous: AgentStatusEntry,
-  next: AgentStatusEntry
-): boolean {
-  return (
-    previous.state === next.state &&
-    previous.prompt === next.prompt &&
-    previous.stateStartedAt === next.stateStartedAt &&
-    previous.agentType === next.agentType &&
-    previous.paneKey === next.paneKey &&
-    previous.terminalTitle === next.terminalTitle &&
-    previous.stateHistory === next.stateHistory &&
-    previous.toolName === next.toolName &&
-    previous.toolInput === next.toolInput &&
-    previous.lastAssistantMessage === next.lastAssistantMessage &&
-    previous.interrupted === next.interrupted
-  )
-}
-
 function pruneMigrationUnsupportedEntries(
   entries: Record<string, MigrationUnsupportedPtyEntry>,
   predicate: (entry: MigrationUnsupportedPtyEntry) => boolean
@@ -305,23 +284,6 @@ export const createAgentStatusSlice: StateCreator<AppState, [], [], AgentStatusS
         // status ping would churn that map reference and force spurious
         // re-renders in any subscriber selecting on it.
         const hasSuppressor = paneKey in s.retentionSuppressedPaneKeys
-        const hasMigrationUnsupportedForPaneKey = Object.values(s.migrationUnsupportedByPtyId).some(
-          (entry) => entry.paneKey === paneKey
-        )
-        // Why: Codex/Claude hook heartbeats can arrive many times per second
-        // with only `updatedAt` changed. Rewriting the whole status map for
-        // those pings wakes sidebar/runtime subscribers without changing what
-        // the user sees, so keep freshness accurate at human-scale cadence.
-        if (
-          existing &&
-          !sortRelevantChange &&
-          !hasSuppressor &&
-          !hasMigrationUnsupportedForPaneKey &&
-          updatedAt - existing.updatedAt < UNCHANGED_AGENT_STATUS_UPDATE_MIN_INTERVAL_MS &&
-          isUnchangedAgentStatusHeartbeat(existing, entry)
-        ) {
-          return s
-        }
         let nextRetentionSuppressedPaneKeys = s.retentionSuppressedPaneKeys
         if (hasSuppressor) {
           nextRetentionSuppressedPaneKeys = { ...s.retentionSuppressedPaneKeys }

tests/e2e/terminal-output-scheduler.spec.ts

@@ -15,13 +15,7 @@ import {
   waitForActiveWorktree,
   waitForSessionReady
 } from './helpers/store'
-import {
-  getTerminalContent,
-  splitActiveTerminalPane,
-  waitForActiveTerminalManager,
-  waitForPaneCount,
-  waitForPaneIdentitySnapshot
-} from './helpers/terminal'
+import { getTerminalContent, waitForActiveTerminalManager } from './helpers/terminal'
 
 type SchedulerDebugSnapshot = {
   backgroundEnqueueCount: number
@@ -255,75 +249,4 @@ test.describe('Terminal output scheduler', () => {
       })
       .toBe(true)
   })
-
-  test('visible inactive split-pane output uses the shared drain @headful', async ({
-    orcaPage
-  }) => {
-    await waitForSessionReady(orcaPage)
-    await waitForActiveWorktree(orcaPage)
-    await ensureTerminalVisible(orcaPage)
-    await waitForActiveTerminalManager(orcaPage, 30_000)
-    await waitForPaneCount(orcaPage, 1, 30_000)
-
-    await splitActiveTerminalPane(orcaPage, 'horizontal')
-    const snapshot = await waitForPaneIdentitySnapshot(orcaPage, 2)
-    const activePane = snapshot.panes[0]
-    const inactivePane = snapshot.panes[1]
-    if (!activePane.ptyId || !inactivePane.ptyId) {
-      throw new Error('Split pane PTY ids were unavailable')
-    }
-
-    await orcaPage.evaluate(
-      ({ tabId, paneId }) => {
-        const manager = window.__paneManagers?.get(tabId)
-        if (!manager) {
-          throw new Error('Active terminal PaneManager is not mounted')
-        }
-        // Why: the active split pane owns keyboard latency even though both
-        // split panes are visible in this headful repro.
-        manager.setActivePane(paneId, { focus: true })
-      },
-      { tabId: snapshot.tabId, paneId: activePane.numericPaneId }
-    )
-
-    await resetSchedulerDebug(orcaPage)
-
-    const runId = Date.now()
-    const activeMarker = `ACTIVE_SPLIT_SCHED_${runId}`
-    const inactiveMarker = `INACTIVE_SPLIT_SCHED_${runId}`
-    await sendPtyCommands(orcaPage, [
-      {
-        ptyId: inactivePane.ptyId,
-        command: nodeConsoleCommand(`'x'.repeat(120000) + ':${inactiveMarker}'`)
-      },
-      {
-        ptyId: activePane.ptyId,
-        command: nodeConsoleCommand(`'${activeMarker}'`)
-      }
-    ])
-
-    await expect
-      .poll(async () => (await getTerminalContent(orcaPage)).includes(activeMarker), {
-        timeout: 5_000,
-        message: 'Active split pane did not render foreground output during inactive burst'
-      })
-      .toBe(true)
-
-    await expect
-      .poll(async () => (await getSchedulerDebug(orcaPage)).backgroundEnqueueCount, {
-        timeout: 5_000,
-        message: 'Visible inactive split-pane output bypassed the shared scheduler'
-      })
-      .toBeGreaterThanOrEqual(1)
-
-    await expect
-      .poll(async () => (await getSchedulerDebug(orcaPage)).backgroundWriteCount, {
-        timeout: 10_000,
-        message: 'Visible inactive split-pane output did not drain from the shared scheduler'
-      })
-      .toBeGreaterThanOrEqual(1)
-
-    const debug = await getSchedulerDebug(orcaPage)
-    expect(debug.foregroundWriteCount).toBeGreaterThan(0)
-  })
 })