Fix terminal stream backpressure and split-pane input lag (#2341)
Co-authored-by: Orca <help@stably.ai>
This commit is contained in:
Jinwoo Hong
GitHub
parent
3333442932
commit
f010bb8619
|
|
@ -50,10 +50,14 @@ 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()
|
||||
|
|
@ -73,10 +77,22 @@ 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) {
|
||||
|
|
@ -252,6 +268,30 @@ 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)', () => {
|
||||
|
|
|
|||
|
|
@ -31,6 +31,7 @@ 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
|
||||
|
|
@ -74,15 +75,20 @@ 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
|
||||
|
|
@ -90,11 +96,18 @@ export class DaemonClient {
|
|||
this.connectionGeneration++
|
||||
|
||||
const gen = this.connectionGeneration
|
||||
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)
|
||||
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}`)
|
||||
)
|
||||
} catch (error) {
|
||||
this.controlSocket?.destroy()
|
||||
this.streamSocket?.destroy()
|
||||
|
|
@ -130,14 +143,19 @@ export class DaemonClient {
|
|||
})
|
||||
}
|
||||
|
||||
notify(type: string, payload: unknown): void {
|
||||
notify(type: string, payload: unknown): boolean {
|
||||
if (!this.connected || !this.controlSocket) {
|
||||
return
|
||||
return false
|
||||
}
|
||||
|
||||
const id = `${NOTIFY_PREFIX}${++this.requestCounter}`
|
||||
const msg = { id, type, ...(payload !== undefined ? { payload } : {}) }
|
||||
this.controlSocket.write(encodeNdjson(msg))
|
||||
try {
|
||||
this.controlSocket.write(encodeNdjson(msg))
|
||||
return true
|
||||
} catch {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
onEvent(listener: (event: unknown) => void): () => void {
|
||||
|
|
@ -161,8 +179,10 @@ 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)
|
||||
|
|
@ -176,6 +196,12 @@ 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)
|
||||
|
|
@ -207,6 +233,11 @@ 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')
|
||||
|
|
@ -214,7 +245,7 @@ export class DaemonClient {
|
|||
return
|
||||
}
|
||||
|
||||
socket.removeListener('data', onData)
|
||||
cleanup()
|
||||
const line = buffer.slice(0, newlineIdx)
|
||||
try {
|
||||
const response = JSON.parse(line) as HelloResponse
|
||||
|
|
@ -229,8 +260,18 @@ 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))
|
||||
})
|
||||
}
|
||||
|
|
@ -282,13 +323,18 @@ export class DaemonClient {
|
|||
this.streamSocket.on('data', (chunk) => parser.feed(chunk.toString()))
|
||||
}
|
||||
|
||||
private handleDisconnect(generation: number): void {
|
||||
private handleDisconnect(generation: number, reason: string): 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'))
|
||||
|
|
|
|||
|
|
@ -114,6 +114,22 @@ 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', () => {
|
||||
|
|
|
|||
|
|
@ -31,6 +31,12 @@ 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) {
|
||||
|
|
@ -53,6 +59,10 @@ 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
|
||||
|
|
@ -60,6 +70,7 @@ 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.
|
||||
|
|
@ -84,6 +95,11 @@ 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 {
|
||||
|
|
@ -151,6 +167,7 @@ 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
|
||||
|
|
@ -230,12 +247,13 @@ export class DaemonPtyAdapter implements IPtyProvider {
|
|||
|
||||
write(id: string, data: string): void {
|
||||
this.markSessionDirty(id)
|
||||
this.client.notify('write', { sessionId: id, data })
|
||||
this.sendNotification('write', { sessionId: id, data })
|
||||
}
|
||||
|
||||
resize(id: string, cols: number, rows: number): void {
|
||||
this.markSessionDirty(id)
|
||||
this.client.notify('resize', { sessionId: id, cols, rows })
|
||||
this.sessionSizes.set(id, { cols, rows })
|
||||
this.sendNotification('resize', { sessionId: id, cols, rows })
|
||||
}
|
||||
|
||||
async shutdown(id: string, opts: { immediate?: boolean; keepHistory?: boolean }): Promise<void> {
|
||||
|
|
@ -243,6 +261,7 @@ 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
|
||||
|
|
@ -359,6 +378,7 @@ 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)
|
||||
}
|
||||
}
|
||||
|
|
@ -403,6 +423,7 @@ 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
|
||||
|
|
@ -458,11 +479,13 @@ 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()
|
||||
|
|
@ -474,6 +497,9 @@ 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.
|
||||
|
|
@ -483,6 +509,7 @@ 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
|
||||
|
|
@ -503,6 +530,9 @@ 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> {
|
||||
|
|
@ -529,6 +559,72 @@ 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
|
||||
|
|
@ -646,6 +742,7 @@ 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)
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import { connect } from 'net'
|
||||
import { connect, type Socket } from 'net'
|
||||
import { tmpdir } from 'os'
|
||||
import { join } from 'path'
|
||||
import { mkdtempSync, rmSync, readFileSync } from 'fs'
|
||||
|
|
@ -66,6 +66,39 @@ 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()
|
||||
|
|
@ -122,6 +155,22 @@ 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()
|
||||
|
|
|
|||
|
|
@ -41,7 +41,12 @@ export class DaemonServer {
|
|||
private tokenPath: string
|
||||
|
||||
private clients = new Map<string, ConnectedClient>()
|
||||
private streamDataBatcher = new DaemonStreamDataBatcher((clientId) => this.clients.get(clientId))
|
||||
private streamDataBatcher = new DaemonStreamDataBatcher(
|
||||
(clientId) => this.clients.get(clientId),
|
||||
{
|
||||
onStreamFailure: (clientId) => this.disconnectClient(clientId)
|
||||
}
|
||||
)
|
||||
|
||||
constructor(opts: DaemonServerOptions) {
|
||||
this.socketPath = opts.socketPath
|
||||
|
|
@ -134,6 +139,7 @@ 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,
|
||||
|
|
@ -144,6 +150,8 @@ 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)
|
||||
|
|
@ -161,8 +169,10 @@ export class DaemonServer {
|
|||
socket.on('data', (chunk) => parser.feed(chunk.toString()))
|
||||
|
||||
socket.on('close', () => {
|
||||
this.streamDataBatcher.clear(clientId)
|
||||
this.clients.delete(clientId)
|
||||
const client = this.clients.get(clientId)
|
||||
if (client?.controlSocket === socket) {
|
||||
this.disconnectClient(clientId, client)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
|
|
@ -212,19 +222,10 @@ export class DaemonServer {
|
|||
this.streamDataBatcher.enqueue(clientId, p.sessionId, data)
|
||||
},
|
||||
onExit: (code) => {
|
||||
// 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 }
|
||||
})
|
||||
)
|
||||
}
|
||||
// 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)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
|
@ -304,19 +305,23 @@ export class DaemonServer {
|
|||
sessionId: string,
|
||||
code: number
|
||||
): void {
|
||||
if (!client?.streamSocket) {
|
||||
if (!client) {
|
||||
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.
|
||||
client.streamSocket.write(
|
||||
encodeNdjson({
|
||||
type: 'event',
|
||||
event: 'exit',
|
||||
sessionId,
|
||||
payload: { code }
|
||||
})
|
||||
)
|
||||
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()
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,309 @@
|
|||
import { describe, expect, it, vi } from 'vitest'
|
||||
import type { Socket } from 'net'
|
||||
import { DaemonStreamDataBatcher } from './daemon-stream-data-batcher'
|
||||
|
||||
function createFakeSocket(writeResults: boolean[]): {
|
||||
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?.()
|
||||
}
|
||||
}
|
||||
|
||||
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()
|
||||
}
|
||||
})
|
||||
})
|
||||
|
|
@ -5,45 +5,115 @@ type StreamDataClient = {
|
|||
streamSocket: Socket | null
|
||||
}
|
||||
|
||||
type PendingStreamEvent =
|
||||
| { kind: 'data'; sessionId: string; data: string }
|
||||
| { kind: 'exit'; sessionId: string; code: number }
|
||||
|
||||
type PendingStreamDataBatch = {
|
||||
timer: ReturnType<typeof setTimeout> | null
|
||||
queue: { sessionId: string; data: string }[]
|
||||
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.
|
||||
const STREAM_DATA_BATCH_INTERVAL_MS = 8
|
||||
const STREAM_DATA_BACKPRESSURE_WARN_BYTES = 512 * 1024
|
||||
const STREAM_DATA_DRAIN_TIMEOUT_MS = 30_000
|
||||
const STREAM_DATA_MAX_QUEUED_BYTES = 8 * 1024 * 1024
|
||||
const STREAM_DATA_MAX_PAYLOAD_CHARS = 64 * 1024
|
||||
const STREAM_DATA_MAX_EVENTS_PER_FLUSH = 1024
|
||||
|
||||
export class DaemonStreamDataBatcher {
|
||||
private pendingByClient = new Map<string, PendingStreamDataBatch>()
|
||||
private getClient: (clientId: string) => StreamDataClient | undefined
|
||||
private onStreamFailure: (clientId: string) => void
|
||||
|
||||
constructor(getClient: (clientId: string) => StreamDataClient | undefined) {
|
||||
constructor(
|
||||
getClient: (clientId: string) => StreamDataClient | undefined,
|
||||
opts: { onStreamFailure?: (clientId: string) => void } = {}
|
||||
) {
|
||||
this.getClient = getClient
|
||||
this.onStreamFailure = opts.onStreamFailure ?? (() => {})
|
||||
}
|
||||
|
||||
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.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
|
||||
return false
|
||||
}
|
||||
|
||||
let batch = this.pendingByClient.get(clientId)
|
||||
if (!batch) {
|
||||
batch = { timer: null, queue: [] }
|
||||
batch = {
|
||||
timer: null,
|
||||
drainTimer: null,
|
||||
cleanupWait: null,
|
||||
queue: [],
|
||||
queueHead: 0,
|
||||
queuedDataBytes: 0,
|
||||
waitingForDrain: false,
|
||||
warnedBackpressure: false
|
||||
}
|
||||
this.pendingByClient.set(clientId, batch)
|
||||
}
|
||||
|
||||
const last = batch.queue.at(-1)
|
||||
if (last?.sessionId === sessionId) {
|
||||
last.data += data
|
||||
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({ sessionId, data })
|
||||
batch.queue.push(event)
|
||||
if (event.kind === 'data') {
|
||||
batch.queuedDataBytes += Buffer.byteLength(event.data, 'utf8')
|
||||
}
|
||||
}
|
||||
|
||||
if (!batch.timer) {
|
||||
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
|
||||
}
|
||||
|
||||
flush(clientId: string): void {
|
||||
|
|
@ -56,23 +126,117 @@ export class DaemonStreamDataBatcher {
|
|||
clearTimeout(batch.timer)
|
||||
batch.timer = null
|
||||
}
|
||||
this.pendingByClient.delete(clientId)
|
||||
|
||||
const client = this.getClient(clientId)
|
||||
if (!client?.streamSocket || client.streamSocket.destroyed) {
|
||||
if (batch.waitingForDrain) {
|
||||
return
|
||||
}
|
||||
|
||||
for (const entry of batch.queue) {
|
||||
client.streamSocket.write(
|
||||
encodeNdjson({
|
||||
type: 'event',
|
||||
event: 'data',
|
||||
sessionId: entry.sessionId,
|
||||
payload: { data: entry.data }
|
||||
})
|
||||
)
|
||||
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 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 }
|
||||
}
|
||||
const ok = streamSocket.write(encodeNdjson(payload))
|
||||
if (!ok) {
|
||||
batch.waitingForDrain = true
|
||||
this.compactQueue(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)
|
||||
return
|
||||
}
|
||||
if (
|
||||
flushedEvents >= STREAM_DATA_MAX_EVENTS_PER_FLUSH &&
|
||||
batch.queueHead < batch.queue.length
|
||||
) {
|
||||
this.compactQueue(batch)
|
||||
batch.timer = setTimeout(() => this.flush(clientId), 0)
|
||||
batch.timer.unref?.()
|
||||
return
|
||||
}
|
||||
}
|
||||
this.pendingByClient.delete(clientId)
|
||||
}
|
||||
|
||||
clear(clientId?: string): void {
|
||||
|
|
@ -82,10 +246,27 @@ 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)
|
||||
}
|
||||
}
|
||||
|
||||
private compactQueue(batch: PendingStreamDataBatch): void {
|
||||
if (batch.queueHead === 0) {
|
||||
return
|
||||
}
|
||||
batch.queue = batch.queue.slice(batch.queueHead)
|
||||
batch.queueHead = 0
|
||||
}
|
||||
|
||||
private failStream(clientId: string): void {
|
||||
this.clear(clientId)
|
||||
this.onStreamFailure(clientId)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2434,6 +2434,41 @@ 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()
|
||||
|
|
|
|||
|
|
@ -646,6 +646,12 @@ 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
|
||||
|
|
@ -653,10 +659,30 @@ export function registerPtyHandlers(
|
|||
pendingData.clear()
|
||||
return
|
||||
}
|
||||
for (const [id, data] of pendingData) {
|
||||
mainWindow.webContents.send('pty:data', { id, data })
|
||||
|
||||
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)
|
||||
}
|
||||
pendingData.clear()
|
||||
}
|
||||
|
||||
const clearFlushTimerIfIdle = (): void => {
|
||||
|
|
@ -710,10 +736,7 @@ export function registerPtyHandlers(
|
|||
clearFlushTimerIfIdle()
|
||||
// Why: agent TUIs redraw small prompt regions after every keystroke.
|
||||
// Waiting for the throughput batch timer adds visible input latency.
|
||||
mainWindow.webContents.send('pty:data', {
|
||||
id: payload.id,
|
||||
data: nextData
|
||||
})
|
||||
sendPtyData(payload.id, nextData)
|
||||
return
|
||||
}
|
||||
pendingData.set(payload.id, nextData)
|
||||
|
|
@ -734,7 +757,7 @@ export function registerPtyHandlers(
|
|||
// tears down the terminal on pty:exit before the batch timer fires.
|
||||
const remaining = pendingData.get(payload.id)
|
||||
if (remaining) {
|
||||
mainWindow.webContents.send('pty:data', { id: payload.id, data: remaining })
|
||||
sendPtyData(payload.id, remaining)
|
||||
pendingData.delete(payload.id)
|
||||
}
|
||||
lastInputAtByPty.delete(payload.id)
|
||||
|
|
|
|||
|
|
@ -990,7 +990,51 @@ describe('connectPanePty', () => {
|
|||
}
|
||||
})
|
||||
|
||||
it('writes visible split-pane PTY bytes immediately even when the tab is not active', async () => {
|
||||
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 () => {
|
||||
const { connectPanePty } = await import('./pty-connection')
|
||||
const transport = createMockTransport()
|
||||
const capturedDataCallback: { current: ((data: string) => void) | null } = { current: null }
|
||||
|
|
@ -1001,9 +1045,9 @@ describe('connectPanePty', () => {
|
|||
transportFactoryQueue.push(transport)
|
||||
|
||||
const pane = createPane(1)
|
||||
const manager = createManager(1)
|
||||
const manager = createManager(2)
|
||||
manager.getActivePane.mockReturnValue({ id: 1 })
|
||||
const deps = createDeps({
|
||||
isActiveRef: { current: false },
|
||||
isVisibleRef: { current: true }
|
||||
})
|
||||
|
||||
|
|
@ -1011,9 +1055,9 @@ describe('connectPanePty', () => {
|
|||
await flushAsyncTicks(6)
|
||||
|
||||
expect(capturedDataCallback.current).not.toBeNull()
|
||||
capturedDataCallback.current?.('visible split output\r\n')
|
||||
capturedDataCallback.current?.('active split output\r\n')
|
||||
|
||||
expect(pane.terminal.write).toHaveBeenCalledWith('visible split output\r\n')
|
||||
expect(pane.terminal.write).toHaveBeenCalledWith('active split output\r\n')
|
||||
})
|
||||
|
||||
it('marks panes that receive Arabic output for DOM rendering', async () => {
|
||||
|
|
|
|||
|
|
@ -877,11 +877,12 @@ export function connectPanePty(
|
|||
if (terminalOutputPrefersDomRenderer(data)) {
|
||||
manager.markPaneHasComplexScriptOutput(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.
|
||||
// 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
|
||||
writeTerminalOutput(pane.terminal, data, {
|
||||
foreground: deps.isVisibleRef.current
|
||||
foreground: deps.isVisibleRef.current && activePaneId === pane.id
|
||||
})
|
||||
|
||||
if (pendingStartupCommand) {
|
||||
|
|
|
|||
|
|
@ -192,6 +192,46 @@ 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()
|
||||
|
|
|
|||
|
|
@ -123,6 +123,27 @@ 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
|
||||
|
|
@ -284,6 +305,23 @@ 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 }
|
||||
|
|
|
|||
|
|
@ -15,7 +15,13 @@ import {
|
|||
waitForActiveWorktree,
|
||||
waitForSessionReady
|
||||
} from './helpers/store'
|
||||
import { getTerminalContent, waitForActiveTerminalManager } from './helpers/terminal'
|
||||
import {
|
||||
getTerminalContent,
|
||||
splitActiveTerminalPane,
|
||||
waitForActiveTerminalManager,
|
||||
waitForPaneCount,
|
||||
waitForPaneIdentitySnapshot
|
||||
} from './helpers/terminal'
|
||||
|
||||
type SchedulerDebugSnapshot = {
|
||||
backgroundEnqueueCount: number
|
||||
|
|
@ -249,4 +255,75 @@ 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)
|
||||
})
|
||||
})
|
||||
|
|
|
|||
Loading…
Reference in New Issue