fix(diagnostics): recover orphaned session activity

Recover idle queued sessions whose diagnostic activity retained stale ownerless model or tool calls by classifying them as recoverable session.stuck after the usual recovery gates. Yield the event loop before stale session-lock process inspection so sync process lookup cannot monopolize lock contention paths.

Docs now describe the widened session.stuck telemetry contract for recoverable stale bookkeeping, including ownerless activity. Thanks @samuelsoaress.

Refs #84903.

Co-authored-by: samuelsoaress <samuelsoares177778@gmail.com>

docs/concepts/agent-loop.md

@@ -165,7 +165,7 @@ surfaces, while Codex native hooks remain a separate lower-level Codex mechanism
 - `agent.wait` default: 30s (just the wait). `timeoutMs` param overrides.
 - Agent runtime: `agents.defaults.timeoutSeconds` default 172800s (48 hours); enforced in `runEmbeddedPiAgent` abort timer.
 - Cron runtime: isolated agent-turn `timeoutSeconds` is owned by cron. The scheduler starts that timer when execution begins, aborts the underlying run at the configured deadline, then runs bounded cleanup before recording the timeout so a stale child session cannot keep the lane stuck.
-- Session liveness diagnostics: with diagnostics enabled, `diagnostics.stuckSessionWarnMs` classifies long `processing` sessions that have no observed reply, tool, status, block, or ACP progress. Active embedded runs, model calls, and tool calls report as `session.long_running`; active work with no recent progress reports as `session.stalled`; `session.stuck` is reserved for stale session bookkeeping with no active work. Stale session bookkeeping releases the affected session lane immediately; stalled embedded runs are abort-drained only after `diagnostics.stuckSessionAbortMs` (default: at least 5 minutes and 3x the warning threshold) so queued work can resume without cutting off merely slow runs. Recovery emits structured requested/completed outcomes, and diagnostic state is marked idle only if the same processing generation is still current. Repeated `session.stuck` diagnostics back off while the session remains unchanged.
+- Session liveness diagnostics: with diagnostics enabled, `diagnostics.stuckSessionWarnMs` classifies long `processing` sessions that have no observed reply, tool, status, block, or ACP progress. Active embedded runs, model calls, and tool calls report as `session.long_running`; active work with no recent progress reports as `session.stalled`; `session.stuck` is reserved for recoverable stale session bookkeeping, including idle queued sessions with stale ownerless model/tool activity. Stale session bookkeeping releases the affected session lane immediately after recovery gates pass; stalled embedded runs are abort-drained only after `diagnostics.stuckSessionAbortMs` (default: at least 5 minutes and 3x the warning threshold) so queued work can resume without cutting off merely slow runs. Recovery emits structured requested/completed outcomes, and diagnostic state is marked idle only if the same processing generation is still current. Repeated `session.stuck` diagnostics back off while the session remains unchanged.
 - Model idle timeout: OpenClaw aborts a model request when no response chunks arrive before the idle window. `models.providers.<id>.timeoutSeconds` extends this idle watchdog for slow local/self-hosted providers, but it is still bounded by any lower `agents.defaults.timeoutSeconds` or run-specific timeout because those control the whole agent run. Otherwise OpenClaw uses `agents.defaults.timeoutSeconds` when configured, capped at 120s by default. Cron-triggered runs with no explicit model or agent timeout disable the idle watchdog and rely on the cron outer timeout.
 - Provider HTTP request timeout: `models.providers.<id>.timeoutSeconds` applies to that provider's model HTTP fetches, including connect, headers, body, SDK request timeout, total guarded-fetch abort handling, and model stream idle watchdog. Use this for slow local/self-hosted providers such as Ollama before raising the whole agent runtime timeout, and keep the agent/runtime timeout at least as high when the model request needs to run longer.
 

docs/concepts/queue.md

@@ -126,7 +126,7 @@ keys.
 - If commands seem stuck, enable verbose logs and look for "queued for ...ms" lines to confirm the queue is draining.
 - If you need queue depth, enable verbose logs and watch for queue timing lines.
 - Codex app-server runs that accept a turn and then stop emitting progress are interrupted by the Codex adapter so the active session lane can release instead of waiting for the outer run timeout.
