diff --git a/docs/concepts/agent-loop.md b/docs/concepts/agent-loop.md index 0b7d6fc945d7..88878b355653 100644 --- a/docs/concepts/agent-loop.md +++ b/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..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..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. diff --git a/docs/concepts/queue.md b/docs/concepts/queue.md index 59789ffeb71a..df3165087944 100644 --- a/docs/concepts/queue.md +++ b/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 diff --git a/docs/gateway/opentelemetry.md b/docs/gateway/opentelemetry.md index 57d6f3fbe704..c5b6d97c10a3 100644 --- a/docs/gateway/opentelemetry.md +++ b/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 diff --git a/src/agents/session-write-lock.ts b/src/agents/session-write-lock.ts index 60f521e94b52..76b08fa190b5 100644 --- a/src/agents/session-write-lock.ts +++ b/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 { + return new Promise((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; }, 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({ diff --git a/src/logging/diagnostic-session-attention.test.ts b/src/logging/diagnostic-session-attention.test.ts index 323754589509..8e127f5bbfb8 100644 --- a/src/logging/diagnostic-session-attention.test.ts +++ b/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, diff --git a/src/logging/diagnostic-session-attention.ts b/src/logging/diagnostic-session-attention.ts index d5a4c5d7335c..1012849467fe 100644 --- a/src/logging/diagnostic-session-attention.ts +++ b/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 && diff --git a/src/logging/diagnostic-stuck-session-recovery.runtime.test.ts b/src/logging/diagnostic-stuck-session-recovery.runtime.test.ts index 5ce3adc57f9c..0dc8ca571dc8 100644 --- a/src/logging/diagnostic-stuck-session-recovery.runtime.test.ts +++ b/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); diff --git a/src/logging/diagnostic.test.ts b/src/logging/diagnostic.test.ts index 6e4c14de8731..c6a502145b02 100644 --- a/src/logging/diagnostic.test.ts +++ b/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, + ); + 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({ diff --git a/src/logging/diagnostic.ts b/src/logging/diagnostic.ts index a7169a1b7739..e0a1b5e8b0f5 100644 --- a/src/logging/diagnostic.ts +++ b/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({