mirror of
https://github.com/stablyai/orca.git
synced 2026-10-26 16:02:42 +00:00
fix(terminal): arm delivery watchdog only with PTYs
Co-authored-by: Orca <help@stably.ai>
This commit is contained in:
@@ -1,5 +1,6 @@
|
||||
import { acquirePtyDeliveryInterest } from './pty-delivery-interest'
|
||||
import { ensurePtyDispatcher, ptyDataSidecars } from './pty-dispatcher'
|
||||
import { syncTerminalDeliveryWatchdogTimer } from './terminal-delivery-watchdog'
|
||||
|
||||
/** Register a side-channel data watcher for a PTY without taking ownership
|
||||
* of the primary handler. Returns an unsubscribe fn. */
|
||||
@@ -15,6 +16,8 @@ export function subscribeToPtyData(ptyId: string, watcher: (data: string) => voi
|
||||
ptyDataSidecars.set(ptyId, set)
|
||||
}
|
||||
set.add(watcher)
|
||||
// Why: a sidecar receives pushed bytes with no primary handler, so it must arm the delivery watchdog.
|
||||
syncTerminalDeliveryWatchdogTimer()
|
||||
return () => {
|
||||
releaseDeliveryInterest()
|
||||
const current = ptyDataSidecars.get(ptyId)
|
||||
@@ -25,5 +28,6 @@ export function subscribeToPtyData(ptyId: string, watcher: (data: string) => voi
|
||||
if (current.size === 0) {
|
||||
ptyDataSidecars.delete(ptyId)
|
||||
}
|
||||
syncTerminalDeliveryWatchdogTimer()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -89,6 +89,31 @@ describe('pty dispatcher delivery interest', () => {
|
||||
expect(setPtyDeliveryInterest).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('arms the delivery watchdog for a sidecar-only PTY and disarms on the last unsubscribe', async () => {
|
||||
// Why: a parked terminal (and draft-readiness / automation observers) receives pushed bytes
|
||||
// through a sidecar with no primary handler — the watchdog must still cover that dead-channel case.
|
||||
vi.useFakeTimers()
|
||||
const reportMock = vi.fn(() =>
|
||||
Promise.resolve({ inFlightTotalChars: 0, inFlightPtyCount: 0, msSinceLastAck: null })
|
||||
)
|
||||
;(window.api.pty as { reportRendererDeliveryState?: unknown }).reportRendererDeliveryState =
|
||||
reportMock
|
||||
try {
|
||||
const { subscribeToPtyData } = await import('./pty-data-sidecar-subscriptions')
|
||||
|
||||
const unsubscribe = subscribeToPtyData('pty-1', vi.fn())
|
||||
await vi.advanceTimersByTimeAsync(15_000)
|
||||
expect(reportMock).toHaveBeenCalled()
|
||||
|
||||
reportMock.mockClear()
|
||||
unsubscribe()
|
||||
await vi.advanceTimersByTimeAsync(15_000 * 2)
|
||||
expect(reportMock).not.toHaveBeenCalled()
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('lets a sidecar exclusively own interest while an eager buffer overlaps', async () => {
|
||||
const { registerEagerPtyBuffer } = await import('./pty-dispatcher')
|
||||
const { subscribeToPtyData } = await import('./pty-data-sidecar-subscriptions')
|
||||
|
||||
@@ -64,6 +64,30 @@ describe('pty dispatcher push-listener reattach and delivery blackhole', () => {
|
||||
}
|
||||
})
|
||||
|
||||
it('arms for eager PTYs and disarms only after the last eager PTY is disposed', async () => {
|
||||
vi.useFakeTimers()
|
||||
try {
|
||||
const { ensurePtyDispatcher, registerEagerPtyBuffer } = await import('./pty-dispatcher')
|
||||
ensurePtyDispatcher()
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
|
||||
const first = registerEagerPtyBuffer('pty-eager-1', vi.fn())
|
||||
expect(vi.getTimerCount()).toBe(1)
|
||||
const second = registerEagerPtyBuffer('pty-eager-2', vi.fn())
|
||||
expect(vi.getTimerCount()).toBe(1)
|
||||
|
||||
await vi.advanceTimersByTimeAsync(30_000)
|
||||
expect(reportMock).toHaveBeenCalledTimes(2)
|
||||
|
||||
first.dispose()
|
||||
expect(vi.getTimerCount()).toBe(1)
|
||||
second.dispose()
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('blackholed delivery reproduces the wedge: no handler dispatch, no ACK ever', async () => {
|
||||
const { ensurePtyDispatcher, ptyDataHandlers } = await import('./pty-dispatcher')
|
||||
ensurePtyDispatcher()
|
||||
|
||||
@@ -18,7 +18,8 @@ import {
|
||||
clearReceivedPtyCharTotal,
|
||||
isPtyPushDeliveryBlackholed,
|
||||
recordPtyDataReceived,
|
||||
startTerminalDeliveryWatchdog
|
||||
startTerminalDeliveryWatchdog,
|
||||
syncTerminalDeliveryWatchdogTimer
|
||||
} from './terminal-delivery-watchdog'
|
||||
import { recordTerminalFreezeBreadcrumb } from './terminal-freeze-breadcrumbs'
|
||||
import { installTerminalFreezeReport } from './terminal-freeze-report'
|
||||
@@ -76,6 +77,7 @@ export function unregisterPtyDataHandlers(ptyIds: string[]): PtyDataHandlerShutd
|
||||
ptyTeardownHandlers.delete(id)
|
||||
clearPreHandlerPtyState(id)
|
||||
}
|
||||
syncTerminalDeliveryWatchdogTimer()
|
||||
return snapshots
|
||||
}
|
||||
|
||||
@@ -93,6 +95,7 @@ export function restorePtyDataHandlersAfterFailedShutdown(
|
||||
ptyTeardownHandlers.set(snapshot.ptyId, snapshot.teardownHandler)
|
||||
}
|
||||
}
|
||||
syncTerminalDeliveryWatchdogTimer()
|
||||
}
|
||||
|
||||
let pushListenerUnsubscribes: (() => void)[] = []
|
||||
@@ -120,7 +123,10 @@ export function ensurePtyDispatcher(): void {
|
||||
attachPtyPushListeners()
|
||||
startTerminalDeliveryWatchdog({
|
||||
reattachPushListeners: reattachPtyDispatcherPushListeners,
|
||||
hasAttachedPtys: () => ptyDataHandlers.size > 0 || eagerPtyHandles.size > 0
|
||||
// Why: sidecar consumers (parked terminals, draft-readiness, automation observers) hold main-side
|
||||
// delivery interest and receive pushed bytes with no primary handler, so they need the watchdog too.
|
||||
hasAttachedPtys: () =>
|
||||
ptyDataHandlers.size > 0 || eagerPtyHandles.size > 0 || ptyDataSidecars.size > 0
|
||||
})
|
||||
}
|
||||
|
||||
@@ -296,13 +302,7 @@ export function registerEagerPtyBuffer(
|
||||
}
|
||||
}
|
||||
const exitHandler = (code: number): void => {
|
||||
// Shell died before attach; identity-guard so we never evict a handler a transport re-registered for this id (#7894 detach/attach race).
|
||||
if (ptyDataHandlers.get(ptyId) === dataHandler) {
|
||||
ptyDataHandlers.delete(ptyId)
|
||||
ptyReplayHandlers.delete(ptyId)
|
||||
}
|
||||
ptyExitHandlers.delete(ptyId)
|
||||
eagerPtyHandles.delete(ptyId)
|
||||
handle.dispose()
|
||||
onExit(ptyId, code)
|
||||
}
|
||||
|
||||
@@ -330,10 +330,12 @@ export function registerEagerPtyBuffer(
|
||||
ptyExitHandlers.delete(ptyId)
|
||||
}
|
||||
eagerPtyHandles.delete(ptyId)
|
||||
syncTerminalDeliveryWatchdogTimer()
|
||||
}
|
||||
}
|
||||
|
||||
eagerPtyHandles.set(ptyId, handle)
|
||||
syncTerminalDeliveryWatchdogTimer()
|
||||
drainPreHandlerPtyData(ptyId, dataHandler)
|
||||
// Why: defer the pre-handler exit one microtask so the caller receives the returned handle before onExit fires.
|
||||
queueMicrotask(() => {
|
||||
|
||||
@@ -79,6 +79,41 @@ describe('createIpcPtyTransport', () => {
|
||||
transport.disconnect()
|
||||
})
|
||||
|
||||
it('arms for mounted PTYs and disarms only after the last transport detaches', async () => {
|
||||
vi.useFakeTimers()
|
||||
const reportRendererDeliveryState = vi.fn().mockResolvedValue({
|
||||
inFlightTotalChars: 0,
|
||||
inFlightPtyCount: 0,
|
||||
msSinceLastAck: null
|
||||
})
|
||||
window.api.pty.reportRendererDeliveryState = reportRendererDeliveryState
|
||||
try {
|
||||
const { createIpcPtyTransport, ensurePtyDispatcher } = await import('./pty-transport')
|
||||
const spawn = window.api.pty.spawn as unknown as ReturnType<typeof vi.fn>
|
||||
spawn.mockResolvedValueOnce({ id: 'pty-1' }).mockResolvedValueOnce({ id: 'pty-2' })
|
||||
|
||||
ensurePtyDispatcher()
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
|
||||
const first = createIpcPtyTransport({})
|
||||
await first.connect({ url: '', callbacks: {} })
|
||||
expect(vi.getTimerCount()).toBe(1)
|
||||
|
||||
const second = createIpcPtyTransport({})
|
||||
await second.connect({ url: '', callbacks: {} })
|
||||
expect(vi.getTimerCount()).toBe(1)
|
||||
|
||||
first.detach?.()
|
||||
expect(vi.getTimerCount()).toBe(1)
|
||||
second.detach?.()
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
} finally {
|
||||
const { stopTerminalDeliveryWatchdog } = await import('./terminal-delivery-watchdog')
|
||||
stopTerminalDeliveryWatchdog()
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('does not create a second kill authority when a mounted pane detaches', async () => {
|
||||
const { createIpcPtyTransport } = await import('./pty-transport')
|
||||
const kill = window.api.pty.kill as unknown as ReturnType<typeof vi.fn>
|
||||
|
||||
@@ -11,6 +11,7 @@ import {
|
||||
iterateTerminalInputChunks
|
||||
} from '../../../../shared/terminal-input'
|
||||
import { isRuntimeOwnedSshTargetId } from '../../../../shared/execution-host'
|
||||
import { syncTerminalDeliveryWatchdogTimer } from './terminal-delivery-watchdog'
|
||||
import {
|
||||
ptyDataHandlers,
|
||||
ptyReplayHandlers,
|
||||
@@ -542,6 +543,7 @@ export function createIpcPtyTransport(opts: IpcPtyTransportOptions = {}): PtyTra
|
||||
}
|
||||
}
|
||||
ownedDataAndReplayHandlers.delete(id)
|
||||
syncTerminalDeliveryWatchdogTimer()
|
||||
}
|
||||
|
||||
function registerPtyDataHandler(id: string): void {
|
||||
@@ -572,6 +574,7 @@ export function createIpcPtyTransport(opts: IpcPtyTransportOptions = {}): PtyTra
|
||||
}
|
||||
ptyDataHandlers.set(id, dataHandler)
|
||||
ownedDataAndReplayHandlers.set(id, { data: dataHandler, replay: replayHandler })
|
||||
syncTerminalDeliveryWatchdogTimer()
|
||||
drainPreHandlerPtyData(id, dataHandler)
|
||||
}
|
||||
|
||||
|
||||
@@ -59,24 +59,26 @@ describe('terminal delivery watchdog', () => {
|
||||
}
|
||||
})
|
||||
|
||||
async function startWatchdog(): Promise<{
|
||||
async function startWatchdog(hasAttachedPtys: () => boolean = () => true): Promise<{
|
||||
recordPtyDataReceived: (ptyId: string, chars: number) => void
|
||||
registerRestoreHandler: (
|
||||
ptyId: string,
|
||||
handler: (event: { id: string; reason: string; markerSeq?: number }) => void
|
||||
) => void
|
||||
syncTimer: () => void
|
||||
}> {
|
||||
const watchdog = await import('./terminal-delivery-watchdog')
|
||||
const restoreChannel = await import('./pty-model-restore-channel')
|
||||
watchdog.startTerminalDeliveryWatchdog({
|
||||
reattachPushListeners: reattachMock,
|
||||
hasAttachedPtys: () => true
|
||||
hasAttachedPtys
|
||||
})
|
||||
return {
|
||||
recordPtyDataReceived: watchdog.recordPtyDataReceived,
|
||||
registerRestoreHandler: (ptyId, handler) => {
|
||||
restoreChannel.registerPtyModelRestoreNeededHandler(ptyId, handler)
|
||||
}
|
||||
},
|
||||
syncTimer: watchdog.syncTerminalDeliveryWatchdogTimer
|
||||
}
|
||||
}
|
||||
|
||||
@@ -182,6 +184,67 @@ describe('terminal delivery watchdog', () => {
|
||||
await vi.advanceTimersByTimeAsync(INTERVAL_MS * 3)
|
||||
|
||||
expect(reportMock).not.toHaveBeenCalled()
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
})
|
||||
|
||||
it('arms on the first attached PTY and disarms on the last detach', async () => {
|
||||
reportMock.mockResolvedValue(HEALTHY)
|
||||
let hasAttachedPty = false
|
||||
const { syncTimer } = await startWatchdog(() => hasAttachedPty)
|
||||
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
|
||||
hasAttachedPty = true
|
||||
syncTimer()
|
||||
expect(vi.getTimerCount()).toBe(1)
|
||||
|
||||
await vi.advanceTimersByTimeAsync(INTERVAL_MS * 2)
|
||||
expect(reportMock).toHaveBeenCalledTimes(2)
|
||||
|
||||
hasAttachedPty = false
|
||||
syncTimer()
|
||||
expect(vi.getTimerCount()).toBe(0)
|
||||
|
||||
await vi.advanceTimersByTimeAsync(INTERVAL_MS * 2)
|
||||
expect(reportMock).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
|
||||
it('discards a stalled tick whose invoke was awaited across a last-PTY detach', async () => {
|
||||
let hasAttachedPty = true
|
||||
const { syncTimer } = await startWatchdog(() => hasAttachedPty)
|
||||
|
||||
// One stalled tick brings the streak to one below the two-tick heal threshold.
|
||||
reportMock.mockResolvedValue(STALLED)
|
||||
await vi.advanceTimersByTimeAsync(INTERVAL_MS)
|
||||
expect(reattachMock).not.toHaveBeenCalled()
|
||||
|
||||
// The next tick's invoke is slow; while it is awaited, the last PTY detaches (disarm).
|
||||
let resolveHealth: (health: PtyRendererDeliveryHealthReply) => void = () => {}
|
||||
reportMock.mockReturnValueOnce(
|
||||
new Promise<PtyRendererDeliveryHealthReply>((resolve) => {
|
||||
resolveHealth = resolve
|
||||
})
|
||||
)
|
||||
const tick = vi.advanceTimersByTimeAsync(INTERVAL_MS)
|
||||
await Promise.resolve()
|
||||
hasAttachedPty = false
|
||||
syncTimer()
|
||||
|
||||
// Even though the (now stale) reply is stalled, the tick must be discarded — no second
|
||||
// confirmation, so no heal. Without the generation guard this would resurrect the streak to the
|
||||
// threshold and heal spuriously while parked.
|
||||
resolveHealth(STALLED)
|
||||
await tick
|
||||
|
||||
expect(reattachMock).not.toHaveBeenCalled()
|
||||
|
||||
// The legitimate first-tick streak is preserved across the flicker (not reset), but the racing
|
||||
// tick contributed nothing — so exactly ONE more real stalled confirmation after re-arm heals.
|
||||
hasAttachedPty = true
|
||||
syncTimer()
|
||||
reportMock.mockResolvedValue(STALLED)
|
||||
await vi.advanceTimersByTimeAsync(INTERVAL_MS)
|
||||
expect(reattachMock).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('never starts without the invoke heal lane (web client, partial mocks)', async () => {
|
||||
|
||||
@@ -59,6 +59,10 @@ let stallStreakTicks = 0
|
||||
let lastHealAtMs: number | null = null
|
||||
let healCount = 0
|
||||
let tickInFlight = false
|
||||
// Why: bumped whenever a live timer is torn down (disarm / stop / reconfigure) so an in-flight tick
|
||||
// whose health invoke was awaited across a last-PTY detach can discard its now-stale result instead
|
||||
// of resurrecting the stall streak while parked.
|
||||
let watchdogGeneration = 0
|
||||
|
||||
/** One Map upsert per received chunk — the watchdog's only hot-path cost.
|
||||
* Counted at dispatcher enqueue, BEFORE parse-deferred ACK crediting, so
|
||||
@@ -104,10 +108,17 @@ async function runWatchdogTick(): Promise<void> {
|
||||
stallStreakTicks = 0
|
||||
return
|
||||
}
|
||||
const generationAtInvoke = watchdogGeneration
|
||||
const health = await report({
|
||||
receivedCharsByPty: Object.fromEntries(receivedPtyCharTotals),
|
||||
processedCharsByPty: getProcessedPtyCharTotals()
|
||||
})
|
||||
// Why: a last-PTY detach (or reconfigure) during the await disarmed/re-armed the timer — this
|
||||
// health reply is for the pre-detach state, so discard it WITHOUT touching the streak: mutating it
|
||||
// now would either resurrect a stale streak while parked or reset progress a live reattach kept.
|
||||
if (watchdogGeneration !== generationAtInvoke || !watchdogTimer) {
|
||||
return
|
||||
}
|
||||
if (!health || !isMainDeliveryStalled(health)) {
|
||||
stallStreakTicks = 0
|
||||
return
|
||||
@@ -172,10 +183,16 @@ async function healDeadPushDelivery(
|
||||
})
|
||||
}
|
||||
|
||||
function scheduleWatchdogTimer(): void {
|
||||
function clearWatchdogTimer(): void {
|
||||
if (watchdogTimer) {
|
||||
clearInterval(watchdogTimer)
|
||||
watchdogTimer = null
|
||||
watchdogGeneration += 1
|
||||
}
|
||||
}
|
||||
|
||||
function scheduleWatchdogTimer(): void {
|
||||
clearWatchdogTimer()
|
||||
watchdogTimer = setInterval(() => {
|
||||
// Why serialized: a heal awaits two invokes; overlapping ticks could
|
||||
// double-heal inside one cooldown window.
|
||||
@@ -189,6 +206,22 @@ function scheduleWatchdogTimer(): void {
|
||||
}, watchdogConfig.intervalMs)
|
||||
}
|
||||
|
||||
export function syncTerminalDeliveryWatchdogTimer(): void {
|
||||
const deps = watchdogDeps
|
||||
if (!deps || !deps.hasAttachedPtys()) {
|
||||
// Why: only park the timer — do NOT reset stallStreakTicks. A wedge mid-confirmation must not
|
||||
// lose its progress across a brief zero-attachment detach/reattach flicker (matching the old
|
||||
// always-armed timer); a healthy tick after re-arm clears the streak. No ticks fire while parked,
|
||||
// so a stale streak can only ever heal one confirmation early, never permanently mis-detect.
|
||||
clearWatchdogTimer()
|
||||
return
|
||||
}
|
||||
if (!watchdogTimer) {
|
||||
eventCountAtLastTick = receivedPtyDataEventCount
|
||||
scheduleWatchdogTimer()
|
||||
}
|
||||
}
|
||||
|
||||
export function startTerminalDeliveryWatchdog(deps: TerminalDeliveryWatchdogDeps): void {
|
||||
if (watchdogDeps) {
|
||||
return
|
||||
@@ -200,17 +233,13 @@ export function startTerminalDeliveryWatchdog(deps: TerminalDeliveryWatchdogDeps
|
||||
return
|
||||
}
|
||||
watchdogDeps = deps
|
||||
eventCountAtLastTick = receivedPtyDataEventCount
|
||||
scheduleWatchdogTimer()
|
||||
syncTerminalDeliveryWatchdogTimer()
|
||||
exposeE2eTerminalDeliveryWatchdog()
|
||||
}
|
||||
|
||||
export function stopTerminalDeliveryWatchdog(): void {
|
||||
watchdogDeps = null
|
||||
if (watchdogTimer) {
|
||||
clearInterval(watchdogTimer)
|
||||
watchdogTimer = null
|
||||
}
|
||||
clearWatchdogTimer()
|
||||
}
|
||||
|
||||
/** Prod-reachable state for the one-paste freeze report (ids redacted). */
|
||||
@@ -264,7 +293,7 @@ function exposeE2eTerminalDeliveryWatchdog(): void {
|
||||
},
|
||||
configure: (config) => {
|
||||
watchdogConfig = { ...watchdogConfig, ...config }
|
||||
if (watchdogDeps) {
|
||||
if (watchdogTimer) {
|
||||
scheduleWatchdogTimer()
|
||||
}
|
||||
},
|
||||
|
||||
Reference in New Issue
Block a user