mirror of
https://github.com/stablyai/orca.git
synced 2026-10-27 16:02:42 +00:00
Improve terminal output backpressure
Adds ACK-backed renderer output admission control for PTY data, bounding per-PTY and global renderer in-flight output while preserving PTY ingestion for runtime state and notifications.
This commit is contained in:
@@ -21,7 +21,9 @@ describe('dispatcher → transport → onTitleChange for Pi spinner', () => {
|
||||
// The singleton dispatcher subscribes a SINGLE global `window.api.pty.onData`
|
||||
// callback on first `ensurePtyDispatcher()`. We simulate the main process
|
||||
// delivering IPC events by invoking that captured callback directly.
|
||||
let dispatcherCallback: ((payload: { id: string; data: string }) => void) | null = null
|
||||
let dispatcherCallback:
|
||||
| ((payload: { id: string; data: string; rawLength?: number }) => void)
|
||||
| null = null
|
||||
|
||||
beforeEach(() => {
|
||||
vi.resetModules()
|
||||
@@ -36,17 +38,20 @@ describe('dispatcher → transport → onTitleChange for Pi spinner', () => {
|
||||
write: vi.fn(),
|
||||
resize: vi.fn(),
|
||||
kill: vi.fn(),
|
||||
onData: vi.fn((cb: (payload: { id: string; data: string }) => void) => {
|
||||
// Only the first subscriber wins in production — the dispatcher
|
||||
// calls onData exactly once (ensurePtyDispatcher guards with the
|
||||
// `ptyDispatcherAttached` flag). Subsequent transport calls go
|
||||
// through the same cached subscription, so we capture the first
|
||||
// one and ignore the rest.
|
||||
if (!dispatcherCallback) {
|
||||
dispatcherCallback = cb
|
||||
ackData: vi.fn(),
|
||||
onData: vi.fn(
|
||||
(cb: (payload: { id: string; data: string; rawLength?: number }) => void) => {
|
||||
// Only the first subscriber wins in production — the dispatcher
|
||||
// calls onData exactly once (ensurePtyDispatcher guards with the
|
||||
// `ptyDispatcherAttached` flag). Subsequent transport calls go
|
||||
// through the same cached subscription, so we capture the first
|
||||
// one and ignore the rest.
|
||||
if (!dispatcherCallback) {
|
||||
dispatcherCallback = cb
|
||||
}
|
||||
return () => {}
|
||||
}
|
||||
return () => {}
|
||||
}),
|
||||
),
|
||||
onReplay: vi.fn(() => () => {}),
|
||||
onExit: vi.fn(() => () => {})
|
||||
}
|
||||
@@ -62,6 +67,20 @@ describe('dispatcher → transport → onTitleChange for Pi spinner', () => {
|
||||
}
|
||||
})
|
||||
|
||||
it('ACKs PTY data after dispatcher consumers accept the chunk', async () => {
|
||||
const { ensurePtyDispatcher, ptyDataHandlers } = await import('./pty-dispatcher')
|
||||
const handler = vi.fn()
|
||||
|
||||
ensurePtyDispatcher()
|
||||
ptyDataHandlers.set('pty-pi', handler)
|
||||
|
||||
dispatcherCallback?.({ id: 'pty-pi', data: 'chunk', rawLength: 10 } as never)
|
||||
|
||||
expect(handler).toHaveBeenCalledWith('chunk', { rawLength: 10 })
|
||||
expect(window.api.pty.ackData).toHaveBeenCalledWith('pty-pi', 10)
|
||||
ptyDataHandlers.delete('pty-pi')
|
||||
})
|
||||
|
||||
it('routes Pi OSC title frames from pty:data → onTitleChange via the dispatcher', async () => {
|
||||
const { createIpcPtyTransport } = await import('./pty-transport')
|
||||
const onTitleChange = vi.fn()
|
||||
|
||||
@@ -94,29 +94,36 @@ export function ensurePtyDispatcher(): void {
|
||||
}
|
||||
ptyDispatcherAttached = true
|
||||
window.api.pty.onData((payload) => {
|
||||
let meta: PtyDataMeta | undefined
|
||||
if (typeof payload.seq === 'number') {
|
||||
meta ??= {}
|
||||
meta.seq = payload.seq
|
||||
}
|
||||
if (typeof payload.rawLength === 'number') {
|
||||
meta ??= {}
|
||||
meta.rawLength = payload.rawLength
|
||||
}
|
||||
ptyDataHandlers.get(payload.id)?.(payload.data, meta)
|
||||
const sidecars = ptyDataSidecars.get(payload.id)
|
||||
if (sidecars && sidecars.size > 0) {
|
||||
// Why: snapshot the Set before iterating because watchers commonly
|
||||
// unsubscribe themselves on the very chunk that satisfies them
|
||||
// (e.g. agent-paste-draft resolves on DECSET 2004 and immediately
|
||||
// tears down). Iterating the live Set in that case can skip a
|
||||
// watcher or — if a watcher synchronously subscribes a sibling —
|
||||
// double-fire. The Set is never large (one watcher per active
|
||||
// ready-wait), so the array allocation is cheap.
|
||||
const snapshot = Array.from(sidecars)
|
||||
for (const watcher of snapshot) {
|
||||
watcher(payload.data)
|
||||
try {
|
||||
let meta: PtyDataMeta | undefined
|
||||
if (typeof payload.seq === 'number') {
|
||||
meta ??= {}
|
||||
meta.seq = payload.seq
|
||||
}
|
||||
if (typeof payload.rawLength === 'number') {
|
||||
meta ??= {}
|
||||
meta.rawLength = payload.rawLength
|
||||
}
|
||||
ptyDataHandlers.get(payload.id)?.(payload.data, meta)
|
||||
const sidecars = ptyDataSidecars.get(payload.id)
|
||||
if (sidecars && sidecars.size > 0) {
|
||||
// Why: snapshot the Set before iterating because watchers commonly
|
||||
// unsubscribe themselves on the very chunk that satisfies them
|
||||
// (e.g. agent-paste-draft resolves on DECSET 2004 and immediately
|
||||
// tears down). Iterating the live Set in that case can skip a
|
||||
// watcher or — if a watcher synchronously subscribes a sibling —
|
||||
// double-fire. The Set is never large (one watcher per active
|
||||
// ready-wait), so the array allocation is cheap.
|
||||
const snapshot = Array.from(sidecars)
|
||||
for (const watcher of snapshot) {
|
||||
watcher(payload.data)
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
// Why: main budgets renderer-bound terminal output by bytes accepted
|
||||
// into this dispatcher. ACK in finally so a bad sidecar cannot leave
|
||||
// a PTY permanently backpressured.
|
||||
window.api.pty.ackData?.(payload.id, payload.rawLength ?? payload.data.length)
|
||||
}
|
||||
})
|
||||
window.api.pty.onReplay((payload) => {
|
||||
|
||||
@@ -2153,6 +2153,7 @@ function createPtyApi(): NonNullable<Partial<PreloadApi>['pty']> {
|
||||
signal: () => {},
|
||||
kill: () => Promise.resolve(),
|
||||
ackColdRestore: () => {},
|
||||
ackData: () => {},
|
||||
hasChildProcesses: () => Promise.resolve(false),
|
||||
getForegroundProcess: () => Promise.resolve(null),
|
||||
getCwd: () => Promise.resolve('~'),
|
||||
|
||||
Reference in New Issue
Block a user