-- When diagnostics are enabled, sessions that remain in `processing` past `diagnostics.stuckSessionWarnMs` with no observed reply, tool, status, block, or ACP progress are classified by current activity. Active work logs as `session.long_running`; active work with no recent progress logs as `session.stalled`; `session.stuck` is reserved for stale session bookkeeping with no active work, and only that path can release the affected session lane so queued work drains. Repeated `session.stuck` diagnostics back off while the session remains unchanged.
+- When diagnostics are enabled, sessions that remain in `processing` past `diagnostics.stuckSessionWarnMs` with no observed reply, tool, status, block, or ACP progress are classified by current activity. Active work logs as `session.long_running`; active work with no recent progress logs as `session.stalled`; `session.stuck` is reserved for recoverable stale session bookkeeping, including idle queued sessions with stale ownerless model/tool activity, and only that path can release the affected session lane so queued work drains. Repeated `session.stuck` diagnostics back off while the session remains unchanged.
 
 ## Related
 

docs/gateway/opentelemetry.md

@@ -223,8 +223,8 @@ message bodies are also approved for export.
 - `openclaw.queue.depth` (histogram, attrs: `openclaw.lane` or `openclaw.channel=heartbeat`)
 - `openclaw.queue.wait_ms` (histogram, attrs: `openclaw.lane`)
 - `openclaw.session.state` (counter, attrs: `openclaw.state`, `openclaw.reason`)
-- `openclaw.session.stuck` (counter, attrs: `openclaw.state`; emitted only for stale session bookkeeping with no active work)
-- `openclaw.session.stuck_age_ms` (histogram, attrs: `openclaw.state`; emitted only for stale session bookkeeping with no active work)
+- `openclaw.session.stuck` (counter, attrs: `openclaw.state`; emitted for recoverable stale session bookkeeping)
+- `openclaw.session.stuck_age_ms` (histogram, attrs: `openclaw.state`; emitted for recoverable stale session bookkeeping)
 - `openclaw.session.turn.created` (counter, attrs: `openclaw.agent`, `openclaw.channel`, `openclaw.trigger`)
 - `openclaw.session.recovery.requested` (counter, attrs: `openclaw.state`, `openclaw.action`, `openclaw.active_work_kind`, `openclaw.reason`)
 - `openclaw.session.recovery.completed` (counter, attrs: `openclaw.state`, `openclaw.action`, `openclaw.status`, `openclaw.active_work_kind`, `openclaw.reason`)
@@ -249,8 +249,9 @@ OpenClaw classifies sessions by the work it can still observe:
   turns behind the lane can resume. When unset, the abort threshold defaults to
   the safer extended window of at least 5 minutes and 3x
   `diagnostics.stuckSessionWarnMs`.
-- `session.stuck`: stale session bookkeeping with no active work. This releases
-  the affected session lane immediately.
+- `session.stuck`: stale session bookkeeping with no active work, or an idle
+  queued session with stale ownerless model/tool activity. This releases the
+  affected session lane immediately after recovery gates pass.
 
 Recovery emits structured `session.recovery.requested` and
 `session.recovery.completed` events. Diagnostic session state is marked idle

src/agents/session-write-lock.ts

@@ -42,6 +42,14 @@ export const DEFAULT_SESSION_WRITE_LOCK_MAX_HOLD_MS = 5 * 60 * 1000;
 export const DEFAULT_SESSION_WRITE_LOCK_ACQUIRE_TIMEOUT_MS = 60_000;
 const DEFAULT_WATCHDOG_INTERVAL_MS = 60_000;
 const DEFAULT_TIMEOUT_GRACE_MS = 2 * 60 * 1000;
+
+/**
+ * Yield control to the event loop so other sessions can make progress
+ * while lock contention callbacks run synchronous I/O.
+ */
+function yieldEventLoop(): Promise<void> {
+  return new Promise<void>((resolve) => setImmediate(resolve));
+}
 // A payload-less lock can be left behind if shutdown lands between open("wx")
 // and the owner metadata write. Keep the grace short so 10s callers recover.
 const ORPHAN_LOCK_PAYLOAD_GRACE_MS = 5_000;
