mirror of
https://github.com/stablyai/orca.git
synced 2026-10-18 16:03:17 +00:00
fix(terminal): make replay progress generation-based
This commit is contained in:
@@ -3,6 +3,7 @@ import { writeForegroundTerminalChunk } from '@/lib/pane-manager/pane-terminal-f
|
||||
import { recordRendererCrashBreadcrumb } from '@/lib/crash-breadcrumb-recorder'
|
||||
import { ensureArabicShapingJoinerForText } from '@/lib/pane-manager/terminal-arabic-shaping-joiner'
|
||||
import {
|
||||
captureTerminalParseProgressGeneration,
|
||||
hasTerminalParseProgressSince,
|
||||
isTerminalWritePipelineCertifiedDead,
|
||||
notifyUndeliverableWrite,
|
||||
@@ -119,7 +120,7 @@ function engageReplayGuard(
|
||||
}
|
||||
onRelease?.()
|
||||
}
|
||||
const armWedgeDeadline = (quietSinceMs: number): void => {
|
||||
const armWedgeDeadline = (quietSinceGeneration: number): void => {
|
||||
timer = setTimeout(() => {
|
||||
if (released) {
|
||||
return
|
||||
@@ -129,8 +130,8 @@ function engageReplayGuard(
|
||||
// wedge verdict here would open the guard while replay bytes are still
|
||||
// parsing — leaking auto-replies into the agent's stdin — and hand a
|
||||
// healthy pane to recovery. Certify only after a fully quiet window.
|
||||
if (hasTerminalParseProgressSince(terminal, quietSinceMs)) {
|
||||
armWedgeDeadline(Date.now())
|
||||
if (hasTerminalParseProgressSince(terminal, quietSinceGeneration)) {
|
||||
armWedgeDeadline(captureTerminalParseProgressGeneration(terminal))
|
||||
return
|
||||
}
|
||||
release('wedged')
|
||||
@@ -140,7 +141,7 @@ function engageReplayGuard(
|
||||
if (released) {
|
||||
return
|
||||
}
|
||||
const probeQueuedAt = Date.now()
|
||||
const probeQueuedAtGeneration = captureTerminalParseProgressGeneration(terminal)
|
||||
try {
|
||||
// FIFO certification: this callback can only run after every replay
|
||||
// byte queued before it has parsed (state 2 above).
|
||||
@@ -154,7 +155,7 @@ function engageReplayGuard(
|
||||
release('wedged')
|
||||
return
|
||||
}
|
||||
armWedgeDeadline(probeQueuedAt)
|
||||
armWedgeDeadline(probeQueuedAtGeneration)
|
||||
}
|
||||
timer = setTimeout(probeForStall, stallCheckMs)
|
||||
return () => {
|
||||
|
||||
@@ -179,9 +179,12 @@ describe('pane terminal output scheduler', () => {
|
||||
it('credits when queued output is discarded', async () => {
|
||||
vi.useFakeTimers()
|
||||
const { writeTerminalOutput, discardTerminalOutput } = await loadScheduler()
|
||||
const { captureTerminalParseProgressGeneration, hasTerminalParseProgressSince } =
|
||||
await import('./terminal-write-pipeline-health')
|
||||
const terminal = createTerminal()
|
||||
terminal.write.mockImplementation(() => {})
|
||||
const credit = makeCredit()
|
||||
const parseGeneration = captureTerminalParseProgressGeneration(terminal)
|
||||
|
||||
writeTerminalOutput(terminal, 'doomed', {
|
||||
foreground: true,
|
||||
@@ -191,6 +194,9 @@ describe('pane terminal output scheduler', () => {
|
||||
expect(credit.count()).toBe(0)
|
||||
discardTerminalOutput(terminal)
|
||||
expect(credit.count()).toBe(1)
|
||||
// Discard settles delivery ownership, not xterm parsing; replay wedge
|
||||
// deadlines must not treat cleanup as evidence that the parser is alive.
|
||||
expect(hasTerminalParseProgressSince(terminal, parseGeneration)).toBe(false)
|
||||
})
|
||||
|
||||
it('credits an empty write immediately', async () => {
|
||||
|
||||
@@ -19,6 +19,7 @@ import {
|
||||
} from './pane-terminal-output-ack-credit'
|
||||
import {
|
||||
armTerminalWriteStallWatch,
|
||||
cancelTerminalWriteStallWatch,
|
||||
settleTerminalWriteStallWatch
|
||||
} from './terminal-write-pipeline-health'
|
||||
import {
|
||||
@@ -1440,9 +1441,9 @@ export function discardTerminalOutput(terminal: TerminalOutputTarget): void {
|
||||
discardInFlightTerminalOutputAckCredits(terminal)
|
||||
queuedByTerminal.delete(terminal)
|
||||
discardForegroundRenderSettle(terminal)
|
||||
// Why: a legitimately disposed pane must not leave a pending stall watch to
|
||||
// probe the dead terminal later and report a false wedge.
|
||||
settleTerminalWriteStallWatch(terminal)
|
||||
// Why: cleanup must cancel the watch without masquerading as parse progress;
|
||||
// replay guards use real completions to distinguish slow from wedged.
|
||||
cancelTerminalWriteStallWatch(terminal)
|
||||
recordQueueDebugPressure()
|
||||
}
|
||||
|
||||
|
||||
@@ -2,6 +2,9 @@ import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import {
|
||||
_resetWritePipelineHealthForTests,
|
||||
armTerminalWriteStallWatch,
|
||||
cancelTerminalWriteStallWatch,
|
||||
captureTerminalParseProgressGeneration,
|
||||
hasTerminalParseProgressSince,
|
||||
isTerminalWritePipelineCertifiedDead,
|
||||
notifyUndeliverableWrite,
|
||||
registerUndeliverableWriteHandler,
|
||||
@@ -43,6 +46,36 @@ afterEach(() => {
|
||||
})
|
||||
|
||||
describe('terminal write pipeline health', () => {
|
||||
it('detects parse progress without relying on wall-clock resolution', () => {
|
||||
vi.useFakeTimers()
|
||||
vi.setSystemTime(1_000)
|
||||
const terminal = makeTerminal()
|
||||
const generation = captureTerminalParseProgressGeneration(terminal)
|
||||
|
||||
settleTerminalWriteStallWatch(terminal)
|
||||
|
||||
expect(Date.now()).toBe(1_000)
|
||||
expect(hasTerminalParseProgressSince(terminal, generation)).toBe(true)
|
||||
_resetWritePipelineHealthForTests(terminal)
|
||||
})
|
||||
|
||||
it('cancels a discarded write watch without recording parse progress', () => {
|
||||
vi.useFakeTimers()
|
||||
const terminal = makeTerminal()
|
||||
const handler = vi.fn()
|
||||
const generation = captureTerminalParseProgressGeneration(terminal)
|
||||
registerUndeliverableWriteHandler(terminal, handler)
|
||||
armTerminalWriteStallWatch(terminal)
|
||||
|
||||
cancelTerminalWriteStallWatch(terminal)
|
||||
vi.advanceTimersByTime(WRITE_PIPELINE_STALL_CHECK_MS * 3)
|
||||
|
||||
expect(hasTerminalParseProgressSince(terminal, generation)).toBe(false)
|
||||
expect(terminal.pendingCallbacks).toHaveLength(0)
|
||||
expect(handler).not.toHaveBeenCalled()
|
||||
_resetWritePipelineHealthForTests(terminal)
|
||||
})
|
||||
|
||||
it('a settled write never probes or notifies', () => {
|
||||
vi.useFakeTimers()
|
||||
const terminal = makeTerminal()
|
||||
|
||||
@@ -21,20 +21,25 @@ type UndeliverableWriteHandler = (reason: UndeliverableWriteReason) => void
|
||||
|
||||
const handlersByTerminal = new WeakMap<object, UndeliverableWriteHandler>()
|
||||
const certifiedDeadTerminals = new WeakSet<object>()
|
||||
// Why: wedge verdicts must distinguish "dead" from "alive but behind". Any
|
||||
// write completion on a terminal — scheduler settles and replay releases —
|
||||
// reports here, and probe deadlines consult it before certifying.
|
||||
const lastParseProgressAtByTerminal = new WeakMap<object, number>()
|
||||
// Why: wedge verdicts must distinguish "dead" from "alive but behind". A
|
||||
// generation avoids same-millisecond misses and wall-clock adjustments while
|
||||
// keeping the completion hot path constant-time and terminal-scoped.
|
||||
const parseProgressGenerationByTerminal = new WeakMap<object, number>()
|
||||
|
||||
/** Report one parsed write completion for this terminal. */
|
||||
export function recordTerminalParseProgress(terminal: object): void {
|
||||
lastParseProgressAtByTerminal.set(terminal, Date.now())
|
||||
const nextGeneration = (parseProgressGenerationByTerminal.get(terminal) ?? 0) + 1
|
||||
parseProgressGenerationByTerminal.set(terminal, nextGeneration)
|
||||
}
|
||||
|
||||
/** Whether any write completion parsed on this terminal after `sinceMs`. */
|
||||
export function hasTerminalParseProgressSince(terminal: object, sinceMs: number): boolean {
|
||||
const at = lastParseProgressAtByTerminal.get(terminal)
|
||||
return at !== undefined && at > sinceMs
|
||||
/** Capture the current parse-progress generation for a later quiet-window check. */
|
||||
export function captureTerminalParseProgressGeneration(terminal: object): number {
|
||||
return parseProgressGenerationByTerminal.get(terminal) ?? 0
|
||||
}
|
||||
|
||||
/** Whether a write completion parsed after `generation` was captured. */
|
||||
export function hasTerminalParseProgressSince(terminal: object, generation: number): boolean {
|
||||
return captureTerminalParseProgressGeneration(terminal) !== generation
|
||||
}
|
||||
|
||||
type StallWatch = {
|
||||
@@ -144,9 +149,8 @@ export function armTerminalWriteStallWatch(
|
||||
stallWatchByTerminal.set(terminal, watch)
|
||||
}
|
||||
|
||||
/** Write completed normally — the pipeline is healthy; drop any pending watch. */
|
||||
export function settleTerminalWriteStallWatch(terminal: object): void {
|
||||
recordTerminalParseProgress(terminal)
|
||||
/** Cancel a pending watch without claiming that any bytes parsed. */
|
||||
export function cancelTerminalWriteStallWatch(terminal: object): void {
|
||||
const watch = stallWatchByTerminal.get(terminal)
|
||||
if (!watch) {
|
||||
return
|
||||
@@ -155,6 +159,12 @@ export function settleTerminalWriteStallWatch(terminal: object): void {
|
||||
stallWatchByTerminal.delete(terminal)
|
||||
}
|
||||
|
||||
/** Write completed normally — the pipeline is healthy; drop any pending watch. */
|
||||
export function settleTerminalWriteStallWatch(terminal: object): void {
|
||||
recordTerminalParseProgress(terminal)
|
||||
cancelTerminalWriteStallWatch(terminal)
|
||||
}
|
||||
|
||||
export function _resetWritePipelineHealthForTests(terminal?: object): void {
|
||||
if (terminal) {
|
||||
const watch = stallWatchByTerminal.get(terminal)
|
||||
@@ -164,6 +174,6 @@ export function _resetWritePipelineHealthForTests(terminal?: object): void {
|
||||
stallWatchByTerminal.delete(terminal)
|
||||
handlersByTerminal.delete(terminal)
|
||||
certifiedDeadTerminals.delete(terminal)
|
||||
lastParseProgressAtByTerminal.delete(terminal)
|
||||
parseProgressGenerationByTerminal.delete(terminal)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user