diff --git a/src/cron/service/ops.regression.test.ts b/src/cron/service/ops.regression.test.ts index 436a6295a77a..b464155ac8aa 100644 --- a/src/cron/service/ops.regression.test.ts +++ b/src/cron/service/ops.regression.test.ts @@ -14,6 +14,13 @@ import { setCommandLaneConcurrency, waitForActiveTasks, } from "../../process/command-queue.js"; +import { + getActiveGatewayRootWorkCount, + isGatewaySubordinateWorkAdmissionClosed, + markGatewayRestartDraining, + resetGatewayWorkAdmission, + tryBeginGatewayRootWorkAdmission, +} from "../../process/gateway-work-admission.js"; import { CommandLane } from "../../process/lanes.js"; import { saveCronStore } from "../store.js"; import { enqueueRun, remove, run, start } from "./ops.js"; @@ -58,6 +65,115 @@ function expectIsolatedRunJobId( } describe("cron service ops regressions", () => { + it("transfers queued manual runs out of the released request root", async () => { + vi.useRealTimers(); + resetGatewayWorkAdmission(); + clearCommandLane(CommandLane.Cron); + setCommandLaneConcurrency(CommandLane.Cron, 1); + + const childLane = "cron-manual-admission-child"; + clearCommandLane(childLane); + setCommandLaneConcurrency(childLane, 1); + const store = opsRegressionFixtures.makeStorePath(); + const now = Date.parse("2026-02-06T10:05:00.000Z"); + const job = createDueIsolatedJob({ + id: "manual-admission-continuation", + nowMs: now, + nextRunAtMs: now, + }); + await saveCronStore(store.storePath, { version: 1, jobs: [job] }); + + const enterRunner = createDeferred(); + const runnerStarted = createDeferred(); + const finished = createDeferred(); + let terminalEvent: CronEvent | undefined; + const state = createCronServiceState({ + cronEnabled: true, + storePath: store.storePath, + log: noopLogger, + nowMs: () => now, + enqueueSystemEvent: vi.fn(), + requestHeartbeat: vi.fn(), + runIsolatedAgentJob: vi.fn(async () => { + runnerStarted.resolve(); + await enterRunner.promise; + expect(isGatewaySubordinateWorkAdmissionClosed()).toBe(false); + await enqueueCommandInLane(childLane, async () => undefined); + return { status: "ok" as const }; + }), + onEvent: (event) => { + if (event.jobId === job.id && event.action === "finished") { + terminalEvent = event; + finished.resolve(); + } + }, + }); + const requestRoot = tryBeginGatewayRootWorkAdmission(); + expect(requestRoot?.ownsRoot).toBe(true); + + try { + await requestRoot?.run(async () => { + expectQueuedRunAck(await enqueueRun(state, job.id, "force")); + await runnerStarted.promise; + expect(getActiveGatewayRootWorkCount()).toBe(2); + }); + requestRoot?.release(); + expect(getActiveGatewayRootWorkCount()).toBe(1); + + enterRunner.resolve(); + await finished.promise; + await waitForActiveTasks(5_000); + expect(terminalEvent).toMatchObject({ status: "ok" }); + await vi.waitFor(() => expect(getActiveGatewayRootWorkCount()).toBe(0)); + } finally { + requestRoot?.release(); + enterRunner.resolve(); + clearCommandLane(childLane); + clearCommandLane(CommandLane.Cron); + resetGatewayWorkAdmission(); + } + }); + + it("emits a terminal error when detached admission is already closed", async () => { + vi.useRealTimers(); + resetGatewayWorkAdmission(); + const store = opsRegressionFixtures.makeStorePath(); + const now = Date.parse("2026-02-06T10:05:00.000Z"); + const job = createDueIsolatedJob({ + id: "manual-admission-closed", + nowMs: now, + nextRunAtMs: now, + }); + await saveCronStore(store.storePath, { version: 1, jobs: [job] }); + + const finished = createDeferred(); + const state = createCronServiceState({ + cronEnabled: true, + storePath: store.storePath, + log: noopLogger, + nowMs: () => now, + enqueueSystemEvent: vi.fn(), + requestHeartbeat: vi.fn(), + runIsolatedAgentJob: vi.fn(async () => ({ status: "ok" as const })), + onEvent: (event) => { + if (event.jobId === job.id && event.action === "finished") { + finished.resolve(event); + } + }, + }); + + try { + markGatewayRestartDraining(); + expectQueuedRunAck(await enqueueRun(state, job.id, "force")); + await expect(finished.promise).resolves.toMatchObject({ + status: "error", + error: expect.stringContaining("gateway is draining for restart"), + }); + } finally { + resetGatewayWorkAdmission(); + } + }); + it("repairs missing job state during startup", async () => { const scheduledAt = Date.now() + 60_000; const store = opsRegressionFixtures.makeStorePath(); diff --git a/src/cron/service/ops.ts b/src/cron/service/ops.ts index 15cf0efa3cae..cf3685ea51ee 100644 --- a/src/cron/service/ops.ts +++ b/src/cron/service/ops.ts @@ -5,6 +5,7 @@ import { normalizeOptionalString, } from "@openclaw/normalization-core/string-coerce"; import { enqueueCommandInLane } from "../../process/command-queue.js"; +import { runWithGatewayIndependentRootWorkContinuation } from "../../process/gateway-work-admission.js"; import { CommandLane } from "../../process/lanes.js"; import { DEFAULT_AGENT_ID } from "../../routing/session-key.js"; import { resolveOpenClawStateSqlitePath } from "../../state/openclaw-state-db.paths.js"; @@ -1332,46 +1333,48 @@ export async function enqueueRun(state: CronServiceState, id: string, mode?: "du const runId = `manual:${id}:${state.deps.nowMs()}:${nextManualRunId++}`; const terminalTracker: ManualRunTerminalTracker = { emitted: false }; - void enqueueCommandInLane( - CommandLane.Cron, - async () => { - const result = await run(state, id, mode, { runId, terminalTracker }); - if (result.ok && "ran" in result && !result.ran) { - if (result.reason !== "invalid-spec") { - const finishedAt = state.deps.nowMs(); - const job = state.store?.jobs.find((entry) => entry.id === id); - emitManualRunFinished( - state, - { - jobId: id, - action: "finished", - job, - status: "skipped", - error: `queued manual run skipped before execution: ${result.reason}`, - runId, - runAtMs: finishedAt, - durationMs: 0, - nextRunAtMs: job?.state.nextRunAtMs, - }, - terminalTracker, + void runWithGatewayIndependentRootWorkContinuation(() => + enqueueCommandInLane( + CommandLane.Cron, + async () => { + const result = await run(state, id, mode, { runId, terminalTracker }); + if (result.ok && "ran" in result && !result.ran) { + if (result.reason !== "invalid-spec") { + const finishedAt = state.deps.nowMs(); + const job = state.store?.jobs.find((entry) => entry.id === id); + emitManualRunFinished( + state, + { + jobId: id, + action: "finished", + job, + status: "skipped", + error: `queued manual run skipped before execution: ${result.reason}`, + runId, + runAtMs: finishedAt, + durationMs: 0, + nextRunAtMs: job?.state.nextRunAtMs, + }, + terminalTracker, + ); + } + state.deps.log.info( + { jobId: id, runId, reason: result.reason }, + "cron: queued manual run skipped before execution", ); } - state.deps.log.info( - { jobId: id, runId, reason: result.reason }, - "cron: queued manual run skipped before execution", - ); - } - return result; - }, - { - warnAfterMs: 5_000, - onWait: (waitMs, queuedAhead) => { - state.deps.log.warn( - { jobId: id, runId, waitMs, queuedAhead }, - "cron: queued manual run waiting for an execution slot", - ); + return result; }, - }, + { + warnAfterMs: 5_000, + onWait: (waitMs, queuedAhead) => { + state.deps.log.warn( + { jobId: id, runId, waitMs, queuedAhead }, + "cron: queued manual run waiting for an execution slot", + ); + }, + }, + ), ).catch((err: unknown) => { if (terminalTracker.emitted) { state.deps.log.error(