@@ -768,6 +776,9 @@ export async function acquireSessionWriteLock(params: {
           return lockPayload as Record<string, unknown>;
         },
         shouldReclaim: async ({ payload, nowMs, heldByThisProcess }) => {
+          // Yield to the event loop before synchronous process inspection
+          // to prevent lock contention retries from starving other sessions.
+          await yieldEventLoop();
           const inspected = inspectLockPayloadForSession({
             payload: payload as LockFilePayload | null,
             staleMs,
@@ -780,6 +791,7 @@ export async function acquireSessionWriteLock(params: {
           return await shouldReclaimContendedLockFile(lockPath, inspected, staleMs, nowMs);
         },
         shouldRemoveStaleLock: async ({ lockPath, normalizedTargetPath, payload }) => {
+          await yieldEventLoop();
           const nowMs = Date.now();
           const heldByThisProcess = sessionLockHeldByThisProcess(normalizedTargetPath);
           const inspected = inspectLockPayloadForSession({

src/logging/diagnostic-session-attention.test.ts

@@ -102,9 +102,62 @@ describe("classifySessionAttention", () => {
         recoveryEligible: false,
       },
     },
-  ])("$name", ({ activity, expected, queueDepth }) => {
+    {
+      name: "idle queued stale model activity without active embedded run",
+      state: "idle" as const,
+      queueDepth: 1,
+      activity: {
+        activeWorkKind: "model_call" as const,
+        hasActiveEmbeddedRun: false,
+        lastProgressAgeMs: 31_000,
+        lastProgressReason: "model_call:started",
+      },
+      expected: {
+        eventType: "session.stuck",
+        reason: "queued_work_without_active_run",
+        classification: "stale_session_state",
+        recoveryEligible: true,
+      },
+    },
+    {
+      name: "idle queued stale tool_call activity without active embedded run",
+      state: "idle" as const,
+      queueDepth: 1,
+      activity: {
+        activeWorkKind: "tool_call" as const,
+        hasActiveEmbeddedRun: false,
+        activeToolAgeMs: 31_000,
+        lastProgressAgeMs: 31_000,
+        lastProgressReason: "tool:shell:started",
+      },
+      expected: {
+        eventType: "session.stuck",
+        reason: "queued_work_without_active_run",
+        classification: "stale_session_state",
+        recoveryEligible: true,
+      },
+    },
+    {
+      name: "processing session with orphaned activity is not recoverable",
+      state: "processing" as const,
+      queueDepth: 1,
+      activity: {
+        activeWorkKind: "model_call" as const,
+        hasActiveEmbeddedRun: false,
+        lastProgressAgeMs: 31_000,
+      },
+      expected: {
+        eventType: "session.stalled",
+        reason: "active_work_without_progress",
+        classification: "stalled_agent_run",
+        activeWorkKind: "model_call",
+        recoveryEligible: false,
+      },
+    },
+  ])("$name", ({ activity, expected, queueDepth, state }) => {
     expect(
       classifySessionAttention({
+        state,
         queueDepth,
         activity,
         staleMs: 30_000,

src/logging/diagnostic-session-attention.ts

@@ -25,11 +25,29 @@ export type SessionAttentionClassification =
     };
 
 export function classifySessionAttention(params: {
+  state?: "idle" | "processing" | "waiting";
   queueDepth: number;
   activity: DiagnosticSessionActivitySnapshot;
   staleMs: number;
 }): SessionAttentionClassification {
   if (params.activity.activeWorkKind) {
+    // Idle session with queued work and stale orphaned activity (no active
+    // embedded owner) should be classified as recoverable stuck state, not as
+    // stalled active work. This prevents orphaned model_call or tool_call
+    // activity from blocking the queue indefinitely.
+    if (
+      params.state === "idle" &&
+      params.queueDepth > 0 &&
+      params.activity.hasActiveEmbeddedRun !== true &&
+      (params.activity.lastProgressAgeMs ?? 0) > params.staleMs
+    ) {
+      return {
+        eventType: "session.stuck",
+        reason: "queued_work_without_active_run",
+        classification: "stale_session_state",
+        recoveryEligible: true,
+      };
+    }
     if (
       params.activity.activeWorkKind === "tool_call" &&
       (params.activity.activeToolAgeMs ?? 0) > params.staleMs &&

src/logging/diagnostic-stuck-session-recovery.runtime.test.ts

@@ -552,6 +552,58 @@ describe("stuck session recovery", () => {
     ]);
   });
 
+  it("releases idle queued work without aborting when stale activity has no active owner", async () => {
+    mocks.resolveActiveEmbeddedRunHandleSessionId.mockReturnValue(undefined);
+    mocks.resolveActiveEmbeddedRunSessionId.mockReturnValue(undefined);
+    mocks.isEmbeddedPiRunActive.mockReturnValue(false);
+    mocks.resetCommandLane.mockReturnValue(0);
+
+    const outcome = await recoverStuckDiagnosticSession({
+      sessionId: "idle-stale-model-session",
+      sessionKey: "agent:main:main",
+      ageMs: 180_000,
+      queueDepth: 1,
+      expectedState: "idle",
+    });
+
+    expect(outcome).toMatchObject({
+      status: "released",
+      action: "release_lane",
+      sessionId: "idle-stale-model-session",
+      sessionKey: "agent:main:main",
+      released: 0,
+    });
+    expect(mocks.abortEmbeddedPiRun).not.toHaveBeenCalled();
+    expect(mocks.forceClearEmbeddedPiRun).not.toHaveBeenCalled();
+    expect(mocks.resetCommandLane).toHaveBeenCalledWith("session:agent:main:main");
+  });
+
+  it("releases idle queued work with orphaned tool_call without aborting active work", async () => {
+    mocks.resolveActiveEmbeddedRunHandleSessionId.mockReturnValue(undefined);
+    mocks.resolveActiveEmbeddedRunSessionId.mockReturnValue(undefined);
+    mocks.isEmbeddedPiRunActive.mockReturnValue(false);
+    mocks.resetCommandLane.mockReturnValue(1);
+
+    const outcome = await recoverStuckDiagnosticSession({
+      sessionId: "idle-stale-tool-session",
+      sessionKey: "agent:sub:tool-runner",
+      ageMs: 180_000,
+      queueDepth: 2,
+      expectedState: "idle",
+    });
+
+    expect(outcome).toMatchObject({
+      status: "released",
+      action: "release_lane",
+      sessionId: "idle-stale-tool-session",
+      sessionKey: "agent:sub:tool-runner",
+      released: 1,
+    });
+    expect(mocks.abortEmbeddedPiRun).not.toHaveBeenCalled();
+    expect(mocks.forceClearEmbeddedPiRun).not.toHaveBeenCalled();
+    expect(mocks.resetCommandLane).toHaveBeenCalledWith("session:agent:sub:tool-runner");
+  });
+
   it("releases a stale session-id lane when no session key is available", async () => {
     mocks.isEmbeddedPiRunHandleActive.mockReturnValue(false);
     mocks.resetCommandLane.mockReturnValue(1);

src/logging/diagnostic.test.ts

@@ -1008,6 +1008,200 @@ describe("stuck session diagnostics threshold", () => {
     );
   });
 
+  it("recovers idle queued work blocked by stale model activity without active ownership", async () => {
+    const events: DiagnosticEventPayload[] = [];
+    const recoverStuckSession = vi.fn().mockResolvedValue({
+      status: "released",
+      action: "release_lane",
+      sessionId: "s1",
+      sessionKey: "main",
+      released: 0,
+    });
+    const unsubscribe = onDiagnosticEvent((event) => {
+      events.push(event);
+    });
+    try {
+      startDiagnosticHeartbeat(
+        {
+          diagnostics: {
+            enabled: true,
+            stuckSessionWarnMs: 30_000,
+            stuckSessionAbortMs: 60_000,
+          },
+        },
+        { recoverStuckSession },
+      );
+      logSessionStateChange({ sessionId: "s1", sessionKey: "main", state: "processing" });
+      markDiagnosticModelStartedForTest({
+        sessionId: "s1",
+        sessionKey: "main",
+        runId: "run-1",
+        provider: "openai",
+        model: "gpt-5",
+      });
+      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();
+    }
+
+    expectRecordFields(
+      requireRecord(
+        events.findLast((event) => event.type === "session.stuck"),
+        "idle stale model activity event",
+      ),
+      {
+        type: "session.stuck",
+        state: "idle",
+        classification: "stale_session_state",
+        reason: "queued_work_without_active_run",
+        queueDepth: 1,
+        lastProgressReason: "model_call:started",
+      },
+    );
+    expectRecoveryCall(
+      recoverStuckSession,
+      {
+        sessionId: "s1",
+        sessionKey: "main",
+        queueDepth: 1,
+        expectedState: "idle",
+      },
+      ["ageMs", "stateGeneration"],
+    );
+    const recoveryParams = requireFirstMockCallArg(recoverStuckSession, "recoverStuckSession");
+    expect(recoveryParams.allowActiveAbort).toBeUndefined();
+  });
+
+  it("recovers idle queued work blocked by stale orphaned tool_call activity", async () => {
+    const events: DiagnosticEventPayload[] = [];
+    const recoverStuckSession = vi.fn().mockResolvedValue({
+      status: "released",
+      action: "release_lane",
+      sessionId: "s1",
+      sessionKey: "main",
+      released: 0,
+    });
+    const unsubscribe = onDiagnosticEvent((event) => {
+      events.push(event);
+    });
+    try {
+      startDiagnosticHeartbeat(
+        {
+          diagnostics: {
+            enabled: true,
+            stuckSessionWarnMs: 30_000,
+            stuckSessionAbortMs: 60_000,
+          },
+        },
+        { recoverStuckSession },
+      );
+      logSessionStateChange({ sessionId: "s1", sessionKey: "main", state: "processing" });
+      markDiagnosticToolStartedForTest({
+        sessionId: "s1",
+        sessionKey: "main",
+        runId: "run-1",
+        toolName: "shell",
+        toolCallId: "tc-1",
+      });
+      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();
+    }
+
+    expectRecordFields(
+      requireRecord(
+        events.findLast((event) => event.type === "session.stuck"),
+        "idle stale tool activity event",
+      ),
+      {
+        type: "session.stuck",
+        state: "idle",
+        classification: "stale_session_state",
+        reason: "queued_work_without_active_run",
+        queueDepth: 1,
+      },
+    );
+    const recoveryParams = requireFirstMockCallArg(recoverStuckSession, "recoverStuckSession");
+    expect(recoveryParams.expectedState).toBe("idle");
+    expect(recoveryParams.allowActiveAbort).toBeUndefined();
+  });
+
+  it("recovers multiple stalled sessions independently without cross-session interference", async () => {
+    const recoverStuckSession = vi.fn().mockImplementation((params: { sessionId: string }) =>
+      Promise.resolve({
+        status: "released",
+        action: "release_lane",
+        sessionId: params.sessionId,
+        sessionKey: params.sessionId === "s1" ? "agent-a" : "agent-b",
+        released: 0,
+      }),
+    );
+    startDiagnosticHeartbeat(
+      {
+        diagnostics: {
+          enabled: true,
+          stuckSessionWarnMs: 30_000,
+          stuckSessionAbortMs: 60_000,
+        },
+      },
+      { recoverStuckSession },
+    );
+
+    // Set up two independent sessions that both stall with orphaned model activity.
+    logSessionStateChange({ sessionId: "s1", sessionKey: "agent-a", state: "processing" });
+    markDiagnosticModelStartedForTest({
+      sessionId: "s1",
+      sessionKey: "agent-a",
+      runId: "run-a",
+      provider: "anthropic",
+      model: "claude-opus-4-6",
+    });
+    logSessionStateChange({ sessionId: "s1", sessionKey: "agent-a", state: "idle" });
+
+    logSessionStateChange({ sessionId: "s2", sessionKey: "agent-b", state: "processing" });
+    markDiagnosticModelStartedForTest({
+      sessionId: "s2",
+      sessionKey: "agent-b",
+      runId: "run-b",
+      provider: "openai",
+      model: "gpt-5.5",
+    });
+    logSessionStateChange({ sessionId: "s2", sessionKey: "agent-b", state: "idle" });
+
+    // Queue work on both sessions.
+    vi.advanceTimersByTime(59_000);
+    logMessageQueued({ sessionId: "s1", sessionKey: "agent-a", source: "user-a" });
+    logMessageQueued({ sessionId: "s2", sessionKey: "agent-b", source: "user-b" });
+    vi.advanceTimersByTime(1_000);
+    await Promise.resolve();
+
+    // Both sessions should get independent recovery calls.
+    expect(recoverStuckSession).toHaveBeenCalledTimes(2);
+    const calls = recoverStuckSession.mock.calls.map(
+      (c: unknown[]) => c[0] as Record<string, unknown>,
+    );
+    const s1Call = calls.find((c) => c.sessionId === "s1");
+    const s2Call = calls.find((c) => c.sessionId === "s2");
+    expect(s1Call).toBeDefined();
+    expect(s2Call).toBeDefined();
+    expect(s1Call!.sessionKey).toBe("agent-a");
+    expect(s2Call!.sessionKey).toBe("agent-b");
+    expect(s1Call!.expectedState).toBe("idle");
+    expect(s2Call!.expectedState).toBe("idle");
+    expect(s1Call!.allowActiveAbort).toBeUndefined();
+    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({

src/logging/diagnostic.ts

@@ -539,7 +539,7 @@ function isActiveAbortRecoveryEligible(params: {
   );
 }
 
-function isIdleQueuedEmbeddedRunStall(params: {
+function isIdleQueuedRecoverableSessionStall(params: {
   state: {
     state: SessionStateValue;
     queueDepth: number;
@@ -550,10 +550,14 @@ function isIdleQueuedEmbeddedRunStall(params: {
   const hasEmbeddedOwner =
     params.activity.activeWorkKind === "embedded_run" ||
     params.activity.hasActiveEmbeddedRun === true;
+  // Also detect orphaned activity (model_call or tool_call left behind
+  // without an active embedded owner) so recovery can pump the stale queue.
+  const hasOrphanedActivity =
+    params.activity.activeWorkKind !== undefined && params.activity.hasActiveEmbeddedRun !== true;
   return (
     params.state.state === "idle" &&
     params.state.queueDepth > 0 &&
-    hasEmbeddedOwner &&
+    (hasEmbeddedOwner || hasOrphanedActivity) &&
     (params.activity.lastProgressAgeMs ?? 0) > params.staleMs
   );
 }
@@ -973,6 +977,7 @@ export function logSessionAttention(
     Date.now(),
   );
   const classification = classifySessionAttention({
+    state: state.state as "idle" | "processing" | "waiting" | undefined,
     queueDepth: state.queueDepth,
     activity,
     staleMs: params.thresholdMs,
@@ -1230,16 +1235,16 @@ export function startDiagnosticHeartbeat(
         { sessionId: state.sessionId, sessionKey: state.sessionKey },
         now,
       );
-      const idleQueuedEmbeddedRunStall = isIdleQueuedEmbeddedRunStall({
+      const idleQueuedRecoverableStall = isIdleQueuedRecoverableSessionStall({
         state,
         activity,
         staleMs: stuckSessionWarnMs,
       });
       if (
         (state.state === "processing" && ageMs > stuckSessionWarnMs) ||
-        idleQueuedEmbeddedRunStall
+        idleQueuedRecoverableStall
       ) {
-        const attentionAgeMs = idleQueuedEmbeddedRunStall
+        const attentionAgeMs = idleQueuedRecoverableStall
           ? (activity.lastProgressAgeMs ?? ageMs)
           : ageMs;
         const classification = logSessionAttention({