From 14cebc477ada8cf5a31324039e5ba661f8f5da3e Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 21 Aug 2026 14:14:26 -0700 Subject: [PATCH] fix(recovery): keep queued replies alive during session watchdog repair (#127510) * fix(recovery): preserve queued turns during session watchdog repair * test(plugins): make stalled response timeout deterministic --- ...diagnostic-session-recovery-coordinator.ts | 12 +-- src/logging/diagnostic-session-recovery.ts | 4 +- ...stuck-session-recovery.integration.test.ts | 68 ++++++++++++++ ...tic-stuck-session-recovery.runtime.test.ts | 59 ++++++++++++ ...agnostic-stuck-session-recovery.runtime.ts | 28 +++--- src/logging/diagnostic.test.ts | 94 ++++++++++--------- test/scripts/plugins-assertions.test.ts | 3 + 7 files changed, 197 insertions(+), 71 deletions(-) diff --git a/src/logging/diagnostic-session-recovery-coordinator.ts b/src/logging/diagnostic-session-recovery-coordinator.ts index 6a830d851d8b..c4734156cce1 100644 --- a/src/logging/diagnostic-session-recovery-coordinator.ts +++ b/src/logging/diagnostic-session-recovery-coordinator.ts @@ -73,10 +73,6 @@ function emitSessionRecoveryCompleted(params: { }); } -function recoveryRequestKey(request: StuckSessionRecoveryRequest): string | undefined { - return resolveStuckSessionRecoveryRef(request); -} - function isRecoveryPromiseLike( value: void | StuckSessionRecoveryOutcome | Promise, ): value is Promise { @@ -85,10 +81,6 @@ function isRecoveryPromiseLike( ); } -function recoveryOutcomeHasQueuedLaneWork(outcome: StuckSessionRecoveryOutcome): boolean { - return outcome.status === "aborted" && (outcome.queuedCount ?? 0) > 0; -} - function applyRecoveryOutcomeToDiagnosticState(params: { request: StuckSessionRecoveryRequest; outcome: StuckSessionRecoveryOutcome | undefined; @@ -152,7 +144,7 @@ function applyRecoveryOutcomeToDiagnosticState(params: { state.lastStuckWarnAgeMs = undefined; state.lastLongRunningWarnAgeMs = undefined; const preserveQueuedIdleWork = - params.request.expectedState === "idle" && recoveryOutcomeHasQueuedLaneWork(params.outcome); + params.request.expectedState === "idle" && (params.outcome.queuedCount ?? 0) > 0; state.queueDepth = recoveryOutcomeClearsQueuedSessionState(params.outcome) ? 0 : preserveQueuedIdleWork @@ -174,7 +166,7 @@ function applyRecoveryOutcomeToDiagnosticState(params: { function requestStuckSessionRecoveryOutcome( params: RequestStuckSessionRecoveryParams, ): Promise { - const inFlightKey = recoveryRequestKey(params.request); + const inFlightKey = resolveStuckSessionRecoveryRef(params.request); if (inFlightKey && recoveryRequestsInFlight.has(inFlightKey)) { const outcome: StuckSessionRecoveryOutcome = { status: "skipped", diff --git a/src/logging/diagnostic-session-recovery.ts b/src/logging/diagnostic-session-recovery.ts index b42343a51aa2..71061c195600 100644 --- a/src/logging/diagnostic-session-recovery.ts +++ b/src/logging/diagnostic-session-recovery.ts @@ -87,8 +87,8 @@ export function recoveryOutcomeClearsQueuedSessionState( outcome: StuckSessionRecoveryOutcome, ): boolean { return ( - outcome.status === "released" || - (outcome.status === "aborted" && outcome.released > 0 && (outcome.queuedCount ?? 0) === 0) + (outcome.status === "released" || (outcome.status === "aborted" && outcome.released > 0)) && + (outcome.queuedCount ?? 0) === 0 ); } diff --git a/src/logging/diagnostic-stuck-session-recovery.integration.test.ts b/src/logging/diagnostic-stuck-session-recovery.integration.test.ts index e74c58b5bec0..8ff89b4bda69 100644 --- a/src/logging/diagnostic-stuck-session-recovery.integration.test.ts +++ b/src/logging/diagnostic-stuck-session-recovery.integration.test.ts @@ -139,6 +139,74 @@ describe("stuck session recovery integration", () => { unsubscribe(); }); + it.each(["preflight_compacting", "memory_flushing"] as const)( + "keeps real queued turns behind healthy %s work", + async (phase) => { + const sessionKey = `agent:main:healthy-${phase}`; + const sessionId = `healthy-${phase}-session`; + const lane = resolveEmbeddedSessionLane(sessionKey); + const operation = createReplyOperation({ sessionKey, sessionId, resetTriggered: false }); + operation.setPhase(phase); + const handle = { + queueMessage: async () => {}, + isStreaming: () => false, + isCompacting: () => phase === "preflight_compacting", + abort: () => {}, + }; + setActiveEmbeddedRun(sessionId, handle, sessionKey); + + let releaseActive!: () => void; + let markActiveStarted!: () => void; + const activeStarted = new Promise((resolve) => { + markActiveStarted = resolve; + }); + const active = enqueueCommandInLane( + lane, + () => + new Promise((resolve) => { + releaseActive = resolve; + markActiveStarted(); + }), + { warnAfterMs: Number.MAX_SAFE_INTEGER }, + ); + const queued = enqueueCommandInLane(lane, async () => "delivered", { + warnAfterMs: Number.MAX_SAFE_INTEGER, + }); + await activeStarted; + operation.abortSignal.addEventListener( + "abort", + () => { + clearActiveEmbeddedRun(sessionId, handle, sessionKey); + operation.complete(); + releaseActive(); + }, + { once: true }, + ); + + try { + const outcome = await recoverStuckDiagnosticSession({ + sessionId, + sessionKey, + ageMs: 720_000, + queueDepth: 1, + compactionSafetyTimeoutMs: 900_000, + allowActiveAbort: true, + }); + + expect(operation.abortSignal.aborted).toBe(false); + expect(outcome.status).toBe("skipped"); + await expectPendingAfterEventLoopTurn(queued); + expect(getQueueSize(lane)).toBe(2); + } finally { + clearActiveEmbeddedRun(sessionId, handle, sessionKey); + operation.complete(); + releaseActive(); + await active; + await queued; + } + }, + ); + it("does not reset a blocked lane while a reply operation is still active", async () => { const sessionKey = "agent:main:active-reply"; const sessionId = "active-reply-session"; diff --git a/src/logging/diagnostic-stuck-session-recovery.runtime.test.ts b/src/logging/diagnostic-stuck-session-recovery.runtime.test.ts index 5c36fe0284f8..4ac6804eac0d 100644 --- a/src/logging/diagnostic-stuck-session-recovery.runtime.test.ts +++ b/src/logging/diagnostic-stuck-session-recovery.runtime.test.ts @@ -520,6 +520,65 @@ describe("stuck session recovery", () => { }, ); + it.each( + (["preflight_compacting", "memory_flushing"] as const) + .flatMap((phase) => + [false, true].flatMap((hasEmbeddedHandle) => + [false, true].map((allowActiveAbort) => ({ + phase, + hasEmbeddedHandle, + allowActiveAbort, + ageMs: 720_000, + })), + ), + ) + .concat([ + { + phase: "preflight_compacting", + hasEmbeddedHandle: false, + allowActiveAbort: false, + ageMs: 915_000, + }, + { + phase: "memory_flushing", + hasEmbeddedHandle: true, + allowActiveAbort: true, + ageMs: 915_000, + }, + ]), + )( + "honors the configured $phase timeout with queued work (handle=$hasEmbeddedHandle, abort=$allowActiveAbort, age=$ageMs)", + async ({ phase, hasEmbeddedHandle, allowActiveAbort, ageMs }) => { + const sessionId = "maintenance-reply-session"; + mocks.resolveActiveEmbeddedRunSessionId.mockReturnValue(sessionId); + mocks.resolveActiveEmbeddedRunHandleSessionId.mockReturnValue( + hasEmbeddedHandle ? sessionId : undefined, + ); + mocks.isEmbeddedAgentRunActive.mockReturnValue(true); + mocks.isEmbeddedAgentRunHandleActive.mockReturnValue(hasEmbeddedHandle); + mocks.resolveEmbeddedAgentReplyRunPhase.mockReturnValue(phase); + mocks.getDiagnosticSessionActivitySnapshot.mockReturnValue({ lastProgressAgeMs: ageMs }); + mocks.abortEmbeddedAgentRun.mockReturnValue(true); + mocks.waitForEmbeddedAgentRunEnd.mockResolvedValue(true); + + const outcome = await recoverStuckDiagnosticSession({ + sessionId, + sessionKey: "agent:main:main", + ageMs, + queueDepth: 1, + allowActiveAbort, + staleActiveProgressAbortMs: 360_000, + compactionSafetyTimeoutMs: 900_000, + }); + + const withinCompactionSafetyWindow = ageMs < 915_000; + expect(outcome.status).toBe(withinCompactionSafetyWindow ? "skipped" : "aborted"); + expect(mocks.abortEmbeddedAgentRun).toHaveBeenCalledTimes( + withinCompactionSafetyWindow ? 0 : 1, + ); + }, + ); + it("keeps reply-only ownership with recent progress even with zero queued backlog", async () => { mocks.resolveActiveEmbeddedRunSessionId.mockReturnValue("live-reply-session"); mocks.resolveActiveEmbeddedRunHandleSessionId.mockReturnValue(undefined); diff --git a/src/logging/diagnostic-stuck-session-recovery.runtime.ts b/src/logging/diagnostic-stuck-session-recovery.runtime.ts index dc3f04e5ae71..8ae3db8e9865 100644 --- a/src/logging/diagnostic-stuck-session-recovery.runtime.ts +++ b/src/logging/diagnostic-stuck-session-recovery.runtime.ts @@ -180,14 +180,19 @@ export async function recoverStuckDiagnosticSession( const activeReplyPhase = activeWorkSessionId ? resolveEmbeddedAgentReplyRunPhase(activeWorkSessionId) : undefined; + const maintenancePhase = + activeReplyPhase === "preflight_compacting" || activeReplyPhase === "memory_flushing"; - if (activeReplyPhase === "waiting_for_global_lane") { - // A global-lane queue owner is healthy pending work. Reclaiming it here - // reintroduces the silent reply drop that the wait phase prevents. + if ( + activeReplyPhase === "waiting_for_global_lane" || + (maintenancePhase && params.ageMs < staleActiveLaneTaskReleaseMs) + ) { + // Queued replies and configured maintenance own their lane until their + // producer finishes or the existing compaction safety window expires. return reportRecoveryOutcome({ status: "skipped", action: "keep_lane", - reason: "global_lane_wait", + reason: maintenancePhase ? "active_reply_work" : "global_lane_wait", sessionId: params.sessionId, sessionKey: params.sessionKey, activeSessionId: activeWorkSessionId, @@ -257,18 +262,9 @@ export async function recoverStuckDiagnosticSession( sessionKey: params.sessionKey, queueDepth: params.queueDepth, staleAbortMs: staleActiveProgressAbortMs, - // Reply-only ownership must expire when proven stale even with zero - // queued backlog; the queue gate exists to protect run handles that - // are actively draining queued turns, and there is no such backlog - // here to protect. Recognized maintenance phases are the exception: - // preflight compaction and memory flush are explicitly allowed to - // run longer than the stale threshold (they honor a configured - // compaction timeout), so they keep the queue-backlog guard and are - // never force-cleared early by this reclaim path. - requireQueueBacklog: - activeReplyPhase === "preflight_compacting" || activeReplyPhase === "memory_flushing" - ? undefined - : false, + // Maintenance retains its backlog gate after the safety window; + // other abandoned reply ownership must expire even without a queue. + requireQueueBacklog: maintenancePhase ? undefined : false, }); if (params.allowActiveAbort === true || reclaimStaleReplyWork) { if (reclaimStaleReplyWork) { diff --git a/src/logging/diagnostic.test.ts b/src/logging/diagnostic.test.ts index b6b1b94e3d51..2c9d737a7308 100644 --- a/src/logging/diagnostic.test.ts +++ b/src/logging/diagnostic.test.ts @@ -1805,57 +1805,65 @@ describe("stuck session diagnostics threshold", () => { expect(s2Call!.allowActiveAbort).toBeUndefined(); }); - it("preserves queued idle work when abort reset releases active lane work", async () => { - const events: DiagnosticEventPayload[] = []; - const recoverStuckSession = vi.fn().mockResolvedValue({ + it.each([ + { status: "aborted", action: "abort_embedded_run", - sessionId: "s1", - sessionKey: "main", - activeSessionId: "s1", - activeWorkKind: "embedded_run", aborted: true, drained: false, forceCleared: true, - released: 1, - queuedCount: 1, - }); - const unsubscribe = onDiagnosticEvent((event) => { - events.push(event); - }); - try { - startDiagnosticHeartbeat( - { - diagnostics: { - enabled: true, + }, + { status: "released", action: "release_lane", reason: "stale_lane_task" }, + ] as const)( + "preserves queued idle work when $status recovery releases active lane work", + async (outcome) => { + const events: DiagnosticEventPayload[] = []; + const recoverStuckSession = vi.fn().mockResolvedValue({ + sessionId: "s1", + sessionKey: "main", + activeSessionId: "s1", + activeWorkKind: "embedded_run", + released: 1, + queuedCount: 1, + ...outcome, + }); + const unsubscribe = onDiagnosticEvent((event) => { + events.push(event); + }); + try { + startDiagnosticHeartbeat( + { + diagnostics: { + enabled: true, + }, }, + { recoverStuckSession }, + ); + logSessionStateChange({ sessionId: "s1", sessionKey: "main", state: "processing" }); + markDiagnosticEmbeddedRunStarted({ sessionId: "s1", sessionKey: "main" }); + logSessionStateChange({ sessionId: "s1", sessionKey: "main", state: "idle" }); + + vi.advanceTimersByTime(59_000); + logMessageQueued({ sessionId: "s1", sessionKey: "main", source: "test-followup" }); + vi.advanceTimersByTime(1_000); + await Promise.resolve(); + } finally { + unsubscribe(); + } + + requireMatchingRecord( + events, + { + type: "session.state", + state: "idle", + reason: `stuck_recovery:${outcome.status}`, + queueDepth: 1, }, - { recoverStuckSession }, + `idle ${outcome.status} preserves queued work`, ); - logSessionStateChange({ sessionId: "s1", sessionKey: "main", state: "processing" }); - markDiagnosticEmbeddedRunStarted({ sessionId: "s1", sessionKey: "main" }); - logSessionStateChange({ sessionId: "s1", sessionKey: "main", state: "idle" }); - - vi.advanceTimersByTime(59_000); - logMessageQueued({ sessionId: "s1", sessionKey: "main", source: "test-followup" }); - vi.advanceTimersByTime(1_000); - await Promise.resolve(); - } finally { - unsubscribe(); - } - - requireMatchingRecord( - events, - { - type: "session.state", - state: "idle", - reason: "stuck_recovery:aborted", - queueDepth: 1, - }, - "idle abort preserves queued work", - ); - expect(getDiagnosticSessionState({ sessionId: "s1", sessionKey: "main" }).queueDepth).toBe(1); - }); + expect(getDiagnosticSessionState({ sessionId: "s1", sessionKey: "main" }).queueDepth).toBe(1); + }, + ); it("marks diagnostic session state idle only after a mutating recovery outcome", async () => { const events: DiagnosticEventPayload[] = []; diff --git a/test/scripts/plugins-assertions.test.ts b/test/scripts/plugins-assertions.test.ts index 62372d9c0c03..aee5ad71c308 100644 --- a/test/scripts/plugins-assertions.test.ts +++ b/test/scripts/plugins-assertions.test.ts @@ -1631,6 +1631,9 @@ ${command} const result = await runAssertionAsync(["clawhub-preflight"], { CLAWHUB_PLUGIN_ID: "openclaw-kitchen-sink-fixture", CLAWHUB_PLUGIN_SPEC: "clawhub:@openclaw/kitchen-sink", + NODE_OPTIONS: `--import=data:text/javascript,${encodeURIComponent( + "const response = await fetch(process.env.OPENCLAW_CLAWHUB_URL); globalThis.fetch = async () => response;", + )}`, OPENCLAW_CLAWHUB_URL: `http://127.0.0.1:${address.port}`, OPENCLAW_PLUGINS_E2E_CLAWHUB_PREFLIGHT_TIMEOUT_MS: "75", });