From 73722c32f09552ef2f68a4b97d840db73143d016 Mon Sep 17 00:00:00 2001 From: Dallin Romney Date: Sat, 18 Jul 2026 12:30:14 -0700 Subject: [PATCH] revert: remove bounded run lifecycle backport Reverts 8286d32da5375470f3fa54b1e560676189e84c0f at operator request. The resulting tree is byte-for-byte identical to the prior canonical extended-stable/2026.6.33 tip c2578949469aacdb05e7b2aa5fbfa27cdd53924c. --- docs/concepts/agent-loop.md | 4 +- .../app-server/attempt-client-cleanup.test.ts | 25 - .../src/app-server/attempt-client-cleanup.ts | 35 +- .../src/app-server/attempt-results.test.ts | 25 +- .../codex/src/app-server/attempt-results.ts | 21 - .../app-server/attempt-turn-watches.test.ts | 69 -- .../src/app-server/attempt-turn-watches.ts | 6 - .../codex/src/app-server/run-attempt.ts | 1 - .../run-attempt.turn-watches.test.ts | 8 +- .../src/app-server/shared-client.test.ts | 102 +-- .../codex/src/app-server/shared-client.ts | 44 +- src/agents/cli-runner.spawn.test.ts | 16 +- src/agents/cli-runner/claude-live-session.ts | 96 ++- .../embedded-agent-runner/run/attempt.ts | 10 - .../run/llm-idle-timeout.test.ts | 117 +--- .../run/llm-idle-timeout.ts | 51 +- src/agents/embedded-agent-runner/runs.test.ts | 31 - src/agents/embedded-agent-runner/runs.ts | 15 +- src/agents/subagent-depth.test.ts | 11 - src/agents/timeout.ts | 9 +- .../reply/agent-runner-cli-dispatch.test.ts | 78 --- .../reply/agent-runner-cli-dispatch.ts | 14 - .../reply/agent-runner-execution.test.ts | 1 - .../reply/agent-runner-execution.ts | 3 - ...nner-memory.preflight-stale-tokens.test.ts | 15 +- .../reply/agent-runner-memory.test.ts | 4 +- ...t-runner.final-media-runreplyagent.test.ts | 3 - .../reply/agent-runner.media-paths.test.ts | 3 - src/auto-reply/reply/agent-runner.ts | 2 - ...ispatch-from-config.stale-recovery.test.ts | 652 ++++++++++++++++-- src/auto-reply/reply/dispatch-from-config.ts | 110 ++- src/auto-reply/reply/followup-runner.ts | 2 - .../reply/reply-run-registry.test.ts | 145 ---- src/auto-reply/reply/reply-run-registry.ts | 224 +----- .../reply/reply-turn-admission.test.ts | 220 ------ src/auto-reply/reply/reply-turn-admission.ts | 44 +- src/config/config-misc.test.ts | 14 - src/config/zod-schema.agent-defaults.ts | 3 +- .../heartbeat-runner.model-override.test.ts | 8 - src/infra/heartbeat-runner.ts | 3 - ...agnostic-stuck-session-recovery.runtime.ts | 3 +- src/logging/diagnostic.ts | 10 - 42 files changed, 853 insertions(+), 1404 deletions(-) diff --git a/docs/concepts/agent-loop.md b/docs/concepts/agent-loop.md index f2d499c3f743..10a752defa4d 100644 --- a/docs/concepts/agent-loop.md +++ b/docs/concepts/agent-loop.md @@ -164,10 +164,10 @@ surfaces, while Codex native hooks remain a separate lower-level Codex mechanism ## Timeouts - `agent.wait` default: 30s (just the wait). `timeoutMs` param overrides. -- Agent runtime: `agents.defaults.timeoutSeconds` default 172800s (48 hours); enforced in `runEmbeddedAgent` abort timer. Set `0` for an unlimited run budget; model-stream liveness watchdogs still apply. +- Agent runtime: `agents.defaults.timeoutSeconds` default 172800s (48 hours); enforced in `runEmbeddedAgent` 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`; owned silent model calls also stay `session.long_running` until `diagnostics.stuckSessionAbortMs` so slow or non-streaming providers are not reported as stalled too early. Active work with no recent progress reports as `session.stalled`; owned model calls switch to `session.stalled` at or after the abort threshold, and ownerless stale model/tool activity is not hidden as long-running. `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. Cloud providers default to 120s and self-hosted providers on network base URLs to 300s; unlimited run budgets retain those liveness watchdogs. `models.providers..timeoutSeconds` extends the watchdog for slow local/self-hosted providers, but any lower finite agent or run timeout still wins. Genuinely local loopback/private endpoints retain the gap-watchdog opt-out while stream creation remains bounded. Cron cloud stalls cap at 60s under a finite cron run timeout; local/self-hosted stalls cap at that explicit cron timeout. +- 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 cloud model runs with no explicit model or agent timeout use the same default idle watchdog; with an explicit cron run timeout, cloud model stream stalls are capped at 60s so configured model fallbacks can run before the outer cron deadline. Cron-triggered local or self-hosted model runs disable the implicit watchdog unless an explicit timeout is configured, and explicit cron run timeouts remain the idle window for local/self-hosted providers, so slow local providers should set `models.providers..timeoutSeconds`. - 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. ## Where things can end early diff --git a/extensions/codex/src/app-server/attempt-client-cleanup.test.ts b/extensions/codex/src/app-server/attempt-client-cleanup.test.ts index dd3f3bdc1f0b..7717583806c2 100644 --- a/extensions/codex/src/app-server/attempt-client-cleanup.test.ts +++ b/extensions/codex/src/app-server/attempt-client-cleanup.test.ts @@ -2,7 +2,6 @@ import { describe, expect, it, vi } from "vitest"; import { interruptCodexTurnBestEffort, - retireCodexAppServerClientAfterTimedOutTurn, unsubscribeCodexThreadBestEffort, } from "./attempt-client-cleanup.js"; @@ -41,28 +40,4 @@ describe("Codex app-server attempt client cleanup", () => { { timeoutMs: 123 }, ); }); - - it("closes only the isolated client after timed-out turn cleanup", async () => { - const request = vi.fn(async () => ({})); - const close = vi.fn(); - - await retireCodexAppServerClientAfterTimedOutTurn({ request, close } as never, { - threadId: "thread-1", - turnId: "turn-1", - reason: "turn_terminal_idle_timeout", - suspectPhysicalClient: true, - }); - - expect(request).toHaveBeenCalledWith( - "turn/interrupt", - { threadId: "thread-1", turnId: "turn-1" }, - { timeoutMs: 5_000 }, - ); - expect(request).toHaveBeenCalledWith( - "thread/unsubscribe", - { threadId: "thread-1" }, - { timeoutMs: 5_000 }, - ); - expect(close).toHaveBeenCalledTimes(1); - }); }); diff --git a/extensions/codex/src/app-server/attempt-client-cleanup.ts b/extensions/codex/src/app-server/attempt-client-cleanup.ts index 145b17c9fd2a..2ab5327ee2bc 100644 --- a/extensions/codex/src/app-server/attempt-client-cleanup.ts +++ b/extensions/codex/src/app-server/attempt-client-cleanup.ts @@ -116,34 +116,19 @@ export async function retireCodexAppServerClientAfterTimedOutTurn( threadId: string; turnId: string; reason: string; - /** - * Only the terminal-idle watch proves the physical client is dead (zero - * notifications for the whole window). Completion/assistant/budget - * timeouts are per-turn conditions on a possibly healthy shared process — - * failing co-leases for those would abort innocent sibling turns. - */ - suspectPhysicalClient: boolean; }, ): Promise { - const retiredSharedClient = retireSharedCodexAppServerClientIfCurrent(client, { - failActiveLeases: params.suspectPhysicalClient, - }); + const retiredSharedClient = retireSharedCodexAppServerClientIfCurrent(client); const detachedSharedClient = Boolean(retiredSharedClient); - const clientAlreadyClosed = - params.suspectPhysicalClient && (retiredSharedClient?.closed ?? false); - // Best-effort interrupt/unsubscribe only make sense while the transport is - // still open; a suspect client was just closed (child gets SIGKILLed). - if (!clientAlreadyClosed) { - interruptCodexTurnBestEffort(client, { - threadId: params.threadId, - turnId: params.turnId, - timeoutMs: CODEX_APP_SERVER_INTERRUPT_TIMEOUT_MS, - }); - await unsubscribeCodexThreadBestEffort(client, { - threadId: params.threadId, - timeoutMs: CODEX_APP_SERVER_UNSUBSCRIBE_TIMEOUT_MS, - }); - } + interruptCodexTurnBestEffort(client, { + threadId: params.threadId, + turnId: params.turnId, + timeoutMs: CODEX_APP_SERVER_INTERRUPT_TIMEOUT_MS, + }); + await unsubscribeCodexThreadBestEffort(client, { + threadId: params.threadId, + timeoutMs: CODEX_APP_SERVER_UNSUBSCRIBE_TIMEOUT_MS, + }); let closedClient = retiredSharedClient?.closed ?? false; if (!detachedSharedClient) { const close = (client as { close?: () => void }).close; diff --git a/extensions/codex/src/app-server/attempt-results.test.ts b/extensions/codex/src/app-server/attempt-results.test.ts index e70fcb6fa481..7f01bcfccbb2 100644 --- a/extensions/codex/src/app-server/attempt-results.test.ts +++ b/extensions/codex/src/app-server/attempt-results.test.ts @@ -67,7 +67,7 @@ describe("Codex app-server attempt results", () => { expect( buildCodexAppServerPromptTimeoutOutcome({ result: createResult({ - assistantTexts: ["Salvaged answer."], + toolMetas: [{ toolName: "exec" }], }), turnCompletionIdleTimedOut: true, turnWatchTimeoutKind: "terminal", @@ -132,29 +132,6 @@ describe("Codex app-server attempt results", () => { }); }); - it("builds an honest terminal-idle outcome instead of budget advice", () => { - expect( - buildCodexAppServerPromptTimeoutOutcome({ - result: createResult({}), - turnCompletionIdleTimedOut: true, - turnWatchTimeoutKind: "terminal", - }), - ).toEqual({ - message: - "Codex stopped responding: no activity arrived for the turn's liveness window, so the turn was ended and the connection was replaced. Retry to continue on a fresh session.", - }); - expect( - buildCodexAppServerPromptTimeoutOutcome({ - result: createResult({ toolMetas: [{ toolName: "exec" }] }), - turnCompletionIdleTimedOut: true, - turnWatchTimeoutKind: "terminal", - }), - ).toMatchObject({ - replayInvalid: true, - livenessState: "abandoned", - }); - }); - it("classifies replay blocked reasons", () => { expect(resolveCodexAppServerReplayBlockedReason(createResult())).toBeUndefined(); expect( diff --git a/extensions/codex/src/app-server/attempt-results.ts b/extensions/codex/src/app-server/attempt-results.ts index 05d2e04d5e5b..437f325ef84b 100644 --- a/extensions/codex/src/app-server/attempt-results.ts +++ b/extensions/codex/src/app-server/attempt-results.ts @@ -14,8 +14,6 @@ const CODEX_APP_SERVER_MISSING_TERMINAL_EVENT_USER_MESSAGE = "Codex stopped before confirming the turn was complete. The response may be incomplete; retry if needed."; const CODEX_APP_SERVER_MISSING_TERMINAL_EVENT_SIDE_EFFECT_USER_MESSAGE = "Codex stopped before confirming the turn was complete. Some work may already have been performed; verify the current state before retrying."; -const CODEX_APP_SERVER_TERMINAL_IDLE_USER_MESSAGE = - "Codex stopped responding: no activity arrived for the turn's liveness window, so the turn was ended and the connection was replaced. Retry to continue on a fresh session."; /** Joins terminal assistant text blocks into the final attempt answer. */ export function collectTerminalAssistantText(result: EmbeddedRunAttemptResult): string { @@ -34,25 +32,6 @@ export function buildCodexAppServerPromptTimeoutOutcome(params: { if (!params.turnCompletionIdleTimedOut) { return undefined; } - // Terminal-idle kills are dead-client events, not slow turns: the generic - // "increase agents.defaults.timeoutSeconds" advice would be wrong because - // that watch deliberately ignores the agent budget. Salvaged assistant - // output still wins over any timeout notice. - if (params.turnWatchTimeoutKind === "terminal") { - if (collectTerminalAssistantText(params.result)) { - return undefined; - } - const terminalReplayBlockedReason = resolveCodexAppServerReplayBlockedReason(params.result); - return { - message: CODEX_APP_SERVER_TERMINAL_IDLE_USER_MESSAGE, - ...(terminalReplayBlockedReason - ? { - replayInvalid: true, - livenessState: "abandoned" as const, - } - : {}), - }; - } if (params.turnWatchTimeoutKind !== undefined && params.turnWatchTimeoutKind !== "completion") { return undefined; } diff --git a/extensions/codex/src/app-server/attempt-turn-watches.test.ts b/extensions/codex/src/app-server/attempt-turn-watches.test.ts index 632e0aaedd34..c19db4890e78 100644 --- a/extensions/codex/src/app-server/attempt-turn-watches.test.ts +++ b/extensions/codex/src/app-server/attempt-turn-watches.test.ts @@ -155,75 +155,6 @@ describe("Codex app-server attempt turn watches", () => { expect(harness.abortController.signal.aborted).toBe(false); }); - it("keeps terminal idle gated while an app-server request is in flight", () => { - const harness = createController(); - harness.activeRequests = 1; - - harness.controller.armTerminalIdleWatch(); - vi.advanceTimersByTime(10); - - expect(harness.timeouts).toEqual([]); - expect(harness.abortController.signal.aborted).toBe(false); - }); - - it("fires terminal idle after the in-flight request settles and silence resumes", () => { - const harness = createController(); - harness.activeRequests = 1; - - harness.controller.armTerminalIdleWatch(); - vi.advanceTimersByTime(10); - expect(harness.timeouts).toEqual([]); - - harness.activeRequests = 0; - harness.controller.touchActivity("request:item/tool/call:response"); - vi.advanceTimersByTime(9); - expect(harness.timeouts).toEqual([]); - - vi.advanceTimersByTime(1); - expect(harness.timeouts).toMatchObject([ - { - kind: "terminal", - idleMs: 10, - timeoutMs: 10, - lastActivityReason: "request:item/tool/call:response", - }, - ]); - expect(harness.abortController.signal.reason).toBe("turn_terminal_idle_timeout"); - }); - - it("keeps completion idle gated while a request is in flight", () => { - const harness = createController(); - harness.activeRequests = 1; - - harness.controller.touchActivity("turn:start", { arm: true }); - vi.advanceTimersByTime(10); - - expect(harness.timeouts).toEqual([]); - expect(harness.abortController.signal.aborted).toBe(false); - - harness.activeRequests = 0; - harness.controller.scheduleProgressWatches(); - vi.advanceTimersByTime(1); - - expect(harness.timeouts).toMatchObject([{ kind: "completion" }]); - }); - - it("keeps assistant-completion gated while a request is in flight", () => { - const harness = createController(); - harness.activeRequests = 1; - - harness.controller.armAssistantCompletionIdleWatch(); - vi.advanceTimersByTime(10); - - expect(harness.completed).toBe(false); - expect(harness.abortController.signal.aborted).toBe(false); - - harness.activeRequests = 0; - vi.advanceTimersByTime(1); - - expect(harness.completed).toBe(true); - }); - it("releases a completed assistant item after the assistant idle guard expires", () => { const harness = createController(); diff --git a/extensions/codex/src/app-server/attempt-turn-watches.ts b/extensions/codex/src/app-server/attempt-turn-watches.ts index cf15422b60d4..0e4cf1a3b037 100644 --- a/extensions/codex/src/app-server/attempt-turn-watches.ts +++ b/extensions/codex/src/app-server/attempt-turn-watches.ts @@ -350,12 +350,6 @@ export function createCodexAttemptTurnWatchController(params: { } function fireTerminalIdleTimeout() { - // Physical-client liveness backstop. A terminal timeout retires the shared - // client, so it must only measure silence the client owns: while a - // server->client request is pending (approval/elicitation/tool call) the - // app-server legitimately says nothing until we respond. The response path - // touches activity when the request settles, so a wedged client is still - // caught within one terminal window after our response. if ( params.isCompleted() || params.isTerminalTurnNotificationQueued() || diff --git a/extensions/codex/src/app-server/run-attempt.ts b/extensions/codex/src/app-server/run-attempt.ts index 6b8119bf5959..dc199d4cf6c6 100644 --- a/extensions/codex/src/app-server/run-attempt.ts +++ b/extensions/codex/src/app-server/run-attempt.ts @@ -2857,7 +2857,6 @@ export async function runCodexAppServerAttempt( threadId: thread.threadId, turnId: activeTurnId, reason: String(runAbortController.signal.reason ?? "timeout"), - suspectPhysicalClient: turnWatchTimeoutKind === "terminal", }); })().finally(() => { resolveCompletion?.(); diff --git a/extensions/codex/src/app-server/run-attempt.turn-watches.test.ts b/extensions/codex/src/app-server/run-attempt.turn-watches.test.ts index c3ee4b058c56..9480d9f3b9d3 100644 --- a/extensions/codex/src/app-server/run-attempt.turn-watches.test.ts +++ b/extensions/codex/src/app-server/run-attempt.turn-watches.test.ts @@ -380,7 +380,7 @@ describe("runCodexAppServerAttempt turn watches", () => { }); }); - it("uses the terminal dead-client outcome for silent terminal timeouts", async () => { + it("does not use completion timeout outcome for terminal timeout with active mutating item", async () => { const harness = createStartedThreadHarness(); const params = createParams( path.join(tempDir, "session.jsonl"), @@ -415,11 +415,7 @@ describe("runCodexAppServerAttempt turn watches", () => { expect(result.codexAppServerFailure?.replaySafe).toBe(false); expect(result.codexAppServerFailure?.replayBlockedReason).toBe("potential_side_effect"); expect(result.codexAppServerFailure?.diagnostics).toBeUndefined(); - expect(result.promptTimeoutOutcome).toMatchObject({ - message: expect.stringContaining("Codex stopped responding"), - replayInvalid: true, - livenessState: "abandoned", - }); + expect(result.promptTimeoutOutcome).toBeUndefined(); }); it("does not use completion timeout outcome for non-completion timeout with assistant output", async () => { diff --git a/extensions/codex/src/app-server/shared-client.test.ts b/extensions/codex/src/app-server/shared-client.test.ts index 67cfda21afcf..f120ccb37199 100644 --- a/extensions/codex/src/app-server/shared-client.test.ts +++ b/extensions/codex/src/app-server/shared-client.test.ts @@ -710,7 +710,7 @@ describe("shared Codex app-server client", () => { expect(second.process.stdin.destroyed).toBe(true); }); - it("closes a retired shared app-server and forces active leases onto the retryable close path", async () => { + it("closes a retired shared app-server after all active leases release", async () => { const first = createClientHarness(); const second = createClientHarness(); vi.spyOn(CodexAppServerClient, "start") @@ -726,15 +726,11 @@ describe("shared Codex app-server client", () => { const releaseSecond = retainSharedCodexAppServerClientIfCurrent(first.client); expect(releaseFirst).toBeTypeOf("function"); expect(releaseSecond).toBeTypeOf("function"); - const activeRequest = first.client.request("test/pending", {}); - expect( - retireSharedCodexAppServerClientIfCurrent(first.client, { failActiveLeases: true }), - ).toEqual({ + expect(retireSharedCodexAppServerClientIfCurrent(first.client)).toEqual({ activeLeases: 2, - closed: true, + closed: false, }); - expect(first.process.stdin.destroyed).toBe(true); - await expect(activeRequest).rejects.toThrow("codex app-server client is closed"); + expect(first.process.stdin.destroyed).toBe(false); const secondList = listCodexAppServerModels({ timeoutMs: 1000 }); await sendInitializeResult(second, "openclaw/0.125.0 (macOS; test)"); @@ -742,6 +738,7 @@ describe("shared Codex app-server client", () => { await expect(secondList).resolves.toEqual({ models: [] }); releaseFirst?.(); + expect(first.process.stdin.destroyed).toBe(false); releaseSecond?.(); expect(first.process.stdin.destroyed).toBe(true); expect(second.process.kill).not.toHaveBeenCalled(); @@ -762,98 +759,23 @@ describe("shared Codex app-server client", () => { await expect(firstLease).resolves.toBe(first.client); await expect(secondLease).resolves.toBe(first.client); - expect( - retireSharedCodexAppServerClientIfCurrent(first.client, { failActiveLeases: true }), - ).toEqual({ - activeLeases: 2, - closed: true, - }); - expect( - retireSharedCodexAppServerClientIfCurrent(first.client, { failActiveLeases: true }), - ).toEqual({ + expect(retireSharedCodexAppServerClientIfCurrent(first.client)).toEqual({ activeLeases: 2, closed: false, }); - expect(first.process.stdin.destroyed).toBe(true); + expect(retireSharedCodexAppServerClientIfCurrent(first.client)).toEqual({ + activeLeases: 2, + closed: false, + }); + expect(first.process.stdin.destroyed).toBe(false); expect(releaseLeasedSharedCodexAppServerClient(first.client)).toBe(true); + expect(first.process.stdin.destroyed).toBe(false); expect(releaseLeasedSharedCodexAppServerClient(first.client)).toBe(true); expect(first.process.stdin.destroyed).toBe(true); expect(releaseLeasedSharedCodexAppServerClient(first.client)).toBe(false); }); - it("rejects pending acquires during shared-client retirement", async () => { - const first = createClientHarness(); - const second = createClientHarness(); - vi.spyOn(CodexAppServerClient, "start") - .mockReturnValueOnce(first.client) - .mockReturnValueOnce(second.client); - - const firstLease = getLeasedSharedCodexAppServerClient(); - const pendingLease = getLeasedSharedCodexAppServerClient(); - await vi.waitFor(() => expect(first.writes.length).toBeGreaterThanOrEqual(1)); - - expect( - retireSharedCodexAppServerClientIfCurrent(first.client, { failActiveLeases: true }), - ).toEqual({ - activeLeases: 0, - closed: true, - }); - await expect(firstLease).rejects.toThrow("codex app-server client is closed"); - await expect(pendingLease).rejects.toThrow("codex app-server client is closed"); - - const freshLease = getLeasedSharedCodexAppServerClient({ timeoutMs: 1000 }); - await sendInitializeResult(second, "openclaw/0.142.0 (macOS; test)"); - await expect(freshLease).resolves.toBe(second.client); - expect(second.process.stdin.destroyed).toBe(false); - }); - - it("suspect retirement closes a client that was already gracefully detached", async () => { - const first = createClientHarness(); - vi.spyOn(CodexAppServerClient, "start").mockReturnValueOnce(first.client); - - const lease = getLeasedSharedCodexAppServerClient({ timeoutMs: 1000 }); - await sendInitializeResult(first, "openclaw/0.142.0 (macOS; test)"); - await expect(lease).resolves.toBe(first.client); - - // Routine cleanup detaches gracefully; a later terminal-idle kill must - // still be able to fail the leaseholders off the poisoned process. - expect(retireSharedCodexAppServerClientIfCurrent(first.client)).toEqual({ - activeLeases: 1, - closed: false, - }); - expect(first.process.stdin.destroyed).toBe(false); - - expect( - retireSharedCodexAppServerClientIfCurrent(first.client, { failActiveLeases: true }), - ).toEqual({ - activeLeases: 1, - closed: true, - }); - expect(first.process.stdin.destroyed).toBe(true); - expect(releaseLeasedSharedCodexAppServerClient(first.client)).toBe(true); - }); - - it("retires gracefully by default: leased clients close on release, not immediately", async () => { - const first = createClientHarness(); - vi.spyOn(CodexAppServerClient, "start").mockReturnValueOnce(first.client); - - const lease = getLeasedSharedCodexAppServerClient({ timeoutMs: 1000 }); - await sendInitializeResult(first, "openclaw/0.142.0 (macOS; test)"); - await expect(lease).resolves.toBe(first.client); - - // Routine cleanup (e.g. one-shot bundle-MCP) must not yank a healthy - // client from co-leased sessions; only suspect retirement does. - expect(retireSharedCodexAppServerClientIfCurrent(first.client)).toEqual({ - activeLeases: 1, - closed: false, - }); - expect(first.process.stdin.destroyed).toBe(false); - - expect(releaseLeasedSharedCodexAppServerClient(first.client)).toBe(true); - expect(first.process.stdin.destroyed).toBe(true); - }); - it("waits only for the shared client that is still current", async () => { const first = createClientHarness(); const second = createClientHarness(); diff --git a/extensions/codex/src/app-server/shared-client.ts b/extensions/codex/src/app-server/shared-client.ts index ce18334b8445..0e6fdb7fad58 100644 --- a/extensions/codex/src/app-server/shared-client.ts +++ b/extensions/codex/src/app-server/shared-client.ts @@ -26,7 +26,6 @@ type SharedCodexAppServerClientEntry = { activeLeases: number; pendingAcquires: number; closeWhenIdle: boolean; - closeError?: Error; }; type SharedCodexAppServerClientState = { @@ -44,9 +43,6 @@ type KeyedSharedCodexAppServerClientState = { leasedReleases?: unknown; }; -// Clients we already force-closed for suspect retirement; a repeat retire must -// report closed:false instead of pretending to close the corpse again. -const suspectClosedClients = new WeakSet(); const SHARED_CODEX_APP_SERVER_CLIENT_STATE = Symbol.for("openclaw.codexAppServerClientState"); function getSharedCodexAppServerClientState(): SharedCodexAppServerClientState { @@ -274,9 +270,6 @@ async function acquireSharedCodexAppServerClient( options?.timeoutMs ?? 0, "codex app-server initialize timed out", ); - if (entry.closeError) { - throw entry.closeError; - } client.setActiveSharedLeaseCountProviderForUnscopedNotifications(() => entry.activeLeases); const release = leaseOptions?.leased ? retainSharedClientEntry(entry) : undefined; return release ? { client, release } : { client }; @@ -405,18 +398,9 @@ export function retainSharedCodexAppServerClientIfCurrent( return undefined; } -/** - * Retires a matching shared client. Default is graceful: detach from the map - * (future acquisitions get a fresh client) and close once leases drain. - * `failActiveLeases` is for suspect clients only (timed-out turns): it closes - * the physical connection immediately so co-leased attempts hit the normal - * client-closed retry path, and pending acquires reject instead of leasing - * the poisoned process. Routine cleanup must NOT use it — it would abort - * healthy sibling turns on a working client. - */ +/** Marks a matching shared client to close after active leases/acquires drain. */ export function retireSharedCodexAppServerClientIfCurrent( client: CodexAppServerClient | undefined, - opts?: { failActiveLeases?: boolean }, ): { activeLeases: number; closed: boolean } | undefined { if (!client) { return undefined; @@ -426,28 +410,12 @@ export function retireSharedCodexAppServerClientIfCurrent( if (entry.client === client) { state.clients.delete(key); entry.closeWhenIdle = true; - if (opts?.failActiveLeases) { - entry.closeError = new Error("codex app-server client is closed"); - const closed = closeRetiredSharedClientEntry(entry); - if (closed) { - suspectClosedClients.add(client); - } - return { activeLeases: entry.activeLeases, closed }; - } const closed = closeRetiredSharedClientEntryIfIdle(entry); return { activeLeases: entry.activeLeases, closed }; } } const activeLeases = state.leasedReleases.get(client)?.length ?? 0; if (activeLeases > 0) { - // A gracefully detached client (e.g. one-shot cleanup) can still be leased - // when a later terminal-idle kill declares it suspect; the map miss must - // not let the poisoned process keep serving those co-leases. - if (opts?.failActiveLeases && !suspectClosedClients.has(client)) { - suspectClosedClients.add(client); - client.close(); - return { activeLeases, closed: true }; - } return { activeLeases, closed: false }; } return undefined; @@ -578,16 +546,6 @@ function closeRetiredSharedClientEntryIfIdle(entry: SharedCodexAppServerClientEn return true; } -function closeRetiredSharedClientEntry(entry: SharedCodexAppServerClientEntry): boolean { - const client = entry.client; - if (!client) { - return false; - } - entry.client = undefined; - client.close(); - return true; -} - function closeSharedClientEntryIfUnclaimed( key: string, entry: SharedCodexAppServerClientEntry, diff --git a/src/agents/cli-runner.spawn.test.ts b/src/agents/cli-runner.spawn.test.ts index 8a41320b6606..b1e7a72f2e97 100644 --- a/src/agents/cli-runner.spawn.test.ts +++ b/src/agents/cli-runner.spawn.test.ts @@ -1970,7 +1970,7 @@ ${JSON.stringify({ expect(parsed.response.response.toolUseID).toBe("tool-allow-1"); }); - it("reports progress without heartbeats and extends the quiet-tool watchdog", async () => { + it("reports Claude live stream progress and keeps native tools fresh while they are running", async () => { vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], }); @@ -1982,7 +1982,6 @@ ${JSON.stringify({ } }); let stdoutListener: ((chunk: string) => void) | undefined; - const cancel = vi.fn(); const stdin = { write: vi.fn((data: string, cb?: (err?: Error | null) => void) => { stdoutListener?.( @@ -2022,7 +2021,7 @@ ${JSON.stringify({ startedAtMs: Date.now(), stdin, wait: vi.fn(() => new Promise(() => {})), - cancel, + cancel: vi.fn(), }; }); @@ -2035,7 +2034,7 @@ ${JSON.stringify({ sessionKey: "agent:main:diagnostics", prompt: "hello", backend: { liveSession: "claude-stdio" }, - timeoutMs: 3_600_000, + timeoutMs: 120_000, }); const resultPromise = runClaudeLiveSessionTurn({ context, @@ -2073,16 +2072,11 @@ ${JSON.stringify({ expect( getDiagnosticSessionActivitySnapshot({ sessionKey: "agent:main:diagnostics" }) .lastProgressReason, - ).toBe("cli_live:tool_started"); + ).toBe("cli_live:tool_running"); expect( getDiagnosticSessionActivitySnapshot({ sessionKey: "agent:main:diagnostics" }) .lastProgressAgeMs, - ).toBeGreaterThanOrEqual(10_000); - - // The 120s byte-silence budget must not kill an observed in-flight tool; - // it is extended to the shared 15-minute blocked-tool floor. - await vi.advanceTimersByTimeAsync(120_000); - expect(cancel).not.toHaveBeenCalled(); + ).toBeLessThan(100); stdoutListener?.( [ diff --git a/src/agents/cli-runner/claude-live-session.ts b/src/agents/cli-runner/claude-live-session.ts index 1cb4f4aab3a7..94cac94beb2c 100644 --- a/src/agents/cli-runner/claude-live-session.ts +++ b/src/agents/cli-runner/claude-live-session.ts @@ -22,7 +22,6 @@ import { type ExecAsk, type ExecSecurity, } from "../../infra/exec-approvals.js"; -import { BLOCKED_TOOL_CALL_ABORT_FLOOR_MS } from "../../logging/diagnostic-run-activity.js"; import { resolveAgentIdFromSessionKey } from "../../routing/session-key.js"; import { createCliJsonlStreamingParser, @@ -53,9 +52,8 @@ type ClaudeLiveTurn = { rawChars: number; sessionId?: string; noOutputTimer: NodeJS.Timeout | null; - /** Last stdout/stderr time; null until the process emits anything this turn. */ - lastOutputAtMs: number | null; timeoutTimer: NodeJS.Timeout | null; + activeToolTimer: NodeJS.Timeout | null; activeTools: Map; streamingParser: ReturnType; execPermission: ClaudeLiveExecPermission; @@ -110,6 +108,7 @@ type ClaudeLiveToolUse = { }; 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_DEFAULT_MAX_TURN_RAW_CHARS = 8 * 1024 * 1024; @@ -377,6 +376,10 @@ function clearTurnTimers(turn: ClaudeLiveTurn): void { clearTimeout(turn.timeoutTimer); turn.timeoutTimer = null; } + if (turn.activeToolTimer) { + clearInterval(turn.activeToolTimer); + turn.activeToolTimer = null; + } } function clearDrainTimer(session: ClaudeLiveSession): void { @@ -585,6 +588,31 @@ function readClaudeLiveToolResultIds(parsed: Record): string[] return toolResultIds; } +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: ClaudeLiveToolUse): void { const now = Date.now(); turn.activeTools.set(tool.toolCallId, { @@ -602,6 +630,7 @@ function markClaudeLiveToolStarted(turn: ClaudeLiveTurn, tool: ClaudeLiveToolUse ...(tool.paramsSummary ? { paramsSummary: tool.paramsSummary } : {}), }); emitClaudeLiveProgress(turn, "cli_live:tool_started"); + startClaudeLiveActiveToolHeartbeat(turn); } function markClaudeLiveToolCompleted(turn: ClaudeLiveTurn, toolCallId: string): void { @@ -624,6 +653,7 @@ function markClaudeLiveToolCompleted(turn: ClaudeLiveTurn, toolCallId: string): ...event, }); emitClaudeLiveProgress(turn, "cli_live:tool_result"); + stopClaudeLiveActiveToolHeartbeatIfIdle(turn); } function completeActiveClaudeLiveTools(turn: ClaudeLiveTurn): void { @@ -672,44 +702,24 @@ function noteClaudeLiveProgress(turn: ClaudeLiveTurn, parsed: Record { - const quietSinceMs = turn.lastOutputAtMs ?? turn.startedAtMs; - if (turn.activeTools.size > 0) { - const quietBudgetMs = Math.max(session.noOutputTimeoutMs, BLOCKED_TOOL_CALL_ABORT_FLOOR_MS); - const remainingMs = quietSinceMs + quietBudgetMs - Date.now(); - if (remainingMs > 0) { - armNoOutputTimer(session, turn, remainingMs); - return; - } - } - closeLiveSession( - session, - "abort", - createTimeoutError( - session, - `CLI produced no output for ${Math.round((Date.now() - quietSinceMs) / 1000)}s and was terminated.`, - // Retryable only when the process never produced any output this turn. - turn.lastOutputAtMs === null ? "cli_no_output_timeout" : undefined, - ), - ); - }, delayMs); -} - function resetNoOutputTimer(session: ClaudeLiveSession): void { const turn = session.currentTurn; if (!turn) { return; } - turn.lastOutputAtMs = Date.now(); - armNoOutputTimer(session, turn, session.noOutputTimeoutMs); + if (turn.noOutputTimer) { + clearTimeout(turn.noOutputTimer); + } + turn.noOutputTimer = setTimeout(() => { + closeLiveSession( + session, + "abort", + createTimeoutError( + session, + `CLI produced no output for ${Math.round(session.noOutputTimeoutMs / 1000)}s and was terminated.`, + ), + ); + }, session.noOutputTimeoutMs); } function parseSessionId(parsed: Record): string | undefined { @@ -1154,8 +1164,8 @@ function createTurn(params: { rawLines: [], rawChars: 0, noOutputTimer: null, - lastOutputAtMs: null, timeoutTimer: null, + activeToolTimer: null, activeTools: new Map(), streamingParser: createCliJsonlStreamingParser({ backend: params.context.preparedBackend.backend, @@ -1169,7 +1179,17 @@ function createTurn(params: { resolve: params.resolve, reject: params.reject, }; - armNoOutputTimer(params.session, turn, params.noOutputTimeoutMs); + turn.noOutputTimer = setTimeout(() => { + closeLiveSession( + params.session, + "abort", + createTimeoutError( + params.session, + `CLI produced no output for ${Math.round(params.noOutputTimeoutMs / 1000)}s and was terminated.`, + "cli_no_output_timeout", + ), + ); + }, params.noOutputTimeoutMs); turn.timeoutTimer = setTimeout(() => { closeLiveSession( params.session, diff --git a/src/agents/embedded-agent-runner/run/attempt.ts b/src/agents/embedded-agent-runner/run/attempt.ts index 7c4bf80baa9a..9dd0ec3e2368 100644 --- a/src/agents/embedded-agent-runner/run/attempt.ts +++ b/src/agents/embedded-agent-runner/run/attempt.ts @@ -3141,16 +3141,6 @@ export async function runEmbeddedAttempt( idleTimeoutMs, (error) => idleTimeoutTrigger?.(error), ); - } else { - // Local providers opt out of inter-chunk gap policing, but a request - // whose headers never arrive must still release the turn. - const localStreamCreationTimeoutMs = 300_000; - activeSession.agent.streamFn = streamWithIdleTimeout( - activeSession.agent.streamFn, - localStreamCreationTimeoutMs, - (error) => idleTimeoutTrigger?.(error), - { scope: "creation-only" }, - ); } let diagnosticModelCallSeq = 0; activeSession.agent.streamFn = wrapStreamFnWithDiagnosticModelCallEvents( diff --git a/src/agents/embedded-agent-runner/run/llm-idle-timeout.test.ts b/src/agents/embedded-agent-runner/run/llm-idle-timeout.test.ts index 86c2daf34669..65a6dd1849ce 100644 --- a/src/agents/embedded-agent-runner/run/llm-idle-timeout.test.ts +++ b/src/agents/embedded-agent-runner/run/llm-idle-timeout.test.ts @@ -12,7 +12,6 @@ import type { StreamFn } from "../../runtime/index.js"; import { resolveLlmIdleTimeoutMs, streamWithIdleTimeout } from "./llm-idle-timeout.js"; const DEFAULT_LLM_IDLE_TIMEOUT_MS = 120_000; -const SELF_HOSTED_LLM_IDLE_TIMEOUT_MS = 300_000; const CRON_LLM_IDLE_TIMEOUT_MS = 60_000; describe("resolveLlmIdleTimeoutMs", () => { @@ -192,28 +191,8 @@ describe("resolveLlmIdleTimeoutMs", () => { ).toBe(CRON_LLM_IDLE_TIMEOUT_MS); }); - it.each([ - [ - "cloud", - { provider: "openai", baseUrl: "https://api.openai.com/v1" }, - DEFAULT_LLM_IDLE_TIMEOUT_MS, - ], - [ - "self-hosted", - { provider: "vllm", baseUrl: "https://gpu.example.com/v1" }, - SELF_HOSTED_LLM_IDLE_TIMEOUT_MS, - ], - ])("uses the provider-class idle default for no-timeout %s models", (_label, model, expected) => { - expect(resolveLlmIdleTimeoutMs({ runTimeoutMs: MAX_TIMER_TIMEOUT_MS, model })).toBe(expected); - }); - - it("keeps local base URLs opted out under no-timeout runs", () => { - expect( - resolveLlmIdleTimeoutMs({ - runTimeoutMs: MAX_TIMER_TIMEOUT_MS, - model: { baseUrl: "http://127.0.0.1:11434" }, - }), - ).toBe(0); + it("disables the idle watchdog when an explicit run timeout disables timeouts", () => { + expect(resolveLlmIdleTimeoutMs({ runTimeoutMs: MAX_TIMER_TIMEOUT_MS })).toBe(0); }); it("honors an explicit models.providers..timeoutSeconds for cloud providers (#77744, #78361)", () => { @@ -467,57 +446,6 @@ describe("resolveLlmIdleTimeoutMs", () => { 30_000, ); }); - - it.each([ - ["local keeps no class ceiling", { baseUrl: "http://127.0.0.1:11434" }, 3_600_000], - [ - "self-hosted keeps the 300s tier", - { provider: "vllm", baseUrl: "https://gpu.example.com/v1" }, - 300_000, - ], - ["cloud keeps the 120s default", { provider: "openai" }, 120_000], - ])("large agents.defaults.timeoutSeconds: %s", (_label, model, expected) => { - const cfg = { agents: { defaults: { timeoutSeconds: 3_600 } } } as OpenClawConfig; - expect(resolveLlmIdleTimeoutMs({ cfg, model })).toBe(expected); - }); - - it.each([ - ["local keeps no class ceiling", { baseUrl: "http://127.0.0.1:11434" }, 900_000], - [ - "self-hosted keeps the 300s tier", - { provider: "vllm", baseUrl: "https://gpu.example.com/v1" }, - 300_000, - ], - ["cloud keeps the 120s default", { provider: "openai" }, 120_000], - ])("explicit run timeout above the tiers: %s", (_label, model, expected) => { - expect(resolveLlmIdleTimeoutMs({ runTimeoutMs: 900_000, model })).toBe(expected); - }); - - it("explicit run timeouts below the class tier still bound self-hosted idle", () => { - expect( - resolveLlmIdleTimeoutMs({ - runTimeoutMs: 90_000, - model: { provider: "vllm", baseUrl: "https://gpu.example.com/v1" }, - }), - ).toBe(90_000); - }); - - it("cron exempts provider-id self-hosted models from the 60s clamp", () => { - expect( - resolveLlmIdleTimeoutMs({ - trigger: "cron", - runTimeoutMs: 900_000, - model: { provider: "vllm", baseUrl: "https://gpu.example.com/v1" }, - }), - ).toBe(900_000); - expect( - resolveLlmIdleTimeoutMs({ - trigger: "cron", - runTimeoutMs: 900_000, - model: { provider: "openai" }, - }), - ).toBe(60_000); - }); }); describe("streamWithIdleTimeout", () => { @@ -608,47 +536,6 @@ describe("streamWithIdleTimeout", () => { await next; }); - it("creation-only scope bounds stream creation but not iterator gaps", async () => { - vi.useFakeTimers(); - // Creation hang: still rejected at the deadline. - const hangingCreate = vi.fn( - () => new Promise(() => {}), - ) as unknown as Parameters[0]; - const onIdleTimeout = vi.fn(); - const wrappedCreate = streamWithIdleTimeout(hangingCreate, 50, onIdleTimeout, { - scope: "creation-only", - }); - const model = {} as Parameters[0]; - const context = {} as Parameters[1]; - const options = {} as Parameters[2]; - const pending = expect(wrappedCreate(model, context, options)).rejects.toThrow( - /LLM idle timeout/, - ); - await vi.advanceTimersByTimeAsync(50); - await pending; - expect(onIdleTimeout).toHaveBeenCalledTimes(1); - - // Iterator gap: never bounded — local providers own their stream pacing. - const slowStream = createNeverYieldingStream(); - const slowFn = vi.fn().mockReturnValue(slowStream); - const wrappedGaps = streamWithIdleTimeout(slowFn, 50, onIdleTimeout, { - scope: "creation-only", - }); - const stream = wrappedGaps( - model as Parameters[0], - context as Parameters[1], - options as Parameters[2], - ) as AsyncIterable; - const iterator = stream[Symbol.asyncIterator](); - let settled = false; - void iterator.next().finally(() => { - settled = true; - }); - await vi.advanceTimersByTimeAsync(5_000); - expect(settled).toBe(false); - expect(onIdleTimeout).toHaveBeenCalledTimes(1); - }); - it("clears the connection timer when stream setup rejects", async () => { vi.useFakeTimers(); const setupError = new Error("provider setup failed"); diff --git a/src/agents/embedded-agent-runner/run/llm-idle-timeout.ts b/src/agents/embedded-agent-runner/run/llm-idle-timeout.ts index f04a0df24b42..af5a50c1e923 100644 --- a/src/agents/embedded-agent-runner/run/llm-idle-timeout.ts +++ b/src/agents/embedded-agent-runner/run/llm-idle-timeout.ts @@ -18,7 +18,6 @@ import type { EmbeddedRunTrigger } from "./params.js"; * Default idle timeout for LLM streaming responses in milliseconds. */ const DEFAULT_LLM_IDLE_TIMEOUT_MS = 120_000; -const SELF_HOSTED_LLM_IDLE_TIMEOUT_MS = 300_000; // Cron has its own outer watchdog; stream stalls must fail early enough for // the existing model fallback chain to try the next configured candidate. const CRON_LLM_IDLE_TIMEOUT_MS = 60_000; @@ -210,6 +209,8 @@ export function resolveLlmIdleTimeoutMs(params?: { model?: { baseUrl?: string; id?: string; provider?: string }; }): number { const clampTimeoutMs = (valueMs: number) => clampTimerTimeoutMs(valueMs) ?? 1; + const clampImplicitTimeoutMs = (valueMs: number) => + clampTimeoutMs(Math.min(valueMs, DEFAULT_LLM_IDLE_TIMEOUT_MS)); const runTimeoutMs = params?.runTimeoutMs; const agentTimeoutSeconds = params?.cfg?.agents?.defaults?.timeoutSeconds; @@ -233,8 +234,6 @@ export function resolveLlmIdleTimeoutMs(params?: { (isSelfHostedProviderId(params?.model?.provider) || hasConfiguredLocalProviderSignal({ cfg: params?.cfg, provider: params?.model?.provider })) && !isOllamaCloudModel(params?.model); - const isSelfHostedRuntimeModel = - isSelfHostedProviderId(params?.model?.provider) && !isOllamaCloudModel(params?.model); const timeoutBounds = [ runTimeoutIsNoTimeout ? undefined : runTimeoutMs, hasExplicitRunTimeout ? undefined : agentTimeoutMs, @@ -246,23 +245,6 @@ export function resolveLlmIdleTimeoutMs(params?: { value < MAX_TIMER_TIMEOUT_MS, ); - // Run/agent budgets bound idle from below the provider-class ceiling; they - // must not shrink class tolerance (local has no ceiling, self-hosted 300s). - // Clamping every class to the cloud default reopened #85826-style kills for - // self-hosted users with explicit budgets above 120s. - const clampToClassIdleCeiling = (budgetMs: number): number => { - if (isLocalRuntimeModel) { - return clampTimeoutMs(budgetMs); - } - const classIdleTimeoutMs = - isSelfHostedRuntimeModel || - isExplicitLocalHostnameRuntimeModel || - isSelfHostedHostnameRuntimeModel - ? SELF_HOSTED_LLM_IDLE_TIMEOUT_MS - : DEFAULT_LLM_IDLE_TIMEOUT_MS; - return clampTimeoutMs(Math.min(budgetMs, classIdleTimeoutMs)); - }; - // Explicit per-model idle timeout (`models.providers..timeoutSeconds`) wins // over the NO_TIMEOUT_MS sentinel that runTimeoutMs may carry when the caller // declared "run is unlimited". The two are independent: an unlimited run does @@ -287,25 +269,25 @@ export function resolveLlmIdleTimeoutMs(params?: { return clampTimeoutMs(boundedTimeoutMs); } - // Unlimited run budgets bound total cost, not stream liveness. Only a finite - // explicit run budget caps the idle watchdog. - if (hasExplicitRunTimeout && runTimeoutMs < MAX_TIMER_TIMEOUT_MS) { + if (typeof runTimeoutMs === "number" && Number.isFinite(runTimeoutMs) && runTimeoutMs > 0) { + if (runTimeoutMs >= MAX_TIMER_TIMEOUT_MS) { + return 0; + } if (params?.trigger === "cron") { if ( isLocalRuntimeModel || isExplicitLocalHostnameRuntimeModel || - isSelfHostedHostnameRuntimeModel || - isSelfHostedRuntimeModel + isSelfHostedHostnameRuntimeModel ) { return clampTimeoutMs(runTimeoutMs); } return clampTimeoutMs(Math.min(runTimeoutMs, CRON_LLM_IDLE_TIMEOUT_MS)); } - return clampToClassIdleCeiling(runTimeoutMs); + return clampImplicitTimeoutMs(runTimeoutMs); } if (agentTimeoutMs !== undefined) { - return clampToClassIdleCeiling(agentTimeoutMs); + return clampImplicitTimeoutMs(agentTimeoutMs); } // The default watchdog is a network-silence-as-hang guard for cloud providers. @@ -319,14 +301,6 @@ export function resolveLlmIdleTimeoutMs(params?: { return 0; } - if ( - isSelfHostedRuntimeModel || - isExplicitLocalHostnameRuntimeModel || - isSelfHostedHostnameRuntimeModel - ) { - return SELF_HOSTED_LLM_IDLE_TIMEOUT_MS; - } - return DEFAULT_LLM_IDLE_TIMEOUT_MS; } @@ -334,17 +308,12 @@ export function resolveLlmIdleTimeoutMs(params?: { * Wraps a stream function with idle timeout detection for both stream creation * and iterator progress. Each successful `next()` resets the timer; a timeout * aborts the provider request and surfaces the same Error to the caller. - * `scope: "creation-only"` bounds only the creation phase: local providers opt - * out of gap policing, but a request whose headers never arrive must still fail - * instead of wedging the turn until the run budget. */ export function streamWithIdleTimeout( baseFn: StreamFn, timeoutMs: number, onIdleTimeout?: (error: Error) => void, - opts?: { scope?: "creation-and-gaps" | "creation-only" }, ): StreamFn { - const guardIterationGaps = opts?.scope !== "creation-only"; return (model, context, options) => { const createIdleTimeoutError = () => new Error(`LLM idle timeout (${Math.floor(timeoutMs / 1000)}s): no response from model`); @@ -409,7 +378,7 @@ export function streamWithIdleTimeout( }; const armTimer = () => { clearTimer(); - if (!guardIterationGaps || !waitingForProvider) { + if (!waitingForProvider) { return; } idleTimer = setTimeout(() => { diff --git a/src/agents/embedded-agent-runner/runs.test.ts b/src/agents/embedded-agent-runner/runs.test.ts index dc8985070c19..3c879c4979d5 100644 --- a/src/agents/embedded-agent-runner/runs.test.ts +++ b/src/agents/embedded-agent-runner/runs.test.ts @@ -560,37 +560,6 @@ describe("embedded-agent runner run registry", () => { } }); - it("expires stuck recovery as run_stalled with a live embedded handle", async () => { - const operation = createReplyOperation({ - sessionKey: "agent:main:reply-stuck-live", - sessionId: "session-reply-stuck-live", - resetTriggered: false, - }); - const handle = createRunHandle({ - abort: () => { - operation.abortByUser(); - }, - }); - operation.attachBackend({ - kind: "embedded", - cancel: handle.abort, - isStreaming: handle.isStreaming, - }); - operation.setPhase("running"); - setActiveEmbeddedRun("session-reply-stuck-live", handle); - - const result = await abortAndDrainEmbeddedAgentRun({ - sessionId: "session-reply-stuck-live", - sessionKey: "agent:main:reply-stuck-live", - reason: "stuck_recovery", - forceClear: true, - settleMs: 50, - }); - - expect(result.aborted).toBe(true); - expect(operation.result).toEqual({ kind: "failed", code: "run_stalled" }); - }); - it("clamps oversized embedded run wait timers", async () => { vi.useFakeTimers(); try { diff --git a/src/agents/embedded-agent-runner/runs.ts b/src/agents/embedded-agent-runner/runs.ts index 716803e1123c..d03818755c55 100644 --- a/src/agents/embedded-agent-runner/runs.ts +++ b/src/agents/embedded-agent-runner/runs.ts @@ -4,7 +4,6 @@ import { abortActiveReplyRuns, abortReplyRunBySessionId, - expireStaleReplyRunBySessionId, forceClearReplyRunBySessionId, isReplyRunEvidenceStaleBySessionId, isReplyRunActiveForSessionId, @@ -727,19 +726,7 @@ export async function abortAndDrainEmbeddedAgentRun(params: { reason?: string; }): Promise { const settleMs = params.settleMs ?? 15_000; - // Stuck recovery is a staleness expiry. Stamp run_stalled before aborting the - // handle so synchronous abort callbacks cannot misattribute it to the user. - const expiredReplyRun = - params.reason === "stuck_recovery" && - expireStaleReplyRunBySessionId(params.sessionId, "stuck_recovery"); - if (expiredReplyRun && !ACTIVE_EMBEDDED_RUNS.has(params.sessionId)) { - await new Promise((resolve) => { - setImmediate(resolve); - }); - const drained = await waitForEmbeddedAgentRunEnd(params.sessionId, settleMs); - return { aborted: true, drained, forceCleared: false }; - } - const aborted = abortEmbeddedAgentRun(params.sessionId) || expiredReplyRun; + const aborted = abortEmbeddedAgentRun(params.sessionId); const drained = aborted ? await waitForEmbeddedAgentRunEnd(params.sessionId, settleMs) : false; const forceCleared = params.forceClear === true && (!aborted || !drained) diff --git a/src/agents/subagent-depth.test.ts b/src/agents/subagent-depth.test.ts index f4b7b99f6de4..609109c7ae3d 100644 --- a/src/agents/subagent-depth.test.ts +++ b/src/agents/subagent-depth.test.ts @@ -5,7 +5,6 @@ import os from "node:os"; import path from "node:path"; import { MAX_TIMER_TIMEOUT_MS } from "@openclaw/normalization-core/number-coercion"; import { describe, expect, it } from "vitest"; -import type { OpenClawConfig } from "../config/config.js"; import { getSubagentDepthFromSessionStore } from "./subagent-depth.js"; import { resolveAgentTimeoutMs } from "./timeout.js"; @@ -147,16 +146,6 @@ describe("resolveAgentTimeoutMs", () => { expect(resolveAgentTimeoutMs({})).toBe(48 * 60 * 60 * 1000); }); - it.each([ - ["unlimited", 0, MAX_TIMER_TIMEOUT_MS], - ["finite", 30, 30_000], - ["negative", -1, 1_000], - ["NaN", Number.NaN, 48 * 60 * 60 * 1000], - ])("resolves config timeoutSeconds %s", (_label, timeoutSeconds, expected) => { - const cfg = { agents: { defaults: { timeoutSeconds } } } as OpenClawConfig; - expect(resolveAgentTimeoutMs({ cfg })).toBe(expected); - }); - it("uses a timer-safe sentinel for no-timeout overrides", () => { expect(resolveAgentTimeoutMs({ overrideSeconds: 0 })).toBe(MAX_TIMER_TIMEOUT_MS); expect(resolveAgentTimeoutMs({ overrideMs: 0 })).toBe(MAX_TIMER_TIMEOUT_MS); diff --git a/src/agents/timeout.ts b/src/agents/timeout.ts index b99ffa65ab88..02c84ec120bf 100644 --- a/src/agents/timeout.ts +++ b/src/agents/timeout.ts @@ -10,19 +10,12 @@ import { import type { OpenClawConfig } from "../config/types.openclaw.js"; const DEFAULT_AGENT_TIMEOUT_SECONDS = 48 * 60 * 60; -const NO_TIMEOUT_MS = MAX_TIMER_TIMEOUT_MS; -const NO_TIMEOUT_SECONDS = Math.floor(NO_TIMEOUT_MS / 1000); const normalizeNumber = (value: unknown): number | undefined => typeof value === "number" && Number.isFinite(value) ? Math.floor(value) : undefined; function resolveAgentTimeoutSeconds(cfg?: OpenClawConfig): number { const raw = normalizeNumber(cfg?.agents?.defaults?.timeoutSeconds); - // Config 0 uses the same unlimited-run sentinel as per-run overrides. Model - // idle watchdogs still enforce liveness under that sentinel. - if (raw === 0) { - return NO_TIMEOUT_SECONDS; - } const seconds = raw ?? DEFAULT_AGENT_TIMEOUT_SECONDS; return Math.max(seconds, 1); } @@ -36,6 +29,8 @@ export function resolveAgentTimeoutMs(opts: { const minMs = Math.max(normalizeNumber(opts.minMs) ?? 1, 1); const clampTimeoutMs = (valueMs: number) => clampTimerTimeoutMs(valueMs, minMs) ?? minMs; const defaultMs = clampTimeoutMs(resolveAgentTimeoutSeconds(opts.cfg) * 1000); + // Use the maximum timer-safe timeout to represent "no timeout" when explicitly set to 0. + const NO_TIMEOUT_MS = MAX_TIMER_TIMEOUT_MS; const overrideMs = normalizeNumber(opts.overrideMs); if (overrideMs !== undefined) { if (overrideMs === 0) { diff --git a/src/auto-reply/reply/agent-runner-cli-dispatch.test.ts b/src/auto-reply/reply/agent-runner-cli-dispatch.test.ts index 8295af79ce5b..d9a2815c28b2 100644 --- a/src/auto-reply/reply/agent-runner-cli-dispatch.test.ts +++ b/src/auto-reply/reply/agent-runner-cli-dispatch.test.ts @@ -166,84 +166,6 @@ describe("runCliAgentWithLifecycle", () => { stopReason: "end_turn", }); }); - - it("stamps activity for every delivered CLI bridge event", async () => { - cliDispatchState.runCliAgentMock.mockImplementationOnce(async (params: { runId: string }) => { - emitAgentEvent({ - runId: params.runId, - stream: "assistant", - data: { text: "Visible answer" }, - }); - emitAgentEvent({ - runId: params.runId, - stream: "tool", - data: { name: "read", phase: "start" }, - }); - return { payloads: [], meta: { durationMs: 1 } } satisfies EmbeddedAgentRunResult; - }); - const onActivity = vi.fn(); - - await runCliAgentWithLifecycle({ - runId: "run-activity", - provider: "codex-cli", - onActivity, - onAssistantText: vi.fn(async () => undefined), - onToolEvent: vi.fn(async () => undefined), - runParams: { - sessionId: "session-1", - sessionFile: "/tmp/session.jsonl", - workspaceDir: "/tmp/workspace", - prompt: "hello", - provider: "codex-cli", - model: "gpt-5.4", - thinkLevel: "off", - timeoutMs: 1_000, - runId: "run-activity", - }, - }); - - expect(onActivity).toHaveBeenCalledTimes(2); - }); - - it("stamps activity when silent runs suppress delivery callbacks", async () => { - cliDispatchState.runCliAgentMock.mockImplementationOnce(async (params: { runId: string }) => { - emitAgentEvent({ - runId: params.runId, - stream: "assistant", - data: { text: "Silent answer" }, - }); - emitAgentEvent({ - runId: params.runId, - stream: "tool", - data: { name: "read", phase: "start" }, - }); - return { payloads: [], meta: { durationMs: 1 } } satisfies EmbeddedAgentRunResult; - }); - const onActivity = vi.fn(); - const onAssistantText = vi.fn(async () => undefined); - - await runCliAgentWithLifecycle({ - runId: "run-activity-suppressed", - provider: "codex-cli", - suppressAssistantBridge: true, - onActivity, - onAssistantText, - runParams: { - sessionId: "session-1", - sessionFile: "/tmp/session.jsonl", - workspaceDir: "/tmp/workspace", - prompt: "hello", - provider: "codex-cli", - model: "gpt-5.4", - thinkLevel: "off", - timeoutMs: 1_000, - runId: "run-activity-suppressed", - }, - }); - - expect(onAssistantText).not.toHaveBeenCalled(); - expect(onActivity).toHaveBeenCalledTimes(2); - }); }); describe("keepCliSessionBindingOnlyWhenReused", () => { diff --git a/src/auto-reply/reply/agent-runner-cli-dispatch.ts b/src/auto-reply/reply/agent-runner-cli-dispatch.ts index 32648178df4e..01ce12b93541 100644 --- a/src/auto-reply/reply/agent-runner-cli-dispatch.ts +++ b/src/auto-reply/reply/agent-runner-cli-dispatch.ts @@ -341,8 +341,6 @@ type RunCliAgentWithLifecycleParams = { emitLifecycleTerminal?: boolean; onAgentRunStart?: () => void; suppressAssistantBridge?: boolean; - /** Stamped before every delivered CLI progress event. */ - onActivity?: () => void; onAssistantText?: (text: string) => Promise; onReasoningText?: (text: string) => Promise; onToolEvent?: (payload: CliToolEventPayload) => Promise; @@ -447,17 +445,6 @@ async function runCliAgentWithLifecycleInternal( }, }); } - // Stamp every CLI event independently of delivery. Silent runs suppress - // callbacks but their real output must still refresh stale-run evidence. - const activityBridge = params.onActivity - ? createAgentEventBridge>({ - runId: params.runId, - read: () => ({}), - deliver: async () => { - params.onActivity?.(); - }, - }) - : undefined; const assistantBridge = createAssistantTextBridge({ runId: params.runId, suppressed: params.suppressAssistantBridge, @@ -486,7 +473,6 @@ async function runCliAgentWithLifecycleInternal( deliver: maybeAnnounceFastModeAutoOff, }); const bridges = [ - activityBridge, assistantBridge, reasoningBridge, toolBridge, diff --git a/src/auto-reply/reply/agent-runner-execution.test.ts b/src/auto-reply/reply/agent-runner-execution.test.ts index 8f249139bc7d..49c01f65f7ac 100644 --- a/src/auto-reply/reply/agent-runner-execution.test.ts +++ b/src/auto-reply/reply/agent-runner-execution.test.ts @@ -424,7 +424,6 @@ function createMockReplyOperation(): { resetTriggered: false, phase: "running", result: null, - startedAtMs: Date.now(), lastActivityAtMs: Date.now(), recordActivity: vi.fn(), setPhase: vi.fn(), diff --git a/src/auto-reply/reply/agent-runner-execution.ts b/src/auto-reply/reply/agent-runner-execution.ts index 6ee63a15aeb4..42f2089d2080 100644 --- a/src/auto-reply/reply/agent-runner-execution.ts +++ b/src/auto-reply/reply/agent-runner-execution.ts @@ -2291,7 +2291,6 @@ export async function runAgentTurnWithFallback(params: { emitLifecycleTerminal: false, onAgentRunStart: notifyAgentRunStart, suppressAssistantBridge: params.followupRun.run.silentExpected, - onActivity: () => params.replyOperation?.recordActivity(), onAssistantText: async (text) => { const textForTyping = await handlePartialForTyping({ text } as ReplyPayload); if (textForTyping === undefined || !params.opts?.onPartialReply) { @@ -2604,7 +2603,6 @@ export async function runAgentTurnWithFallback(params: { : undefined, onReasoningEnd: params.opts?.onReasoningEnd, onAgentEvent: async (evt) => { - params.replyOperation?.recordActivity(); lifecycleBackstop.note(evt); // Signal run start only after the embedded agent emits real activity. const hasLifecyclePhase = @@ -2867,7 +2865,6 @@ export async function runAgentTurnWithFallback(params: { return (payload: ReplyPayload) => { toolResultChain = toolResultChain .then(async () => { - params.replyOperation?.recordActivity(); const { text, skip } = normalizeStreamingText(payload); if (skip) { return; diff --git a/src/auto-reply/reply/agent-runner-memory.preflight-stale-tokens.test.ts b/src/auto-reply/reply/agent-runner-memory.preflight-stale-tokens.test.ts index d5ec6b1e6ced..9d3467845433 100644 --- a/src/auto-reply/reply/agent-runner-memory.preflight-stale-tokens.test.ts +++ b/src/auto-reply/reply/agent-runner-memory.preflight-stale-tokens.test.ts @@ -26,9 +26,6 @@ function createReplyOperation(): ReplyOperation { resetTriggered: false, phase: "queued", result: null, - startedAtMs: Date.now(), - lastActivityAtMs: Date.now(), - recordActivity: vi.fn(), setPhase: vi.fn(), updateSessionId: vi.fn(), attachBackend: vi.fn(), @@ -116,11 +113,7 @@ describe("runPreflightCompactionIfNeeded stale totalTokens gating", () => { totalTokens: 200_000, totalTokensFresh: false, }; - await writeTestSessionStore( - path.join(rootDir, "sessions.json"), - "agent:main:main", - sessionEntry, - ); + await writeTestSessionStore(path.join(rootDir, "sessions.json"), "agent:main:main", sessionEntry); const entry = await runWithEntry(sessionEntry, sessionFile); @@ -142,11 +135,7 @@ describe("runPreflightCompactionIfNeeded stale totalTokens gating", () => { totalTokens: 200_000, totalTokensFresh: true, }; - await writeTestSessionStore( - path.join(rootDir, "sessions.json"), - "agent:main:main", - sessionEntry, - ); + await writeTestSessionStore(path.join(rootDir, "sessions.json"), "agent:main:main", sessionEntry); await runWithEntry(sessionEntry, sessionFile); diff --git a/src/auto-reply/reply/agent-runner-memory.test.ts b/src/auto-reply/reply/agent-runner-memory.test.ts index f53161f5b165..c21dd20aa282 100644 --- a/src/auto-reply/reply/agent-runner-memory.test.ts +++ b/src/auto-reply/reply/agent-runner-memory.test.ts @@ -41,16 +41,14 @@ type TestReplyOperation = ReplyOperation & { }; function createReplyOperation(): TestReplyOperation { - const now = Date.now(); return { key: "test", sessionId: "session", abortSignal: new AbortController().signal, resetTriggered: false, - startedAtMs: now, phase: "queued", result: null, - lastActivityAtMs: now, + lastActivityAtMs: Date.now(), recordActivity: vi.fn(), setPhase: vi.fn(), updateSessionId: vi.fn(), diff --git a/src/auto-reply/reply/agent-runner.final-media-runreplyagent.test.ts b/src/auto-reply/reply/agent-runner.final-media-runreplyagent.test.ts index 5dc6e1a5b425..b66404b9b78b 100644 --- a/src/auto-reply/reply/agent-runner.final-media-runreplyagent.test.ts +++ b/src/auto-reply/reply/agent-runner.final-media-runreplyagent.test.ts @@ -102,9 +102,6 @@ const { runReplyAgent } = await import("./agent-runner.js"); function createReplyOperation(): ReplyOperation { return { result: undefined, - startedAtMs: Date.now(), - lastActivityAtMs: Date.now(), - recordActivity: vi.fn(), setPhase: vi.fn(), fail: vi.fn(), complete: vi.fn(), diff --git a/src/auto-reply/reply/agent-runner.media-paths.test.ts b/src/auto-reply/reply/agent-runner.media-paths.test.ts index 7d43baa4aa5b..86ddd72fe089 100644 --- a/src/auto-reply/reply/agent-runner.media-paths.test.ts +++ b/src/auto-reply/reply/agent-runner.media-paths.test.ts @@ -259,9 +259,6 @@ const { runReplyAgent } = await import("./agent-runner.js"); function createReplyOperation(): ReplyOperation { return { result: undefined, - startedAtMs: Date.now(), - lastActivityAtMs: Date.now(), - recordActivity: vi.fn(), setPhase: vi.fn(), fail: vi.fn(), complete: vi.fn(), diff --git a/src/auto-reply/reply/agent-runner.ts b/src/auto-reply/reply/agent-runner.ts index 16b4177ae5e5..f999dd0a03cb 100644 --- a/src/auto-reply/reply/agent-runner.ts +++ b/src/auto-reply/reply/agent-runner.ts @@ -1293,8 +1293,6 @@ export async function runReplyAgent(params: { }, ); if (steerOutcome.queued) { - const activeReplyOperation = sessionKey ? replyRunRegistry.get(sessionKey) : undefined; - activeReplyOperation?.recordActivity(); await touchActiveSessionEntry(); typing.cleanup(); return undefined; diff --git a/src/auto-reply/reply/dispatch-from-config.stale-recovery.test.ts b/src/auto-reply/reply/dispatch-from-config.stale-recovery.test.ts index f63bf2ddd235..53f7fa476891 100644 --- a/src/auto-reply/reply/dispatch-from-config.stale-recovery.test.ts +++ b/src/auto-reply/reply/dispatch-from-config.stale-recovery.test.ts @@ -4,12 +4,12 @@ import type { ReplyPayload } from "../types.js"; import { createDispatcher, diagnosticMocks, + emptyConfig, mocks, noAbortResult, resetPluginTtsAndThreadMocks, runtimePluginMocks, } from "./dispatch-from-config.shared.test-harness.js"; -import { REPLY_RUN_STALE_TAKEOVER_MS } from "./reply-run-registry.js"; import { buildTestCtx } from "./test-ctx.js"; let dispatchReplyFromConfig: typeof import("./dispatch-from-config.js").dispatchReplyFromConfig; @@ -17,35 +17,10 @@ let createReplyOperation: typeof import("./reply-run-registry.js").createReplyOp let replyRunTesting: typeof import("./reply-run-registry.js").__testing; let resetInboundDedupe: typeof import("./inbound-dedupe.js").resetInboundDedupe; -const sessionKey = "agent:main:telegram:direct:1"; - function setNoAbort() { mocks.tryFastAbortFromMessage.mockResolvedValue(noAbortResult); } -function createVisibleDispatchParams(replyResolver: () => Promise) { - return { - ctx: buildTestCtx({ - Provider: "telegram", - Surface: "telegram", - OriginatingChannel: "telegram", - OriginatingTo: "user:1", - ChatType: "direct", - SessionKey: sessionKey, - MessageThreadId: "501.000", - BodyForAgent: "second telegram direct turn", - }), - cfg: { - diagnostics: { - stuckSessionWarnMs: 1_000, - stuckSessionAbortMs: 1_000, - }, - } as OpenClawConfig, - dispatcher: createDispatcher(), - replyResolver, - }; -} - describe("dispatchReplyFromConfig stale visible admission recovery", () => { beforeEach(async () => { ({ dispatchReplyFromConfig } = await import("./dispatch-from-config.js")); @@ -61,6 +36,11 @@ describe("dispatchReplyFromConfig stale visible admission recovery", () => { mocks.tryFastAbortFromMessage.mockReset(); setNoAbort(); diagnosticMocks.requestStuckDiagnosticSessionRecovery.mockReset(); + diagnosticMocks.requestStuckDiagnosticSessionRecovery.mockResolvedValue({ + status: "skipped", + action: "keep_lane", + reason: "active_reply_work", + }); }); afterEach(() => { @@ -69,28 +49,571 @@ describe("dispatchReplyFromConfig stale visible admission recovery", () => { resetInboundDedupe(); }); - it("waits for fresh visible reply work without invoking diagnostic recovery", async () => { + it("recovers stale visible reply work and retries dispatch admission", async () => { vi.useFakeTimers(); + const sessionKey = "agent:main:telegram:direct:1"; const activeOperation = createReplyOperation({ sessionKey, sessionId: "active-session", resetTriggered: false, }); activeOperation.setPhase("running"); + const dispatcher = createDispatcher(); const replyResolver = vi.fn(async () => ({ text: "telegram reply" }) satisfies ReplyPayload); - const dispatchParams = createVisibleDispatchParams(replyResolver); - let settled = false; + diagnosticMocks.requestStuckDiagnosticSessionRecovery.mockImplementationOnce(async () => { + activeOperation.fail("run_failed", new Error("stale reply operation")); + return { + status: "aborted", + action: "abort_embedded_run", + sessionId: "active-session", + sessionKey, + activeSessionId: "active-session", + activeWorkKind: "embedded_run", + aborted: true, + drained: true, + forceCleared: false, + released: 0, + }; + }); - const resultPromise = dispatchReplyFromConfig(dispatchParams).then((result) => { - settled = true; - return result; + const resultPromise = dispatchReplyFromConfig({ + ctx: buildTestCtx({ + Provider: "telegram", + Surface: "telegram", + OriginatingChannel: "telegram", + OriginatingTo: "user:1", + ChatType: "direct", + SessionKey: sessionKey, + MessageThreadId: "501.000", + BodyForAgent: "second telegram direct turn", + }), + cfg: { + diagnostics: { + stuckSessionWarnMs: 1_000, + stuckSessionAbortMs: 1_000, + }, + } as OpenClawConfig, + dispatcher, + replyResolver, }); await vi.advanceTimersByTimeAsync(1_000); + const result = await resultPromise; - expect(settled).toBe(false); + expect(diagnosticMocks.requestStuckDiagnosticSessionRecovery).toHaveBeenCalledWith( + expect.objectContaining({ + sessionId: "active-session", + sessionKey, + queueDepth: 1, + staleActiveProgressAbortMs: 1_000, + }), + ); + expect(result).toMatchObject({ + queuedFinal: true, + counts: { tool: 0, block: 0, final: 0 }, + }); + expect(replyResolver).toHaveBeenCalledTimes(1); + expect(dispatcher.sendFinalReply).toHaveBeenCalledTimes(1); + }); + + it("reclaims a pure stale reply registry lock when recovery finds no active work", async () => { + vi.useFakeTimers(); + const sessionKey = "agent:main:telegram:direct:pure-stale-registry"; + const activeOperation = createReplyOperation({ + sessionKey, + sessionId: "active-session", + resetTriggered: false, + }); + activeOperation.setPhase("running"); + const dispatcher = createDispatcher(); + const replyResolver = vi.fn(async () => ({ text: "telegram reply" }) satisfies ReplyPayload); + diagnosticMocks.requestStuckDiagnosticSessionRecovery.mockResolvedValue({ + status: "noop", + action: "none", + reason: "no_active_work", + sessionId: "active-session", + sessionKey, + }); + + const resultPromise = dispatchReplyFromConfig({ + ctx: buildTestCtx({ + Provider: "telegram", + Surface: "telegram", + OriginatingChannel: "telegram", + OriginatingTo: "user:1", + ChatType: "direct", + SessionKey: sessionKey, + MessageThreadId: "501.000", + BodyForAgent: "second telegram direct turn", + }), + cfg: { + diagnostics: { + stuckSessionWarnMs: 1_000, + stuckSessionAbortMs: 1_000, + }, + } as OpenClawConfig, + dispatcher, + replyResolver, + }); + + await vi.advanceTimersByTimeAsync(1_000); + const result = await resultPromise; + + expect(result).toMatchObject({ + queuedFinal: true, + counts: { tool: 0, block: 0, final: 0 }, + }); + expect(replyResolver).toHaveBeenCalledTimes(1); + expect(dispatcher.sendFinalReply).toHaveBeenCalledTimes(1); + expect(activeOperation.result).toMatchObject({ + kind: "failed", + code: "run_failed", + }); + }); + + it("does not clear a fresh reply operation with the same session id after recovery", async () => { + vi.useFakeTimers(); + const sessionKey = "agent:main:telegram:direct:fresh-same-session"; + const activeOperation = createReplyOperation({ + sessionKey, + sessionId: "active-session", + resetTriggered: false, + }); + activeOperation.setPhase("running"); + let freshOperation: ReturnType | undefined; + const dispatcher = createDispatcher(); + const replyResolver = vi.fn(async () => ({ text: "telegram reply" }) satisfies ReplyPayload); + diagnosticMocks.requestStuckDiagnosticSessionRecovery + .mockImplementationOnce(async () => { + activeOperation.complete(); + freshOperation = createReplyOperation({ + sessionKey, + sessionId: "active-session", + resetTriggered: false, + }); + freshOperation.setPhase("running"); + return { + status: "noop", + action: "none", + reason: "no_active_work", + sessionId: "active-session", + sessionKey, + }; + }) + .mockImplementationOnce(async () => { + freshOperation?.fail("run_failed", new Error("fresh operation later became stale")); + return { + status: "aborted", + action: "abort_embedded_run", + sessionId: "active-session", + sessionKey, + activeSessionId: "active-session", + activeWorkKind: "embedded_run", + aborted: true, + drained: true, + forceCleared: false, + released: 0, + }; + }); + + const resultPromise = dispatchReplyFromConfig({ + ctx: buildTestCtx({ + Provider: "telegram", + Surface: "telegram", + OriginatingChannel: "telegram", + OriginatingTo: "user:1", + ChatType: "direct", + SessionKey: sessionKey, + MessageThreadId: "501.000", + BodyForAgent: "second telegram direct turn", + }), + cfg: { + diagnostics: { + stuckSessionWarnMs: 1_000, + stuckSessionAbortMs: 1_000, + }, + } as OpenClawConfig, + dispatcher, + replyResolver, + }); + + await vi.advanceTimersByTimeAsync(1_000); + expect(freshOperation?.result).toBeNull(); expect(replyResolver).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1_000); + const result = await resultPromise; + + expect(diagnosticMocks.requestStuckDiagnosticSessionRecovery).toHaveBeenCalledTimes(2); + expect(result).toMatchObject({ + queuedFinal: true, + counts: { tool: 0, block: 0, final: 0 }, + }); + expect(replyResolver).toHaveBeenCalledTimes(1); + expect(dispatcher.sendFinalReply).toHaveBeenCalledTimes(1); + }); + + it("keeps waiting when recovery observes active reply work", async () => { + vi.useFakeTimers(); + const sessionKey = "agent:main:telegram:direct:active-reply-work"; + const activeOperation = createReplyOperation({ + sessionKey, + sessionId: "active-session", + resetTriggered: false, + }); + activeOperation.setPhase("running"); + const dispatcher = createDispatcher(); + const replyResolver = vi.fn(async () => ({ text: "telegram reply" }) satisfies ReplyPayload); + diagnosticMocks.requestStuckDiagnosticSessionRecovery + .mockResolvedValueOnce({ + status: "skipped", + action: "keep_lane", + reason: "active_reply_work", + sessionId: "active-session", + sessionKey, + activeSessionId: "active-session", + activeWorkKind: "embedded_run", + }) + .mockImplementationOnce(async () => { + activeOperation.fail("run_failed", new Error("stale reply operation")); + return { + status: "aborted", + action: "abort_embedded_run", + sessionId: "active-session", + sessionKey, + activeSessionId: "active-session", + activeWorkKind: "embedded_run", + aborted: true, + drained: true, + forceCleared: false, + released: 0, + }; + }); + + const resultPromise = dispatchReplyFromConfig({ + ctx: buildTestCtx({ + Provider: "telegram", + Surface: "telegram", + OriginatingChannel: "telegram", + OriginatingTo: "user:1", + ChatType: "direct", + SessionKey: sessionKey, + MessageThreadId: "501.000", + BodyForAgent: "second telegram direct turn", + }), + cfg: { + diagnostics: { + stuckSessionWarnMs: 1_000, + stuckSessionAbortMs: 1_000, + }, + } as OpenClawConfig, + dispatcher, + replyResolver, + }); + + await vi.advanceTimersByTimeAsync(1_000); + expect(activeOperation.result).toBeNull(); + expect(replyResolver).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1_000); + const result = await resultPromise; + + expect(diagnosticMocks.requestStuckDiagnosticSessionRecovery).toHaveBeenCalledTimes(2); + expect(result).toMatchObject({ + queuedFinal: true, + counts: { tool: 0, block: 0, final: 0 }, + }); + expect(replyResolver).toHaveBeenCalledTimes(1); + expect(dispatcher.sendFinalReply).toHaveBeenCalledTimes(1); + }); + + it("keeps waiting when another recovery is already in flight", async () => { + vi.useFakeTimers(); + const sessionKey = "agent:main:telegram:direct:in-flight"; + const activeOperation = createReplyOperation({ + sessionKey, + sessionId: "active-session", + resetTriggered: false, + }); + activeOperation.setPhase("running"); + const dispatcher = createDispatcher(); + const replyResolver = vi.fn(async () => ({ text: "telegram reply" }) satisfies ReplyPayload); + diagnosticMocks.requestStuckDiagnosticSessionRecovery + .mockResolvedValueOnce({ + status: "skipped", + action: "observe_only", + reason: "already_in_flight", + sessionId: "active-session", + sessionKey, + }) + .mockImplementationOnce(async () => { + activeOperation.fail("run_failed", new Error("stale reply operation")); + return { + status: "aborted", + action: "abort_embedded_run", + sessionId: "active-session", + sessionKey, + activeSessionId: "active-session", + activeWorkKind: "embedded_run", + aborted: true, + drained: true, + forceCleared: false, + released: 0, + }; + }); + + const resultPromise = dispatchReplyFromConfig({ + ctx: buildTestCtx({ + Provider: "telegram", + Surface: "telegram", + OriginatingChannel: "telegram", + OriginatingTo: "user:1", + ChatType: "direct", + SessionKey: sessionKey, + MessageThreadId: "501.000", + BodyForAgent: "second telegram direct turn", + }), + cfg: { + diagnostics: { + stuckSessionWarnMs: 1_000, + stuckSessionAbortMs: 1_000, + }, + } as OpenClawConfig, + dispatcher, + replyResolver, + }); + + await vi.advanceTimersByTimeAsync(1_000); + expect(replyResolver).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1_000); + const result = await resultPromise; + + expect(diagnosticMocks.requestStuckDiagnosticSessionRecovery).toHaveBeenCalledTimes(2); + expect(result).toMatchObject({ + queuedFinal: true, + counts: { tool: 0, block: 0, final: 0 }, + }); + expect(replyResolver).toHaveBeenCalledTimes(1); + expect(dispatcher.sendFinalReply).toHaveBeenCalledTimes(1); + }); + + it("keeps waiting when recovery observes an active lane task", async () => { + vi.useFakeTimers(); + const sessionKey = "agent:main:telegram:direct:active-lane-task"; + const activeOperation = createReplyOperation({ + sessionKey, + sessionId: "active-session", + resetTriggered: false, + }); + activeOperation.setPhase("running"); + const dispatcher = createDispatcher(); + const replyResolver = vi.fn(async () => ({ text: "telegram reply" }) satisfies ReplyPayload); + diagnosticMocks.requestStuckDiagnosticSessionRecovery + .mockResolvedValueOnce({ + status: "skipped", + action: "keep_lane", + reason: "active_lane_task", + sessionId: "active-session", + sessionKey, + activeCount: 1, + queuedCount: 1, + }) + .mockImplementationOnce(async () => { + activeOperation.fail("run_failed", new Error("stale reply operation")); + return { + status: "aborted", + action: "abort_embedded_run", + sessionId: "active-session", + sessionKey, + activeSessionId: "active-session", + activeWorkKind: "embedded_run", + aborted: true, + drained: true, + forceCleared: false, + released: 0, + }; + }); + + const resultPromise = dispatchReplyFromConfig({ + ctx: buildTestCtx({ + Provider: "telegram", + Surface: "telegram", + OriginatingChannel: "telegram", + OriginatingTo: "user:1", + ChatType: "direct", + SessionKey: sessionKey, + MessageThreadId: "501.000", + BodyForAgent: "second telegram direct turn", + }), + cfg: { + diagnostics: { + stuckSessionWarnMs: 1_000, + stuckSessionAbortMs: 1_000, + }, + } as OpenClawConfig, + dispatcher, + replyResolver, + }); + + await vi.advanceTimersByTimeAsync(1_000); + expect(replyResolver).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1_000); + const result = await resultPromise; + + expect(diagnosticMocks.requestStuckDiagnosticSessionRecovery).toHaveBeenCalledTimes(2); + expect(result).toMatchObject({ + queuedFinal: true, + counts: { tool: 0, block: 0, final: 0 }, + }); + expect(replyResolver).toHaveBeenCalledTimes(1); + expect(dispatcher.sendFinalReply).toHaveBeenCalledTimes(1); + }); + + it("does not clear active reply work when recovery fails", async () => { + vi.useFakeTimers(); + const sessionKey = "agent:main:telegram:direct:recovery-failed"; + const activeOperation = createReplyOperation({ + sessionKey, + sessionId: "active-session", + resetTriggered: false, + }); + activeOperation.setPhase("running"); + const dispatcher = createDispatcher(); + const replyResolver = vi.fn(async () => ({ text: "telegram reply" }) satisfies ReplyPayload); + diagnosticMocks.requestStuckDiagnosticSessionRecovery.mockResolvedValue({ + status: "failed", + action: "none", + reason: "exception", + sessionId: "active-session", + sessionKey, + error: "recovery failed", + }); + + const resultPromise = dispatchReplyFromConfig({ + ctx: buildTestCtx({ + Provider: "telegram", + Surface: "telegram", + OriginatingChannel: "telegram", + OriginatingTo: "user:1", + ChatType: "direct", + SessionKey: sessionKey, + MessageThreadId: "501.000", + BodyForAgent: "second telegram direct turn", + }), + cfg: { + diagnostics: { + stuckSessionWarnMs: 1_000, + stuckSessionAbortMs: 1_000, + }, + } as OpenClawConfig, + dispatcher, + replyResolver, + }); + + await vi.advanceTimersByTimeAsync(1_000); + const result = await resultPromise; + + expect(diagnosticMocks.requestStuckDiagnosticSessionRecovery).toHaveBeenCalledTimes(1); + expect(result).toMatchObject({ + queuedFinal: false, + counts: { tool: 0, block: 0, final: 0 }, + }); + expect(activeOperation.result).toBeNull(); + expect(replyResolver).not.toHaveBeenCalled(); + expect(dispatcher.sendFinalReply).not.toHaveBeenCalled(); + }); + + it("clears stale reply work after recovery releases lane state", async () => { + vi.useFakeTimers(); + const sessionKey = "agent:main:telegram:direct:released-lane"; + const activeOperation = createReplyOperation({ + sessionKey, + sessionId: "active-session", + resetTriggered: false, + }); + activeOperation.setPhase("running"); + const dispatcher = createDispatcher(); + const replyResolver = vi.fn(async () => ({ text: "telegram reply" }) satisfies ReplyPayload); + diagnosticMocks.requestStuckDiagnosticSessionRecovery.mockResolvedValue({ + status: "released", + action: "release_lane", + sessionId: "active-session", + sessionKey, + released: 1, + }); + + const resultPromise = dispatchReplyFromConfig({ + ctx: buildTestCtx({ + Provider: "telegram", + Surface: "telegram", + OriginatingChannel: "telegram", + OriginatingTo: "user:1", + ChatType: "direct", + SessionKey: sessionKey, + MessageThreadId: "501.000", + BodyForAgent: "second telegram direct turn", + }), + cfg: { + diagnostics: { + stuckSessionWarnMs: 1_000, + stuckSessionAbortMs: 1_000, + }, + } as OpenClawConfig, + dispatcher, + replyResolver, + }); + + await vi.advanceTimersByTimeAsync(1_000); + expect(diagnosticMocks.requestStuckDiagnosticSessionRecovery).toHaveBeenCalledTimes(1); + const result = await resultPromise; + + expect(result).toMatchObject({ + queuedFinal: true, + counts: { tool: 0, block: 0, final: 0 }, + }); + expect(replyResolver).toHaveBeenCalledTimes(1); + expect(dispatcher.sendFinalReply).toHaveBeenCalledTimes(1); + expect(activeOperation.result).toMatchObject({ + kind: "failed", + code: "run_failed", + }); + }); + + it("does not run visible stuck recovery when diagnostics are disabled", async () => { + vi.useFakeTimers(); + const sessionKey = "agent:main:telegram:direct:diagnostics-disabled"; + const activeOperation = createReplyOperation({ + sessionKey, + sessionId: "active-session", + resetTriggered: false, + }); + activeOperation.setPhase("running"); + const dispatcher = createDispatcher(); + const replyResolver = vi.fn(async () => ({ text: "telegram reply" }) satisfies ReplyPayload); + + const resultPromise = dispatchReplyFromConfig({ + ctx: buildTestCtx({ + Provider: "telegram", + Surface: "telegram", + OriginatingChannel: "telegram", + OriginatingTo: "user:1", + ChatType: "direct", + SessionKey: sessionKey, + MessageThreadId: "501.000", + BodyForAgent: "second telegram direct turn", + }), + cfg: { + diagnostics: { + enabled: false, + stuckSessionWarnMs: 1_000, + stuckSessionAbortMs: 1_000, + }, + } as OpenClawConfig, + dispatcher, + replyResolver, + }); + + await vi.advanceTimersByTimeAsync(1_000); expect(diagnosticMocks.requestStuckDiagnosticSessionRecovery).not.toHaveBeenCalled(); + expect(replyResolver).not.toHaveBeenCalled(); activeOperation.complete(); const result = await resultPromise; @@ -100,31 +623,68 @@ describe("dispatchReplyFromConfig stale visible admission recovery", () => { counts: { tool: 0, block: 0, final: 0 }, }); expect(replyResolver).toHaveBeenCalledTimes(1); - expect(dispatchParams.dispatcher.sendFinalReply).toHaveBeenCalledTimes(1); + expect(dispatcher.sendFinalReply).toHaveBeenCalledTimes(1); }); - it("reclaims stale visible reply work through admission and dispatches the turn", async () => { - vi.useFakeTimers(); - const startedAt = Date.now(); + it("releases inbound dedupe when active reply admission is aborted before processing", async () => { + const sessionKey = "agent:main:telegram:direct:dedupe"; const activeOperation = createReplyOperation({ sessionKey, sessionId: "active-session", resetTriggered: false, }); activeOperation.setPhase("running"); - const replyResolver = vi.fn(async () => ({ text: "telegram reply" }) satisfies ReplyPayload); - const dispatchParams = createVisibleDispatchParams(replyResolver); - vi.setSystemTime(startedAt + REPLY_RUN_STALE_TAKEOVER_MS + 1); + const abortController = new AbortController(); + const ctx = buildTestCtx({ + Provider: "telegram", + Surface: "telegram", + OriginatingChannel: "telegram", + OriginatingTo: "telegram:user-1", + To: "telegram:user-1", + ChatType: "direct", + SessionKey: sessionKey, + MessageSid: "message-1", + BodyForAgent: "second visible turn", + }); + const firstDispatcher = createDispatcher(); + const firstReplyResolver = vi.fn( + async () => ({ text: "should not run" }) satisfies ReplyPayload, + ); - const result = await dispatchReplyFromConfig(dispatchParams); + const firstResult = dispatchReplyFromConfig({ + ctx, + cfg: emptyConfig, + dispatcher: firstDispatcher, + replyOptions: { abortSignal: abortController.signal }, + replyResolver: firstReplyResolver, + }); + setTimeout(() => abortController.abort(), 0); - expect(diagnosticMocks.requestStuckDiagnosticSessionRecovery).not.toHaveBeenCalled(); - expect(activeOperation.result).toEqual({ kind: "failed", code: "run_stalled" }); - expect(result).toMatchObject({ + await expect(firstResult).resolves.toMatchObject({ + queuedFinal: false, + counts: { tool: 0, block: 0, final: 0 }, + }); + expect(firstReplyResolver).not.toHaveBeenCalled(); + expect(firstDispatcher.sendFinalReply).not.toHaveBeenCalled(); + + activeOperation.complete(); + + const secondDispatcher = createDispatcher(); + const secondReplyResolver = vi.fn( + async () => ({ text: "runs after dedupe release" }) satisfies ReplyPayload, + ); + await expect( + dispatchReplyFromConfig({ + ctx, + cfg: emptyConfig, + dispatcher: secondDispatcher, + replyResolver: secondReplyResolver, + }), + ).resolves.toMatchObject({ queuedFinal: true, counts: { tool: 0, block: 0, final: 0 }, }); - expect(replyResolver).toHaveBeenCalledTimes(1); - expect(dispatchParams.dispatcher.sendFinalReply).toHaveBeenCalledTimes(1); + expect(secondReplyResolver).toHaveBeenCalledTimes(1); + expect(secondDispatcher.sendFinalReply).toHaveBeenCalledTimes(1); }); }); diff --git a/src/auto-reply/reply/dispatch-from-config.ts b/src/auto-reply/reply/dispatch-from-config.ts index f565feb32af1..0107b3a1aedc 100644 --- a/src/auto-reply/reply/dispatch-from-config.ts +++ b/src/auto-reply/reply/dispatch-from-config.ts @@ -67,10 +67,15 @@ import { measureDiagnosticsTimelineSpan } from "../../infra/diagnostics-timeline import { formatErrorMessage } from "../../infra/errors.js"; import { getSessionBindingService } from "../../infra/outbound/session-binding-service.js"; import { isAbortError } from "../../infra/unhandled-rejections.js"; +import type { StuckSessionRecoveryOutcome } from "../../logging/diagnostic-session-recovery.js"; import { logMessageDispatchCompleted, logMessageDispatchStarted, + isStuckSessionRecoveryEnabled, markDiagnosticSessionProgress, + requestStuckDiagnosticSessionRecovery, + resolveStuckSessionAbortMs, + resolveStuckSessionWarnMs, } from "../../logging/diagnostic.js"; import { createDiagnosticMessageLifecycle } from "../../logging/message-lifecycle.js"; import { createSubsystemLogger } from "../../logging/subsystem.js"; @@ -150,7 +155,11 @@ import type { ReplyDispatcher, } from "./reply-dispatcher.types.js"; import { readDispatcherFailedCounts } from "./reply-dispatcher.types.js"; -import { replyRunRegistry, type ReplyOperation } from "./reply-run-registry.js"; +import { + forceClearReplyRunBySessionId, + replyRunRegistry, + type ReplyOperation, +} from "./reply-run-registry.js"; import { createReplyDeliveryContext, resolveReplyDeliveryAccountId, @@ -803,6 +812,31 @@ export function getDispatcherFinalOutcomeCounts(dispatcher: DispatcherOutcomeCou }; } +function visibleRecoveryClearedActiveWork(outcome: StuckSessionRecoveryOutcome): boolean { + return ( + outcome.status === "aborted" || + outcome.status === "released" || + (outcome.status === "noop" && outcome.reason === "no_active_work") + ); +} + +function isSameReplyOperation( + left: ReplyOperation | undefined, + right: ReplyOperation | undefined, +): boolean { + return Boolean(left && right && left === right); +} + +function visibleRecoveryShouldKeepWaiting(outcome: StuckSessionRecoveryOutcome): boolean { + return ( + outcome.status === "skipped" && + (outcome.reason === "active_reply_work" || + outcome.reason === "active_embedded_run" || + outcome.reason === "active_lane_task" || + outcome.reason === "already_in_flight") + ); +} + function transcriptMirrorForDeliveredPayload( metadata: TranscriptMirror, payload: ReplyPayload, @@ -1203,6 +1237,10 @@ export async function dispatchReplyFromConfig( markDiagnosticSessionProgress({ sessionKey: acpDispatchSessionKey }); } }; + const visibleReplyRecoveryWaitMs = (() => { + const warnMs = resolveStuckSessionWarnMs(cfg); + return resolveStuckSessionAbortMs(cfg, warnMs); + })(); const sessionStoreEntry = boundAcpDispatchSessionKey ? resolveSessionStoreLookup({ ...ctx, SessionKey: boundAcpDispatchSessionKey }, cfg) : initialSessionStoreEntry; @@ -1294,7 +1332,7 @@ export async function dispatchReplyFromConfig( if (!dispatchOperationSessionKey) { return { status: "ready" }; } - const operationSessionId = + let operationSessionId = dispatchAbortOperation?.sessionId ?? initialSessionStoreEntry.entry?.sessionId ?? sessionStoreEntry.entry?.sessionId ?? @@ -1308,7 +1346,23 @@ export async function dispatchReplyFromConfig( ctx, routeThreadId, }); - const admission = await admitReplyTurn({ + const shouldRecoverStaleVisibleOperation = + phase === "dispatch" && + replyTurnKind === "visible" && + !allowSlackRoutedThreadBypass && + isStuckSessionRecoveryEnabled(cfg) && + params.replyOptions?.abortSignal?.aborted !== true; + const recoverStaleVisibleOperation = async ( + activeOperation: ReplyOperation, + ): Promise => + requestStuckDiagnosticSessionRecovery({ + sessionId: activeOperation.sessionId, + sessionKey: dispatchOperationSessionKey, + ageMs: visibleReplyRecoveryWaitMs, + queueDepth: 1, + staleActiveProgressAbortMs: visibleReplyRecoveryWaitMs, + }); + let admission = await admitReplyTurn({ sessionKey: dispatchOperationSessionKey, sessionId: operationSessionId, kind: replyTurnKind, @@ -1316,7 +1370,55 @@ export async function dispatchReplyFromConfig( routeThreadId, upstreamAbortSignal: params.replyOptions?.abortSignal, waitForActive: !allowActivePreDispatch && !allowSlackRoutedThreadBypass, + ...(shouldRecoverStaleVisibleOperation ? { waitTimeoutMs: visibleReplyRecoveryWaitMs } : {}), }); + if (shouldRecoverStaleVisibleOperation) { + while ( + admission.status === "skipped" && + admission.reason === "active-run" && + admission.activeOperation + ) { + operationSessionId = admission.activeOperation.sessionId; + const recovery = await recoverStaleVisibleOperation(admission.activeOperation); + let activeAfterRecovery = replyRunRegistry.get(dispatchOperationSessionKey); + if ( + recovery && + visibleRecoveryClearedActiveWork(recovery) && + isSameReplyOperation(activeAfterRecovery, admission.activeOperation) + ) { + forceClearReplyRunBySessionId( + admission.activeOperation.sessionId, + new Error("Stale visible reply operation recovered without clearing reply registry"), + ); + activeAfterRecovery = replyRunRegistry.get(dispatchOperationSessionKey); + if (isSameReplyOperation(activeAfterRecovery, admission.activeOperation)) { + break; + } + } + const replyOperationStillActive = Boolean(activeAfterRecovery); + if ( + replyOperationStillActive && + (!recovery || + (!visibleRecoveryClearedActiveWork(recovery) && + !visibleRecoveryShouldKeepWaiting(recovery))) + ) { + break; + } + if (activeAfterRecovery) { + operationSessionId = activeAfterRecovery.sessionId; + } + admission = await admitReplyTurn({ + sessionKey: dispatchOperationSessionKey, + sessionId: operationSessionId, + kind: replyTurnKind, + resetTriggered: false, + routeThreadId, + upstreamAbortSignal: params.replyOptions?.abortSignal, + waitForActive: replyOperationStillActive, + waitTimeoutMs: visibleReplyRecoveryWaitMs, + }); + } + } if (admission.status === "skipped") { if (allowActivePreDispatch && admission.reason === "active-run") { preDispatchAbortOperation = admission.activeOperation; @@ -2735,7 +2837,6 @@ export async function dispatchReplyFromConfig( if (isDispatchOperationAborted()) { return; } - dispatchReplyOperation?.recordActivity(); markProgress(); if (options?.waitForDirectBlockReplyDelivery) { await waitForPendingDirectBlockReplyDelivery(dispatchAbortOperation?.abortSignal); @@ -2871,7 +2972,6 @@ export async function dispatchReplyFromConfig( waitForDirectBlockReplyDelivery: true, }), onToolResult: (payload: ReplyPayload) => { - dispatchReplyOperation?.recordActivity(); markProgress(); const run = async () => { if (isDispatchOperationAborted()) { diff --git a/src/auto-reply/reply/followup-runner.ts b/src/auto-reply/reply/followup-runner.ts index 296593990986..17a75331b144 100644 --- a/src/auto-reply/reply/followup-runner.ts +++ b/src/auto-reply/reply/followup-runner.ts @@ -1011,7 +1011,6 @@ export function createFollowupRunner(params: { emitLifecycleTerminal: false, onAgentRunStart: () => opts?.onAgentRunStart?.(runId), suppressAssistantBridge: run.silentExpected, - onActivity: () => replyOperation?.recordActivity(), onToolEvent: async (payload) => { await cliToolSummaryTracker.noteToolEvent(payload); if (payload.phase === "result") { @@ -1253,7 +1252,6 @@ export function createFollowupRunner(params: { shouldEmitToolOutput: shouldEmitToolOutputProgress, onToolResult: deliverFollowupToolSummary, onAgentEvent: (evt) => { - replyOperation?.recordActivity(); lifecycleBackstop.note(evt); return enqueueProgressDelivery(async () => { await forwardFollowupProgressEvent({ diff --git a/src/auto-reply/reply/reply-run-registry.test.ts b/src/auto-reply/reply/reply-run-registry.test.ts index ddf415b5985e..72fdbd258b0b 100644 --- a/src/auto-reply/reply/reply-run-registry.test.ts +++ b/src/auto-reply/reply/reply-run-registry.test.ts @@ -6,26 +6,22 @@ import { markDiagnosticRunProgressForTest, resetDiagnosticRunActivityForTest, } from "../../logging/diagnostic-run-activity.js"; -import { diagnosticLogger } from "../../logging/diagnostic-runtime.js"; import { MAX_TIMER_TIMEOUT_MS } from "../../shared/number-coercion.js"; import { testing, abortActiveReplyRuns, createReplyOperation, - expireStaleReplyOperation, forceClearReplyRunBySessionId, isReplyRunEvidenceStaleBySessionId, isReplyRunActiveForSessionId, isReplyRunAbortableForCompaction, queueReplyRunMessage, REPLY_RUN_IDLE_SETTLE_TIMEOUT_MS, - REPLY_RUN_TERMINAL_SETTLE_TIMEOUT_MS, replyRunRegistry, runAfterReplyOperationClear, resolveActiveReplyRunSessionId, waitForReplyRunEndBySessionId, } from "./reply-run-registry.js"; -import { admitReplyTurn } from "./reply-turn-admission.js"; describe("reply run registry", () => { afterEach(() => { @@ -389,147 +385,6 @@ describe("reply run registry", () => { expect(afterClear).toHaveBeenCalledTimes(1); }); - it("force-releases a running aborted operation when the owner never returns", async () => { - vi.useFakeTimers(); - try { - const cancel = vi.fn(); - const operation = createReplyOperation({ - sessionKey: "agent:main:hung-abort", - sessionId: "session-hung-abort", - resetTriggered: false, - }); - operation.attachBackend({ - kind: "embedded", - cancel, - isStreaming: () => true, - }); - operation.setPhase("running"); - const afterClear = vi.fn(); - runAfterReplyOperationClear(operation, afterClear); - const waitPromise = replyRunRegistry.waitForIdle("agent:main:hung-abort"); - - operation.abortByUser(); - - await vi.advanceTimersByTimeAsync(REPLY_RUN_TERMINAL_SETTLE_TIMEOUT_MS - 1); - expect(replyRunRegistry.get("agent:main:hung-abort")).toBe(operation); - expect(afterClear).not.toHaveBeenCalled(); - - await vi.advanceTimersByTimeAsync(1); - - expect(replyRunRegistry.get("agent:main:hung-abort")).toBeUndefined(); - await expect(waitPromise).resolves.toBe(true); - expect(afterClear).toHaveBeenCalledTimes(1); - const next = await admitReplyTurn({ - sessionKey: "agent:main:hung-abort", - sessionId: "session-after-hung-abort", - kind: "visible", - resetTriggered: false, - }); - expect(next.status).toBe("owned"); - if (next.status === "owned") { - next.operation.complete(); - } - } finally { - await vi.runOnlyPendingTimersAsync(); - vi.useRealTimers(); - } - }); - - it("keeps late owner complete harmless after forced terminal release", async () => { - vi.useFakeTimers(); - try { - const operation = createReplyOperation({ - sessionKey: "agent:main:late-complete", - sessionId: "session-late-complete", - resetTriggered: false, - }); - operation.setPhase("running"); - const afterClear = vi.fn(); - runAfterReplyOperationClear(operation, afterClear); - - operation.abortByUser(); - await vi.advanceTimersByTimeAsync(REPLY_RUN_TERMINAL_SETTLE_TIMEOUT_MS); - operation.complete(); - - expect(replyRunRegistry.isActive("agent:main:late-complete")).toBe(false); - expect(afterClear).toHaveBeenCalledTimes(1); - } finally { - await vi.runOnlyPendingTimersAsync(); - vi.useRealTimers(); - } - }); - - it("force-releases retained failures when the owner never completes", async () => { - vi.useFakeTimers(); - try { - const operation = createReplyOperation({ - sessionKey: "agent:main:retained-hung-failure", - sessionId: "session-retained-hung-failure", - resetTriggered: false, - }); - operation.retainFailureUntilComplete(); - const afterClear = vi.fn(); - runAfterReplyOperationClear(operation, afterClear); - - operation.fail("run_failed", new Error("delivery payload pending")); - await vi.advanceTimersByTimeAsync(REPLY_RUN_TERMINAL_SETTLE_TIMEOUT_MS); - - expect(replyRunRegistry.get("agent:main:retained-hung-failure")).toBeUndefined(); - expect(afterClear).toHaveBeenCalledTimes(1); - expect(operation.result).toMatchObject({ kind: "failed", code: "run_failed" }); - } finally { - await vi.runOnlyPendingTimersAsync(); - vi.useRealTimers(); - } - }); - - it("keeps run_stalled attribution when backend cancel re-enters abortByUser", () => { - const operation = createReplyOperation({ - sessionKey: "agent:main:reentrant-expire", - sessionId: "reentrant-session", - resetTriggered: false, - }); - operation.attachBackend({ - kind: "embedded", - // Mirrors the run loop's abort handler: backend cancellation propagates - // synchronously back into a user-shaped abort on the same operation. - cancel: () => { - operation.abortByUser(); - }, - isStreaming: () => true, - }); - operation.setPhase("running"); - - expect(expireStaleReplyOperation(operation, "no_activity")).toBe(true); - expect(operation.result).toEqual({ kind: "failed", code: "run_stalled" }); - expect(replyRunRegistry.get("agent:main:reentrant-expire")).toBeUndefined(); - }); - - it("cancels terminal settle when the owner clears state first", async () => { - vi.useFakeTimers(); - try { - const warnSpy = vi.spyOn(diagnosticLogger, "warn").mockImplementation(() => undefined); - const operation = createReplyOperation({ - sessionKey: "agent:main:owner-clears", - sessionId: "session-owner-clears", - resetTriggered: false, - }); - operation.setPhase("running"); - - operation.abortByUser(); - operation.complete(); - await vi.advanceTimersByTimeAsync(REPLY_RUN_TERMINAL_SETTLE_TIMEOUT_MS); - - expect(replyRunRegistry.isActive("agent:main:owner-clears")).toBe(false); - expect(warnSpy).not.toHaveBeenCalledWith( - expect.stringContaining("reply run terminal settle: forced release"), - ); - } finally { - await vi.runOnlyPendingTimersAsync(); - vi.useRealTimers(); - } - }); - it("force-clears retained failed operations", () => { const operation = createReplyOperation({ sessionKey: "agent:main:main", diff --git a/src/auto-reply/reply/reply-run-registry.ts b/src/auto-reply/reply/reply-run-registry.ts index d235bc9e8363..ecaea212a5d1 100644 --- a/src/auto-reply/reply/reply-run-registry.ts +++ b/src/auto-reply/reply/reply-run-registry.ts @@ -1,9 +1,6 @@ // Tracks active reply runs so stop, queue, and status commands can coordinate. import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; -import { - createAgentRunRestartAbortError, - isAgentRunRestartAbortReason, -} from "../../agents/run-termination.js"; +import { createAgentRunRestartAbortError } from "../../agents/run-termination.js"; import { areDiagnosticsEnabledForProcess } from "../../infra/diagnostic-events.js"; import { BLOCKED_TOOL_CALL_ABORT_FLOOR_MS, @@ -11,7 +8,6 @@ import { markDiagnosticEmbeddedRunEnded, markDiagnosticEmbeddedRunStarted, } from "../../logging/diagnostic-run-activity.js"; -import { diagnosticLogger as diag } from "../../logging/diagnostic-runtime.js"; import { resolveGlobalSingleton } from "../../shared/global-singleton.js"; import { resolveTimerTimeoutMs } from "../../shared/number-coercion.js"; import type { ReplyFollowupAdmissionBarrierTimeoutPolicy } from "./reply-dispatcher.types.js"; @@ -48,7 +44,6 @@ export type ReplyOperationFailureCode = | "command_lane_cleared" | "aborted_by_user" | "session_corruption_reset" - | "run_stalled" | "run_failed"; export type ReplyOperationAbortCode = "aborted_by_user" | "aborted_for_restart"; @@ -66,7 +61,6 @@ export type ReplyOperation = { readonly resetTriggered: boolean; readonly phase: ReplyOperationPhase; readonly result: ReplyOperationResult | null; - readonly startedAtMs: number; readonly lastActivityAtMs: number; recordActivity(): void; setPhase(next: "queued" | "preflight_compacting" | "memory_flushing" | "running"): void; @@ -149,14 +143,7 @@ const replyRunState = resolveGlobalSingleton(REPLY_RUN_STATE_KEY, replyRunState.followupAdmissionBarriersByKey ??= new Map(); export const REPLY_RUN_IDLE_SETTLE_TIMEOUT_MS = 15_000; -// Terminal results must release the lane even if the owner never resumes. -// Without this, abort/failure can leave the session wedged until process restart. -export const REPLY_RUN_TERMINAL_SETTLE_TIMEOUT_MS = 60_000; -// Visible human turns may reclaim only runs with no real progress for this window. -// 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" | "stuck_recovery"; +const REPLY_RUN_STALE_TAKEOVER_MS = 10 * 60_000; export class ReplyRunAlreadyActiveError extends Error { constructor(sessionKey: string) { @@ -240,12 +227,6 @@ const afterClearCallbacksByOperation = new WeakMap< ReplyOperation, Set<(sessionId: string) => void> >(); -const terminalSettleTimersByOperation = new WeakMap(); -const terminalSettleTimers = new Set(); -const expireReplyOperationByOperation = new WeakMap< - ReplyOperation, - (reason: ReplyOperationStaleReason) => boolean ->(); function getAttachedBackend(operation: ReplyOperation): ReplyBackendHandle | undefined { return attachedBackendByOperation.get(operation); @@ -277,23 +258,6 @@ function flushReplyOperationAfterClear(operation: ReplyOperation, sessionId: str } } -function formatReplyOperationResult(result: ReplyOperationResult | null): string { - if (!result) { - return "none"; - } - return result.kind === "completed" ? result.kind : `${result.kind}:${result.code}`; -} - -function clearTerminalSettleTimer(operation: ReplyOperation): void { - const timer = terminalSettleTimersByOperation.get(operation); - if (!timer) { - return; - } - clearTimeout(timer); - terminalSettleTimers.delete(timer); - terminalSettleTimersByOperation.delete(operation); -} - function registerFollowupAdmissionBarrier( sessionKey: string, sessionId: string, @@ -431,28 +395,13 @@ export function createReplyOperation(params: { let currentSessionId = sessionId; let phase: ReplyOperationPhase = "queued"; let result: ReplyOperationResult | null = null; - const startedAtMs = Date.now(); - let lastActivityAtMs = startedAtMs; + let lastActivityAtMs = Date.now(); let stateCleared = false; let retainFailureUntilComplete = false; - const upstreamAbortSignal = params.upstreamAbortSignal; - let upstreamAbortHandler: (() => void) | undefined; - - const detachUpstreamAbort = () => { - if (!upstreamAbortHandler) { - return; - } - upstreamAbortSignal?.removeEventListener("abort", upstreamAbortHandler); - upstreamAbortHandler = undefined; - }; const recordActivity = () => { lastActivityAtMs = Date.now(); }; - const setResult = (next: ReplyOperationResult) => { - result = next; - recordActivity(); - }; const clearState = ( afterClearBarrier?: PromiseLike, @@ -462,9 +411,6 @@ export function createReplyOperation(params: { return; } stateCleared = true; - clearTerminalSettleTimer(operation); - expireReplyOperationByOperation.delete(operation); - detachUpstreamAbort(); const registeredBarrier = afterClearBarrier ? registerFollowupAdmissionBarrier( sessionKey, @@ -494,45 +440,33 @@ export function createReplyOperation(params: { } }; - const scheduleTerminalSettle = () => { - if (stateCleared || terminalSettleTimersByOperation.has(operation)) { - return; - } - // Retained terminal results get one delivery grace window, not a second - // lifetime. Expiry frees the lane and flushes lifecycle after-clear work — - // including skipping any delivery barrier the owner never got to register; - // followups may then interleave with a still-draining terminal delivery. - const timer = setTimeout(() => { - if (replyRunState.activeRunsByKey.get(sessionKey) !== operation) { - clearTerminalSettleTimer(operation); - return; - } - diag.warn( - `reply run terminal settle: forced release sessionKey=${sessionKey} phase=${phase} result=${formatReplyOperationResult( - result, - )} ageMs=${Date.now() - lastActivityAtMs} ranForMs=${Date.now() - startedAtMs}`, - ); - clearState(); - }, REPLY_RUN_TERMINAL_SETTLE_TIMEOUT_MS); - timer.unref?.(); - terminalSettleTimersByOperation.set(operation, timer); - terminalSettleTimers.add(timer); - }; - const abortWithReason = ( reason: ReplyBackendCancelReason, abortReason: unknown, opts?: { abortedCode?: ReplyOperationAbortCode }, ) => { if (opts?.abortedCode && !result) { - setResult({ kind: "aborted", code: opts.abortedCode }); - detachUpstreamAbort(); + result = { kind: "aborted", code: opts.abortedCode }; } phase = "aborted"; abortInternally(abortReason); getAttachedBackend(operation)?.cancel(reason); }; + if (params.upstreamAbortSignal) { + if (params.upstreamAbortSignal.aborted) { + abortInternally(params.upstreamAbortSignal.reason); + } else { + params.upstreamAbortSignal.addEventListener( + "abort", + () => { + abortInternally(params.upstreamAbortSignal?.reason); + }, + { once: true }, + ); + } + } + const operation: ReplyOperation = { get key() { return sessionKey; @@ -555,9 +489,6 @@ export function createReplyOperation(params: { get result() { return result; }, - get startedAtMs() { - return startedAtMs; - }, get lastActivityAtMs() { return lastActivityAtMs; }, @@ -622,7 +553,7 @@ export function createReplyOperation(params: { }, complete() { if (!result) { - setResult({ kind: "completed" }); + result = { kind: "completed" }; phase = "completed"; } clearState(); @@ -633,122 +564,49 @@ export function createReplyOperation(params: { }, completeWithAfterClearBarrier(barrier, timeoutMs) { if (!result) { - setResult({ kind: "completed" }); + result = { kind: "completed" }; phase = "completed"; } clearState(barrier, timeoutMs); }, fail(code, cause) { if (!result) { - setResult({ kind: "failed", code, cause }); + result = { kind: "failed", code, cause }; phase = "failed"; } if (!retainFailureUntilComplete) { clearState(); - } else { - scheduleTerminalSettle(); } }, abortByUser() { - if (result) { - return; - } const phaseBeforeAbort = phase; abortWithReason("user_abort", createUserAbortError(), { abortedCode: "aborted_by_user", }); if (phaseBeforeAbort === "queued") { clearState(); - } else { - scheduleTerminalSettle(); } }, abortForRestart() { - if (result) { - return; - } const phaseBeforeAbort = phase; abortWithReason("restart", createAgentRunRestartAbortError(), { abortedCode: "aborted_for_restart", }); if (phaseBeforeAbort === "queued") { clearState(); - } else { - scheduleTerminalSettle(); } }, }; - expireReplyOperationByOperation.set(operation, (reason) => { - if (replyRunState.activeRunsByKey.get(sessionKey) !== operation) { - return false; - } - // Set the terminal result BEFORE cancelling the backend: cancel can - // synchronously re-enter abortByUser() from the run loop's abort handler, - // which would stamp aborted_by_user and misattribute a watchdog expiry. - if (!result) { - setResult({ kind: "failed", code: "run_stalled" }); - phase = "failed"; - } - getAttachedBackend(operation)?.cancel("superseded"); - abortInternally(createUserAbortError()); - diag.warn( - `reply run stale takeover: forced release sessionKey=${sessionKey} reason=${reason} phase=${phase} result=${formatReplyOperationResult( - result, - )} ageMs=${Date.now() - lastActivityAtMs} ranForMs=${Date.now() - startedAtMs}`, - ); - clearState(); - return true; - }); - replyRunState.activeRunsByKey.set(sessionKey, operation); replyRunState.activeSessionIdsByKey.set(sessionKey, currentSessionId); replyRunState.activeKeysBySessionId.set(currentSessionId, sessionKey); registerWaitSessionId(sessionKey, currentSessionId); markReplyRunDiagnosticWorkStarted({ sessionKey, sessionId: currentSessionId }); - if (upstreamAbortSignal) { - const abortFromUpstream = () => { - if (result) { - return; - } - const restart = isAgentRunRestartAbortReason(upstreamAbortSignal.reason); - const phaseBeforeAbort = phase; - abortWithReason(restart ? "restart" : "user_abort", upstreamAbortSignal.reason, { - abortedCode: restart ? "aborted_for_restart" : "aborted_by_user", - }); - if (phaseBeforeAbort === "queued") { - clearState(); - } else { - scheduleTerminalSettle(); - } - }; - if (upstreamAbortSignal.aborted) { - abortFromUpstream(); - } else { - upstreamAbortHandler = abortFromUpstream; - upstreamAbortSignal.addEventListener("abort", upstreamAbortHandler, { once: true }); - } - } - return operation; } -export function expireStaleReplyOperation( - operation: ReplyOperation, - reason: ReplyOperationStaleReason, -): boolean { - 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); @@ -864,38 +722,25 @@ export function isReplyRunStreamingForSessionId(sessionId: string): boolean { return getAttachedBackend(operation)?.isStreaming() ?? false; } -export function resolveReplyRunStaleThresholdMs(operation: ReplyOperation): number { - const activity = getDiagnosticSessionActivitySnapshot({ - sessionId: operation.sessionId, - sessionKey: operation.key, - }); - return activity.activeWorkKind === "tool_call" - ? Math.max(REPLY_RUN_STALE_TAKEOVER_MS, BLOCKED_TOOL_CALL_ABORT_FLOOR_MS) - : REPLY_RUN_STALE_TAKEOVER_MS; -} - -export function isReplyRunEvidenceStale(operation: ReplyOperation): boolean { - if (operation.result) { +export function isReplyRunEvidenceStaleBySessionId(sessionId: string): boolean { + const operation = resolveReplyRunForCurrentSessionId(sessionId); + if (!operation || operation.result || !areDiagnosticsEnabledForProcess()) { return false; } const activity = getDiagnosticSessionActivitySnapshot({ sessionId: operation.sessionId, sessionKey: operation.key, }); + const staleThresholdMs = + activity.activeWorkKind === "tool_call" + ? Math.max(REPLY_RUN_STALE_TAKEOVER_MS, BLOCKED_TOOL_CALL_ABORT_FLOOR_MS) + : REPLY_RUN_STALE_TAKEOVER_MS; const replyActivityAgeMs = Date.now() - operation.lastActivityAtMs; const evidenceAgeMs = typeof activity.lastProgressAgeMs === "number" ? Math.min(replyActivityAgeMs, activity.lastProgressAgeMs) : replyActivityAgeMs; - return evidenceAgeMs > resolveReplyRunStaleThresholdMs(operation); -} - -export function isReplyRunEvidenceStaleBySessionId(sessionId: string): boolean { - if (!areDiagnosticsEnabledForProcess()) { - return false; - } - const operation = resolveReplyRunForCurrentSessionId(sessionId); - return operation ? isReplyRunEvidenceStale(operation) : false; + return evidenceAgeMs > staleThresholdMs; } export function queueReplyRunMessage(sessionId: string, text: string): boolean { @@ -912,12 +757,7 @@ export function queueReplyRunMessage(sessionId: string, text: string): boolean { if (!backend.isStreaming()) { return false; } - // Injection is user input, not run evidence: stamping activity here would let - // sub-10-minute user messages re-arm a wedged run's staleness window forever. - const queued = backend.queueMessage(text); - queued.catch((error: unknown) => { - diag.debug(`queued reply run message rejected: sessionId=${sessionId} error=${String(error)}`); - }); + void backend.queueMessage(text); return true; } @@ -1038,10 +878,6 @@ export const testing = { replyRunState.activeSessionIdsByKey.clear(); replyRunState.activeKeysBySessionId.clear(); replyRunState.waitKeysBySessionId.clear(); - for (const timer of terminalSettleTimers) { - clearTimeout(timer); - } - terminalSettleTimers.clear(); for (const waiters of replyRunState.waitersByKey.values()) { for (const waiter of waiters) { waiter.finish(false); diff --git a/src/auto-reply/reply/reply-turn-admission.test.ts b/src/auto-reply/reply/reply-turn-admission.test.ts index 15423f5a4263..8b148456ca74 100644 --- a/src/auto-reply/reply/reply-turn-admission.test.ts +++ b/src/auto-reply/reply/reply-turn-admission.test.ts @@ -1,15 +1,7 @@ // Tests reply turn admission decisions for active, queued, and aborted runs. import { afterEach, describe, expect, it, vi } from "vitest"; -import { - markDiagnosticToolStartedForTest, - resetDiagnosticRunActivityForTest, -} from "../../logging/diagnostic-run-activity.js"; import { createReplyOperation, - REPLY_RUN_IDLE_SETTLE_TIMEOUT_MS, - REPLY_RUN_STALE_TAKEOVER_MS, - REPLY_RUN_TERMINAL_SETTLE_TIMEOUT_MS, - replyRunRegistry, runAfterReplyOperationClear, testing, } from "./reply-run-registry.js"; @@ -18,7 +10,6 @@ import { admitReplyTurn } from "./reply-turn-admission.js"; describe("reply turn admission", () => { afterEach(() => { testing.resetReplyRunRegistry(); - resetDiagnosticRunActivityForTest(); }); it("waits for visible turns and reuses the active session id", async () => { @@ -302,217 +293,6 @@ describe("reply turn admission", () => { active.complete(); }); - it("lets visible turns reclaim a stale active operation", async () => { - vi.useFakeTimers(); - try { - const cancel = vi.fn(); - const startedAt = Date.now(); - const active = createReplyOperation({ - sessionKey: "agent:main:telegram:topic:stale-visible", - sessionId: "stale-session", - resetTriggered: false, - }); - active.attachBackend({ - kind: "embedded", - cancel, - isStreaming: () => true, - }); - active.setPhase("running"); - vi.setSystemTime(startedAt + REPLY_RUN_STALE_TAKEOVER_MS + 1); - - const result = await admitReplyTurn({ - sessionKey: "agent:main:telegram:topic:stale-visible", - sessionId: "replacement-session", - kind: "visible", - resetTriggered: false, - }); - - expect(active.result).toEqual({ kind: "failed", code: "run_stalled" }); - expect(active.abortSignal.aborted).toBe(true); - expect(cancel).toHaveBeenCalledWith("superseded"); - expect(result.status).toBe("owned"); - if (result.status === "owned") { - result.operation.complete(); - } - } finally { - await vi.runOnlyPendingTimersAsync(); - vi.useRealTimers(); - } - }); - - it("keeps visible turns waiting while an active operation is still fresh", async () => { - vi.useFakeTimers(); - try { - const active = createReplyOperation({ - sessionKey: "agent:main:telegram:topic:fresh-visible", - sessionId: "fresh-session", - resetTriggered: false, - }); - active.setPhase("running"); - active.recordActivity(); - const abortController = new AbortController(); - let settled = false; - const result = admitReplyTurn({ - sessionKey: "agent:main:telegram:topic:fresh-visible", - sessionId: "waiting-session", - kind: "visible", - resetTriggered: false, - upstreamAbortSignal: abortController.signal, - }).then((admission) => { - settled = true; - return admission; - }); - - await vi.advanceTimersByTimeAsync(REPLY_RUN_IDLE_SETTLE_TIMEOUT_MS); - expect(settled).toBe(false); - expect(replyRunRegistry.get("agent:main:telegram:topic:fresh-visible")).toBe(active); - - abortController.abort(); - await expect(result).resolves.toMatchObject({ - status: "skipped", - reason: "aborted", - activeOperation: active, - }); - } finally { - await vi.runOnlyPendingTimersAsync(); - vi.useRealTimers(); - } - }); - - it("defers takeover to the blocked-tool floor while a quiet tool is active", async () => { - vi.useFakeTimers(); - try { - const cancel = vi.fn(); - const startedAt = Date.now(); - const active = createReplyOperation({ - sessionKey: "agent:main:telegram:topic:quiet-tool", - sessionId: "quiet-tool-session", - resetTriggered: false, - }); - active.attachBackend({ - kind: "embedded", - cancel, - isStreaming: () => true, - }); - active.setPhase("running"); - markDiagnosticToolStartedForTest({ - sessionId: "quiet-tool-session", - sessionKey: "agent:main:telegram:topic:quiet-tool", - toolName: "exec", - toolCallId: "tool-quiet-1", - }); - - vi.setSystemTime(startedAt + 12 * 60_000); - let settled = false; - const waiting = admitReplyTurn({ - sessionKey: "agent:main:telegram:topic:quiet-tool", - sessionId: "replacement-quiet-tool", - kind: "visible", - resetTriggered: false, - }).then((admission) => { - settled = true; - return admission; - }); - await vi.advanceTimersByTimeAsync(REPLY_RUN_IDLE_SETTLE_TIMEOUT_MS); - expect(settled).toBe(false); - expect(cancel).not.toHaveBeenCalled(); - - vi.setSystemTime(startedAt + 16 * 60_000); - await vi.advanceTimersByTimeAsync(REPLY_RUN_IDLE_SETTLE_TIMEOUT_MS); - const result = await waiting; - expect(active.result).toEqual({ kind: "failed", code: "run_stalled" }); - expect(result.status).toBe("owned"); - if (result.status === "owned") { - result.operation.complete(); - } - } finally { - await vi.runOnlyPendingTimersAsync(); - vi.useRealTimers(); - } - }); - - it.each(["heartbeat", "queued_followup"] as const)( - "does not let %s turns reclaim a stale active operation", - async (kind) => { - vi.useFakeTimers(); - try { - const cancel = vi.fn(); - const startedAt = Date.now(); - const active = createReplyOperation({ - sessionKey: `agent:main:telegram:topic:stale-${kind}`, - sessionId: `stale-${kind}-session`, - resetTriggered: false, - }); - active.attachBackend({ - kind: "embedded", - cancel, - isStreaming: () => true, - }); - active.setPhase("running"); - vi.setSystemTime(startedAt + REPLY_RUN_STALE_TAKEOVER_MS + 1); - - const admission = admitReplyTurn({ - sessionKey: `agent:main:telegram:topic:stale-${kind}`, - sessionId: `replacement-${kind}-session`, - kind, - resetTriggered: false, - waitTimeoutMs: 1, - }); - if (kind === "queued_followup") { - await Promise.resolve(); - await vi.advanceTimersByTimeAsync(100); - } - const result = await admission; - - expect(result).toMatchObject({ - status: "skipped", - reason: "active-run", - activeOperation: active, - }); - expect(cancel).not.toHaveBeenCalled(); - expect(replyRunRegistry.get(`agent:main:telegram:topic:stale-${kind}`)).toBe(active); - active.complete(); - } finally { - await vi.runOnlyPendingTimersAsync(); - vi.useRealTimers(); - } - }, - ); - - it("lets visible turns reclaim terminal operations after settle grace elapsed", async () => { - vi.useFakeTimers(); - try { - const startedAt = Date.now(); - const active = createReplyOperation({ - sessionKey: "agent:main:telegram:topic:terminal-unreleased", - sessionId: "terminal-unreleased-session", - resetTriggered: false, - }); - active.setPhase("running"); - active.abortByUser(); - vi.setSystemTime(startedAt + REPLY_RUN_TERMINAL_SETTLE_TIMEOUT_MS); - - const result = await admitReplyTurn({ - sessionKey: "agent:main:telegram:topic:terminal-unreleased", - sessionId: "replacement-terminal-session", - kind: "visible", - resetTriggered: false, - }); - - expect(active.result).toEqual({ kind: "aborted", code: "aborted_by_user" }); - expect(replyRunRegistry.get("agent:main:telegram:topic:terminal-unreleased")).not.toBe( - active, - ); - expect(result.status).toBe("owned"); - if (result.status === "owned") { - result.operation.complete(); - } - } finally { - await vi.runOnlyPendingTimersAsync(); - vi.useRealTimers(); - } - }); - it("stops waiting when the caller aborts", async () => { const active = createReplyOperation({ sessionKey: "agent:main:telegram:topic:42", diff --git a/src/auto-reply/reply/reply-turn-admission.ts b/src/auto-reply/reply/reply-turn-admission.ts index 5813998875f7..e4bcf8be94c5 100644 --- a/src/auto-reply/reply/reply-turn-admission.ts +++ b/src/auto-reply/reply/reply-turn-admission.ts @@ -1,14 +1,10 @@ // Decides whether an inbound turn may start, queue, or abort a reply run. import { createReplyOperation, - expireStaleReplyOperation, - isReplyRunEvidenceStale, REPLY_RUN_IDLE_SETTLE_TIMEOUT_MS, - REPLY_RUN_TERMINAL_SETTLE_TIMEOUT_MS, replyRunRegistry, ReplyRunAlreadyActiveError, ReplyRunFollowupAdmissionBlockedError, - resolveReplyRunStaleThresholdMs, type ReplyOperation, waitForReplyRunFollowupAdmission, } from "./reply-run-registry.js"; @@ -29,31 +25,6 @@ function isAbortSignalAborted(signal: AbortSignal | undefined): boolean { return signal?.aborted === true; } -function expireVisibleStaleOperation(operation: ReplyOperation | undefined): boolean { - if (!operation) { - return false; - } - const idleMs = Date.now() - operation.lastActivityAtMs; - if (operation.result) { - return ( - idleMs >= REPLY_RUN_TERMINAL_SETTLE_TIMEOUT_MS && - expireStaleReplyOperation(operation, "terminal_unreleased") - ); - } - return isReplyRunEvidenceStale(operation) && expireStaleReplyOperation(operation, "no_activity"); -} - -function resolveVisibleActiveWaitMs(operation: ReplyOperation | undefined): number { - if (!operation) { - return REPLY_RUN_IDLE_SETTLE_TIMEOUT_MS; - } - const ageMs = Date.now() - operation.lastActivityAtMs; - const remainingMs = operation.result - ? REPLY_RUN_TERMINAL_SETTLE_TIMEOUT_MS - ageMs - : resolveReplyRunStaleThresholdMs(operation) - ageMs; - return Math.min(REPLY_RUN_IDLE_SETTLE_TIMEOUT_MS, Math.max(1, remainingMs)); -} - /** Waits for or claims the per-session reply run slot. */ export async function admitReplyTurn(params: { sessionKey: string; @@ -109,9 +80,6 @@ export async function admitReplyTurn(params: { throw error; } const activeOperation = replyRunRegistry.get(params.sessionKey); - if (params.kind === "visible" && expireVisibleStaleOperation(activeOperation)) { - continue; - } if (params.kind === "heartbeat" || params.kind === "control_abort") { return { status: "skipped", reason: "active-run", activeOperation }; } @@ -119,20 +87,10 @@ export async function admitReplyTurn(params: { if (params.waitForActive === false) { return { status: "skipped", reason: "active-run", activeOperation }; } - const activeWaitTimeoutMs = - params.kind === "visible" ? resolveVisibleActiveWaitMs(activeOperation) : waitTimeoutMs; - const ended = await replyRunRegistry.waitForIdle(params.sessionKey, activeWaitTimeoutMs, { + const ended = await replyRunRegistry.waitForIdle(params.sessionKey, waitTimeoutMs, { signal: params.upstreamAbortSignal, }); if (!ended) { - if (params.kind === "visible" && !isAbortSignalAborted(params.upstreamAbortSignal)) { - // Visible turns block on active work like before, but in bounded wait - // slices: each wake reclaims the owner once it is provably stale, - // otherwise loops back to keep waiting. - const latestActiveOperation = replyRunRegistry.get(params.sessionKey); - expireVisibleStaleOperation(latestActiveOperation ?? activeOperation); - continue; - } return { status: "skipped", reason: isAbortSignalAborted(params.upstreamAbortSignal) ? "aborted" : "active-run", diff --git a/src/config/config-misc.test.ts b/src/config/config-misc.test.ts index da6bada9376e..d1d4f6445bbe 100644 --- a/src/config/config-misc.test.ts +++ b/src/config/config-misc.test.ts @@ -59,20 +59,6 @@ describe("boolean config validation", () => { }); }); -describe("agent timeoutSeconds config", () => { - it.each([ - ["unlimited opt-in", 0, true], - ["finite", 600, true], - ["negative", -1, false], - ["fractional", 1.5, false], - ])("agents.defaults.timeoutSeconds %s", (_label, timeoutSeconds, ok) => { - const result = OpenClawSchema.safeParse({ - agents: { defaults: { timeoutSeconds } }, - }); - expect(result.success).toBe(ok); - }); -}); - describe("model provider localService config", () => { it("accepts standalone timeout overlays for bundled model providers", () => { const result = OpenClawSchema.safeParse({ diff --git a/src/config/zod-schema.agent-defaults.ts b/src/config/zod-schema.agent-defaults.ts index c7e755199a5b..f1f4c229c19f 100644 --- a/src/config/zod-schema.agent-defaults.ts +++ b/src/config/zod-schema.agent-defaults.ts @@ -228,8 +228,7 @@ export const AgentDefaultsSchema = z blockStreamingChunk: BlockStreamingChunkSchema.optional(), blockStreamingCoalesce: BlockStreamingCoalesceSchema.optional(), humanDelay: HumanDelaySchema.optional(), - // 0 = unlimited run budget; stream liveness watchdogs still apply. - timeoutSeconds: z.number().int().nonnegative().optional(), + timeoutSeconds: z.number().int().positive().optional(), mediaMaxMb: z.number().positive().optional(), imageMaxDimensionPx: z.number().int().positive().optional(), imageQuality: z.enum(["auto", "efficient", "balanced", "high"]).optional(), diff --git a/src/infra/heartbeat-runner.model-override.test.ts b/src/infra/heartbeat-runner.model-override.test.ts index b8c4dd5beaf9..509de8cd146f 100644 --- a/src/infra/heartbeat-runner.model-override.test.ts +++ b/src/infra/heartbeat-runner.model-override.test.ts @@ -237,14 +237,6 @@ describe("runHeartbeatOnce – heartbeat model override", () => { }); }); - it("preserves an unlimited default agent timeout for heartbeat runs", async () => { - const replyOpts = await runDefaultsHeartbeat({ defaultTimeoutSeconds: 0, every: "30m" }); - expectReplyOptions(replyOpts, { - isHeartbeat: true, - timeoutOverrideSeconds: 0, - }); - }); - it("passes bootstrapContextMode when heartbeat lightContext is enabled", async () => { const replyOpts = await runDefaultsHeartbeat({ lightContext: true }); expectReplyOptions(replyOpts, { diff --git a/src/infra/heartbeat-runner.ts b/src/infra/heartbeat-runner.ts index fb2176fbcc5e..bdec9dfb3c56 100644 --- a/src/infra/heartbeat-runner.ts +++ b/src/infra/heartbeat-runner.ts @@ -272,9 +272,6 @@ function resolveHeartbeatTimeoutOverrideSeconds(cfg: OpenClawConfig, heartbeat?: typeof agentDefaultTimeoutSeconds === "number" && Number.isFinite(agentDefaultTimeoutSeconds) ) { - if (agentDefaultTimeoutSeconds === 0) { - return 0; - } return Math.max(1, Math.floor(agentDefaultTimeoutSeconds)); } // The wake dispatcher awaits heartbeat turns serially. Keep unset heartbeat diff --git a/src/logging/diagnostic-stuck-session-recovery.runtime.ts b/src/logging/diagnostic-stuck-session-recovery.runtime.ts index 98498cababf5..befc18c183cf 100644 --- a/src/logging/diagnostic-stuck-session-recovery.runtime.ts +++ b/src/logging/diagnostic-stuck-session-recovery.runtime.ts @@ -177,8 +177,7 @@ export async function recoverStuckDiagnosticSession( `stuck session recovery reclaiming stale active run: ${formatRecoveryContext(params, { activeSessionId })}`, ); } - // Active embedded runs own their cleanup; registry terminal settle bounds - // lane release if the owner never drains after this abort. + // Active embedded runs own their cleanup; recovery asks them to abort and drain first. const result = await abortAndDrainEmbeddedAgentRun({ sessionId: activeSessionId, sessionKey: params.sessionKey, diff --git a/src/logging/diagnostic.ts b/src/logging/diagnostic.ts index e9c238c08059..da5bfc4397ea 100644 --- a/src/logging/diagnostic.ts +++ b/src/logging/diagnostic.ts @@ -178,20 +178,10 @@ async function recoverStuckSession( }); } -/** - * @deprecated Unused by core since the dispatch-side recovery loop was removed - * (#101910); reply admission owns stale-run reclaim now. Kept only because the - * plugin SDK re-exports this module; scheduled for removal in the next SDK major. - */ export function isStuckSessionRecoveryEnabled(config?: OpenClawConfig): boolean { return areDiagnosticsEnabledForProcess() && isDiagnosticsEnabled(config); } -/** - * @deprecated Unused by core since the dispatch-side recovery loop was removed - * (#101910); reply admission owns stale-run reclaim now. Kept only because the - * plugin SDK re-exports this module; scheduled for removal in the next SDK major. - */ export async function requestStuckDiagnosticSessionRecovery( params: StuckSessionRecoveryRequest, ): Promise {