From fa4f1abb290cafee615e2d9671df73977809108f Mon Sep 17 00:00:00 2001 From: Vincent Koc <25068+vincentkoc@users.noreply.github.com> Date: Thu, 18 Jun 2026 18:56:20 +0800 Subject: [PATCH] fix(cron): release cancelled embedded lanes --- .../run.compaction-loop-guard.test.ts | 200 ++++++++++++++++++ src/agents/embedded-agent-runner/run.ts | 67 +++++- .../embedded-agent-runner/run/attempt.ts | 6 +- src/agents/embedded-agent-runner/run/types.ts | 6 + src/process/command-queue.test.ts | 63 ++++++ src/process/command-queue.ts | 68 +++++- src/process/command-queue.types.ts | 4 + 7 files changed, 401 insertions(+), 13 deletions(-) diff --git a/src/agents/embedded-agent-runner/run.compaction-loop-guard.test.ts b/src/agents/embedded-agent-runner/run.compaction-loop-guard.test.ts index 8e5f442b038b..04f948117d44 100644 --- a/src/agents/embedded-agent-runner/run.compaction-loop-guard.test.ts +++ b/src/agents/embedded-agent-runner/run.compaction-loop-guard.test.ts @@ -187,6 +187,206 @@ describe("post-compaction loop guard wired into runEmbeddedAgent", () => { expect(attemptSignalReason).toBeInstanceOf(PostCompactionLoopPersistedError); }); + it("releases the lane after a post-compaction abort when the backend ignores cancellation", async () => { + vi.useFakeTimers(); + let settleIgnoredAttempt: ((value: ReturnType) => void) | undefined; + let resolveAttemptAborted: (() => void) | undefined; + const attemptAbortedPromise = new Promise((resolve) => { + resolveAttemptAborted = resolve; + }); + try { + const overflowError = makeOverflowError(); + let attemptAborted = false; + mockedRunEmbeddedAttempt.mockResolvedValueOnce( + makeAttemptResult({ promptError: overflowError }), + ); + mockedRunEmbeddedAttempt.mockImplementationOnce(async (attemptParams: unknown) => { + const { abortSignal, onToolOutcome } = attemptParams as { + abortSignal?: AbortSignal; + onToolOutcome?: ToolOutcomeObserver; + }; + for (let i = 0; i < 3; i += 1) { + await executeWrappedToolOutcome( + "gateway", + { action: "lookup", path: "x" }, + "identical-result", + onToolOutcome, + ); + } + attemptAborted = abortSignal?.aborted ?? false; + resolveAttemptAborted?.(); + return await new Promise((resolve) => { + settleIgnoredAttempt = resolve; + }); + }); + mockedCompactDirect.mockResolvedValueOnce( + makeCompactionSuccess({ + summary: "Compacted session", + firstKeptEntryId: "entry-5", + tokensBefore: 150000, + }), + ); + + const run = runEmbeddedAgent({ + ...baseParams, + runId: "run-post-compaction-abort-lane-release", + timeoutMs: 48 * 60 * 60 * 1000, + }); + let settled = false; + void run + .finally(() => { + settled = true; + }) + .catch(() => {}); + + await attemptAbortedPromise; + expect(attemptAborted).toBe(true); + await vi.advanceTimersByTimeAsync(30_001); + + expect(settled).toBe(true); + await expect(run).rejects.toMatchObject({ name: "CommandLaneTaskTimeoutError" }); + } finally { + settleIgnoredAttempt?.(makeAttemptResult()); + vi.useRealTimers(); + } + }); + + it("keeps a native lane alive while a tool is still running", async () => { + vi.useFakeTimers(); + let settleAttempt: ((value: ReturnType) => void) | undefined; + let resolveAttemptStarted: (() => void) | undefined; + const attemptStarted = new Promise((resolve) => { + resolveAttemptStarted = resolve; + }); + try { + mockedRunEmbeddedAttempt.mockImplementationOnce(async (attemptParams: unknown) => { + const { onAttemptTimeoutArmed } = attemptParams as { + onAttemptTimeoutArmed?: () => void; + }; + resolveAttemptStarted?.(); + onAttemptTimeoutArmed?.(); + return await new Promise((resolve) => { + settleAttempt = resolve; + }); + }); + + const run = runEmbeddedAgent({ + ...baseParams, + runId: "run-native-tool-heartbeat", + timeoutMs: 1, + agentHarnessRuntimeOverride: "openclaw", + }); + let settled = false; + void run + .finally(() => { + settled = true; + }) + .catch(() => {}); + + await vi.advanceTimersByTimeAsync(0); + await attemptStarted; + await vi.advanceTimersByTimeAsync(30_001); + + expect(settled).toBe(false); + settleAttempt?.(makeAttemptResult()); + await expect(run).resolves.toMatchObject({ meta: { aborted: false } }); + } finally { + vi.useRealTimers(); + } + }); + + it("releases a native lane when a timed-out attempt ignores cancellation", async () => { + vi.useFakeTimers(); + let settleAttempt: ((value: ReturnType) => void) | undefined; + let resolveAttemptStarted: (() => void) | undefined; + const attemptStarted = new Promise((resolve) => { + resolveAttemptStarted = resolve; + }); + try { + mockedRunEmbeddedAttempt.mockImplementationOnce(async (attemptParams: unknown) => { + const { onAttemptTimeoutArmed, onAttemptTimeout } = attemptParams as { + onAttemptTimeoutArmed?: () => void; + onAttemptTimeout?: (reason: Error) => void; + }; + resolveAttemptStarted?.(); + onAttemptTimeoutArmed?.(); + onAttemptTimeout?.(new Error("attempt timed out")); + return await new Promise((resolve) => { + settleAttempt = resolve; + }); + }); + + const run = runEmbeddedAgent({ + ...baseParams, + runId: "run-native-timeout-lane-release", + timeoutMs: 48 * 60 * 60 * 1000, + agentHarnessRuntimeOverride: "openclaw", + }); + let settled = false; + void run + .finally(() => { + settled = true; + }) + .catch(() => {}); + + await vi.advanceTimersByTimeAsync(0); + await attemptStarted; + await vi.advanceTimersByTimeAsync(30_001); + + expect(settled).toBe(true); + await expect(run).rejects.toMatchObject({ name: "CommandLaneTaskTimeoutError" }); + } finally { + settleAttempt?.(makeAttemptResult()); + vi.useRealTimers(); + } + }); + + it("releases a native lane when an explicit abort ignores cancellation", async () => { + vi.useFakeTimers(); + let settleAttempt: ((value: ReturnType) => void) | undefined; + let resolveAttemptStarted: (() => void) | undefined; + const attemptStarted = new Promise((resolve) => { + resolveAttemptStarted = resolve; + }); + try { + mockedRunEmbeddedAttempt.mockImplementationOnce(async (attemptParams: unknown) => { + const { onAttemptTimeoutArmed, onAttemptAbort } = attemptParams as { + onAttemptTimeoutArmed?: () => void; + onAttemptAbort?: () => void; + }; + resolveAttemptStarted?.(); + onAttemptTimeoutArmed?.(); + onAttemptAbort?.(); + return await new Promise((resolve) => { + settleAttempt = resolve; + }); + }); + + const run = runEmbeddedAgent({ + ...baseParams, + runId: "run-native-abort-lane-release", + timeoutMs: 48 * 60 * 60 * 1000, + agentHarnessRuntimeOverride: "openclaw", + }); + let settled = false; + void run + .finally(() => { + settled = true; + }) + .catch(() => {}); + + await vi.advanceTimersByTimeAsync(0); + await attemptStarted; + await vi.advanceTimersByTimeAsync(30_001); + + expect(settled).toBe(true); + await expect(run).rejects.toMatchObject({ name: "CommandLaneTaskTimeoutError" }); + } finally { + settleAttempt?.(makeAttemptResult()); + vi.useRealTimers(); + } + }); + it("does not abort when the result hash changes across post-compaction attempts (progress was made)", async () => { const overflowError = makeOverflowError(); // Attempt 1: overflow → triggers compaction. diff --git a/src/agents/embedded-agent-runner/run.ts b/src/agents/embedded-agent-runner/run.ts index 492fb3c0293a..efeb56cc59db 100644 --- a/src/agents/embedded-agent-runner/run.ts +++ b/src/agents/embedded-agent-runner/run.ts @@ -237,6 +237,7 @@ type ApiKeyInfo = ResolvedProviderAuth; const MAX_SAME_MODEL_IDLE_TIMEOUT_RETRIES = 1; const EMBEDDED_RUN_LANE_TIMEOUT_GRACE_MS = 30_000; +const EMBEDDED_RUN_LANE_HEARTBEAT_MS = EMBEDDED_RUN_LANE_TIMEOUT_GRACE_MS / 2; const MID_TURN_PRECHECK_CONTINUATION_PROMPT = "Continue from the current transcript after the latest tool result. Do not repeat the original user request, and do not rerun completed tools unless the transcript shows they are still needed."; const COMPACTION_CONTINUATION_RETRY_INSTRUCTION = @@ -637,6 +638,8 @@ async function runEmbeddedAgentInternal( }; const sessionQueuePriority = resolveEmbeddedRunSessionQueuePriority(params.trigger); const laneTaskTimeoutMs = resolveEmbeddedRunLaneTimeoutMs(params.timeoutMs); + const laneTaskAbortController = new AbortController(); + const laneTaskReleaseController = new AbortController(); let laneTaskProgressAtMs = Date.now(); const noteLaneTaskProgress = () => { laneTaskProgressAtMs = Date.now(); @@ -661,6 +664,9 @@ async function runEmbeddedAgentInternal( { ...opts, taskTimeoutProgressAtMs: () => laneTaskProgressAtMs, + taskTimeoutAbortSignal: laneTaskAbortController.signal, + taskTimeoutAbortGraceMs: EMBEDDED_RUN_LANE_TIMEOUT_GRACE_MS, + taskTimeoutReleaseSignal: laneTaskReleaseController.signal, }, laneTaskTimeoutMs, ); @@ -1480,6 +1486,7 @@ async function runEmbeddedAgentInternal( const verdict = postCompactionGuard.observe(observation); if (verdict.shouldAbort) { postCompactionAbortError ??= PostCompactionLoopPersistedError.fromVerdict(verdict); + laneTaskAbortController.abort(postCompactionAbortError); postCompactionAbortController?.abort(postCompactionAbortError); } }; @@ -1846,6 +1853,7 @@ async function runEmbeddedAgentInternal( postCompactionAbortController = attemptAbortController; const parentAbortSignal = params.abortSignal; const relayParentAbort = (): void => { + laneTaskAbortController.abort(parentAbortSignal?.reason); attemptAbortController.abort(parentAbortSignal?.reason); }; if (parentAbortSignal?.aborted) { @@ -1853,10 +1861,48 @@ async function runEmbeddedAgentInternal( } else { parentAbortSignal?.addEventListener("abort", relayParentAbort, { once: true }); } - // Periodically report progress during long-running tool execution - // so the lane timeout does not expire while a tool is still running - // (e.g., exec commands taking >5 minutes) (#94033). - const progressInterval = setInterval(() => noteLaneTaskProgress(), 30_000); + // Native attempts start the heartbeat only after their own timeout + // watchdog is armed, keeping preflight inside the requested deadline. + let progressInterval: ReturnType | undefined; + const stopLaneProgressHeartbeat = () => { + if (progressInterval) { + clearInterval(progressInterval); + progressInterval = undefined; + } + attemptAbortController.signal.removeEventListener("abort", stopLaneProgressHeartbeat); + }; + const startLaneProgressHeartbeat = () => { + if (progressInterval || attemptAbortController.signal.aborted) { + return; + } + progressInterval = setInterval( + () => noteLaneTaskProgress(), + EMBEDDED_RUN_LANE_HEARTBEAT_MS, + ); + progressInterval.unref?.(); + attemptAbortController.signal.addEventListener("abort", stopLaneProgressHeartbeat, { + once: true, + }); + }; + // Timeout recovery can continue after an attempt returns, but a native + // transport that ignores its timeout releases the lane after one grace. + let timeoutReleaseTimer: ReturnType | undefined; + const clearAttemptTimeoutRelease = () => { + if (timeoutReleaseTimer) { + clearTimeout(timeoutReleaseTimer); + timeoutReleaseTimer = undefined; + } + }; + const armAttemptTimeoutRelease = (reason: Error) => { + if (timeoutReleaseTimer) { + return; + } + timeoutReleaseTimer = setTimeout( + () => laneTaskReleaseController.abort(reason), + EMBEDDED_RUN_LANE_TIMEOUT_GRACE_MS, + ); + timeoutReleaseTimer.unref?.(); + }; const rawAttempt = await runEmbeddedAttemptWithBackend({ sessionId: activeSessionId, sessionKey: resolvedSessionKey, @@ -1960,6 +2006,16 @@ async function runEmbeddedAgentInternal( runId: params.runId, lifecycleGeneration, abortSignal: attemptAbortController.signal, + onAttemptTimeoutArmed: pluginHarnessOwnsTransport + ? undefined + : startLaneProgressHeartbeat, + onAttemptTimeout: pluginHarnessOwnsTransport ? undefined : armAttemptTimeoutRelease, + onAttemptAbort: pluginHarnessOwnsTransport + ? undefined + : () => { + stopLaneProgressHeartbeat(); + laneTaskAbortController.abort(); + }, replyOperation: params.replyOperation, shouldEmitToolResult: params.shouldEmitToolResult, shouldEmitToolOutput: params.shouldEmitToolOutput, @@ -2015,7 +2071,8 @@ async function runEmbeddedAgentInternal( throw postCompactionAbortError ?? err; }) .finally(() => { - clearInterval(progressInterval); + clearAttemptTimeoutRelease(); + stopLaneProgressHeartbeat(); parentAbortSignal?.removeEventListener?.("abort", relayParentAbort); if (postCompactionAbortController === attemptAbortController) { postCompactionAbortController = undefined; diff --git a/src/agents/embedded-agent-runner/run/attempt.ts b/src/agents/embedded-agent-runner/run/attempt.ts index 83bb97042217..601a1c2d7752 100644 --- a/src/agents/embedded-agent-runner/run/attempt.ts +++ b/src/agents/embedded-agent-runner/run/attempt.ts @@ -3391,7 +3391,9 @@ export async function runEmbeddedAttempt( } } if (isTimeout) { - runAbortController.abort(reason ?? makeTimeoutAbortReason()); + const timeoutReason = reason instanceof Error ? reason : makeTimeoutAbortReason(); + params.onAttemptTimeout?.(timeoutReason); + runAbortController.abort(timeoutReason); } else { runAbortController.abort(reason); } @@ -3701,6 +3703,7 @@ export async function runEmbeddedAttempt( const abortActiveRunExternally = (reason?: "user_abort" | "restart" | "superseded") => { externalAbort = true; + params.onAttemptAbort?.(); abortRun(false, reason === "restart" ? createAgentRunRestartAbortError() : undefined); }; const queueHandle: EmbeddedAgentQueueHandle & { @@ -3796,6 +3799,7 @@ export async function runEmbeddedAttempt( ); }; scheduleAbortTimer(params.timeoutMs, "initial"); + params.onAttemptTimeoutArmed?.(); let messagesSnapshot: AgentMessage[] = []; let sessionIdUsed = activeSession.sessionId; diff --git a/src/agents/embedded-agent-runner/run/types.ts b/src/agents/embedded-agent-runner/run/types.ts index 8ce28dd53a06..cac31c19136b 100644 --- a/src/agents/embedded-agent-runner/run/types.ts +++ b/src/agents/embedded-agent-runner/run/types.ts @@ -73,6 +73,12 @@ export type EmbeddedRunAttemptParams = EmbeddedRunAttemptBase & { agentHarnessTaskRuntimeScope?: AgentHarnessTaskRuntimeScope; /** Live observer called after wrapped tool outcomes are recorded. */ onToolOutcome?: ToolOutcomeObserver; + /** Signals that the attempt's own run-timeout watchdog is active. */ + onAttemptTimeoutArmed?: () => void; + /** Signals that this attempt's timeout has fired and must unwind promptly. */ + onAttemptTimeout?: (reason: Error) => void; + /** Signals an explicit cancellation through the active native run handle. */ + onAttemptAbort?: () => void; /** Supplies run-global model-call ordering for parallel tool outcomes. */ allocateToolOutcomeOrdinal?: (toolCallId?: string) => number; model: Model; diff --git a/src/process/command-queue.test.ts b/src/process/command-queue.test.ts index 3d1a94e98580..66c5c8964f88 100644 --- a/src/process/command-queue.test.ts +++ b/src/process/command-queue.test.ts @@ -543,6 +543,69 @@ describe("command queue", () => { } }); + it("task timeout switches to a short abort grace period", async () => { + const lane = `timeout-abort-lane-${Date.now()}-${Math.random().toString(16).slice(2)}`; + setCommandLaneConcurrency(lane, 1); + + vi.useFakeTimers(); + try { + const abortController = new AbortController(); + const first = enqueueCommandInLane(lane, async () => new Promise(() => {}), { + taskTimeoutMs: 48 * 60 * 60 * 1000, + taskTimeoutAbortSignal: abortController.signal, + taskTimeoutAbortGraceMs: 25, + }); + const firstRejected = expect(first).rejects.toBeInstanceOf(CommandLaneTaskTimeoutError); + let secondRan = false; + const second = enqueueCommandInLane(lane, async () => { + secondRan = true; + return "second"; + }); + + abortController.abort(); + await vi.advanceTimersByTimeAsync(24); + expect(secondRan).toBe(false); + await vi.advanceTimersByTimeAsync(1); + + await firstRejected; + await expect(second).resolves.toBe("second"); + expect(secondRan).toBe(true); + } finally { + vi.useRealTimers(); + } + }); + + it("task timeout release signal skips the abort grace period", async () => { + const lane = `timeout-release-lane-${Date.now()}-${Math.random().toString(16).slice(2)}`; + setCommandLaneConcurrency(lane, 1); + + vi.useFakeTimers(); + try { + const releaseController = new AbortController(); + const first = enqueueCommandInLane(lane, async () => new Promise(() => {}), { + taskTimeoutMs: 48 * 60 * 60 * 1000, + taskTimeoutProgressAtMs: () => Date.now(), + taskTimeoutAbortGraceMs: 25, + taskTimeoutReleaseSignal: releaseController.signal, + }); + const firstRejected = expect(first).rejects.toBeInstanceOf(CommandLaneTaskTimeoutError); + let secondRan = false; + const second = enqueueCommandInLane(lane, async () => { + secondRan = true; + return "second"; + }); + + releaseController.abort(); + await vi.advanceTimersByTimeAsync(0); + + await firstRejected; + await expect(second).resolves.toBe("second"); + expect(secondRan).toBe(true); + } finally { + vi.useRealTimers(); + } + }); + it("task timeout falls back when progress timestamp callback throws", async () => { const lane = `timeout-progress-throw-lane-${Date.now()}-${Math.random().toString(16).slice(2)}`; setCommandLaneConcurrency(lane, 1); diff --git a/src/process/command-queue.ts b/src/process/command-queue.ts index 27b2a8f71d27..97ce602d023a 100644 --- a/src/process/command-queue.ts +++ b/src/process/command-queue.ts @@ -69,6 +69,9 @@ type QueueEntry = { activeAheadAtEnqueue: number; taskTimeoutMs?: number; taskTimeoutProgressAtMs?: () => number | undefined; + taskTimeoutAbortSignal?: AbortSignal; + taskTimeoutAbortGraceMs?: number; + taskTimeoutReleaseSignal?: AbortSignal; onWait?: (waitMs: number, queuedAhead: number) => void; }; @@ -277,6 +280,8 @@ async function runQueueEntryTask(lane: string, entry: QueueEntry): Promise { let value: number | undefined; @@ -290,20 +295,64 @@ async function runQueueEntryTask(lane: string, entry: QueueEntry): Promise | undefined; + let removeAbortListener: (() => void) | undefined; + let removeReleaseListener: (() => void) | undefined; let timedOut = false; const timeoutPromise = new Promise((_, reject) => { - const armTimeout = () => { + const rejectForTimeout = () => { + timedOut = true; + reject(new CommandLaneTaskTimeoutError(lane, taskTimeoutMs)); + }; + const armTimer = (delayMs: number, onTimeout: () => void) => { + if (timeoutHandle) { + clearTimeout(timeoutHandle); + } + if (delayMs <= 0) { + onTimeout(); + return; + } + timeoutHandle = setTimeout(onTimeout, delayMs); + timeoutHandle.unref?.(); + }; + const armProgressTimeout = () => { const elapsedMs = Math.max(0, Date.now() - readLastProgressAtMs()); const remainingMs = taskTimeoutMs - elapsedMs; if (remainingMs <= 0) { - timedOut = true; - reject(new CommandLaneTaskTimeoutError(lane, taskTimeoutMs)); + rejectForTimeout(); return; } - timeoutHandle = setTimeout(armTimeout, remainingMs); - timeoutHandle.unref?.(); + armTimer(remainingMs, armProgressTimeout); }; - armTimeout(); + const armAbortTimeout = () => { + armTimer(taskTimeoutAbortGraceMs, rejectForTimeout); + }; + const abortSignal = entry.taskTimeoutAbortSignal; + const releaseSignal = entry.taskTimeoutReleaseSignal; + const onRelease = () => { + removeReleaseListener?.(); + rejectForTimeout(); + }; + if (releaseSignal?.aborted) { + onRelease(); + return; + } + if (abortSignal?.aborted) { + armAbortTimeout(); + return; + } + armProgressTimeout(); + if (abortSignal) { + const onAbort = () => { + removeAbortListener?.(); + armAbortTimeout(); + }; + abortSignal.addEventListener("abort", onAbort, { once: true }); + removeAbortListener = () => abortSignal.removeEventListener("abort", onAbort); + } + if (releaseSignal) { + releaseSignal.addEventListener("abort", onRelease, { once: true }); + removeReleaseListener = () => releaseSignal.removeEventListener("abort", onRelease); + } }); try { @@ -318,9 +367,11 @@ async function runQueueEntryTask(lane: string, entry: QueueEntry): Promise( activeAheadAtEnqueue: 0, taskTimeoutMs: normalizeTaskTimeoutMs(opts?.taskTimeoutMs), taskTimeoutProgressAtMs: opts?.taskTimeoutProgressAtMs, + taskTimeoutAbortSignal: opts?.taskTimeoutAbortSignal, + taskTimeoutAbortGraceMs: normalizeTaskTimeoutMs(opts?.taskTimeoutAbortGraceMs), + taskTimeoutReleaseSignal: opts?.taskTimeoutReleaseSignal, onWait: opts?.onWait, }); logLaneEnqueue(cleaned, getLaneDepth(state)); diff --git a/src/process/command-queue.types.ts b/src/process/command-queue.types.ts index e46f31d45a11..fe38d4d3099a 100644 --- a/src/process/command-queue.types.ts +++ b/src/process/command-queue.types.ts @@ -7,6 +7,10 @@ export type CommandQueueEnqueueOptions = { onWait?: (waitMs: number, queuedAhead: number) => void; taskTimeoutMs?: number; taskTimeoutProgressAtMs?: () => number | undefined; + taskTimeoutAbortSignal?: AbortSignal; + taskTimeoutAbortGraceMs?: number; + /** Ends the task after a caller-owned timeout cleanup grace has already elapsed. */ + taskTimeoutReleaseSignal?: AbortSignal; priority?: "foreground" | "normal" | "background"; };