mirror of
https://github.com/stablyai/orca.git
synced 2026-10-19 16:03:26 +00:00
fix(codex): give each swept pane its own retry rung
One shared timer drained the whole queue, but each PTY tracks its own rung, so a pane already waiting on its 1500ms rung had the next rung consumed by a newer pane's shorter delay: two panes bound ~300ms apart burned pane A's three attempts by t=3300 instead of t=5800, dropping a Windows daemon reattach that had not settled yet. `arm()` also refused to move a pending timer earlier, so a fresh bind waited out whatever retry delay happened to be armed. Queue a due time per PTY and keep one timer aimed at the earliest of them. A PTY that has never been inspected still folds into an earlier sweep, so a staggered startup burst is still one registry round-trip.
This commit is contained in:
@@ -14,6 +14,10 @@ const STALE_PANE = {
|
||||
activeAccountId: 'account-b'
|
||||
}
|
||||
|
||||
function inspectCallCountFor(ptyId: string): number {
|
||||
return vi.mocked(window.api.pty.inspectProcess).mock.calls.filter(([id]) => id === ptyId).length
|
||||
}
|
||||
|
||||
describe('notifyCodexPaneBoundForStaleSweep', () => {
|
||||
const originalWindow = (globalThis as { window?: typeof window }).window
|
||||
|
||||
@@ -104,6 +108,22 @@ describe('notifyCodexPaneBoundForStaleSweep', () => {
|
||||
})
|
||||
})
|
||||
|
||||
it('still coalesces a burst whose binds land milliseconds apart', async () => {
|
||||
// Real startup binds are staggered, so per-PTY due times must not turn one
|
||||
// sweep into one registry round-trip per pane.
|
||||
useAppStore.setState({ ptyIdsByTabId: { 'tab-1': ['pty-1', 'pty-2'] } })
|
||||
vi.mocked(window.api.codexAccounts.listStalePanes).mockResolvedValue([STALE_PANE])
|
||||
|
||||
notifyCodexPaneBoundForStaleSweep('pty-1')
|
||||
await vi.advanceTimersByTimeAsync(10)
|
||||
notifyCodexPaneBoundForStaleSweep('pty-2')
|
||||
await vi.advanceTimersByTimeAsync(290)
|
||||
|
||||
expect(window.api.codexAccounts.listStalePanes).toHaveBeenCalledExactlyOnceWith({
|
||||
ptyIds: ['pty-1', 'pty-2']
|
||||
})
|
||||
})
|
||||
|
||||
it('retries a PTY whose process read is unusable until the reattach settles', async () => {
|
||||
vi.mocked(window.api.codexAccounts.listStalePanes).mockResolvedValue([STALE_PANE])
|
||||
vi.mocked(window.api.pty.inspectProcess).mockRejectedValueOnce(new Error('terminal_gone'))
|
||||
@@ -129,6 +149,39 @@ describe('notifyCodexPaneBoundForStaleSweep', () => {
|
||||
expect(window.api.pty.inspectProcess).toHaveBeenCalledTimes(3)
|
||||
})
|
||||
|
||||
it('spends a pane its full retry ladder even when another pane binds later', async () => {
|
||||
// Regression: one shared timer used to drain every queued PTY, so a pane
|
||||
// already waiting on its 1500ms rung had the next rung consumed by the newer
|
||||
// pane's shorter delay — exhausting its 3 attempts ~2.5s early and dropping
|
||||
// a Windows daemon reattach that had not settled yet.
|
||||
useAppStore.setState({ ptyIdsByTabId: { 'tab-1': ['pty-1', 'pty-2'] } })
|
||||
vi.mocked(window.api.pty.inspectProcess).mockRejectedValue(new Error('terminal_gone'))
|
||||
|
||||
notifyCodexPaneBoundForStaleSweep('pty-1')
|
||||
await vi.advanceTimersByTimeAsync(310)
|
||||
notifyCodexPaneBoundForStaleSweep('pty-2')
|
||||
|
||||
// t = 3310: past pty-1's second look (1800), well before its third (5800).
|
||||
await vi.advanceTimersByTimeAsync(3_000)
|
||||
expect(inspectCallCountFor('pty-1')).toBe(2)
|
||||
|
||||
await vi.advanceTimersByTimeAsync(3_000)
|
||||
expect(inspectCallCountFor('pty-1')).toBe(3)
|
||||
})
|
||||
|
||||
it('does not park a fresh bind behind a pending retry wait', async () => {
|
||||
useAppStore.setState({ ptyIdsByTabId: { 'tab-1': ['pty-1', 'pty-2'] } })
|
||||
vi.mocked(window.api.pty.inspectProcess).mockRejectedValue(new Error('terminal_gone'))
|
||||
|
||||
notifyCodexPaneBoundForStaleSweep('pty-1')
|
||||
await vi.advanceTimersByTimeAsync(310)
|
||||
notifyCodexPaneBoundForStaleSweep('pty-2')
|
||||
await vi.advanceTimersByTimeAsync(300)
|
||||
|
||||
// The pending timer was pty-1's 1500ms rung; pty-2 must not inherit that wait.
|
||||
expect(inspectCallCountFor('pty-2')).toBe(1)
|
||||
})
|
||||
|
||||
it('stops retrying a pane the registry reports as not stale', async () => {
|
||||
notifyCodexPaneBoundForStaleSweep('pty-1')
|
||||
await vi.advanceTimersByTimeAsync(60_000)
|
||||
|
||||
@@ -10,10 +10,15 @@ import {
|
||||
// a hint, and a pane that never resolves must not become a polling loop.
|
||||
const SWEEP_ATTEMPT_DELAYS_MS = [300, 1500, 4000] as const
|
||||
|
||||
const queuedPtyIds = new Set<string>()
|
||||
// Why: each PTY owns its rung, so the queue has to carry a per-PTY due time. One
|
||||
// shared timer coalesces the startup burst, but it must always target the
|
||||
// EARLIEST due entry — a single timer that drained everything let one pane's
|
||||
// short delay consume another's later rung and cut its ladder short.
|
||||
const dueAtByPtyId = new Map<string, number>()
|
||||
const attemptsByPtyId = new Map<string, number>()
|
||||
const notifiedPtyIds = new Set<string>()
|
||||
let flushTimer: ReturnType<typeof setTimeout> | null = null
|
||||
let flushTimerDueAt: number | null = null
|
||||
|
||||
/**
|
||||
* Queues a stale-account check for a PTY that has just bound to a pane.
|
||||
@@ -27,8 +32,8 @@ export function notifyCodexPaneBoundForStaleSweep(ptyId: string): void {
|
||||
if (notifiedPtyIds.has(ptyId)) {
|
||||
return
|
||||
}
|
||||
queuedPtyIds.add(ptyId)
|
||||
arm(SWEEP_ATTEMPT_DELAYS_MS[0])
|
||||
queue(ptyId, SWEEP_ATTEMPT_DELAYS_MS[0])
|
||||
armForEarliestDue()
|
||||
}
|
||||
|
||||
export function resetCodexStalePaneSweepForTests(): void {
|
||||
@@ -36,19 +41,61 @@ export function resetCodexStalePaneSweepForTests(): void {
|
||||
clearTimeout(flushTimer)
|
||||
flushTimer = null
|
||||
}
|
||||
queuedPtyIds.clear()
|
||||
flushTimerDueAt = null
|
||||
dueAtByPtyId.clear()
|
||||
attemptsByPtyId.clear()
|
||||
notifiedPtyIds.clear()
|
||||
}
|
||||
|
||||
function arm(delayMs: number): void {
|
||||
if (flushTimer !== null) {
|
||||
function queue(ptyId: string, delayMs: number): void {
|
||||
const dueAt = Date.now() + delayMs
|
||||
const existing = dueAtByPtyId.get(ptyId)
|
||||
// Why: a rebind of an already-queued PTY must never push its look later.
|
||||
dueAtByPtyId.set(ptyId, existing === undefined ? dueAt : Math.min(existing, dueAt))
|
||||
}
|
||||
|
||||
function armForEarliestDue(): void {
|
||||
let earliestDueAt: number | null = null
|
||||
for (const dueAt of dueAtByPtyId.values()) {
|
||||
if (earliestDueAt === null || dueAt < earliestDueAt) {
|
||||
earliestDueAt = dueAt
|
||||
}
|
||||
}
|
||||
if (earliestDueAt === null) {
|
||||
return
|
||||
}
|
||||
flushTimer = setTimeout(() => {
|
||||
flushTimer = null
|
||||
void flush()
|
||||
}, delayMs)
|
||||
if (flushTimer !== null) {
|
||||
if (flushTimerDueAt !== null && flushTimerDueAt <= earliestDueAt) {
|
||||
return
|
||||
}
|
||||
clearTimeout(flushTimer)
|
||||
}
|
||||
flushTimerDueAt = earliestDueAt
|
||||
flushTimer = setTimeout(
|
||||
() => {
|
||||
flushTimer = null
|
||||
flushTimerDueAt = null
|
||||
void flush()
|
||||
},
|
||||
Math.max(0, earliestDueAt - Date.now())
|
||||
)
|
||||
}
|
||||
|
||||
function takeDuePtyIds(): string[] {
|
||||
const now = Date.now()
|
||||
const duePtyIds: string[] = []
|
||||
for (const [ptyId, dueAt] of dueAtByPtyId) {
|
||||
// Why: folding a never-inspected PTY into an earlier sweep is what coalesces
|
||||
// the startup burst, and costs it nothing. A PTY already waiting on a retry
|
||||
// rung must wait for its own due time, or that rung is spent for free.
|
||||
if (dueAt <= now || !attemptsByPtyId.has(ptyId)) {
|
||||
duePtyIds.push(ptyId)
|
||||
}
|
||||
}
|
||||
for (const ptyId of duePtyIds) {
|
||||
dueAtByPtyId.delete(ptyId)
|
||||
}
|
||||
return duePtyIds
|
||||
}
|
||||
|
||||
function shouldRetry(scan: CodexPaneScanResult): boolean {
|
||||
@@ -59,9 +106,10 @@ function shouldRetry(scan: CodexPaneScanResult): boolean {
|
||||
}
|
||||
|
||||
async function flush(): Promise<void> {
|
||||
const ptyIds = [...queuedPtyIds]
|
||||
queuedPtyIds.clear()
|
||||
const ptyIds = takeDuePtyIds()
|
||||
if (ptyIds.length === 0) {
|
||||
// Why: a timer can fire a hair early; re-aim it rather than dropping the queue.
|
||||
armForEarliestDue()
|
||||
return
|
||||
}
|
||||
|
||||
@@ -74,7 +122,6 @@ async function flush(): Promise<void> {
|
||||
}
|
||||
|
||||
const scanByPtyId = new Map(scans.map((scan) => [scan.ptyId, scan]))
|
||||
let nextDelayMs: number | null = null
|
||||
for (const ptyId of ptyIds) {
|
||||
const scan = scanByPtyId.get(ptyId)
|
||||
if (scan?.notified === true) {
|
||||
@@ -95,11 +142,8 @@ async function flush(): Promise<void> {
|
||||
continue
|
||||
}
|
||||
attemptsByPtyId.set(ptyId, attempt)
|
||||
queuedPtyIds.add(ptyId)
|
||||
nextDelayMs = nextDelayMs === null ? delayMs : Math.min(nextDelayMs, delayMs)
|
||||
queue(ptyId, delayMs)
|
||||
}
|
||||
|
||||
if (nextDelayMs !== null) {
|
||||
arm(nextDelayMs)
|
||||
}
|
||||
armForEarliestDue()
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user