diff --git a/src/agents/embedded-agent-runner/run-orchestrator.ts b/src/agents/embedded-agent-runner/run-orchestrator.ts index abb308e58312..060124455693 100644 --- a/src/agents/embedded-agent-runner/run-orchestrator.ts +++ b/src/agents/embedded-agent-runner/run-orchestrator.ts @@ -60,6 +60,7 @@ import type { RunEmbeddedAgentParamsWithSessionFile, } from "./run/internal-params.js"; import { createEmbeddedRunLaneController } from "./run/lane-controller.js"; +import { withEmbeddedRunLaneProgressHeartbeat } from "./run/lane-runtime.js"; import type { RunEmbeddedAgentParams } from "./run/params.js"; import { bindRunToPreparedModelRuntime } from "./run/prepared-runtime-context.js"; import { createEmbeddedRunProgressController } from "./run/progress-controller.js"; @@ -213,10 +214,15 @@ async function runEmbeddedAgentInternal( }; // Configless direct hosts reuse one bounded idle generation. Gateway and explicitly // configured runs release dynamic workspaces so one-off paths cannot accumulate owners. - const preparedModelRuntimeLease = - params.preparedModelRuntimeMode === "isolated-read-only" - ? await acquireReadOnlyPreparedModelRuntime(preparedInput) - : await acquireAgentRunPreparedModelRuntime(preparedInput, { retainIdleRunOwner }); + // Cold plugin loading and provider discovery can exceed the lane no-progress budget. + // Active runtime acquisition is progress, not a hung lane task. + const preparedModelRuntimeLease = await withEmbeddedRunLaneProgressHeartbeat( + noteLaneTaskProgress, + () => + params.preparedModelRuntimeMode === "isolated-read-only" + ? acquireReadOnlyPreparedModelRuntime(preparedInput) + : acquireAgentRunPreparedModelRuntime(preparedInput, { retainIdleRunOwner }), + ); const preparedModelRuntimeOwnerSnapshot = preparedModelRuntimeLease.snapshot; try { // A reload may complete while admission waits. The committed generation owns config, diff --git a/src/agents/embedded-agent-runner/run/lane-runtime.test.ts b/src/agents/embedded-agent-runner/run/lane-runtime.test.ts index e12717129966..65fbffbfa286 100644 --- a/src/agents/embedded-agent-runner/run/lane-runtime.test.ts +++ b/src/agents/embedded-agent-runner/run/lane-runtime.test.ts @@ -1,14 +1,19 @@ -import { afterEach, describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { enqueueCommandInLane, setCommandLaneConcurrency } from "../../../process/command-queue.js"; import { resetCommandQueueStateForTest } from "../../../process/command-queue.test-support.js"; import { MAIN_SESSION_RESTART_RECOVERY_SOURCE_TOOL } from "../../../sessions/input-provenance.js"; -import { resolveEmbeddedRunSessionQueuePriority } from "./lane-runtime.js"; +import { + EMBEDDED_RUN_LANE_HEARTBEAT_MS, + resolveEmbeddedRunSessionQueuePriority, + withEmbeddedRunLaneProgressHeartbeat, +} from "./lane-runtime.js"; + +afterEach(() => { + vi.useRealTimers(); + resetCommandQueueStateForTest(); +}); describe("embedded run lane priority", () => { - afterEach(() => { - resetCommandQueueStateForTest(); - }); - it("runs a foreground user turn before queued restart recovery", async () => { const lane = "test:restart-recovery-priority"; setCommandLaneConcurrency(lane, 1); @@ -46,3 +51,83 @@ describe("embedded run lane priority", () => { expect(order).toEqual(["foreground-user", "restart-recovery"]); }); }); + +describe("embedded run lane progress heartbeat", () => { + it("notes progress immediately and at the heartbeat cadence", async () => { + vi.useFakeTimers(); + const noteLaneTaskProgress = vi.fn(); + let finish: ((value: string) => void) | undefined; + const task = withEmbeddedRunLaneProgressHeartbeat( + noteLaneTaskProgress, + () => + new Promise((resolve) => { + finish = resolve; + }), + ); + + expect(noteLaneTaskProgress).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(EMBEDDED_RUN_LANE_HEARTBEAT_MS - 1); + expect(noteLaneTaskProgress).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(EMBEDDED_RUN_LANE_HEARTBEAT_MS * 2 + 1); + expect(noteLaneTaskProgress).toHaveBeenCalledTimes(4); + + finish?.("done"); + await expect(task).resolves.toBe("done"); + expect(noteLaneTaskProgress).toHaveBeenCalledTimes(5); + + await vi.advanceTimersByTimeAsync(EMBEDDED_RUN_LANE_HEARTBEAT_MS * 2); + expect(noteLaneTaskProgress).toHaveBeenCalledTimes(5); + }); + + it("clears the heartbeat after rejection and propagates the error", async () => { + vi.useFakeTimers(); + const noteLaneTaskProgress = vi.fn(); + const expectedError = new Error("runtime acquisition failed"); + let fail: ((error: Error) => void) | undefined; + const task = withEmbeddedRunLaneProgressHeartbeat( + noteLaneTaskProgress, + () => + new Promise((_resolve, reject) => { + fail = reject; + }), + ); + + await vi.advanceTimersByTimeAsync(EMBEDDED_RUN_LANE_HEARTBEAT_MS); + expect(noteLaneTaskProgress).toHaveBeenCalledTimes(2); + fail?.(expectedError); + await expect(task).rejects.toBe(expectedError); + expect(noteLaneTaskProgress).toHaveBeenCalledTimes(3); + + await vi.advanceTimersByTimeAsync(EMBEDDED_RUN_LANE_HEARTBEAT_MS * 2); + expect(noteLaneTaskProgress).toHaveBeenCalledTimes(3); + }); + + it("keeps a slow lane task alive while runtime acquisition makes progress", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-07-29T00:00:00.000Z")); + const lane = `test:embedded-run-runtime-heartbeat:${Date.now()}`; + setCommandLaneConcurrency(lane, 1); + let progressAtMs = Date.now(); + + const task = enqueueCommandInLane( + lane, + () => + withEmbeddedRunLaneProgressHeartbeat( + () => { + progressAtMs = Date.now(); + }, + () => + new Promise((resolve) => { + setTimeout(() => resolve("completed"), 52_000); + }), + ), + { + taskTimeoutMs: 25_000, + taskTimeoutProgressAtMs: () => progressAtMs, + }, + ); + + await vi.advanceTimersByTimeAsync(52_000); + await expect(task).resolves.toBe("completed"); + }); +}); diff --git a/src/agents/embedded-agent-runner/run/lane-runtime.ts b/src/agents/embedded-agent-runner/run/lane-runtime.ts index 99540127f464..40d2326b1052 100644 --- a/src/agents/embedded-agent-runner/run/lane-runtime.ts +++ b/src/agents/embedded-agent-runner/run/lane-runtime.ts @@ -10,6 +10,21 @@ import type { RunEmbeddedAgentParams } from "./params.js"; export const EMBEDDED_RUN_LANE_TIMEOUT_GRACE_MS = 30_000; export const EMBEDDED_RUN_LANE_HEARTBEAT_MS = EMBEDDED_RUN_LANE_TIMEOUT_GRACE_MS / 2; +export async function withEmbeddedRunLaneProgressHeartbeat( + noteLaneTaskProgress: () => void, + fn: () => Promise, +): Promise { + noteLaneTaskProgress(); + const progressInterval = setInterval(noteLaneTaskProgress, EMBEDDED_RUN_LANE_HEARTBEAT_MS); + progressInterval.unref?.(); + try { + return await fn(); + } finally { + clearInterval(progressInterval); + noteLaneTaskProgress(); + } +} + export function resolveEmbeddedRunLaneTimeoutMs(timeoutMs: number): number { const defaultLaneTimeoutMs = DEFAULT_AGENT_TIMEOUT_MS + EMBEDDED_RUN_LANE_TIMEOUT_GRACE_MS; // "No timeout" resolves to the timer-safe MAX_TIMER sentinel upstream. diff --git a/src/agents/embedded-agent-runner/run/settled-turn-finalization.ts b/src/agents/embedded-agent-runner/run/settled-turn-finalization.ts index 73ca32277738..eaf50920eac7 100644 --- a/src/agents/embedded-agent-runner/run/settled-turn-finalization.ts +++ b/src/agents/embedded-agent-runner/run/settled-turn-finalization.ts @@ -10,7 +10,7 @@ import { mergeUsageIntoAccumulator, } from "../usage-accumulator.js"; import { runEmbeddedSettledTurnFinalizationWithBackend } from "./backend.js"; -import { EMBEDDED_RUN_LANE_HEARTBEAT_MS } from "./lane-runtime.js"; +import { withEmbeddedRunLaneProgressHeartbeat } from "./lane-runtime.js"; import { resolveEmbeddedRunAttemptTerminalOutcome, type EmbeddedRunTerminalState, @@ -167,10 +167,7 @@ async function runPreparedSettledTurnFinalization(input: { prompt: string; noteLaneTaskProgress: () => void; }): Promise { - input.noteLaneTaskProgress(); - const progressInterval = setInterval(input.noteLaneTaskProgress, EMBEDDED_RUN_LANE_HEARTBEAT_MS); - progressInterval.unref?.(); - try { + return await withEmbeddedRunLaneProgressHeartbeat(input.noteLaneTaskProgress, async () => { const result = await runEmbeddedSettledTurnFinalizationWithBackend( { ...input.attempt, @@ -189,10 +186,7 @@ async function runPreparedSettledTurnFinalization(input: { prompt: input.prompt, agentHarnessId: input.attempt.agentHarnessId, }); - } finally { - clearInterval(progressInterval); - input.noteLaneTaskProgress(); - } + }); } function buildSettledTurnFinalizationAttemptResult(input: { diff --git a/src/skills/workshop/experience-review.test.ts b/src/skills/workshop/experience-review.test.ts index bb08df8d01c8..610ebbbd1270 100644 --- a/src/skills/workshop/experience-review.test.ts +++ b/src/skills/workshop/experience-review.test.ts @@ -72,6 +72,13 @@ function completedRun( }; } +async function flushMicrotasks(): Promise { + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); +} + afterEach(() => { vi.useRealTimers(); }); @@ -368,6 +375,65 @@ describe("skill experience review scheduler", () => { scheduler.clear(); }); + it("drops terminal auth-migration failures without re-arming", async () => { + const callbacks: Array<() => void> = []; + const setTimer = vi.fn((callback: () => void) => { + callbacks.push(callback); + return { unref: vi.fn() } as unknown as ReturnType; + }); + const clearTimer = vi.fn(); + const runReview = vi.fn().mockRejectedValue( + Object.assign(new Error("Auth migration required; run openclaw doctor --fix."), { + code: "AUTH_PROFILE_MIGRATION_REQUIRED" as const, + }), + ); + const scheduler = createSkillExperienceReviewScheduler({ + isSystemActive: () => false, + runReview, + setTimer, + clearTimer, + }); + + scheduler.schedule(completedRun()); + callbacks[0]?.(); + await flushMicrotasks(); + + expect(runReview).toHaveBeenCalledTimes(1); + expect(setTimer).toHaveBeenCalledTimes(1); + expect(clearTimer).not.toHaveBeenCalled(); + + scheduler.schedule(completedRun()); + expect(setTimer).toHaveBeenCalledTimes(2); + expect(clearTimer).not.toHaveBeenCalled(); + scheduler.clear(); + }); + + it("re-arms after a generic review failure", async () => { + const callbacks: Array<() => void> = []; + const setTimer = vi.fn((callback: () => void, _delayMs: number) => { + callbacks.push(callback); + return { unref: vi.fn() } as unknown as ReturnType; + }); + const clearTimer = vi.fn(); + const runReview = vi.fn().mockRejectedValue(new Error("provider unavailable")); + const scheduler = createSkillExperienceReviewScheduler({ + isSystemActive: () => false, + runReview, + setTimer, + clearTimer, + }); + + scheduler.schedule(completedRun()); + callbacks[0]?.(); + await flushMicrotasks(); + + expect(runReview).toHaveBeenCalledTimes(1); + expect(setTimer).toHaveBeenCalledTimes(2); + expect(setTimer).toHaveBeenLastCalledWith(expect.any(Function), 30_000); + expect(clearTimer).not.toHaveBeenCalled(); + scheduler.clear(); + }); + it("serializes reviews across sessions", async () => { vi.useFakeTimers(); let finishFirst: (() => void) | undefined; diff --git a/src/skills/workshop/experience-review.ts b/src/skills/workshop/experience-review.ts index b08971bc20b5..5e8f9dbe60ef 100644 --- a/src/skills/workshop/experience-review.ts +++ b/src/skills/workshop/experience-review.ts @@ -99,6 +99,16 @@ type PendingExperienceReview = { timer?: ExperienceReviewTimer; }; +function isAuthProfileMigrationRequiredError( + error: unknown, +): error is { code: "AUTH_PROFILE_MIGRATION_REQUIRED" } { + return ( + typeof error === "object" && + error !== null && + (error as { code?: unknown }).code === "AUTH_PROFILE_MIGRATION_REQUIRED" + ); +} + function isEligibleContext(ctx: ExperienceReviewAgentContext): boolean { // Only harnesses that report both the resolved model and actual host-side // Workshop availability may schedule. Other runtimes fail closed here. @@ -254,14 +264,22 @@ export function createSkillExperienceReviewScheduler(deps: ExperienceReviewSched if (pendingBySession.get(sessionKey) !== pending || pending.generation !== generation) { return; } - pendingBySession.delete(sessionKey); await deps.runReview(candidate); + if (pendingBySession.get(sessionKey) === pending && pending.generation === generation) { + pendingBySession.delete(sessionKey); + } } finally { reviewInFlight = false; } }) .catch((error: unknown) => { log.warn(`skill experience review failed: ${String(error)}`); + if (isAuthProfileMigrationRequiredError(error)) { + if (pendingBySession.get(sessionKey) === pending && pending.generation === generation) { + pendingBySession.delete(sessionKey); + } + return; + } if (pendingBySession.get(sessionKey) === pending && pending.generation === generation) { arm(sessionKey, pending, EXPERIENCE_REVIEW_RETRY_IDLE_MS); }