diff --git a/TRACK_B1_BRIEF.md b/TRACK_B1_BRIEF.md new file mode 100644 index 000000000000..2207cacf2809 --- /dev/null +++ b/TRACK_B1_BRIEF.md @@ -0,0 +1,70 @@ +# Track B1+B4+B5: truthful run liveness — delete fake evidence, honest recovery attribution, evidence-aged steer capture + +Design issue: https://github.com/openclaw/openclaw/issues/101863 (Track B design comment). This branch is stacked on `fix/reply-run-terminal-release` (PR #101910, Track A) and uses its primitives: `ReplyOperation.lastActivityAtMs`, `recordActivity()`, `expireStaleReplyOperation()`, failure code `"run_stalled"`, constant `REPLY_RUN_STALE_TAKEOVER_MS` — all in `src/auto-reply/reply/reply-run-registry.ts`. This brief is authoritative for scope. + +## B1 — delete the CLI timer heartbeat; phase-aware tool-stall floor + +Problem (verified): `startClaudeLiveActiveToolHeartbeat` (`src/agents/cli-runner/claude-live-session.ts:551-566`) emits `cli_live:tool_running` on a 10s `setInterval` while a tool is merely _marked_ active (`turn.activeTools.size > 0`). That flows via `emitClaudeLiveProgress` → `run.progress` → `touchSessionActivity` (`src/logging/diagnostic-run-activity.ts:185`), resetting `lastProgressAt` with zero evidence the CLI child is alive. Consequence (#96168): the `blocked_tool_call` classifier branch (`src/logging/diagnostic-session-attention.ts:55-67`) — which correctly requires BOTH `activeToolAgeMs > staleMs` AND `lastProgressAgeMs > staleMs` — can never fire for a wedged CLI tool, and neither can any stuck-recovery abort gate keyed on `lastProgressAgeMs`. + +Fix: + +1. Delete the heartbeat entirely: `startClaudeLiveActiveToolHeartbeat`, `CLAUDE_LIVE_ACTIVE_TOOL_PROGRESS_MS` (`claude-live-session.ts:112`), the start call (~`:598`), the stop calls (~`:568`, `:654`), turn-cleanup clearing (~`:390`), and the `cli_live:tool_running` reason string. Real frames already stamp progress: tool start (`:597`), tool result (`:612`, `:653`), terminal result (`:711`), and any stdout frame via `noteClaudeLiveProgress` (`:705`, `:717`). Do NOT touch the byte-level no-output watchdog (`resetNoOutputTimer` `:720-738`) — it is evidence-based and correct. +2. Deleting the fake stamp re-exposes the #88870 hazard: a legitimately quiet long tool would become abort-eligible at the default `stuckSessionAbortMs` (~6 min). Add a tool-phase abort floor in `src/logging/diagnostic.ts`: + - New constant next to the other thresholds (`diagnostic.ts:80-85`): `BLOCKED_TOOL_CALL_ABORT_FLOOR_MS = 15 * 60_000`, with a 2-3 line contract comment: quiet-but-alive tools are normal agent behavior; the CLI byte watchdog kills truly-silent children at ≤600s; this floor only governs diagnostic recovery aborts for chatty-but-stuck turns; too low re-creates #88870, removal re-creates #96168's motivation for fake heartbeats. + - Apply in `isBlockedToolCallRecoveryEligible` (`diagnostic.ts:514`): eligibility requires `lastProgressAgeMs >= Math.max(stuckSessionAbortMs, BLOCKED_TOOL_CALL_ABORT_FLOOR_MS)` (only for the tool_call kind — do not change model_call or embedded_run gates). + - The warn-level classification (`classifySessionAttention` blocked_tool_call branch) keeps current thresholds — early warning naming the tool is good UX and cheap. +3. Delete tests asserting the removed heartbeat behavior (they protect a removed path); update wedged-tool tests so a stuck tool with no frames becomes stale → classified → recovery-eligible after the floor. + +## B4 — honest recovery attribution + stall-proof heartbeat math + +Problem 1 (verified): a diagnostic stuck-recovery abort of a reply-backed run routes `abortAndDrainEmbeddedAgentRun` → `abortReplyRunBySessionId` (`reply-run-registry.ts:856`) → `operation.abortByUser()` (`:664`) which stamps `abortedCode: "aborted_by_user"` (`:671`). Watchdog kills surface to the user as their own abort (#88870's misattribution). + +Fix: recovery is a staleness expiry, and Track A already built the primitive. In `src/logging/diagnostic-stuck-session-recovery.runtime.ts` (and the runs.ts abort leg it calls), when recovery decides to abort/force-clear a **reply-registry-owned** operation, route it through `expireStaleReplyOperation(operation, ...)` instead of `abortByUser()`/plain `fail("run_failed")`, so the terminal result is `{kind:"failed", code:"run_stalled"}` and the takeover log names the reason. Extend the `ReplyOperationStaleReason` union with `"stuck_recovery"` (closed union — check every consumer). For embedded (non-reply) runs, keep `abortAndDrainEmbeddedAgentRun` but ensure the abort reason threading does not produce a user-abort-shaped terminal outcome: check `resolveAgentRunAbortLifecycleFields` / terminal-outcome normalization (`src/agents/agent-run-terminal-outcome.ts`) and make the stuck-recovery abort distinguishable (a recovery-tagged abort error is acceptable; do not invent new lifecycle phases). Do not change the recovery module's detection logic beyond this routing. + +Problem 2 (verified, #101670): all staleness ages are raw `Date.now()` deltas; the 30s heartbeat (`diagnostic.ts:1211`) computes `ageMs = now - state.lastActivity` (`:1281`) after an event-loop stall and reads the stall as session staleness. + +Fix: the heartbeat records its own last tick timestamp. When a tick arrives later than ~3× the interval (>90s), that tick logs a warn (`liveness heartbeat delayed ms; deferring recovery decisions`) and performs NO recovery/abort scheduling — classification/warn logging may still run. The next on-time tick proceeds normally. One module-local variable, one guard, one contract comment (a delayed tick means the process stalled, not the sessions; acting on inflated ages aborts healthy runs). Test with fake timers. + +## B5 — steer capture refuses evidence-stale runs + +Problem (proven live in PR #101910's E2E): steer acceptance checks only streaming/stopped state, so a wedged run swallows its own rescue messages before reply admission can evaluate Track A's stale takeover. + +Fix, two gates: + +1. `queueReplyRunMessage` (`reply-run-registry.ts:839-854`): refuse (return false) when `Date.now() - operation.lastActivityAtMs > REPLY_RUN_STALE_TAKEOVER_MS`. A refused caller already falls back to normal followup queueing → admission → Track A reclaim. Contract comment: steering into an evidence-dead run swallows the human message that would otherwise trigger stale takeover. +2. `prepareEmbeddedAgentQueueMessage` (`src/agents/embedded-agent-runner/runs.ts:437-499`): after the injectable check, refuse when the session's diagnostic activity evidence is stale: use `getDiagnosticSessionActivitySnapshot` (`src/logging/diagnostic-run-activity.ts:572`) `lastProgressAgeMs` (when available) `>` a local `EMBEDDED_STEER_STALE_CAPTURE_MS = 10 * 60_000` constant (src/agents must not import src/auto-reply; duplicate the value with a comment naming `REPLY_RUN_STALE_TAKEOVER_MS` as the paired constant). New closed failure reason `"stale_run"` added to `EmbeddedAgentQueueFailureReason` (`runs.ts:56-62`) — check every consumer of that union handles it (sessions-send-tool.ts, subagent-announce-delivery.ts, etc.; they should treat it like `no_active_run`/`not_streaming` fall-through). If no diagnostic snapshot exists (diagnostics disabled), skip the gate (status quo). + +- Non-goal: draining already-captured steer entries from a dead handle (Track C durability). + +## Non-goals (do not touch) + +- Codex app-server watches (`extensions/codex`) — separate Track B2 PR. +- `llm-idle-timeout.ts` / `timeout.ts` semantics — separate Track B3 PR. +- No new config keys or env vars; constants only. `stuckSessionWarnMs`/`stuckSessionAbortMs` keep their meaning. +- No changes to the CLI byte-level no-output watchdog or supervisor. +- No user-visible message wording work beyond what attribution requires (Track D owns messaging). + +## Repo conventions + +- TS ESM strict, no `any`, closed unions/codes. Comments: 1-3 lines, contract + bad outcome if removed, at lifecycle/threshold/ownership points only. +- Tests colocated, Vitest, fake timers cleaned up, `--isolate=false` safe. Delete tests of removed paths rather than porting them. +- Keep LOC tight; B1 should be clearly net-negative (a whole heartbeat mechanism disappears). + +## Required tests + +1. CLI wedged-tool: tool started, no further frames → `lastProgressAgeMs` grows (no timer stamps exist anymore), `blocked_tool_call` classification fires, recovery becomes eligible only after `max(stuckSessionAbortMs, BLOCKED_TOOL_CALL_ABORT_FLOOR_MS)`. +2. CLI active-tool with real frames (tool result / stream frames arriving) → never classified stalled. +3. Recovery abort of a reply-owned stale run → operation result `{failed, run_stalled}`, NOT `aborted_by_user`; user-abort path still yields `aborted_by_user`. +4. Delayed heartbeat tick (fake timers, simulate >90s gap) → no recovery scheduled that tick, warn logged, next tick recovers normally. +5. `queueReplyRunMessage` refuses when `lastActivityAtMs` stale, accepts when fresh (recordActivity just called). +6. `prepareEmbeddedAgentQueueMessage` returns `"stale_run"` when diagnostic evidence stale; accepts when fresh; skips gate when no snapshot. +7. Consumers of `EmbeddedAgentQueueFailureReason` handle `"stale_run"` (compile-time exhaustiveness + behavior fall-through where tested). + +## Validation before you commit + +- `node scripts/run-vitest.mjs src/logging src/agents/cli-runner src/auto-reply/reply src/agents/embedded-agent-runner` relevant test files (start with the directly touched test files, then the import sweep below). Never bare `pnpm test`/`vitest`. +- Import sweep: `rg -l 'claude-live-session|diagnostic-run-activity|diagnostic-session-attention|stuck-session-recovery|queueEmbeddedAgentMessage|queueReplyRunMessage|EmbeddedAgentQueueFailureReason' --glob '*.test.ts' src test` and run those files. +- Typecheck: `node scripts/run-tsgo.mjs -p tsconfig.core.json --incremental --tsBuildInfoFile .artifacts/tsgo-cache/core.tsbuildinfo` and the core-test lane; check exit codes directly, no pipes. +- Format touched files with `node_modules/.bin/oxfmt`; lint with `node scripts/run-oxlint.mjs `. + +Commit on this branch with a conventional message (e.g. `fix(agents): replace fake CLI liveness with evidence-based stall detection`). Do not push. Do not open a PR. Write design decisions + test results to `TRACK_B1_NOTES.md` at the worktree root. diff --git a/src/agents/cli-runner.spawn.test.ts b/src/agents/cli-runner.spawn.test.ts index 36256b0e0c34..b3e7a1d6a180 100644 --- a/src/agents/cli-runner.spawn.test.ts +++ b/src/agents/cli-runner.spawn.test.ts @@ -17,6 +17,7 @@ import { onAgentEvent, resetAgentEventsForTest } from "../infra/agent-events.js" import { onInternalDiagnosticEvent, onTrustedToolExecutionEvent, + setDiagnosticsEnabledForProcess, waitForDiagnosticEventsDrained, } from "../infra/diagnostic-events.js"; import { @@ -66,6 +67,7 @@ type ProcessSupervisor = ReturnType; type SupervisorSpawnFn = ProcessSupervisor["spawn"]; beforeEach(() => { + setDiagnosticsEnabledForProcess(true); resetAgentEventsForTest(); resetDiagnosticRunActivityForTest(); resetClaudeLiveSessionsForTest(); @@ -2142,7 +2144,7 @@ ${JSON.stringify({ expect(parsed.response.response.updatedInput).toEqual({ command: "ls" }); }); - it("reports Claude live stream progress and keeps native tools fresh while they are running", async () => { + it("reports Claude live stream progress without timer heartbeats", async () => { vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], }); @@ -2250,11 +2252,11 @@ ${JSON.stringify({ expect( getDiagnosticSessionActivitySnapshot({ sessionKey: "agent:main:diagnostics" }) .lastProgressReason, - ).toBe("cli_live:tool_running"); + ).toBe("cli_live:tool_started"); expect( getDiagnosticSessionActivitySnapshot({ sessionKey: "agent:main:diagnostics" }) .lastProgressAgeMs, - ).toBeLessThan(100); + ).toBeGreaterThanOrEqual(10_000); stdoutListener?.( [ diff --git a/src/agents/cli-runner/claude-live-session.ts b/src/agents/cli-runner/claude-live-session.ts index 2b47e25e9ade..f0c7cc1acb10 100644 --- a/src/agents/cli-runner/claude-live-session.ts +++ b/src/agents/cli-runner/claude-live-session.ts @@ -58,7 +58,6 @@ type ClaudeLiveTurn = { sessionId?: string; noOutputTimer: NodeJS.Timeout | null; timeoutTimer: NodeJS.Timeout | null; - activeToolTimer: NodeJS.Timeout | null; activeTools: Map; observedStdout: boolean; completedToolCallIds: Set; @@ -109,7 +108,6 @@ type ClaudeLiveToolTerminalOutcome = | { outcome: "blocked"; deniedReason: string; reason?: string } | { outcome: "cancelled" | "failed" | "timed_out" | "unknown" }; const CLAUDE_LIVE_IDLE_TIMEOUT_MS = 10 * 60 * 1_000; -const CLAUDE_LIVE_ACTIVE_TOOL_PROGRESS_MS = 10_000; const CLAUDE_LIVE_MAX_SESSIONS = 16; const CLAUDE_LIVE_MAX_STDERR_CHARS = 64 * 1024; const CLAUDE_LIVE_CLOSE_WAIT_TIMEOUT_MS = 5_000; @@ -386,10 +384,6 @@ function clearTurnTimers(turn: ClaudeLiveTurn): void { clearTimeout(turn.timeoutTimer); turn.timeoutTimer = null; } - if (turn.activeToolTimer) { - clearInterval(turn.activeToolTimer); - turn.activeToolTimer = null; - } } function finishTurn(session: ClaudeLiveSession, output: CliOutput): void { @@ -548,31 +542,6 @@ function summarizeClaudeLiveToolInput(input: unknown): DiagnosticToolParamsSumma } } -function startClaudeLiveActiveToolHeartbeat(turn: ClaudeLiveTurn): void { - if (turn.activeToolTimer || turn.activeTools.size === 0) { - return; - } - turn.activeToolTimer = setInterval(() => { - if (turn.activeTools.size === 0) { - if (turn.activeToolTimer) { - clearInterval(turn.activeToolTimer); - turn.activeToolTimer = null; - } - return; - } - emitClaudeLiveProgress(turn, "cli_live:tool_running"); - }, CLAUDE_LIVE_ACTIVE_TOOL_PROGRESS_MS); - turn.activeToolTimer.unref?.(); -} - -function stopClaudeLiveActiveToolHeartbeatIfIdle(turn: ClaudeLiveTurn): void { - if (turn.activeTools.size > 0 || !turn.activeToolTimer) { - return; - } - clearInterval(turn.activeToolTimer); - turn.activeToolTimer = null; -} - function markClaudeLiveToolStarted(turn: ClaudeLiveTurn, tool: CliToolUseStartDelta): void { if (turn.completedToolCallIds.has(tool.toolCallId) || turn.activeTools.has(tool.toolCallId)) { return; @@ -595,7 +564,6 @@ function markClaudeLiveToolStarted(turn: ClaudeLiveTurn, tool: CliToolUseStartDe paramsSummary: summarizeClaudeLiveToolInput(tool.args), }); emitClaudeLiveProgress(turn, "cli_live:tool_started"); - startClaudeLiveActiveToolHeartbeat(turn); } function markClaudeLiveToolCompleted( @@ -651,7 +619,6 @@ function markClaudeLiveToolCompleted( }); } emitClaudeLiveProgress(turn, "cli_live:tool_result"); - stopClaudeLiveActiveToolHeartbeatIfIdle(turn); } function markClaudeLiveToolDenied(turn: ClaudeLiveTurn, tool: CliToolUseStartDelta): void { @@ -1170,7 +1137,6 @@ function createTurn(params: { rawChars: 0, noOutputTimer: null, timeoutTimer: null, - activeToolTimer: null, activeTools: new Map(), observedStdout: false, completedToolCallIds: new Set(), diff --git a/src/agents/embedded-agent-runner/runs.test.ts b/src/agents/embedded-agent-runner/runs.test.ts index c0129d54a197..8e29d3f6f6fb 100644 --- a/src/agents/embedded-agent-runner/runs.test.ts +++ b/src/agents/embedded-agent-runner/runs.test.ts @@ -11,6 +11,7 @@ import { isReplyRunActiveForSessionId, } from "../../auto-reply/reply/reply-run-registry.js"; import { setDiagnosticsEnabledForProcess } from "../../infra/diagnostic-events.js"; +import { resetDiagnosticRunActivityForTest } from "../../logging/diagnostic-run-activity.js"; import { getDiagnosticSessionState, resetDiagnosticSessionStateForTest, @@ -236,6 +237,32 @@ describe("embedded-agent runner run registry", () => { expect(abort).toHaveBeenCalledWith("restart"); }); + it("expires reply-owned stuck recovery as run_stalled instead of user abort", async () => { + const cancel = vi.fn(); + const operation = createReplyOperation({ + sessionKey: "agent:main:reply-stuck", + sessionId: "session-reply-stuck", + resetTriggered: false, + }); + operation.attachBackend({ + kind: "embedded", + cancel, + isStreaming: () => true, + }); + operation.setPhase("running"); + + const result = await abortAndDrainEmbeddedAgentRun({ + sessionId: "session-reply-stuck", + sessionKey: "agent:main:reply-stuck", + reason: "stuck_recovery", + forceClear: true, + }); + + expect(result).toEqual({ aborted: true, drained: true, forceCleared: false }); + expect(operation.result).toEqual({ kind: "failed", code: "run_stalled" }); + expect(cancel).toHaveBeenCalledWith("superseded"); + }); + it("claims shared restart ownership before invoking an attached handle", () => { const abort = vi.fn(); const handle = createRunHandle({ abort }); @@ -458,6 +485,53 @@ describe("embedded-agent runner run registry", () => { expect(queueMessage).toHaveBeenCalledWith("continue", { steeringMode: "all" }); }); + it("refuses embedded steering when diagnostic evidence is stale", () => { + vi.useFakeTimers(); + try { + const queueMessage = vi.fn(async () => {}); + setActiveEmbeddedRun("session-stale-steer", createRunHandle({ queueMessage })); + + vi.advanceTimersByTime(10 * 60_000 + 1); + + const outcome = queueEmbeddedAgentMessageWithOutcome("session-stale-steer", "continue"); + + expect(outcome).toEqual({ + queued: false, + sessionId: "session-stale-steer", + reason: "stale_run", + gatewayHealth: "live", + }); + expect(queueMessage).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }); + + it("accepts embedded steering with fresh or missing diagnostic evidence", () => { + const freshQueueMessage = vi.fn(async () => {}); + setActiveEmbeddedRun( + "session-fresh-steer", + createRunHandle({ queueMessage: freshQueueMessage }), + ); + + expect(queueEmbeddedAgentMessageWithOutcome("session-fresh-steer", "continue").queued).toBe( + true, + ); + expect(freshQueueMessage).toHaveBeenCalledWith("continue", { steeringMode: "all" }); + + const missingSnapshotQueueMessage = vi.fn(async () => {}); + setActiveEmbeddedRun( + "session-no-diagnostic-snapshot", + createRunHandle({ queueMessage: missingSnapshotQueueMessage }), + ); + resetDiagnosticRunActivityForTest(); + + expect( + queueEmbeddedAgentMessageWithOutcome("session-no-diagnostic-snapshot", "continue").queued, + ).toBe(true); + expect(missingSnapshotQueueMessage).toHaveBeenCalledWith("continue", { steeringMode: "all" }); + }); + it("does not queue into stopped handles", () => { const queueMessage = vi.fn(async () => {}); setActiveEmbeddedRun( diff --git a/src/agents/embedded-agent-runner/runs.ts b/src/agents/embedded-agent-runner/runs.ts index 96b3d4ec06a9..fb421f3834ee 100644 --- a/src/agents/embedded-agent-runner/runs.ts +++ b/src/agents/embedded-agent-runner/runs.ts @@ -4,6 +4,7 @@ import { abortActiveReplyRuns, abortReplyRunBySessionId, + expireStaleReplyRunBySessionId, forceClearReplyRunBySessionId, isReplyRunActiveForSessionId, isReplyRunAbortableForCompaction, @@ -13,6 +14,7 @@ import { waitForReplyRunEndBySessionId, } from "../../auto-reply/reply/reply-run-registry.js"; import { + getDiagnosticSessionActivitySnapshot, markDiagnosticEmbeddedRunEnded, markDiagnosticEmbeddedRunStarted, } from "../../logging/diagnostic-run-activity.js"; @@ -56,6 +58,7 @@ export { export type EmbeddedAgentQueueFailureReason = | "no_active_run" | "not_streaming" + | "stale_run" | "compacting" | "source_reply_delivery_mode_mismatch" | "transcript_commit_wait_unsupported" @@ -88,6 +91,10 @@ type PreparedEmbeddedAgentQueueMessage = handle: EmbeddedAgentQueueHandle; }; +// Paired with REPLY_RUN_STALE_TAKEOVER_MS in the reply registry; src/agents +// keeps its own constant to avoid importing auto-reply policy into this owner. +const EMBEDDED_STEER_STALE_CAPTURE_MS = 10 * 60_000; + function createQueueFailureOutcome( sessionId: string, reason: EmbeddedAgentQueueFailureReason, @@ -471,6 +478,14 @@ function prepareEmbeddedAgentQueueMessage( diag.debug(`queue message failed: sessionId=${sessionId} reason=not_streaming`); return { kind: "complete", outcome: createQueueFailureOutcome(sessionId, "not_streaming") }; } + const activity = getDiagnosticSessionActivitySnapshot({ sessionId }); + if ( + typeof activity.lastProgressAgeMs === "number" && + activity.lastProgressAgeMs > EMBEDDED_STEER_STALE_CAPTURE_MS + ) { + diag.debug(`queue message failed: sessionId=${sessionId} reason=stale_run`); + return { kind: "complete", outcome: createQueueFailureOutcome(sessionId, "stale_run") }; + } if (handle.isCompacting()) { diag.debug(`queue message failed: sessionId=${sessionId} reason=compacting`); return { kind: "complete", outcome: createQueueFailureOutcome(sessionId, "compacting") }; @@ -755,6 +770,19 @@ export async function abortAndDrainEmbeddedAgentRun(params: { reason?: string; }): Promise { const settleMs = params.settleMs ?? 15_000; + if ( + params.reason === "stuck_recovery" && + !ACTIVE_EMBEDDED_RUNS.has(params.sessionId) && + expireStaleReplyRunBySessionId(params.sessionId, "stuck_recovery") + ) { + // Reply expiry aborts synchronously and clears registry ownership. Let the + // command lane observe that abort before recovery decides whether to reset it. + await new Promise((resolve) => { + setImmediate(resolve); + }); + const drained = await waitForEmbeddedAgentRunEnd(params.sessionId, settleMs); + return { aborted: true, drained, forceCleared: false }; + } const aborted = abortEmbeddedAgentRun(params.sessionId); const drained = aborted ? await waitForEmbeddedAgentRunEnd(params.sessionId, settleMs) : false; const forceCleared = diff --git a/src/agents/tools/sessions-send-tool.ts b/src/agents/tools/sessions-send-tool.ts index 1f9ba2ae6b94..279b13f719aa 100644 --- a/src/agents/tools/sessions-send-tool.ts +++ b/src/agents/tools/sessions-send-tool.ts @@ -235,7 +235,10 @@ function shouldFallbackCronRunScopedActiveDelivery( outcome: EmbeddedAgentQueueMessageOutcome, ): boolean { return ( - !outcome.queued && (outcome.reason === "not_streaming" || outcome.reason === "no_active_run") + !outcome.queued && + (outcome.reason === "not_streaming" || + outcome.reason === "no_active_run" || + outcome.reason === "stale_run") ); } diff --git a/src/auto-reply/reply/reply-run-registry.test.ts b/src/auto-reply/reply/reply-run-registry.test.ts index 50e25fd873e1..09c7beb222ef 100644 --- a/src/auto-reply/reply/reply-run-registry.test.ts +++ b/src/auto-reply/reply/reply-run-registry.test.ts @@ -18,6 +18,7 @@ import { isReplyRunAbortableForSignal, queueReplyRunMessage, REPLY_RUN_IDLE_SETTLE_TIMEOUT_MS, + REPLY_RUN_STALE_TAKEOVER_MS, REPLY_RUN_TERMINAL_SETTLE_TIMEOUT_MS, replyRunRegistry, runAfterReplyOperationClear, @@ -849,6 +850,37 @@ describe("reply run registry", () => { expect(queueMessage).toHaveBeenCalledWith("hello"); }); + it("refuses stale reply-run steering until real activity resumes", () => { + vi.useFakeTimers(); + try { + const queueMessage = vi.fn(async () => {}); + const operation = createReplyOperation({ + sessionKey: "agent:main:main", + sessionId: "session-running", + resetTriggered: false, + }); + operation.attachBackend({ + kind: "embedded", + cancel: vi.fn(), + isStreaming: () => true, + queueMessage, + }); + operation.setPhase("running"); + + vi.advanceTimersByTime(REPLY_RUN_STALE_TAKEOVER_MS + 1); + + expect(queueReplyRunMessage("session-running", "stale")).toBe(false); + expect(queueMessage).not.toHaveBeenCalled(); + + operation.recordActivity(); + + expect(queueReplyRunMessage("session-running", "fresh")).toBe(true); + expect(queueMessage).toHaveBeenCalledWith("fresh"); + } finally { + vi.useRealTimers(); + } + }); + it("does not queue messages through stopped backends", () => { const queueMessage = vi.fn(async () => {}); const operation = createReplyOperation({ diff --git a/src/auto-reply/reply/reply-run-registry.ts b/src/auto-reply/reply/reply-run-registry.ts index 54c57a09e492..5a464bac0815 100644 --- a/src/auto-reply/reply/reply-run-registry.ts +++ b/src/auto-reply/reply/reply-run-registry.ts @@ -187,7 +187,7 @@ export const REPLY_RUN_TERMINAL_SETTLE_TIMEOUT_MS = 60_000; // Timers and user-message injection never refresh activity; agent events do. export const REPLY_RUN_STALE_TAKEOVER_MS = 10 * 60_000; -export type ReplyOperationStaleReason = "terminal_unreleased" | "no_activity"; +export type ReplyOperationStaleReason = "terminal_unreleased" | "no_activity" | "stuck_recovery"; export class ReplyRunAlreadyActiveError extends Error { constructor(sessionKey: string) { @@ -843,6 +843,14 @@ export function expireStaleReplyOperation( return expireReplyOperationByOperation.get(operation)?.(reason) ?? false; } +export function expireStaleReplyRunBySessionId( + sessionId: string, + reason: ReplyOperationStaleReason, +): boolean { + const operation = resolveReplyRunForCurrentSessionId(sessionId); + return operation ? expireStaleReplyOperation(operation, reason) : false; +} + export const replyRunRegistry: ReplyRunRegistry = { begin(params) { return createReplyOperation(params); @@ -969,6 +977,11 @@ export function queueReplyRunMessage( if (!operation || operation.phase !== "running" || !backend?.queueMessage) { return false; } + // Steering into an evidence-dead run swallows the human message that would + // otherwise trigger stale takeover through normal reply admission. + if (Date.now() - operation.lastActivityAtMs > REPLY_RUN_STALE_TAKEOVER_MS) { + return false; + } if (!isReplyBackendMessageInjectable(backend)) { return false; } diff --git a/src/logging/diagnostic.test.ts b/src/logging/diagnostic.test.ts index 6eca92ab6ba1..64121d973aa7 100644 --- a/src/logging/diagnostic.test.ts +++ b/src/logging/diagnostic.test.ts @@ -423,6 +423,37 @@ describe("stuck session diagnostics threshold", () => { ); }); + it("defers recovery on delayed heartbeat ticks and recovers on the next on-time tick", () => { + const recoverStuckSession = vi.fn(); + const warnSpy = vi.spyOn(diagnosticLogger, "warn").mockImplementation(() => undefined); + + vi.setSystemTime(0); + startDiagnosticHeartbeat( + { + diagnostics: { + enabled: true, + stuckSessionWarnMs: 30_000, + }, + }, + { recoverStuckSession }, + ); + logSessionStateChange({ sessionId: "s1", sessionKey: "main", state: "processing" }); + + vi.setSystemTime(120_001); + vi.advanceTimersByTime(30_000); + + expectLoggerMessageContaining(warnSpy, "liveness heartbeat delayed"); + expect(recoverStuckSession).not.toHaveBeenCalled(); + + vi.advanceTimersByTime(30_000); + + expectRecoveryCall( + recoverStuckSession, + { sessionId: "s1", sessionKey: "main", queueDepth: 0 }, + ["ageMs", "stateGeneration"], + ); + }); + it("does not warn while a processing session continues reporting progress", () => { const events: DiagnosticEventPayload[] = []; const unsubscribe = onDiagnosticEvent((event) => { @@ -747,10 +778,10 @@ describe("stuck session diagnostics threshold", () => { toolCallId: "cmd-1", }); - vi.advanceTimersByTime(stuckSessionAbortMs - 30_000); + vi.advanceTimersByTime(stuckSessionAbortMs); expect(recoverStuckSession).not.toHaveBeenCalled(); - vi.advanceTimersByTime(30_000); + vi.advanceTimersByTime(15 * 60_000 - stuckSessionAbortMs); } finally { unsubscribe(); } @@ -775,6 +806,51 @@ describe("stuck session diagnostics threshold", () => { ); }); + it("does not classify active tool calls stalled while real progress frames arrive", () => { + const events: DiagnosticEventPayload[] = []; + const recoverStuckSession = vi.fn(); + 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" }); + markDiagnosticEmbeddedRunStarted({ sessionId: "s1", sessionKey: "main" }); + markDiagnosticToolStartedForTest({ + sessionId: "s1", + sessionKey: "main", + runId: "run-1", + toolName: "bash", + toolCallId: "cmd-1", + }); + + for (let i = 0; i < 20; i += 1) { + vi.advanceTimersByTime(29_000); + markDiagnosticRunProgressForTest({ + sessionId: "s1", + sessionKey: "main", + runId: "run-1", + reason: "cli_live:stream_progress", + }); + vi.advanceTimersByTime(1_000); + } + } finally { + unsubscribe(); + } + + expect(events.some((event) => event.type === "session.stalled")).toBe(false); + expect(recoverStuckSession).not.toHaveBeenCalled(); + }); + it("recovers stale model calls through the active embedded-run abort path", async () => { const events: DiagnosticEventPayload[] = []; const recoverStuckSession = vi.fn(); @@ -1070,6 +1146,9 @@ describe("stuck session diagnostics threshold", () => { expect(recoverStuckSession).not.toHaveBeenCalled(); vi.advanceTimersByTime(30_000); + expect(recoverStuckSession).not.toHaveBeenCalled(); + + vi.advanceTimersByTime(15 * 60_000 - stuckSessionAbortMs); expectRecoveryCall( recoverStuckSession, { sessionId: "s1", sessionKey: "main", queueDepth: 0, allowActiveAbort: true }, diff --git a/src/logging/diagnostic.ts b/src/logging/diagnostic.ts index e301e363106b..59a0816d422f 100644 --- a/src/logging/diagnostic.ts +++ b/src/logging/diagnostic.ts @@ -82,11 +82,17 @@ const MIN_STUCK_SESSION_WARN_MS = 1_000; const MAX_STUCK_SESSION_WARN_MS = 24 * 60 * 60 * 1000; const MIN_STALLED_EMBEDDED_RUN_ABORT_MS = 5 * 60_000; const STALLED_EMBEDDED_RUN_ABORT_WARN_MULTIPLIER = 3; +// Quiet-but-alive tools are normal agent behavior; the CLI byte watchdog kills +// truly silent children within its own deadline. This floor only bounds +// diagnostic recovery aborts; lowering it reopens #88870, removing it reopens #96168. +const BLOCKED_TOOL_CALL_ABORT_FLOOR_MS = 15 * 60_000; const RECENT_DIAGNOSTIC_ACTIVITY_MS = 120_000; const DEFAULT_LIVENESS_EVENT_LOOP_DELAY_WARN_MS = 1_000; const DEFAULT_LIVENESS_EVENT_LOOP_UTILIZATION_WARN = 0.95; const DEFAULT_LIVENESS_CPU_CORE_RATIO_WARN = 0.9; const DEFAULT_LIVENESS_WARN_COOLDOWN_MS = 120_000; +const DIAGNOSTIC_HEARTBEAT_INTERVAL_MS = 30_000; +const DIAGNOSTIC_HEARTBEAT_DELAY_RECOVERY_SKIP_MS = 3 * DIAGNOSTIC_HEARTBEAT_INTERVAL_MS; const loadStuckSessionRecoveryRuntime = createLazyRuntimeModule( () => import("./diagnostic-stuck-session-recovery.runtime.js"), ); @@ -528,14 +534,15 @@ function isBlockedToolCallRecoveryEligible(params: { }): boolean { const toolAgeMs = params.activity?.activeToolAgeMs; const lastProgressAgeMs = params.activity?.lastProgressAgeMs; + const abortMs = Math.max(params.stuckSessionAbortMs, BLOCKED_TOOL_CALL_ABORT_FLOOR_MS); return ( params.classification?.eventType === "session.stalled" && params.classification.classification === "blocked_tool_call" && params.classification.activeWorkKind === "tool_call" && typeof toolAgeMs === "number" && typeof lastProgressAgeMs === "number" && - toolAgeMs >= params.stuckSessionAbortMs && - lastProgressAgeMs >= params.stuckSessionAbortMs + toolAgeMs >= abortMs && + lastProgressAgeMs >= abortMs ); } @@ -1202,6 +1209,7 @@ export function logActiveRuns() { } let heartbeatInterval: NodeJS.Timeout | null = null; +let lastDiagnosticHeartbeatTickAt: number | undefined; export function startDiagnosticHeartbeat( config?: OpenClawConfig, @@ -1218,6 +1226,7 @@ export function startDiagnosticHeartbeat( startDiagnosticLivenessSampler(); const livenessGraceUntil = opts?.startupGraceMs != null && opts.startupGraceMs > 0 ? Date.now() + opts.startupGraceMs : 0; + lastDiagnosticHeartbeatTickAt = Date.now(); heartbeatInterval = setInterval(() => { let heartbeatConfig = config; if (!heartbeatConfig) { @@ -1231,6 +1240,17 @@ export function startDiagnosticHeartbeat( const stuckSessionAbortMs = resolveStuckSessionAbortMs(heartbeatConfig, stuckSessionWarnMs); const compactionSafetyTimeoutMs = resolveCompactionTimeoutMs(heartbeatConfig); const now = Date.now(); + const tickDelayMs = + lastDiagnosticHeartbeatTickAt === undefined ? 0 : now - lastDiagnosticHeartbeatTickAt; + lastDiagnosticHeartbeatTickAt = now; + // A late interval tick means the process was stalled, not the sessions. + // Acting on inflated ages here can abort healthy runs; the next on-time tick decides. + const skipRecoveryThisTick = tickDelayMs > DIAGNOSTIC_HEARTBEAT_DELAY_RECOVERY_SKIP_MS; + if (skipRecoveryThisTick) { + diag.warn( + `liveness heartbeat delayed ${Math.round(tickDelayMs)}ms; deferring recovery decisions`, + ); + } pruneDiagnosticSessionStates(now, true); const work = getDiagnosticWorkSnapshot(now); const inStartupGrace = livenessGraceUntil > 0 && now < livenessGraceUntil; @@ -1313,7 +1333,7 @@ export function startDiagnosticHeartbeat( thresholdMs: stuckSessionWarnMs, abortThresholdMs: stuckSessionAbortMs, }); - if (classification?.recoveryEligible) { + if (classification?.recoveryEligible && !skipRecoveryThisTick) { requestStuckSessionRecovery({ recover: opts?.recoverStuckSession ?? recoverStuckSession, classification, @@ -1331,6 +1351,7 @@ export function startDiagnosticHeartbeat( }); } else if ( classification && + !skipRecoveryThisTick && isActiveAbortRecoveryEligible({ classification, activity, @@ -1355,7 +1376,7 @@ export function startDiagnosticHeartbeat( } } } - }, 30_000); + }, DIAGNOSTIC_HEARTBEAT_INTERVAL_MS); heartbeatInterval.unref?.(); } @@ -1364,6 +1385,7 @@ export function stopDiagnosticHeartbeat() { clearInterval(heartbeatInterval); heartbeatInterval = null; } + lastDiagnosticHeartbeatTickAt = undefined; stopDiagnosticLivenessSampler(); stopDiagnosticStabilityRecorder(); uninstallDiagnosticStabilityFatalHook();