diff --git a/apps/desktop/src/app/session/hooks/use-prompt-actions/single-flight-resume.test.ts b/apps/desktop/src/app/session/hooks/use-prompt-actions/single-flight-resume.test.ts index 8e3986add1f90..8982519e8a625 100644 --- a/apps/desktop/src/app/session/hooks/use-prompt-actions/single-flight-resume.test.ts +++ b/apps/desktop/src/app/session/hooks/use-prompt-actions/single-flight-resume.test.ts @@ -1,14 +1,18 @@ import { afterEach, describe, expect, it, vi } from 'vitest' +import { isTimeoutError } from '@/lib/with-timeout' + import { clearSingleFlightSessionResumeState, registerRecoveredRuntime, + SESSION_RESUME_SETTLEMENT_TIMEOUT_MS, singleFlightSessionResume, takeRecoveredRuntime } from './single-flight-resume' import { resumeStoredRuntimeSession, SessionRecoveryAborted, withSessionNotFoundResume } from './utils' afterEach(() => { + vi.useRealTimers() clearSingleFlightSessionResumeState() vi.restoreAllMocks() }) @@ -66,6 +70,140 @@ describe('singleFlightSessionResume', () => { }) }) +describe('bounded settlement and straggler adoption', () => { + it('a never-settling resume rejects at the deadline instead of wedging the slot forever', async () => { + vi.useFakeTimers() + + const flight = singleFlightSessionResume('stored-wedged', () => new Promise(() => {})) + const settled = flight.catch(error => error) + + await vi.advanceTimersByTimeAsync(SESSION_RESUME_SETTLEMENT_TIMEOUT_MS) + + expect(isTimeoutError(await settled)).toBe(true) + }) + + it('a later caller for the same stored id gets a FRESH attempt, not the dead flight', async () => { + vi.useFakeTimers() + + const first = singleFlightSessionResume('stored-wedged', () => new Promise(() => {})) + const firstSettled = first.catch(() => 'timed-out') + + await vi.advanceTimersByTimeAsync(SESSION_RESUME_SETTLEMENT_TIMEOUT_MS) + expect(await firstSettled).toBe('timed-out') + + const run = vi.fn(async () => ({ session_id: 'rt-second' })) + + await expect(singleFlightSessionResume('stored-wedged', run)).resolves.toEqual({ session_id: 'rt-second' }) + expect(run).toHaveBeenCalledTimes(1) + }) + + it('a joiner of an already-wedged flight inherits the same deadline', async () => { + vi.useFakeTimers() + + const first = singleFlightSessionResume('stored-wedged', () => new Promise(() => {})) + // The joiner passes no run() of its own — it must not hang past the ceiling. + const joiner = singleFlightSessionResume('stored-wedged', async () => ({ session_id: 'never-used' })) + + expect(joiner).toBe(first) + + const settled = joiner.catch(error => error) + + await vi.advanceTimersByTimeAsync(SESSION_RESUME_SETTLEMENT_TIMEOUT_MS) + + expect(isTimeoutError(await settled)).toBe(true) + }) + + it('a slow-but-legitimate resume (profile probe + RPC) still settles inside the ceiling', async () => { + vi.useFakeTimers() + + // The worst healthy shape the ceiling is derived from: an active-profile + // probe, one cross-profile probe, then the resume RPC — each on its own + // 30s budget. A tighter ceiling would abort this and re-mint a runtime. + const flight = singleFlightSessionResume( + 'stored-slow', + () => new Promise<{ session_id: string }>(resolve => setTimeout(() => resolve({ session_id: 'rt-slow' }), 89_000)) + ) + + await vi.advanceTimersByTimeAsync(89_000) + + await expect(flight).resolves.toEqual({ session_id: 'rt-slow' }) + }) + + it('a runtime minted by a timed-out resume is adopted, not stranded on the gateway', async () => { + vi.useFakeTimers() + + let land: (value: { session_id: string }) => void = () => {} + const flight = singleFlightSessionResume( + 'stored-late', + () => + new Promise<{ session_id: string }>(resolve => { + land = resolve + }) + ) + const settled = flight.catch(() => 'timed-out') + + await vi.advanceTimersByTimeAsync(SESSION_RESUME_SETTLEMENT_TIMEOUT_MS) + expect(await settled).toBe('timed-out') + + // The RPC was never cancelled: it lands a real, registered runtime. + land({ session_id: 'rt-late' }) + await vi.advanceTimersByTimeAsync(0) + + // The next resume-shaped action reuses it instead of minting a second one. + expect(takeRecoveredRuntime('stored-late')).toBe('rt-late') + }) + + it('a straggler is NOT cached when a newer flight already owns the stored id', async () => { + vi.useFakeTimers() + + let land: (value: { session_id: string }) => void = () => {} + const first = singleFlightSessionResume( + 'stored-late', + () => + new Promise<{ session_id: string }>(resolve => { + land = resolve + }) + ) + const settled = first.catch(() => 'timed-out') + + await vi.advanceTimersByTimeAsync(SESSION_RESUME_SETTLEMENT_TIMEOUT_MS) + expect(await settled).toBe('timed-out') + + // A new flight claims the slot before the straggler lands. + const second = singleFlightSessionResume('stored-late', () => new Promise(() => {})) + + void second.catch(() => undefined) + land({ session_id: 'rt-late' }) + await vi.advanceTimersByTimeAsync(0) + + // Its caller adopts the second flight's result; caching the older runtime + // here would aim the next action at the wrong one. + expect(takeRecoveredRuntime('stored-late')).toBeUndefined() + }) + + it('a straggler that fails minted nothing and caches nothing', async () => { + vi.useFakeTimers() + + let fail: (error: unknown) => void = () => {} + const flight = singleFlightSessionResume( + 'stored-late', + () => + new Promise<{ session_id: string }>((_resolve, reject) => { + fail = reject + }) + ) + const settled = flight.catch(() => 'timed-out') + + await vi.advanceTimersByTimeAsync(SESSION_RESUME_SETTLEMENT_TIMEOUT_MS) + expect(await settled).toBe('timed-out') + + fail(new Error('resume failed')) + await vi.advanceTimersByTimeAsync(0) + + expect(takeRecoveredRuntime('stored-late')).toBeUndefined() + }) +}) + describe('drift-abort recovered-runtime cache', () => { it('drift-abort does not strand the recovered runtime — it is registered in the cache', async () => { const requestGateway = vi.fn(async (method: string) => { diff --git a/apps/desktop/src/app/session/hooks/use-prompt-actions/single-flight-resume.ts b/apps/desktop/src/app/session/hooks/use-prompt-actions/single-flight-resume.ts index 96184ba35dc1d..4aae3ad1bde76 100644 --- a/apps/desktop/src/app/session/hooks/use-prompt-actions/single-flight-resume.ts +++ b/apps/desktop/src/app/session/hooks/use-prompt-actions/single-flight-resume.ts @@ -13,9 +13,52 @@ * `session_id`); joiners receive whatever the winning call returns. */ +import { withTimeout } from '@/lib/with-timeout' + const _inFlightResumeByStoredSessionId = new Map>() -export function singleFlightSessionResume(storedSessionId: string, run: () => Promise): Promise { +/** + * Ceiling on how long ONE flight may hold the shared slot. + * + * Sharing the promise is what makes an unsettled resume so expensive: the + * entry is only released by `.finally()`, so while it hangs every later + * caller for that stored id joins the same dead promise and the conversation + * is unrecoverable for the life of the window. The deadline exists to break + * that wedge, NOT to make a slow resume feel responsive — the RPC's own 30s + * budget already covers responsiveness. + * + * Derived from the longest LEGITIMATE settlement rather than picked round, so + * a slow-but-healthy resume is never aborted. `run()` bodies resolve the + * owning profile before they send anything, and `resolveStoredSession()` + * probes backends sequentially on a cache miss: + * + * 30s active-profile `getSession` probe (Electron DEFAULT_FETCH_TIMEOUT_MS) + * + 30s one cross-profile `getSession` probe (same budget, per profile) + * + 30s the `session.resume` RPC (HermesGateway DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS) + * = 90s + * + * An install with more profiles can still exceed this, and that is deliberate: + * past this point the flight has held the slot longer than any single healthy + * recovery ever needs, and breaking it is better than stranding the window. + */ +export const SESSION_RESUME_SETTLEMENT_TIMEOUT_MS = 90_000 + +/** Best-effort `session_id` off a `session.resume`-shaped response. */ +function resumedRuntimeId(result: unknown): string { + if (typeof result !== 'object' || result === null) { + return '' + } + + const sessionId = (result as { session_id?: unknown }).session_id + + return typeof sessionId === 'string' ? sessionId : '' +} + +export function singleFlightSessionResume( + storedSessionId: string, + run: () => Promise, + timeoutMs = SESSION_RESUME_SETTLEMENT_TIMEOUT_MS +): Promise { const existing = _inFlightResumeByStoredSessionId.get(storedSessionId) if (existing) { @@ -25,13 +68,41 @@ export function singleFlightSessionResume(storedSessionId: string, run: () => // Promise.resolve().then(run) tolerates run() being synchronous, returning a // bare value, or throwing synchronously (test doubles and legacy callers do // all three) — a raw run().finally() would crash on a non-promise return. - const flight = Promise.resolve() - .then(run) - .finally(() => { - if (_inFlightResumeByStoredSessionId.get(storedSessionId) === flight) { - _inFlightResumeByStoredSessionId.delete(storedSessionId) - } - }) + const work = Promise.resolve().then(run) + + // withTimeout does NOT cancel the work it bounds, and here the straggler is + // not inert: `session.resume` can still land a REAL runtime, registered on + // the gateway, that no client is holding. Dropping it on the floor is the + // exact orphan-per-resume shape this module exists to prevent (#91276), so + // hand a late arrival to the same recovered-runtime cache the drift-abort + // path uses and let the next resume-shaped action adopt it. + const adoptStraggler = () => { + void work.then( + result => { + // A newer flight already owns this stored id — its caller will adopt + // whatever it returns, so caching an older runtime here would just + // aim the next action at the wrong one. + if (_inFlightResumeByStoredSessionId.has(storedSessionId)) { + return + } + + registerRecoveredRuntime(storedSessionId, resumedRuntimeId(result)) + }, + // A straggler that failed minted nothing — there is nothing to adopt. + () => undefined + ) + } + + const flight = withTimeout( + work, + timeoutMs, + `Timed out resuming session ${storedSessionId}`, + adoptStraggler + ).finally(() => { + if (_inFlightResumeByStoredSessionId.get(storedSessionId) === flight) { + _inFlightResumeByStoredSessionId.delete(storedSessionId) + } + }) _inFlightResumeByStoredSessionId.set(storedSessionId, flight)