From 2f26181c3f36528fd8e0cb5c11aa2cb262ee6368 Mon Sep 17 00:00:00 2001 From: baixiaohang Date: Thu, 6 Aug 2026 15:57:40 +0800 Subject: [PATCH 1/2] fix(client): quarantine timed-out operator suspend --- .../session-manager-edge-coverage.test.ts | 254 +++++++++++++++++- .../client/src/runtime/session-manager.ts | 182 +++++++++++-- 2 files changed, 413 insertions(+), 23 deletions(-) diff --git a/packages/client/src/__tests__/session-manager-edge-coverage.test.ts b/packages/client/src/__tests__/session-manager-edge-coverage.test.ts index 6372744b1..1ab31157b 100644 --- a/packages/client/src/__tests__/session-manager-edge-coverage.test.ts +++ b/packages/client/src/__tests__/session-manager-edge-coverage.test.ts @@ -11,7 +11,13 @@ import type { } from "@first-tree/shared"; import { encodeProviderRetryEventMessage, parseProviderRetryEventMessage } from "@first-tree/shared"; import { afterEach, describe, expect, it, vi } from "vitest"; -import type { AgentHandler, HandlerFactory, SessionContext, SessionMessage } from "../runtime/handler.js"; +import type { + AgentHandler, + DeliveryToken, + HandlerFactory, + SessionContext, + SessionMessage, +} from "../runtime/handler.js"; import type { DeliveryDecision, DeliveryRouteOwnership, DeliveryWork } from "../runtime/inbox-delivery-coordinator.js"; import type { SubprocessProbe } from "../runtime/process-tree-probe.js"; import { SessionManager } from "../runtime/session-manager.js"; @@ -54,6 +60,15 @@ type SessionManagerInternals = { terminatePersistFailures: Set; awaitingResetFenceRelease: Set; pendingTeardowns: Map>; + quarantinedSessions: Map< + string, + { + handler: AgentHandler; + generation: number; + reason: "operator_suspend_timeout"; + routeTransitionInFlight: boolean; + } + >; routeProducers: Map>>; registry: SessionRegistry | null; pendingQueue: Array<{ message: SessionMessage | null; chatId: string; deliveryKind: string }>; @@ -3257,6 +3272,243 @@ describe("SessionManager edge coverage", () => { expect(i.sessions.has(chatId)).toBe(false); }); + it("quarantines a timed-out operator suspend and recovers real inbox debt before a fresh handler", async () => { + vi.useFakeTimers(); + let initialCtx: SessionContext | undefined; + let initialHead: SessionMessage | undefined; + const oldHandler = handler({ + start: vi.fn().mockImplementation(async (message, ctx, token) => { + initialCtx = ctx; + initialHead = message; + token?.processingStarted(message); + return { sessionId: "established-session", route: { kind: "owned" as const, mode: "queued" as const } }; + }), + inject: vi.fn().mockReturnValue({ kind: "owned", mode: "queued" } as const), + suspend: vi.fn(() => new Promise(() => {})), + shutdown: vi.fn(() => new Promise(() => {})), + }); + let freshCtx: SessionContext | undefined; + let freshMessage: SessionMessage | undefined; + const freshHandler = handler({ + resume: vi.fn().mockImplementation(async (message, _sessionId, ctx, token) => { + freshCtx = ctx; + freshMessage = message; + if (message) token?.processingStarted(message); + return { sessionId: "fresh-session", route: { kind: "owned" as const, mode: "queued" as const } }; + }), + }); + const ackEntry = vi.fn<(entryId: number) => Promise>().mockResolvedValue(undefined); + const recoverChat = vi.fn<(chatId: string) => Promise>().mockResolvedValue(undefined); + const onSessionEvent = vi.fn<(chatId: string, event: SessionEvent) => void>(); + const sm = makeManager({ handlers: [oldHandler, freshHandler], ackEntry, recoverChat, onSessionEvent }); + const i = internals(sm); + const chatId = "chat-timeout-recovery-before-fresh-handler"; + const headEntry = mockEntry({ id: 9100, chatId, messageId: "msg-timeout-head" }); + const queuedTailEntry = mockEntry({ id: 9101, chatId, messageId: "msg-timeout-queued-tail" }); + + await sm.dispatch(headEntry); + if (!initialCtx || !initialHead) throw new Error("initial route was not captured"); + await sm.dispatch(queuedTailEntry); + expect(i.sessions.get(chatId)?.routeTransition).toBe(null); + + await sm.handleCommand(chatId, "session:suspend"); + const laterDispatch = sm.dispatch(mockEntry({ id: 9102, chatId, messageId: "msg-after-timeout" })); + + await vi.advanceTimersByTimeAsync(29_999); + expect(freshHandler.resume).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(1); + await laterDispatch; + + expect(ackEntry).toHaveBeenCalledWith(9100); + expect(ackEntry).not.toHaveBeenCalledWith(9101); + expect(recoverChat).toHaveBeenCalledTimes(1); + expect(freshHandler.resume).not.toHaveBeenCalled(); + expect(oldHandler.shutdown).not.toHaveBeenCalled(); + expect(i.pendingTeardowns.has(chatId)).toBe(false); + expect(i.quarantinedSessions.get(chatId)).toEqual( + expect.objectContaining({ + handler: oldHandler, + reason: "operator_suspend_timeout", + routeTransitionInFlight: false, + }), + ); + expect(onSessionEvent).toHaveBeenCalledWith( + chatId, + expect.objectContaining({ + payload: expect.objectContaining({ + message: expect.stringContaining('"routeTransitionInFlight":false'), + }), + }), + ); + + await sm.dispatch(queuedTailEntry); + + const recoveryOrder = recoverChat.mock.invocationCallOrder[0]; + const resumeOrder = vi.mocked(freshHandler.resume).mock.invocationCallOrder[0]; + expect(recoveryOrder).toBeDefined(); + expect(resumeOrder).toBeDefined(); + expect(Number(recoveryOrder)).toBeLessThan(Number(resumeOrder)); + expect(i.sessions.get(chatId)?.handler).toBe(freshHandler); + if (!freshCtx || !freshMessage) throw new Error("fresh route was not captured"); + await freshCtx.finishTurn(freshMessage, { status: "success", terminal: true }); + + await sm.shutdown(); + }); + + it("keeps later suspend and resume cycles healthy while the quarantined callback never settles", async () => { + vi.useFakeTimers(); + const oldHandler = handler({ + suspend: vi.fn(() => new Promise(() => {})), + shutdown: vi.fn(() => new Promise(() => {})), + }); + const freshHandler = handler({ + resume: vi + .fn() + .mockResolvedValue({ sessionId: "fresh-session", route: { kind: "owned" as const, mode: "queued" as const } }), + }); + const sm = makeManager({ handlers: [freshHandler] }); + const i = internals(sm); + const chatId = "chat-quarantine-repeated-resume"; + i.sessions.set(chatId, makeSessionRecord(chatId, { handler: oldHandler, status: "active" })); + i._activeCount = 1; + + await sm.handleCommand(chatId, "session:suspend"); + await vi.advanceTimersByTimeAsync(30_000); + await i.sessions.get(chatId)?.suspending; + await sm.handleCommand(chatId, "session:resume"); + + expect(freshHandler.resume).toHaveBeenCalledTimes(1); + expect(i.sessions.get(chatId)?.handler).toBe(freshHandler); + expect(oldHandler.shutdown).not.toHaveBeenCalled(); + + await sm.handleCommand(chatId, "session:suspend"); + await i.sessions.get(chatId)?.suspending; + await sm.handleCommand(chatId, "session:resume"); + + expect(freshHandler.resume).toHaveBeenCalledTimes(2); + expect(i.sessions.get(chatId)?.handler).toBe(freshHandler); + expect(i.quarantinedSessions.get(chatId)?.handler).toBe(oldHandler); + expect(oldHandler.shutdown).not.toHaveBeenCalled(); + + await sm.shutdown(); + }); + + it("fences late output and route adoption from a quarantined generation", async () => { + vi.useFakeTimers(); + let signalResumeStarted: (() => void) | undefined; + let resolveResume: (() => void) | undefined; + let resolveSuspend: (() => void) | undefined; + let oldCtx: SessionContext | undefined; + let oldToken: DeliveryToken | undefined; + let oldMessage: SessionMessage | undefined; + const resumeStarted = new Promise((resolve) => { + signalResumeStarted = resolve; + }); + const resumeGate = new Promise((resolve) => { + resolveResume = resolve; + }); + const suspendGate = new Promise((resolve) => { + resolveSuspend = resolve; + }); + const oldHandler = handler({ + resume: vi.fn().mockImplementation(async (message, _sessionId, ctx, token) => { + oldCtx = ctx; + oldToken = token; + oldMessage = message; + token?.processingStarted(message); + signalResumeStarted?.(); + await resumeGate; + return { sessionId: "late-session", route: { kind: "owned" as const, mode: "queued" as const } }; + }), + suspend: vi.fn(() => suspendGate), + shutdown: vi.fn(() => new Promise(() => {})), + }); + const freshHandler = handler(); + const ackEntry = vi.fn<(entryId: number) => Promise>().mockResolvedValue(undefined); + const onSessionEvent = vi.fn<(chatId: string, event: SessionEvent) => void>(); + const sm = makeManager({ handlers: [freshHandler], ackEntry, onSessionEvent }); + const i = internals(sm); + const chatId = "chat-quarantined-late-generation"; + i.sessions.set( + chatId, + makeSessionRecord(chatId, { + handler: oldHandler, + status: "suspended", + claudeSessionId: "existing-session", + }), + ); + + const resumeDispatch = sm.dispatch(mockEntry({ id: 9120, chatId, messageId: "msg-late-generation" })); + await resumeStarted; + await sm.handleCommand(chatId, "session:suspend"); + await vi.advanceTimersByTimeAsync(30_000); + await i.sessions.get(chatId)?.suspending; + + expect(i.quarantinedSessions.get(chatId)).toEqual( + expect.objectContaining({ + handler: oldHandler, + routeTransitionInFlight: true, + }), + ); + if (!oldCtx || !oldToken || !oldMessage) throw new Error("old route output handles were not captured"); + const eventCount = onSessionEvent.mock.calls.length; + const ackCount = ackEntry.mock.calls.length; + const lastActivity = i.sessions.get(chatId)?.lastActivity; + const trigger = i.currentTrigger.get(chatId); + + oldCtx.emitEvent({ kind: "assistant_text", payload: { text: "late output" } }); + oldCtx.recordProviderActivity(); + await oldCtx.forwardResult("late result"); + await oldToken.complete(oldMessage, { status: "success", terminal: true }); + resolveSuspend?.(); + + expect(onSessionEvent).toHaveBeenCalledTimes(eventCount); + expect(ackEntry).toHaveBeenCalledTimes(ackCount); + expect(i.sessions.get(chatId)?.lastActivity).toBe(lastActivity); + expect(i.currentTrigger.get(chatId)).toEqual(trigger); + expect(i.sessions.get(chatId)?.claudeSessionId).toBe("existing-session"); + + await sm.dispatch(mockEntry({ id: 9121, chatId, messageId: "msg-provider-admission-fenced" })); + expect(freshHandler.start).not.toHaveBeenCalled(); + expect(freshHandler.resume).not.toHaveBeenCalled(); + + await sm.shutdown(); + expect(oldHandler.shutdown).not.toHaveBeenCalled(); + + resolveResume?.(); + await resumeDispatch; + expect(i.sessions.has(chatId)).toBe(false); + expect(i.pendingTeardowns.has(chatId)).toBe(false); + }); + + it("returns restart-required Reset failure and bounds manager shutdown after suspend timeout", async () => { + vi.useFakeTimers(); + const oldHandler = handler({ + suspend: vi.fn(() => new Promise(() => {})), + shutdown: vi.fn(() => new Promise(() => {})), + }); + const sm = makeManager(); + const i = internals(sm); + const chatId = "chat-quarantine-reset-restart-required"; + i.sessions.set(chatId, makeSessionRecord(chatId, { handler: oldHandler, status: "active" })); + i._activeCount = 1; + + await sm.handleCommand(chatId, "session:suspend"); + await vi.advanceTimersByTimeAsync(30_000); + await i.sessions.get(chatId)?.suspending; + + await expect(sm.handleCommand(chatId, "session:terminate", { resetRef: "ref-restart" })).rejects.toThrow( + "Reset blocked: operator_suspend_timeout; provider teardown was not confirmed", + ); + expect(i.sessions.has(chatId)).toBe(true); + expect(i.quarantinedSessions.has(chatId)).toBe(true); + expect(oldHandler.shutdown).not.toHaveBeenCalled(); + + await expect(sm.shutdown()).resolves.toBeUndefined(); + expect(oldHandler.shutdown).not.toHaveBeenCalled(); + }); + it("manager shutdown stops a handler after a completed failed suspend with a falsey rejection", async () => { const targetHandler = handler({ suspend: vi.fn().mockRejectedValue(undefined), diff --git a/packages/client/src/runtime/session-manager.ts b/packages/client/src/runtime/session-manager.ts index 9611f4f75..ca8cf6ad5 100644 --- a/packages/client/src/runtime/session-manager.ts +++ b/packages/client/src/runtime/session-manager.ts @@ -193,6 +193,13 @@ type RouteTransitionToken = RouteLeaseToken & { phase: "start" | "resume"; }; +type QuarantinedSession = { + handler: AgentHandler; + generation: number; + reason: "operator_suspend_timeout"; + routeTransitionInFlight: boolean; +}; + type SessionFailureHandling = | { kind: "retry" } | { kind: "terminal"; reasonCode: string; terminalEventPersisted: boolean }; @@ -468,6 +475,34 @@ type SessionManagerConfig = { /** Maximum number of evicted session mappings to retain for resume recovery. */ const MAX_EVICTED_MAPPINGS = 500; +/** + * Provider suspend/drain callbacks can disappear across host sleep or + * transport loss. Bound an operator pause so later inbox work can recover. + */ +const OPERATOR_SUSPEND_TIMEOUT_MS = 30_000; + +class HandlerSuspendTimeoutError extends Error { + constructor(chatId: string) { + super(`handler suspend timed out after ${OPERATOR_SUSPEND_TIMEOUT_MS}ms for chat ${chatId}`); + this.name = "HandlerSuspendTimeoutError"; + } +} + +async function waitForHandlerSuspend(chatId: string, suspend: () => Promise): Promise { + let timer: ReturnType | null = null; + try { + await Promise.race([ + Promise.resolve().then(suspend), + new Promise((_resolve, reject) => { + timer = setTimeout(() => reject(new HandlerSuspendTimeoutError(chatId)), OPERATOR_SUSPEND_TIMEOUT_MS); + timer.unref?.(); + }), + ]); + } finally { + if (timer) clearTimeout(timer); + } +} + /** * Minimum spacing between gate-triggered replay-fence reconciliations for * one chat. Withheld dispatches retry the server-truth readback so a @@ -648,6 +683,13 @@ export class SessionManager { * `detachHandlerWithPendingTeardown` / `registerPendingTeardown`). */ private readonly pendingTeardowns = new Map>(); + /** + * Chat-level authority for provider generations whose operator suspend + * callback exceeded its bound. The entry lives for this manager's lifetime: + * ordinary routing may replace the exact handler, while Reset remains + * restart-required because provider teardown was never confirmed. + */ + private readonly quarantinedSessions = new Map(); /** * In-flight route producers (start/resume/retry provider calls), per chat. * Tracked from `beginRouteTransition` until the route settles: a canceled @@ -988,6 +1030,9 @@ export class SessionManager { * or `session:command:aborted`) can lift the fence. */ async handleCommand(chatId: string, command: SessionCommandType, options?: { resetRef?: string }): Promise { + if (command === "session:terminate" && this.quarantinedSessions.has(chatId)) { + throw this.quarantineRestartRequiredError(chatId, "Reset"); + } const inFlightTermination = this.terminatingChats.get(chatId); if (inFlightTermination) { // A duplicate terminate joins the in-flight cleanup instead of @@ -1032,6 +1077,9 @@ export class SessionManager { if (command === "session:resume") { const session = this.sessions.get(chatId); if (session?.suspending) await session.suspending; + if (this.isProviderAdmissionRestartRequired(chatId)) { + throw this.quarantineRestartRequiredError(chatId, "provider admission"); + } if (await this.recoverDebtBeforeResume(chatId, "session_resume:recovery_debt")) { this.drainPendingQueue(); return; @@ -1078,6 +1126,9 @@ export class SessionManager { // apply instead of acking over a possibly-live handler. const joinedSuspend = session?.suspending != null; if (session?.suspending) await session.suspending; + if (this.quarantinedSessions.has(chatId)) { + throw this.quarantineRestartRequiredError(chatId, "Reset"); + } if (joinedSuspend && session?.suspendError) throw asTerminateError("suspend", session.suspendError.error); const activeSlotHeld = session?.activeSlotHeld === true; if (session) this.releaseActiveSlot(session); @@ -1234,6 +1285,9 @@ export class SessionManager { for (const id of this.routeProducers.keys()) { if (this.shouldIncludeInRuntimeSync(id, activeChatIds)) ids.add(id); } + for (const id of this.quarantinedSessions.keys()) { + if (this.shouldIncludeInRuntimeSync(id, activeChatIds)) ids.add(id); + } // Unresolved Reset retirement (in-flight terminate, failed durable flush, // or parked Reset-fence debt awaiting an exact receipted terminal // disposition) force-keeps the chat: the server must retain reconcile @@ -1345,6 +1399,7 @@ export class SessionManager { if (!session.activeSlotHeld && session.suspending === null && !stopUnconfirmedAfterFailedBoundary) { return Promise.resolve(); } + if (this.isCurrentHandlerQuarantined(session)) return Promise.resolve(); attemptedHandlers.add(session.handler); return this.shutdownHandler(session.handler, reason ?? "manager_shutdown", { ...(session.activeSlotHeld ? { settleProviderEntered: true } : {}), @@ -1370,6 +1425,7 @@ export class SessionManager { } } for (const [pendingHandler, chatIds] of debtChatsByHandler) { + if (chatIds.some((chatId) => this.quarantinedSessions.get(chatId)?.handler === pendingHandler)) continue; if (attemptedHandlers.has(pendingHandler)) continue; attemptedHandlers.add(pendingHandler); shutdowns.push( @@ -1412,6 +1468,7 @@ export class SessionManager { const retriedHandlers = new Set(); for (const pending of this.pendingTeardowns.values()) { for (const pendingHandler of [...pending]) { + if ([...this.quarantinedSessions.values()].some((entry) => entry.handler === pendingHandler)) continue; if (retriedHandlers.has(pendingHandler)) continue; retriedHandlers.add(pendingHandler); // Each attempt joins a still in-flight shutdown when one exists; a @@ -1548,6 +1605,7 @@ export class SessionManager { } private hasRuntimeSyncForceKeep(chatId: string): boolean { + if (this.quarantinedSessions.has(chatId)) return true; // Unresolved teardown debt force-keeps the chat: its handler is not // confirmed stopped, so dropping the chat from the held report would // lose the reconcile retry channel for the debt. @@ -1578,14 +1636,17 @@ export class SessionManager { * last one fences the post-apply / pre-disposition window even when nothing * was parked at apply time. Must NOT be used for the duplicate-terminate * join lookup — a genuine retry terminate must still execute and clear the - * persistence failure. + * persistence failure. A quarantined generation normally still permits a + * fresh route, but a timeout that caught an unresolved start/resume + * transition stays fail closed until daemon restart. */ private isProviderRouteAdmissionFenced(chatId: string): boolean { return ( this.terminatingChats.has(chatId) || this.terminatePersistFailures.has(chatId) || this.awaitingResetFenceRelease.has(chatId) || - this.hasArmedResetGeneration(chatId) + this.hasArmedResetGeneration(chatId) || + this.isProviderAdmissionRestartRequired(chatId) ); } @@ -2043,11 +2104,66 @@ export class SessionManager { return this.config.handlerFactory(handlerCfg); } + private quarantineMatches(chatId: string, lease: RouteLeaseToken): boolean { + const quarantined = this.quarantinedSessions.get(chatId); + return quarantined?.handler === lease.handler && quarantined.generation === lease.generation; + } + + private isCurrentHandlerQuarantined(entry: SessionEntry): boolean { + return this.quarantinedSessions.get(entry.chatId)?.handler === entry.handler; + } + + private isProviderAdmissionRestartRequired(chatId: string): boolean { + return this.quarantinedSessions.get(chatId)?.routeTransitionInFlight === true; + } + + private quarantineRestartRequiredError(chatId: string, operation: "Reset" | "provider admission"): Error { + const quarantined = this.quarantinedSessions.get(chatId); + const reason = quarantined?.reason ?? "operator_suspend_timeout"; + return new Error( + `${operation} blocked: ${reason}; provider teardown was not confirmed for chat ${chatId}. Restart this agent daemon and retry.`, + ); + } + + private quarantineTimedOutSuspend(entry: SessionEntry, transition: RouteTransitionToken | null): void { + const generation = transition?.generation ?? entry.routeTransitionGeneration; + const handler = transition?.handler ?? entry.handler; + const routeTransitionInFlight = transition !== null; + const quarantine: QuarantinedSession = { + handler, + generation, + reason: "operator_suspend_timeout", + routeTransitionInFlight, + }; + this.quarantinedSessions.set(entry.chatId, quarantine); + this.retiredHandlers.add(handler); + // The quarantine is now the sole authority for this lost generation. + // Remove its unresolved producer from ordinary lifecycle joins; the + // producer's own finally remains safe when the set is already absent. + if (transition) this.routeProducers.delete(entry.chatId); + + const provider = this.runtimeProvider(); + const details = { + provider, + chatId: entry.chatId, + generation, + routeTransitionInFlight, + routeTransitionPhase: transition?.phase ?? null, + reason: quarantine.reason, + }; + this.config.log.error(details, "operator suspend timed out; provider session quarantined"); + this.emitResilienceEvent(entry.chatId, "resilience.session.operator_suspend_timeout", details); + } + private handlerForRouteTransition(entry: SessionEntry): AgentHandler { if (!this.retiredHandlers.has(entry.handler)) return entry.handler; const previous = entry.handler; const handler = this.createHandler(); entry.handler = handler; + // The quarantined generation has no trustworthy teardown join. Keep it + // exclusively in quarantinedSessions: ordinary pendingTeardowns would + // make later routes and manager shutdown wait on the lost callback again. + if (this.quarantinedSessions.get(entry.chatId)?.handler === previous) return handler; // The retired handler's shutdown was fire-and-forget — never confirmed. // Record the debt so a ref'd terminate strictly confirms the stop before // it may ack (the join resolves immediately when the stop already @@ -2216,6 +2332,13 @@ export class SessionManager { * of routing. */ private async settleTeardownDebtBeforeRoute(chatId: string): Promise { + if (this.isProviderAdmissionRestartRequired(chatId)) { + this.config.log.error( + { chatId, provider: this.runtimeProvider(), reason: "operator_suspend_timeout" }, + "provider admission blocked by unresolved quarantined route transition; daemon restart required", + ); + return false; + } // Quiesce in-flight route producers FIRST: a canceled start/resume can // still materialize late, and its discard registers teardown debt only // when the producer settles — debt drained before this point would be @@ -2277,6 +2400,19 @@ export class SessionManager { private discardStaleRouteTransition(chatId: string, transition: RouteTransitionToken, reason: string): void { this.retiredHandlers.add(transition.handler); + if (this.quarantineMatches(chatId, transition)) { + this.config.log.warn( + { + chatId, + provider: this.runtimeProvider(), + generation: transition.generation, + phase: transition.phase, + reason, + }, + "late quarantined route completion ignored", + ); + return; + } // A stale route completion MATERIALIZES the handler (the canceled // start/resume returned late), so it needs a NEW shutdown chained after // any prior (a pre-materialization shutdown was a no-op). That stop's @@ -3106,6 +3242,9 @@ export class SessionManager { } if (stopForManagerShutdown("session_resume:manager_shutdown_after_suspend")) return; if (this.sessions.get(entry.chatId) !== entry) return; + if (this.isProviderAdmissionRestartRequired(entry.chatId)) { + throw this.quarantineRestartRequiredError(entry.chatId, "provider admission"); + } if (await this.recoverDebtBeforeResume(entry.chatId, "session_resume:recovery_debt")) return; if (stopForManagerShutdown("session_resume:manager_shutdown_after_recovery")) return; if ( @@ -3117,20 +3256,17 @@ export class SessionManager { return; } - // A failed suspend left the current handler never confirmed stopped. - // Stop it strictly (joining any in-flight shutdown's raw face) BEFORE - // the route below reuses or replaces the reference — reusing or - // overwriting an unconfirmed-stop handler loses the teardown authority - // while the old run may still be alive. A teardown failure propagates - // into resume's existing error semantics (routeMessage retries the - // delivery / operator resume logs the command error); it must not - // silently continue. On success the handler is retired so the route - // installs a fresh one, and the recorded suspend failure is cleared. - if (entry.suspendError && entry.handlerStoppedBySuspend !== entry.handler) { - await this.shutdownHandler(entry.handler, "session_resume_after_failed_suspend", { observeFailure: true }); - entry.handlerStoppedBySuspend = entry.handler; + // Ordinary suspend failure still requires a confirmed strict stop before + // replacement. A quarantined timeout has deliberately lost that join: + // its exact handler/generation stays fenced by quarantinedSessions and the + // route below installs a fresh handler without registering ordinary debt. + if (entry.suspendError) { + if (entry.handlerStoppedBySuspend !== entry.handler && !this.isCurrentHandlerQuarantined(entry)) { + await this.shutdownHandler(entry.handler, "session_resume_after_failed_suspend", { observeFailure: true }); + entry.handlerStoppedBySuspend = entry.handler; + this.retiredHandlers.add(entry.handler); + } entry.suspendError = null; - this.retiredHandlers.add(entry.handler); } // Route admission fence: settle this chat's teardown authority before @@ -4133,11 +4269,14 @@ export class SessionManager { this.recomputeRuntimeState(); entry.suspending = (async () => { let settled = false; + let timedOut = false; try { // settleProviderEntered keeps already-issued DeliveryTokens on the // settlement lease (including active/deferred inject) so they can // post durable notice+ACK before prepareOperatorSuspend runs. - await entry.handler.suspend(opts.reason, { settleProviderEntered: true }); + await waitForHandlerSuspend(entry.chatId, () => + entry.handler.suspend(opts.reason, { settleProviderEntered: true }), + ); // If settle captured a terminal notice but could not persist it (or // could not claim token settlement), transfer the obligation onto // the inbox ledger so prepareOperatorSuspend / recovery cannot ACK @@ -4147,10 +4286,9 @@ export class SessionManager { } settled = true; } catch (err) { - // Settle failure leaves the handler joinable for a strict terminate / - // resume stop — do not start teardown here or suspend will hang on a - // gated shutdown the terminate owns. entry.suspendError = { error: err }; + timedOut = err instanceof HandlerSuspendTimeoutError; + if (timedOut) this.quarantineTimedOutSuspend(entry, inFlightTransition); try { this.config.log.warn({ chatId: entry.chatId, err }, "operator suspend settlement error"); } catch (logErr) { @@ -4167,7 +4305,7 @@ export class SessionManager { this.retiredHandlers.add(inFlightTransition.handler); } - if (!settled) { + if (!settled && !timedOut) { entry.suspending = null; if (unestablishedStart) { if (entry.handlerStoppedBySuspend !== entry.handler) { @@ -4182,7 +4320,7 @@ export class SessionManager { } const stopPromise = - inFlightTransition || !this.retiredHandlers.has(entry.handler) + settled && (inFlightTransition || !this.retiredHandlers.has(entry.handler)) ? this.shutdownHandler(target, opts.reason, { observeFailure: true }) : Promise.resolve(); @@ -4194,7 +4332,7 @@ export class SessionManager { await Promise.resolve(); await this.inboxDelivery.prepareOperatorSuspend(entry.chatId); await stopPromise; - if (target === entry.handler) { + if (settled && target === entry.handler) { entry.handlerStoppedBySuspend = entry.handler; } } catch (err) { From 81b3e616c9bc6fd1276e7e51182bf9f9a029a74f Mon Sep 17 00:00:00 2001 From: baixiaohang Date: Thu, 6 Aug 2026 16:48:14 +0800 Subject: [PATCH 2/2] fix(client): preserve suspend timeout notice debt --- .../session-manager-edge-coverage.test.ts | 73 +++++++++++++++++++ .../client/src/runtime/session-manager.ts | 16 ++-- 2 files changed, 82 insertions(+), 7 deletions(-) diff --git a/packages/client/src/__tests__/session-manager-edge-coverage.test.ts b/packages/client/src/__tests__/session-manager-edge-coverage.test.ts index 1ab31157b..fade97f0d 100644 --- a/packages/client/src/__tests__/session-manager-edge-coverage.test.ts +++ b/packages/client/src/__tests__/session-manager-edge-coverage.test.ts @@ -3356,6 +3356,79 @@ describe("SessionManager edge coverage", () => { await sm.shutdown(); }); + it("preserves terminal notice debt when operator suspend emits failure and then times out", async () => { + vi.useFakeTimers(); + let initialCtx: SessionContext | undefined; + const oldHandler = handler({ + start: vi.fn().mockImplementation(async (message, ctx, token) => { + initialCtx = ctx; + token?.processingStarted(message); + return { sessionId: "established-session", route: { kind: "owned" as const, mode: "queued" as const } }; + }), + suspend: vi.fn().mockImplementation(() => { + if (!initialCtx) throw new Error("initial route context was not captured"); + initialCtx.emitEvent({ + kind: "error", + payload: { + source: "runtime", + message: encodeProviderRetryEventMessage({ + event: "provider_failure_terminal", + provider: "codex", + scope: "provider_turn", + category: "credential", + reasonCode: "provider_credential_required", + replaySafety: "provider_entered", + userSeverity: "error", + messagePreview: "refresh token revoked while suspending", + }), + }, + }); + return new Promise(() => {}); + }), + shutdown: vi.fn(() => new Promise(() => {})), + }); + const ackEntry = vi.fn<(entryId: number) => Promise>().mockResolvedValue(undefined); + const recoverChat = vi.fn<(chatId: string) => Promise>().mockResolvedValue(undefined); + const sendMessage = vi.fn().mockResolvedValue({ id: "runtime-notice-after-timeout" }); + const sdk = { ...mockSdk(), sendMessage } as unknown as FirstTreeHubSDK; + const sm = makeManager({ handlers: [oldHandler], ackEntry, recoverChat, sdk }); + const i = internals(sm); + const chatId = "chat-timeout-terminal-notice-debt"; + const headEntry = mockEntry({ id: 9110, chatId, messageId: "msg-timeout-terminal-notice" }); + + await sm.dispatch(headEntry); + await sm.handleCommand(chatId, "session:suspend"); + await vi.advanceTimersByTimeAsync(30_000); + await i.sessions.get(chatId)?.suspending; + + expect(ackEntry).not.toHaveBeenCalled(); + expect(i.inboxDelivery.snapshot(chatId)).toMatchObject({ + entries: [{ entryId: 9110, messageId: headEntry.message.id, phase: "owned" }], + recoveryDebt: "required", + }); + + // The first frame opens recovery; the server's redelivery then settles + // the retained notice debt without re-entering the provider. + await sm.dispatch(headEntry); + expect(recoverChat).toHaveBeenCalledTimes(1); + expect(sendMessage).not.toHaveBeenCalled(); + expect(ackEntry).not.toHaveBeenCalled(); + + await sm.dispatch(headEntry); + await vi.waitFor(() => expect(ackEntry).toHaveBeenCalledTimes(1)); + + expect(sendMessage).toHaveBeenCalledTimes(1); + expect(ackEntry).toHaveBeenCalledWith(9110); + expect(oldHandler.start).toHaveBeenCalledTimes(1); + const noticeOrder = sendMessage.mock.invocationCallOrder[0]; + const ackOrder = ackEntry.mock.invocationCallOrder[0]; + if (noticeOrder === undefined || ackOrder === undefined) throw new Error("expected notice and ACK order"); + expect(noticeOrder).toBeLessThan(ackOrder); + expect(oldHandler.shutdown).not.toHaveBeenCalled(); + + await sm.shutdown(); + }); + it("keeps later suspend and resume cycles healthy while the quarantined callback never settles", async () => { vi.useFakeTimers(); const oldHandler = handler({ diff --git a/packages/client/src/runtime/session-manager.ts b/packages/client/src/runtime/session-manager.ts index ca8cf6ad5..f2b043c0e 100644 --- a/packages/client/src/runtime/session-manager.ts +++ b/packages/client/src/runtime/session-manager.ts @@ -4277,13 +4277,6 @@ export class SessionManager { await waitForHandlerSuspend(entry.chatId, () => entry.handler.suspend(opts.reason, { settleProviderEntered: true }), ); - // If settle captured a terminal notice but could not persist it (or - // could not claim token settlement), transfer the obligation onto - // the inbox ledger so prepareOperatorSuspend / recovery cannot ACK - // without durable notice evidence. - if (entry.pendingRuntimeFailureNotice) { - this.inboxDelivery.markNoticeRequiredForProcessingPrefix(entry.chatId, entry.pendingRuntimeFailureNotice); - } settled = true; } catch (err) { entry.suspendError = { error: err }; @@ -4296,6 +4289,15 @@ export class SessionManager { } } + // Suspend may emit a terminal provider-failure event and then either + // settle or lose its completion callback. Transfer that durable-notice + // obligation before invalidating the generation and before + // prepareOperatorSuspend can promote the provider-entered prefix to + // ACK-eligible terminal work. + if ((settled || timedOut) && entry.pendingRuntimeFailureNotice) { + this.inboxDelivery.markNoticeRequiredForProcessingPrefix(entry.chatId, entry.pendingRuntimeFailureNotice); + } + // Bump adoption generation only after settle. Kick observeFailure // teardown before awaiting prepare so a gated prepare still leaves an // in-flight shutdown that strict terminate can join (main #2125).