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

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<void | StuckSessionRecoveryOutcome>,
 ): value is Promise<void | StuckSessionRecoveryOutcome> {
@@ -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<StuckSessionRecoveryOutcome | undefined> {
-  const inFlightKey = recoveryRequestKey(params.request);
+  const inFlightKey = resolveStuckSessionRecoveryRef(params.request);
   if (inFlightKey && recoveryRequestsInFlight.has(inFlightKey)) {
     const outcome: StuckSessionRecoveryOutcome = {
       status: "skipped",

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
   );
 }
 

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<void>((resolve) => {
+        markActiveStarted = resolve;
+      });
+      const active = enqueueCommandInLane(
+        lane,
+        () =>
+          new Promise<void>((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";

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);

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) {

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[] = [];

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",
       });