diff --git a/CHANGELOG.md b/CHANGELOG.md index 8872f5870e7c..56ce32a7abc1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -68,6 +68,7 @@ Docs: https://docs.openclaw.ai ### Fixes +- **Cron execution ownership:** record durable SQLite run receipts, fence overlapping gateway schedulers by exact process ownership, and supersede live results when the job's effective agent changes. - **Control UI session companion:** load bounded visible session context before answering, keep unavailable questions retryable, and prevent private companion reference wrappers from appearing as answers. Fixes #120746. Thanks @shakkernerd. - **Telegram live locations:** expose initial, moving, and stopped live-location updates through the channel-neutral `message_received` hook without starting agent turns for edits. - **Updater plugin convergence:** keep pre-plugin doctor passes from installing configured plugins before the updater's plugin sweep, while preserving the final post-plugin migration pass and preventing ambient update-phase state from leaking into fresh doctor processes. diff --git a/src/cron/service/ops-lifecycle.ts b/src/cron/service/ops-lifecycle.ts index fe5e806bd2a4..48e03ac9a7e6 100644 --- a/src/cron/service/ops-lifecycle.ts +++ b/src/cron/service/ops-lifecycle.ts @@ -1,4 +1,5 @@ import { materializeLegacyDefaultCronJobOwners } from "../legacy-default-agent-owner-migration.js"; +import { reconcileCronRunReceiptForStartup } from "../store/run-receipt-store.js"; import { failureNotificationDeliveryFromJobState } from "./failure-alerts.js"; import { nextWakeAtMs, recomputeNextRunsForMaintenance } from "./jobs-scheduling.js"; import { locked } from "./locked.js"; @@ -66,6 +67,18 @@ export async function start(state: CronServiceState) { // Older releases used runningAtMs for both queued and active work. Those // rows are intentionally recovered conservatively to avoid replaying side effects. const runningAtMs = job.state.runningAtMs; + const liveReceipt = reconcileCronRunReceiptForStartup({ + storePath: state.deps.storePath, + jobId: job.id, + startedAtMs: runningAtMs, + nowMs: state.deps.nowMs(), + }); + if (liveReceipt) { + // An overlapping replacement gateway must not retire work whose + // exact process incarnation is still alive. + interruptedJobIds.add(job.id); + continue; + } const taskRunId = tryFindCronTaskRunIdForRecovery(state, job.id, runningAtMs); const finalized = tryFindFinalizedCronTaskRun(state, job.id, runningAtMs); if (finalized) { diff --git a/src/cron/service/ops-run-preparation.ts b/src/cron/service/ops-run-preparation.ts index 6e040f0e71b6..cea9b62759f0 100644 --- a/src/cron/service/ops-run-preparation.ts +++ b/src/cron/service/ops-run-preparation.ts @@ -1,6 +1,7 @@ import type { CommandLaneTaskMarker } from "../../process/command-queue.js"; import type { CronActiveJobMarker } from "../active-jobs.js"; import { createCronRunDiagnosticsFromError } from "../run-diagnostics.js"; +import type { CronRunReceiptHandle } from "../store/run-receipt-store.js"; import type { CronJob, CronPayload, CronRunErrorClassification } from "../types.js"; import { normalizeCronRunErrorText } from "./execution-errors.js"; import { failureNotificationDeliveryFromJobState } from "./failure-alerts.js"; @@ -71,6 +72,7 @@ export type ActivatedManualRun = Extract & { activeJobMarker?: CronActiveJobMarker; admittedJob: CronJob; executionJob: CronJob; + runReceipt: CronRunReceiptHandle; }; export type ManualRunOptions = { @@ -483,13 +485,17 @@ export async function activatePreparedManualRun( if (activation.kind === "unavailable") { return { ok: true, ran: false, reason: activation.reason } as const; } + if (activation.kind === "fenced") { + await releasePreparedManualReservationWithRetry(state, prepared); + return { ok: true, ran: false, reason: "already-running" } as const; + } const { startedAt } = activation; emit(state, { jobId: job.id, action: "started", job, runAtMs: startedAt }); const taskRunId = tryCreateCronTaskRun({ state, job, startedAt, - publicRunId: prepared.runId, + publicRunId: prepared.runId ?? activation.runReceipt.receiptId, }); const activeJobMarker = markManualCronJobActive(state, job); // Execute against a snapshot so later reload/merge can preserve delivery @@ -512,6 +518,7 @@ export async function activatePreparedManualRun( activeJobMarker, admittedJob, executionJob, + runReceipt: activation.runReceipt, } as const; }); } diff --git a/src/cron/service/ops-run.ts b/src/cron/service/ops-run.ts index e68e64b0a734..9cc472f6d98c 100644 --- a/src/cron/service/ops-run.ts +++ b/src/cron/service/ops-run.ts @@ -2,6 +2,7 @@ import { enqueueCommandInLane } from "../../process/command-queue.js"; import { runWithGatewayIndependentRootWorkContinuation } from "../../process/gateway-work-admission.js"; import { CommandLane } from "../../process/lanes.js"; import { isCronActiveJobMarkerCurrent } from "../active-jobs.js"; +import { CronRunReceiptRevisionError, finishCronRunReceipt } from "../store/run-receipt-store.js"; import { normalizeCronRunErrorText } from "./execution-errors.js"; import { failureNotificationDeliveryFromJobState } from "./failure-alerts.js"; import { recomputeNextRunsForMaintenance } from "./jobs-scheduling.js"; @@ -19,6 +20,11 @@ import { } from "./ops-run-preparation.js"; import { clearManualCronJobActive, maybeNotifyManualIsolatedSetupTimeout } from "./ops-shared.js"; import { releaseQueuedCronRun, runWithCronAdmission } from "./run-admission.js"; +import { + assertServiceCronRunReceiptCurrent, + cronRunReceiptPersistHooks, + supersedeServiceCronRunReceipt, +} from "./run-receipts.js"; import { mergeManualRunSnapshotAfterReload } from "./startup-run-repair.js"; import type { CronServiceState, CronWakeMode, DeferredCronNotifications } from "./state.js"; import { emit } from "./state.js"; @@ -55,6 +61,7 @@ async function finishPreparedManualRun( const jobId = prepared.jobId; const taskRunId = prepared.taskRunId; const runId = prepared.runId; + let finalized = false; try { let coreResult: Awaited>; @@ -66,6 +73,7 @@ async function finishPreparedManualRun( streamBatch: prepared.streamBatch, streamScheduleKey: prepared.streamScheduleKey, streamSourceIdentity: prepared.streamSourceIdentity, + runReceipt: prepared.runReceipt, }); } catch (err) { coreResult = { status: "error", error: normalizeCronRunErrorText(err) }; @@ -147,7 +155,6 @@ async function finishPreparedManualRun( return; } - let finalized = false; let notifySetupTimeout = coreResult.isolatedAgentSetupTimeout !== undefined; await locked(state, async () => { await ensureLoaded(state, { skipRecompute: true }); @@ -162,6 +169,16 @@ async function finishPreparedManualRun( if (!job) { return; } + try { + assertServiceCronRunReceiptCurrent(state, prepared.runReceipt); + } catch (error) { + if (error instanceof CronRunReceiptRevisionError) { + supersedeServiceCronRunReceipt(prepared.runReceipt, state.deps.nowMs(), error.message); + notifySetupTimeout = false; + return; + } + throw error; + } const scheduleOwnership = resolveCronRunScheduleOwnership({ admittedJob: prepared.admittedJob, @@ -305,9 +322,27 @@ async function finishPreparedManualRun( } : {}), }); - await persistOrRestore(state, rollbackSnapshot, { - postPersistNotifications, - }); + try { + await persistOrRestore(state, rollbackSnapshot, { + postPersistNotifications, + transactionHooks: cronRunReceiptPersistHooks({ + state, + handle: prepared.runReceipt, + terminal: { + status: triggerSkipped ? "skipped" : coreResult.status, + finishedAtMs: endedAt, + error: coreResult.error, + }, + }), + }); + } catch (error) { + if (error instanceof CronRunReceiptRevisionError) { + supersedeServiceCronRunReceipt(prepared.runReceipt, state.deps.nowMs(), error.message); + notifySetupTimeout = false; + return; + } + throw error; + } if (removedJob) { pruneCronJobScratchAfterCommit(state, [removedJob.id]); emit(state, { jobId: removedJob.id, action: "removed", job: removedJob }); @@ -336,6 +371,14 @@ async function finishPreparedManualRun( } emitMissingQueuedTerminal(); } finally { + if (!finalized) { + finishCronRunReceipt({ + handle: prepared.runReceipt, + status: "superseded", + finishedAtMs: state.deps.nowMs(), + error: "cron run result was not applied to the current job revision", + }); + } releaseQueuedCronRun(state, prepared.jobId, prepared.reservationIdentity); clearManualCronJobActive(state, jobId, prepared.activeJobMarker); } diff --git a/src/cron/service/owner-hardening.test.ts b/src/cron/service/owner-hardening.test.ts new file mode 100644 index 000000000000..ebf64216ee58 --- /dev/null +++ b/src/cron/service/owner-hardening.test.ts @@ -0,0 +1,271 @@ +import { type ChildProcess, spawn } from "node:child_process"; +import fs from "node:fs"; +import fsPromises from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { pathToFileURL } from "node:url"; +import { afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { openOpenClawStateDatabase } from "../../state/openclaw-state-db.js"; +import { resolveOpenClawStateDirForDatabasePath } from "../../state/openclaw-state-db.paths.js"; +import { CronService } from "../service.js"; +import { createCronStoreHarness } from "../service.test-harness.js"; +import { loadCronStore, saveCronStore } from "../store.js"; +import { listCronRunReceipts } from "../store/run-receipt-store.js"; +import type { CronJob } from "../types.js"; + +const { makeStorePath } = createCronStoreHarness({ prefix: "cron-owner-hardening-" }); +const children = new Set(); +let scriptRoot = ""; +let runnerScript = ""; + +beforeAll(async () => { + scriptRoot = await fsPromises.mkdtemp(path.join(os.tmpdir(), "cron-owner-hardening-script-")); + runnerScript = path.join(scriptRoot, "runner.mts"); + const serviceUrl = pathToFileURL(path.resolve("src/cron/service.ts")).href; + await fsPromises.writeFile( + runnerScript, + ` + import fs from "node:fs"; + import { CronService } from ${JSON.stringify(serviceUrl)}; + const [storePath, jobId, mode, releasePath, outputPath] = process.argv.slice(2); + const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); + const logger = { debug() {}, info() {}, warn() {}, error() {} }; + const cron = new CronService({ + storePath, + cronEnabled: true, + log: logger, + enqueueSystemEvent() {}, + requestHeartbeat() {}, + evaluateCronTrigger: async () => { + process.stdout.write("trigger\\n"); + while (!fs.existsSync(releasePath)) await sleep(10); + return { kind: "evaluated", fire: true }; + }, + runIsolatedAgentJob: async () => ({ status: "ok" }), + runCommandJob: async () => { + fs.appendFileSync(outputPath, process.pid + "\\n"); + process.stdout.write("started\\n"); + if (mode === "block") await new Promise(() => {}); + await sleep(150); + return { status: "ok", summary: "done" }; + }, + }); + await cron.start(); + if (mode === "block") await cron.run(jobId, "force"); + if (mode === "due") await sleep(350); + cron.stop(); + `, + ); +}); + +afterEach(async () => { + for (const child of children) { + if (child.exitCode === null && child.signalCode === null) { + child.kill("SIGKILL"); + } + } + children.clear(); +}); + +function makeCommandJob(id: string, nextRunAtMs: number, trigger = false): CronJob { + return { + id, + agentId: "alpha", + name: id, + enabled: true, + createdAtMs: nextRunAtMs - 1, + updatedAtMs: nextRunAtMs - 1, + schedule: { kind: "every", everyMs: 60_000, anchorMs: nextRunAtMs }, + sessionTarget: "isolated", + wakeMode: "next-heartbeat", + ...(trigger ? { trigger: { script: "return true" } } : {}), + payload: { kind: "command", argv: ["true"] }, + state: { nextRunAtMs }, + }; +} + +function spawnRunner(params: { + storePath: string; + jobId: string; + mode: "block" | "trigger" | "due"; + releasePath: string; + outputPath: string; +}): ChildProcess { + const stateDir = resolveOpenClawStateDirForDatabasePath(openOpenClawStateDatabase().path); + const child = spawn( + process.execPath, + [ + "--import", + "tsx", + runnerScript, + params.storePath, + params.jobId, + params.mode, + params.releasePath, + params.outputPath, + ], + { + env: { ...process.env, OPENCLAW_STATE_DIR: stateDir }, + stdio: ["ignore", "pipe", "pipe"], + }, + ); + children.add(child); + return child; +} + +async function waitForLine(child: ChildProcess, expected: string): Promise { + let stdout = ""; + let stderr = ""; + child.stdout?.on("data", (chunk) => { + stdout += String(chunk); + }); + child.stderr?.on("data", (chunk) => { + stderr += String(chunk); + }); + await vi.waitFor( + () => { + if (child.exitCode !== null || child.signalCode !== null) { + throw new Error(`cron child exited before ${expected}: ${stderr || stdout}`); + } + expect(stdout.split("\n")).toContain(expected); + }, + { timeout: 10_000, interval: 20 }, + ); +} + +async function waitForExit(child: ChildProcess): Promise { + if (child.exitCode !== null || child.signalCode !== null) { + return; + } + await new Promise((resolve, reject) => { + child.once("exit", () => resolve()); + child.once("error", reject); + }); +} + +function makeParentService(storePath: string, runCommandJob = vi.fn()) { + return new CronService({ + storePath, + cronEnabled: true, + log: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + enqueueSystemEvent: vi.fn(), + requestHeartbeat: vi.fn(), + runIsolatedAgentJob: vi.fn(async () => ({ status: "ok" as const })), + runCommandJob, + }); +} + +describe("cron durable run ownership", () => { + it("does not execute when the durable receipt cannot be recorded", async () => { + vi.useRealTimers(); + const { storePath } = await makeStorePath(); + const now = Date.now(); + const job = makeCommandJob("receipt-required", now + 60_000); + await saveCronStore(storePath, { version: 1, jobs: [job] }); + listCronRunReceipts(storePath, job.id); + const database = openOpenClawStateDatabase().db; + database.exec(` + CREATE TRIGGER reject_cron_run_receipt + BEFORE INSERT ON cron_run_receipts + BEGIN + SELECT RAISE(ABORT, 'receipt unavailable'); + END; + `); + const runner = vi.fn(async () => ({ status: "ok" as const })); + const cron = makeParentService(storePath, runner); + try { + await expect(cron.run(job.id, "force")).rejects.toThrow("receipt unavailable"); + expect(runner).not.toHaveBeenCalled(); + } finally { + cron.stop(); + database.exec("DROP TRIGGER IF EXISTS reject_cron_run_receipt"); + } + }); + + it("keeps a live run fenced across an overlapping gateway start", async () => { + vi.useRealTimers(); + const { storePath } = await makeStorePath(); + const now = Date.now(); + const job = makeCommandJob("restart-mid-run", now + 60_000); + await saveCronStore(storePath, { version: 1, jobs: [job] }); + const releasePath = path.join(scriptRoot, `release-${now}`); + const outputPath = path.join(scriptRoot, `output-${now}`); + const owner = spawnRunner({ storePath, jobId: job.id, mode: "block", releasePath, outputPath }); + await waitForLine(owner, "started"); + + const replacementRunner = vi.fn(async () => ({ status: "ok" as const })); + const replacement = makeParentService(storePath, replacementRunner); + await replacement.start(); + await expect(replacement.run(job.id, "force")).resolves.toEqual({ + ok: true, + ran: false, + reason: "already-running", + }); + expect(replacementRunner).not.toHaveBeenCalled(); + expect(listCronRunReceipts(storePath, job.id)).toMatchObject([{ status: "running" }]); + replacement.stop(); + + owner.kill("SIGKILL"); + await waitForExit(owner); + const recovered = makeParentService(storePath); + await recovered.start(); + recovered.stop(); + + expect(listCronRunReceipts(storePath, job.id)[0]).toMatchObject({ status: "interrupted" }); + expect((await loadCronStore(storePath)).jobs[0]?.state.lastError).toContain( + "interrupted by gateway restart", + ); + }); + + it("admits one payload across overlapping scheduler processes", async () => { + vi.useRealTimers(); + const { storePath } = await makeStorePath(); + const now = Date.now(); + const job = makeCommandJob("overlapping-ticks", now - 1); + await saveCronStore(storePath, { version: 1, jobs: [job] }); + const releasePath = path.join(scriptRoot, `barrier-${now}`); + const outputPath = path.join(scriptRoot, `ticks-${now}`); + const first = spawnRunner({ storePath, jobId: job.id, mode: "due", releasePath, outputPath }); + const second = spawnRunner({ storePath, jobId: job.id, mode: "due", releasePath, outputPath }); + await Promise.all([waitForExit(first), waitForExit(second)]); + + const invocations = fs.existsSync(outputPath) + ? fs.readFileSync(outputPath, "utf8").trim().split("\n").filter(Boolean) + : []; + expect(invocations).toHaveLength(1); + expect(listCronRunReceipts(storePath, job.id)).toMatchObject([{ status: "ok" }]); + }); + + it("supersedes a live run before payload effects after its owner changes", async () => { + vi.useRealTimers(); + const { storePath } = await makeStorePath(); + const now = Date.now(); + const job = makeCommandJob("owner-change-live", now - 1, true); + await saveCronStore(storePath, { version: 1, jobs: [job] }); + const releasePath = path.join(scriptRoot, `owner-release-${now}`); + const outputPath = path.join(scriptRoot, `owner-output-${now}`); + const owner = spawnRunner({ + storePath, + jobId: job.id, + mode: "trigger", + releasePath, + outputPath, + }); + await waitForLine(owner, "trigger"); + + const editor = makeParentService(storePath); + await editor.update(job.id, { agentId: "beta" }); + editor.stop(); + await fsPromises.writeFile(releasePath, "release"); + await waitForExit(owner); + + expect(fs.existsSync(outputPath)).toBe(false); + expect(listCronRunReceipts(storePath, job.id)[0]).toMatchObject({ + agentId: "alpha", + status: "superseded", + }); + const current = (await loadCronStore(storePath)).jobs[0]; + expect(current?.agentId).toBe("beta"); + expect(current?.state.lastRunAtMs).toBeUndefined(); + }); +}); diff --git a/src/cron/service/run-admission.ts b/src/cron/service/run-admission.ts index 5ae6a7ee8e9b..4932a7ca0d8f 100644 --- a/src/cron/service/run-admission.ts +++ b/src/cron/service/run-admission.ts @@ -1,10 +1,20 @@ import { DEFAULT_CRON_MAX_CONCURRENT_RUNS } from "../../config/cron-limits.js"; import { markCronJobActive } from "../active-jobs.js"; import { createCronRunDiagnosticsFromError } from "../run-diagnostics.js"; +import { + CronRunReceiptConflictError, + CronRunReceiptRevisionError, + finishCronRunReceipt, +} from "../store/run-receipt-store.js"; import type { CronJob } from "../types.js"; import { normalizeCronRunErrorText } from "./execution-errors.js"; import { recomputeNextRunsForMaintenance } from "./jobs-scheduling.js"; import { locked } from "./locked.js"; +import { + claimServiceCronRunReceipt, + cronRunReceiptPersistHooks, + supersedeServiceCronRunReceipt, +} from "./run-receipts.js"; import { type CronServiceState, type DeferredCronNotifications, emit } from "./state.js"; import { ensureLoaded, persistOrRestore, snapshotStoreForRollback } from "./store.js"; import { tryCreateCronTaskRun } from "./task-runs.js"; @@ -243,11 +253,28 @@ export async function activateQueuedCronRun(params: { onUnavailable?: () => void; onUnavailableRollbackError?: () => Promise; }): Promise< - | { kind: "activated"; startedAt: number } + | { + kind: "activated"; + startedAt: number; + runReceipt: ReturnType; + } + | { kind: "fenced" } | { kind: "unavailable"; reason: "stopped" | "restart-recovery-pending" } > { const { state, job, reservationIdentity } = params; const startedAt = state.deps.nowMs(); + let runReceipt: ReturnType; + try { + runReceipt = claimServiceCronRunReceipt({ state, job, startedAtMs: startedAt }); + } catch (error) { + if ( + error instanceof CronRunReceiptConflictError || + error instanceof CronRunReceiptRevisionError + ) { + return { kind: "fenced" }; + } + throw error; + } const previousLastError = job.state.lastError; const activationRollbackSnapshot = snapshotStoreForRollback(state); delete job.state.queuedAtMs; @@ -255,14 +282,30 @@ export async function activateQueuedCronRun(params: { job.state.lastError = undefined; // Persist running ownership before execution. A failed write restores the // durable queued marker so the caller can release or recover that claim. - await persistOrRestore(state, activationRollbackSnapshot); + try { + await persistOrRestore(state, activationRollbackSnapshot, { + transactionHooks: cronRunReceiptPersistHooks({ state, handle: runReceipt }), + }); + } catch (error) { + if (error instanceof CronRunReceiptRevisionError) { + supersedeServiceCronRunReceipt(runReceipt, state.deps.nowMs(), error.message); + return { kind: "fenced" }; + } + finishCronRunReceipt({ + handle: runReceipt, + status: "error", + finishedAtMs: state.deps.nowMs(), + error: normalizeCronRunErrorText(error), + }); + throw error; + } const reservation = state.queuedRunReservationsByJobId.get(job.id); if (reservation?.identity === reservationIdentity) { reservation.markerAtMs = startedAt; reservation.activationPreviousLastError = { value: previousLastError }; } if (!state.stopped && !state.restartRecoveryPending) { - return { kind: "activated", startedAt }; + return { kind: "activated", startedAt, runReceipt }; } params.onUnavailable?.(); @@ -270,7 +313,17 @@ export async function activateQueuedCronRun(params: { const rollbackSnapshot = snapshotStoreForRollback(state); delete job.state.runningAtMs; try { - await persistOrRestore(state, rollbackSnapshot); + await persistOrRestore(state, rollbackSnapshot, { + transactionHooks: cronRunReceiptPersistHooks({ + state, + handle: runReceipt, + terminal: { + status: "skipped", + finishedAtMs: state.deps.nowMs(), + error: state.stopped ? "cron service stopped" : "cron restart recovery pending", + }, + }), + }); } catch (error) { await params.onUnavailableRollbackError?.(); throw error; @@ -360,7 +413,7 @@ export async function executeQueuedCronRun(params: { } activated = true; params.onActivated?.(); - return { job, startedAt: activation.startedAt }; + return { job, startedAt: activation.startedAt, runReceipt: activation.runReceipt }; }); if (!started) { return undefined; @@ -372,6 +425,7 @@ export async function executeQueuedCronRun(params: { state, job: executionJob, startedAt: started.startedAt, + publicRunId: started.runReceipt.receiptId, }); const activeJobMarker = markCronJobActive(executionJob.id, { preserveAcrossGenerationAdvance: !runsDetachedFromMainSession(executionJob), @@ -389,12 +443,14 @@ export async function executeQueuedCronRun(params: { activeJobMarker, reservationIdentity: params.reservationIdentity, startedAt: started.startedAt, + runReceipt: started.runReceipt, }; let outcome: TimedCronRunOutcome; try { const result = await executeJobCoreWithTimeout(state, executionJob, { runId: taskRunId, activeJobMarker, + runReceipt: started.runReceipt, }); outcome = { ...base, ...result, endedAt: state.deps.nowMs() }; } catch (error) { diff --git a/src/cron/service/run-receipts.ts b/src/cron/service/run-receipts.ts new file mode 100644 index 000000000000..921598d96a86 --- /dev/null +++ b/src/cron/service/run-receipts.ts @@ -0,0 +1,113 @@ +import type { CronStoreTransactionHooks } from "../store.js"; +import { + assertCronRunReceiptCurrent, + assertCronRunReceiptCurrentInDatabase, + claimCronRunReceipt, + CronRunReceiptRevisionError, + finishCronRunReceipt, + finishCronRunReceiptInDatabase, + type CronRunReceiptHandle, + type CronRunReceiptStatus, +} from "../store/run-receipt-store.js"; +import type { CronJob, CronRunStatus } from "../types.js"; +import { resolveEffectiveJobAgentId } from "./ops-shared.js"; +import type { CronServiceState } from "./state.js"; + +function currentDefaultAgentId(state: CronServiceState): string | undefined { + return state.deps.resolveDefaultAgentId?.() ?? state.deps.defaultAgentId; +} + +export function resolveCronRunReceiptAgentId(state: CronServiceState, job: CronJob): string { + return resolveEffectiveJobAgentId(job, currentDefaultAgentId(state)); +} + +function resolveAgentId(state: CronServiceState) { + return (job: CronJob) => resolveCronRunReceiptAgentId(state, job); +} + +export function claimServiceCronRunReceipt(params: { + state: CronServiceState; + job: CronJob; + startedAtMs: number; + requestRunId?: string; +}): CronRunReceiptHandle { + return claimCronRunReceipt({ + storePath: params.state.deps.storePath, + job: params.job, + agentId: resolveCronRunReceiptAgentId(params.state, params.job), + startedAtMs: params.startedAtMs, + requestRunId: params.requestRunId, + resolveAgentId: resolveAgentId(params.state), + }); +} + +export function assertServiceCronRunReceiptCurrent( + state: CronServiceState, + handle: CronRunReceiptHandle, +): void { + assertCronRunReceiptCurrent({ + handle, + resolveAgentId: resolveAgentId(state), + isAgentAvailable: state.deps.isAgentAvailable, + }); +} + +function terminalReceiptStatus(status: CronRunStatus): Exclude { + if (status === "ok") { + return "ok"; + } + if (status === "skipped") { + return "skipped"; + } + return "error"; +} + +export function cronRunReceiptPersistHooks(params: { + state: CronServiceState; + handle: CronRunReceiptHandle; + terminal?: { status: CronRunStatus; finishedAtMs: number; error?: string }; +}): CronStoreTransactionHooks { + return { + beforeWrite: (database) => { + if (params.state.deps.isAgentAvailable?.(params.handle.agentId) === false) { + throw new CronRunReceiptRevisionError( + params.handle.receiptId, + `cron run owner ${params.handle.agentId} is no longer configured`, + ); + } + assertCronRunReceiptCurrentInDatabase({ + database, + handle: params.handle, + resolveAgentId: resolveAgentId(params.state), + }); + }, + ...(params.terminal + ? { + afterWrite: ( + database: Parameters>[0], + ) => { + finishCronRunReceiptInDatabase({ + database, + handle: params.handle, + status: terminalReceiptStatus(params.terminal!.status), + finishedAtMs: params.terminal!.finishedAtMs, + error: params.terminal!.error, + }); + }, + } + : {}), + }; +} + +export function supersedeServiceCronRunReceipt( + handle: CronRunReceiptHandle, + finishedAtMs: number, + error: string, +): void { + finishCronRunReceipt({ + handle, + status: "superseded", + finishedAtMs, + error, + }); +} diff --git a/src/cron/service/store.ts b/src/cron/service/store.ts index dbb8865781d2..9ae05b5c4a29 100644 --- a/src/cron/service/store.ts +++ b/src/cron/service/store.ts @@ -9,6 +9,7 @@ import { getCronJobsStoreRevision, loadCronJobsStoreWithConfigJobs, saveCronJobsStore, + type CronStoreTransactionHooks, type QuarantinedCronConfigJob, } from "../store.js"; import type { CronJob, CronStoreFile } from "../types.js"; @@ -22,6 +23,7 @@ type PersistOptions = { stateOnly?: boolean; suppressScheduledJobId?: string; postPersistNotifications?: DeferredCronNotifications; + transactionHooks?: CronStoreTransactionHooks; }; export type CronRollbackSnapshot = { @@ -282,7 +284,13 @@ export async function persist(state: CronServiceState, opts?: PersistOptions) { await saveCronJobsStore( state.deps.storePath, store, - quarantine ? { quarantine } : stateOnly ? { stateOnly: true } : undefined, + quarantine + ? { quarantine, transactionHooks: opts?.transactionHooks } + : stateOnly + ? { stateOnly: true, transactionHooks: opts?.transactionHooks } + : opts?.transactionHooks + ? { transactionHooks: opts.transactionHooks } + : undefined, ); } catch (error) { if (!quarantine) { diff --git a/src/cron/service/timer-execution-timeout.ts b/src/cron/service/timer-execution-timeout.ts index 5ba31f7cbc92..0c8f80fe665f 100644 --- a/src/cron/service/timer-execution-timeout.ts +++ b/src/cron/service/timer-execution-timeout.ts @@ -4,6 +4,7 @@ import { normalizeAgentId, resolveAgentIdFromSessionKey } from "../../routing/se import { deliveryContextFromSession } from "../../utils/delivery-context.shared.js"; import type { DeliveryContext } from "../../utils/delivery-context.types.js"; import type { CronActiveJobMarker } from "../active-jobs.js"; +import type { CronRunReceiptHandle } from "../store/run-receipt-store.js"; import type { CronAgentExecutionPhaseUpdate, CronAgentExecutionStarted, @@ -46,6 +47,7 @@ export type TimedCronRunOutcome = CronRunOutcome & isolatedAgentSetupTimeout?: IsolatedAgentSetupTimeoutSignal; activeJobMarker?: CronActiveJobMarker; reservationIdentity?: object; + runReceipt?: CronRunReceiptHandle; startedAt: number; endedAt: number; triggerEval?: CronTriggerEvalOutcome; @@ -110,6 +112,8 @@ export type ExecuteJobCoreOptions = { onExecutionStarted?: (info?: CronAgentExecutionStarted) => void; onExecutionPhase?: (info: CronAgentExecutionPhaseUpdate) => void; onLaneWait?: (info?: { waiting?: boolean }) => void; + /** Revalidates the durable run fence after awaited planning and before effects. */ + assertRunCurrent?: () => void; streamBatch?: string; // Source definition and logical identity are an inseparable admission claim. // The key catches edits; the identity catches disable→re-enable and A→B→A. diff --git a/src/cron/service/timer-execution.ts b/src/cron/service/timer-execution.ts index 8f03a2f0178d..c602a1954395 100644 --- a/src/cron/service/timer-execution.ts +++ b/src/cron/service/timer-execution.ts @@ -141,6 +141,7 @@ export async function executeJobCore( effectiveJob = { ...job, payload: appendCronPayloadText(job.payload, evaluation.message) }; } } + options?.assertRunCurrent?.(); if (effectiveJob.payload.kind === "script") { const result = await executeScriptCronJob( state, @@ -148,6 +149,7 @@ export async function executeJobCore( abortSignal, options?.activeJobMarker, options?.streamBatch, + options?.assertRunCurrent, ); return triggerEval ? { ...result, triggerEval } : result; } @@ -493,6 +495,7 @@ async function executeScriptCronJob( abortSignal: AbortSignal | undefined, activeJobMarker?: CronActiveJobMarker, streamBatch?: string, + assertRunCurrent?: () => void, ) { if (state.deps.cronConfig?.triggers?.enabled !== true) { return { @@ -513,6 +516,7 @@ async function executeScriptCronJob( if (abortSignal?.aborted) { return { status: "error" as const, error: abortErrorMessage(abortSignal) }; } + assertRunCurrent?.(); if (result.status !== "ok") { return result; } diff --git a/src/cron/service/timer-job-runner.ts b/src/cron/service/timer-job-runner.ts index 987bee65b599..3354173f8fa4 100644 --- a/src/cron/service/timer-job-runner.ts +++ b/src/cron/service/timer-job-runner.ts @@ -18,6 +18,7 @@ import { isSetupTimeoutErrorText, timeoutErrorMessage, } from "./execution-errors.js"; +import { assertServiceCronRunReceiptCurrent } from "./run-receipts.js"; import type { CronServiceState } from "./state.js"; import { tryUpdateCronTaskRunSession, withCronTaskRunId } from "./task-runs.js"; import { resolveCronJobTimeoutMs } from "./timeout-policy.js"; @@ -43,6 +44,7 @@ type CronCoreRunOptions = { streamBatch?: string; streamScheduleKey?: string; streamSourceIdentity?: string; + runReceipt?: import("../store/run-receipt-store.js").CronRunReceiptHandle; }; async function deliverPrimaryWebhook( @@ -52,6 +54,7 @@ async function deliverPrimaryWebhook( abortSignal: AbortSignal, progress: CronRunProgress, deadlineAtMs?: number, + assertRunCurrent?: () => void, ): Promise { const settle = (settledResult: CronCoreRunOutcome) => { // Publish the terminal delivery fact before this async function resolves; @@ -97,6 +100,8 @@ async function deliverPrimaryWebhook( }); } + assertRunCurrent?.(); + const startedAt = job.state.runningAtMs; const deliveredResult = withPrimaryWebhookTrace({ job, result, delivered: true }); try { @@ -183,6 +188,9 @@ export async function executeJobCoreWithTimeout( opts?: CronCoreRunOptions, ): Promise { const runAbortController = new AbortController(); + const assertRunCurrent = opts?.runReceipt + ? () => assertServiceCronRunReceiptCurrent(state, opts.runReceipt!) + : undefined; const operatorCancellationMarker = Symbol("cron-operator-cancelled"); let resolveOperatorCancellation: ((value: typeof operatorCancellationMarker) => void) | undefined; const operatorCancellationPromise = new Promise((resolve) => { @@ -244,13 +252,22 @@ export async function executeJobCoreWithTimeout( streamSourceIdentity: opts?.streamSourceIdentity, onExecutionStarted: noteExecutionStarted, onExecutionPhase: accumulateExecution, + assertRunCurrent, }; const corePromise = withCronTaskRunId(opts?.runId, () => executeJobCore(state, job, runAbortController.signal, coreOptions), ); const runPromise = corePromise.then(async (result) => { progress.completedCoreResult = result; - return await deliverPrimaryWebhook(state, job, result, runAbortController.signal, progress); + return await deliverPrimaryWebhook( + state, + job, + result, + runAbortController.signal, + progress, + undefined, + assertRunCurrent, + ); }); trackActiveCronTaskRunSettlement(runPromise, runAbortController.signal); void runPromise.catch((err: unknown) => { @@ -322,6 +339,7 @@ export async function executeJobCoreWithTimeout( onExecutionStarted: deferTimeoutUntilExecutionStart ? noteRunnerStarted : undefined, onExecutionPhase: deferTimeoutUntilExecutionStart ? watchdog.notePhase : undefined, onLaneWait: deferTimeoutUntilExecutionStart ? noteLaneState : undefined, + assertRunCurrent, }; const corePromise = withCronTaskRunId(opts?.runId, () => executeJobCore(state, job, runAbortController.signal, coreOptions), @@ -336,6 +354,7 @@ export async function executeJobCoreWithTimeout( runAbortController.signal, progress, watchdog.deadlineAtMs(), + assertRunCurrent, ); }); trackActiveCronTaskRunSettlement(runPromise, runAbortController.signal); diff --git a/src/cron/service/timer-outcome-finalization.ts b/src/cron/service/timer-outcome-finalization.ts index ba7ac17df3a6..d2b7c74597c7 100644 --- a/src/cron/service/timer-outcome-finalization.ts +++ b/src/cron/service/timer-outcome-finalization.ts @@ -1,10 +1,16 @@ /** Finalizes cron task rows and active markers after timer outcome persistence. */ import { clearCronJobActive, isCronActiveJobMarkerCurrent } from "../active-jobs.js"; import type { CronActiveJobMarker } from "../active-jobs.js"; +import { + CronRunReceiptRevisionError, + releaseLocalCronRunReceiptOwnership, + type CronRunReceiptHandle, +} from "../store/run-receipt-store.js"; import type { CronJob } from "../types.js"; import { recomputeNextRunsForMaintenance } from "./jobs-scheduling.js"; import { locked } from "./locked.js"; import { releaseQueuedCronRun } from "./run-admission.js"; +import { cronRunReceiptPersistHooks, supersedeServiceCronRunReceipt } from "./run-receipts.js"; import { emit, type CronServiceState, type DeferredCronNotifications } from "./state.js"; import { ensureLoaded, @@ -26,6 +32,7 @@ type CronTaskRunFinalizationOutcome = { childSessionKey?: string; triggerEval?: { fired: boolean }; activeJobMarker?: CronActiveJobMarker; + runReceipt?: CronRunReceiptHandle; }; type CompletedCronRunOutcomeFinalizationOptions = { @@ -158,8 +165,37 @@ export async function finalizeCompletedCronRunOutcomes( ); // Run notifications describe durable state. Drain them only after the // terminal write succeeds so rollback cannot publish a false outcome. + const receiptHooks = finalizedOutcomes + .filter((outcome) => outcome.runReceipt) + .map((outcome) => + cronRunReceiptPersistHooks({ + state, + handle: outcome.runReceipt!, + terminal: { + status: outcome.status, + finishedAtMs: outcome.endedAt, + error: outcome.error, + }, + }), + ); await persistOrRestore(state, rollbackSnapshot, { postPersistNotifications, + ...(receiptHooks.length > 0 + ? { + transactionHooks: { + beforeWrite: (database) => { + for (const hooks of receiptHooks) { + hooks.beforeWrite?.(database); + } + }, + afterWrite: (database) => { + for (const hooks of receiptHooks) { + hooks.afterWrite?.(database); + } + }, + }, + } + : {}), }); pruneCronJobScratchAfterCommit( state, @@ -173,6 +209,22 @@ export async function finalizeCompletedCronRunOutcomes( finalizationSucceeded ||= finalizedOutcomes.length > 0; return finalizedOutcomes; + } catch (error) { + if (error instanceof CronRunReceiptRevisionError) { + const stale = outcomes.find((outcome) => outcome.runReceipt?.receiptId === error.receiptId); + if (stale?.runReceipt) { + supersedeServiceCronRunReceipt(stale.runReceipt, state.deps.nowMs(), error.message); + tryFinishCronTaskRunWithoutHistory(state, { + taskRunId: stale.taskRunId, + status: "skipped", + error: error.message, + endedAt: state.deps.nowMs(), + }); + const remaining = outcomes.filter((outcome) => outcome !== stale); + return await finalizeCompletedCronRunOutcomes(state, remaining, opts); + } + } + throw error; } finally { for (const outcome of outcomes) { if (outcome.reservationIdentity) { @@ -182,6 +234,11 @@ export async function finalizeCompletedCronRunOutcomes( if (opts?.clearOnFailure !== false || finalizationSucceeded) { clearActiveMarkersForOutcomes(outcomes); } + for (const outcome of outcomes) { + if (outcome.runReceipt) { + releaseLocalCronRunReceiptOwnership(outcome.runReceipt); + } + } } } @@ -216,6 +273,13 @@ function finishRetiredCronTaskRuns( const current = new Set(currentOutcomes); for (const outcome of outcomes) { if (!current.has(outcome)) { + if (outcome.runReceipt) { + supersedeServiceCronRunReceipt( + outcome.runReceipt, + state.deps.nowMs(), + "cron run retired before its result became durable", + ); + } tryFinishCronTaskRunWithoutHistory(state, outcome); } } diff --git a/src/cron/store.ts b/src/cron/store.ts index b22d0e57d9c1..96db2571712d 100644 --- a/src/cron/store.ts +++ b/src/cron/store.ts @@ -236,11 +236,19 @@ type SaveCronStoreOptions = { stateOnly?: boolean; }; +export type CronStoreTransactionHooks = { + /** Runs inside the authoritative write transaction before cron rows change. */ + beforeWrite?: (db: DatabaseSync) => void; + /** Runs after row changes but before the same transaction commits. */ + afterWrite?: (db: DatabaseSync) => void; +}; + type SaveCronJobsStoreOptions = SaveCronStoreOptions & { quarantine?: { entries: readonly (QuarantinedCronConfigJob | CronQuarantinedJob)[]; nowMs: number; }; + transactionHooks?: CronStoreTransactionHooks; }; /** Persists cron jobs, or only mutable runtime state when stateOnly is set. */ @@ -256,6 +264,7 @@ export async function saveCronJobsStore( assertCronStoreCanPersist(store); } runOpenClawStateWriteTransaction((database) => { + opts?.transactionHooks?.beforeWrite?.(database.db); if (opts?.quarantine?.entries.length) { saveCronQuarantinedJobs({ storePath: resolvedStorePath, @@ -268,10 +277,12 @@ export async function saveCronJobsStore( // quarantine and full replacement commit together or roll back together. if (stateOnly) { updateCronRuntimeRows(database.db, storeKey, store); + opts?.transactionHooks?.afterWrite?.(database.db); return; } const normalizedJobs = replaceCronRows(database.db, storeKey, store); replaceCronRuntimeAuthorityRows({ db: database.db, storeKey, jobs: normalizedJobs }); + opts?.transactionHooks?.afterWrite?.(database.db); }); noteCronJobsStoreCommit(storeKey); } diff --git a/src/cron/store/run-receipt-store.test.ts b/src/cron/store/run-receipt-store.test.ts new file mode 100644 index 000000000000..a6ca681276c8 --- /dev/null +++ b/src/cron/store/run-receipt-store.test.ts @@ -0,0 +1,111 @@ +import { describe, expect, it } from "vitest"; +import { openOpenClawStateDatabase } from "../../state/openclaw-state-db.js"; +import { setupCronServiceSuite } from "../service.test-harness.js"; +import { saveCronStore } from "../store.js"; +import type { CronJob } from "../types.js"; +import { + assertCronRunReceiptCurrent, + claimCronRunReceipt, + CronRunReceiptConflictError, + CronRunReceiptRevisionError, + finishCronRunReceipt, + listCronRunReceipts, +} from "./run-receipt-store.js"; + +const { makeStorePath } = setupCronServiceSuite({ prefix: "cron-run-receipt-" }); + +function makeJob(id: string, agentId = "alpha"): CronJob { + return { + id, + agentId, + name: id, + enabled: true, + createdAtMs: 1, + updatedAtMs: 1, + schedule: { kind: "every", everyMs: 60_000 }, + sessionTarget: "isolated", + wakeMode: "next-heartbeat", + payload: { kind: "agentTurn", message: id }, + state: {}, + }; +} + +function claim(storePath: string, job: CronJob, startedAtMs: number) { + return claimCronRunReceipt({ + storePath, + job, + agentId: job.agentId!, + startedAtMs, + resolveAgentId: (current) => current.agentId!, + }); +} + +describe("cron run receipt store", () => { + it("records one durable active run and rejects an overlapping claimant", async () => { + const { storePath } = await makeStorePath(); + const job = makeJob("overlap"); + await saveCronStore(storePath, { version: 1, jobs: [job] }); + + const first = claim(storePath, job, 100); + + expect(() => claim(storePath, job, 101)).toThrow(CronRunReceiptConflictError); + expect(listCronRunReceipts(storePath, job.id)).toMatchObject([ + { receiptId: first.receiptId, status: "running", startedAtMs: 100 }, + ]); + + finishCronRunReceipt({ handle: first, status: "ok", finishedAtMs: 110 }); + const second = claim(storePath, job, 120); + finishCronRunReceipt({ handle: second, status: "skipped", finishedAtMs: 121 }); + + expect(listCronRunReceipts(storePath, job.id).map((receipt) => receipt.status)).toEqual([ + "skipped", + "ok", + ]); + }); + + it("retires a provably dead process claim before admitting its successor", async () => { + const { storePath } = await makeStorePath(); + const job = makeJob("restart"); + await saveCronStore(storePath, { version: 1, jobs: [job] }); + const abandoned = claim(storePath, job, 200); + openOpenClawStateDatabase() + .db.prepare("UPDATE cron_run_receipts SET owner_pid = ? WHERE receipt_id = ?") + .run(2_147_483_647, abandoned.receiptId); + + const replacement = claim(storePath, job, 220); + + expect(replacement.receiptId).not.toBe(abandoned.receiptId); + expect(listCronRunReceipts(storePath, job.id)).toMatchObject([ + { receiptId: replacement.receiptId, status: "running" }, + { receiptId: abandoned.receiptId, status: "interrupted" }, + ]); + }); + + it("rejects a live run after its durable owner revision changes", async () => { + const { storePath } = await makeStorePath(); + const admitted = makeJob("owner-change", "alpha"); + await saveCronStore(storePath, { version: 1, jobs: [admitted] }); + const receipt = claim(storePath, admitted, 300); + const reassigned = { ...admitted, agentId: "beta", updatedAtMs: 2 }; + await saveCronStore(storePath, { version: 1, jobs: [reassigned] }); + + expect(() => + assertCronRunReceiptCurrent({ + handle: receipt, + resolveAgentId: (job) => job.agentId!, + }), + ).toThrow(CronRunReceiptRevisionError); + + finishCronRunReceipt({ + handle: receipt, + status: "superseded", + finishedAtMs: 310, + error: "owner changed", + }); + expect(listCronRunReceipts(storePath, admitted.id)[0]).toMatchObject({ + status: "superseded", + agentId: "alpha", + error: "owner changed", + }); + }); +}); diff --git a/src/cron/store/run-receipt-store.ts b/src/cron/store/run-receipt-store.ts new file mode 100644 index 000000000000..796f7d5ae362 --- /dev/null +++ b/src/cron/store/run-receipt-store.ts @@ -0,0 +1,463 @@ +import crypto from "node:crypto"; +import type { DatabaseSync } from "node:sqlite"; +import type { Selectable } from "kysely"; +import { + executeSqliteQuerySync, + executeSqliteQueryTakeFirstSync, + getNodeSqliteKysely, +} from "../../infra/kysely-sync.js"; +import { getFileLockProcessStartTime, isPidDefinitelyDead } from "../../shared/pid-alive.js"; +import type { DB as OpenClawStateDatabase } from "../../state/openclaw-state-db.generated.js"; +import { + runOpenClawStateWriteTransaction, + type OpenClawStateDatabaseOptions, +} from "../../state/openclaw-state-db.js"; +import { OPENCLAW_STATE_SCHEMA_SQL } from "../../state/openclaw-state-schema.js"; +import { resolveCronJobConfigRevision } from "../config-revision.js"; +import type { CronJob } from "../types.js"; +import { cronStoreKey } from "./key.js"; +import { loadedCronStoreFromRows, loadCronRows } from "./row-codec.js"; + +type CronRunReceiptDatabase = Pick; +type CronRunReceiptRow = Selectable; + +export type CronRunReceiptStatus = + | "running" + | "ok" + | "error" + | "skipped" + | "interrupted" + | "superseded"; + +export type CronRunReceipt = { + receiptId: string; + storeKey: string; + jobId: string; + configRevision: string; + agentId: string; + requestRunId?: string; + status: CronRunReceiptStatus; + ownerPid: number; + ownerStartTime: number | null; + startedAtMs: number; + finishedAtMs: number | null; + error?: string; +}; + +export type CronRunReceiptHandle = Pick< + CronRunReceipt, + | "agentId" + | "configRevision" + | "jobId" + | "ownerPid" + | "ownerStartTime" + | "receiptId" + | "startedAtMs" + | "storeKey" +>; + +type ResolveReceiptAgentId = (job: CronJob) => string; + +const CRON_RUN_RECEIPT_SCHEMA_START = "CREATE TABLE IF NOT EXISTS cron_run_receipts ("; +const CRON_RUN_RECEIPT_SCHEMA_END = + "ON cron_run_receipts(store_key, job_id, started_at_ms DESC, receipt_id DESC);"; +const initializedDatabases = new WeakSet(); +const locallyOwnedReceipts = new Set(); + +export class CronRunReceiptConflictError extends Error { + constructor(readonly receipt: CronRunReceipt) { + super(`cron job ${receipt.jobId} is already running in process ${receipt.ownerPid}`); + this.name = "CronRunReceiptConflictError"; + } +} + +export class CronRunReceiptRevisionError extends Error { + constructor( + readonly receiptId: string, + message = "cron run configuration changed", + ) { + super(message); + this.name = "CronRunReceiptRevisionError"; + } +} + +function ensureCronRunReceiptSchema(database: DatabaseSync): void { + const start = OPENCLAW_STATE_SCHEMA_SQL.indexOf(CRON_RUN_RECEIPT_SCHEMA_START); + const endMarker = OPENCLAW_STATE_SCHEMA_SQL.indexOf(CRON_RUN_RECEIPT_SCHEMA_END, start); + if (start < 0 || endMarker < start) { + throw new Error("OpenClaw cron run receipt schema marker is missing."); + } + database.exec( + OPENCLAW_STATE_SCHEMA_SQL.slice(start, endMarker + CRON_RUN_RECEIPT_SCHEMA_END.length), + ); // sqlite-allow-raw -- Canonical feature-local additive DDL only. +} + +function query(database: DatabaseSync) { + return getNodeSqliteKysely(database); +} + +function withReceiptWrite( + operationLabel: string, + options: OpenClawStateDatabaseOptions, + operation: (database: DatabaseSync) => T, +): T { + let initializedDatabase: DatabaseSync | undefined; + const result = runOpenClawStateWriteTransaction( + ({ db }) => { + if (!initializedDatabases.has(db)) { + ensureCronRunReceiptSchema(db); + initializedDatabase = db; + } + return operation(db); + }, + options, + { operationLabel }, + ); + if (initializedDatabase) { + initializedDatabases.add(initializedDatabase); + } + return result; +} + +function isReceiptStatus(value: string): value is CronRunReceiptStatus { + return ( + value === "running" || + value === "ok" || + value === "error" || + value === "skipped" || + value === "interrupted" || + value === "superseded" + ); +} + +function receiptFromRow(row: CronRunReceiptRow): CronRunReceipt { + if (!isReceiptStatus(row.status)) { + throw new Error(`invalid cron run receipt status ${row.status}`); + } + return { + receiptId: row.receipt_id, + storeKey: row.store_key, + jobId: row.job_id, + configRevision: row.config_revision, + agentId: row.agent_id, + ...(row.request_run_id ? { requestRunId: row.request_run_id } : {}), + status: row.status, + ownerPid: row.owner_pid, + ownerStartTime: row.owner_start_time, + startedAtMs: row.started_at_ms, + finishedAtMs: row.finished_at_ms, + ...(row.error_text ? { error: row.error_text } : {}), + }; +} + +function activeRow(database: DatabaseSync, storeKey: string, jobId: string) { + return executeSqliteQueryTakeFirstSync( + database, + query(database) + .selectFrom("cron_run_receipts") + .selectAll() + .where("store_key", "=", storeKey) + .where("job_id", "=", jobId) + .where("status", "=", "running"), + ); +} + +function currentJob(database: DatabaseSync, storeKey: string, jobId: string): CronJob | undefined { + const rows = loadCronRows(database, storeKey); + if (rows.length === 0) { + return undefined; + } + return loadedCronStoreFromRows(rows).store.jobs.find((job) => job.id === jobId); +} + +function sameOwner(left: CronRunReceiptRow, right: CronRunReceiptRow): boolean { + return ( + left.receipt_id === right.receipt_id && + left.owner_pid === right.owner_pid && + left.owner_start_time === right.owner_start_time && + left.status === right.status + ); +} + +function ownerDefinitelyStale(row: CronRunReceiptRow): boolean { + if (row.owner_pid === process.pid) { + return !locallyOwnedReceipts.has(row.receipt_id); + } + if (isPidDefinitelyDead(row.owner_pid)) { + return true; + } + const observedStartTime = getFileLockProcessStartTime(row.owner_pid); + return ( + row.owner_start_time !== null && + observedStartTime !== null && + row.owner_start_time !== observedStartTime + ); +} + +function validateCurrentJob(params: { + database: DatabaseSync; + handle: Pick< + CronRunReceiptHandle, + "agentId" | "configRevision" | "jobId" | "receiptId" | "storeKey" + >; + resolveAgentId: ResolveReceiptAgentId; +}): CronJob { + const job = currentJob(params.database, params.handle.storeKey, params.handle.jobId); + if (!job) { + throw new CronRunReceiptRevisionError(params.handle.receiptId, "cron job was removed"); + } + if (params.resolveAgentId(job) !== params.handle.agentId) { + throw new CronRunReceiptRevisionError(params.handle.receiptId); + } + return job; +} + +function receiptHandle(receipt: CronRunReceipt): CronRunReceiptHandle { + return { + receiptId: receipt.receiptId, + storeKey: receipt.storeKey, + jobId: receipt.jobId, + configRevision: receipt.configRevision, + agentId: receipt.agentId, + ownerPid: receipt.ownerPid, + ownerStartTime: receipt.ownerStartTime, + startedAtMs: receipt.startedAtMs, + }; +} + +/** Atomically records a run start and acquires the one-active-run durable fence. */ +export function claimCronRunReceipt(params: { + storePath: string; + job: CronJob; + agentId: string; + startedAtMs: number; + requestRunId?: string; + resolveAgentId: ResolveReceiptAgentId; + env?: NodeJS.ProcessEnv; +}): CronRunReceiptHandle { + const storeKey = cronStoreKey(params.storePath); + const options = params.env ? { env: params.env } : {}; + const observed = withReceiptWrite("cron.run-receipt.inspect", options, (database) => + activeRow(database, storeKey, params.job.id), + ); + const observedStale = observed ? ownerDefinitelyStale(observed) : false; + const receiptId = crypto.randomUUID(); + const handle: CronRunReceiptHandle = { + receiptId, + storeKey, + jobId: params.job.id, + configRevision: resolveCronJobConfigRevision(params.job), + agentId: params.agentId, + ownerPid: process.pid, + ownerStartTime: getFileLockProcessStartTime(process.pid), + startedAtMs: params.startedAtMs, + }; + + const claimed = withReceiptWrite("cron.run-receipt.claim", options, (database) => { + const current = activeRow(database, storeKey, params.job.id); + if (current) { + if (observed && observedStale && sameOwner(current, observed)) { + executeSqliteQuerySync( + database, + query(database) + .updateTable("cron_run_receipts") + .set({ + status: "interrupted", + finished_at_ms: params.startedAtMs, + error_text: "cron: job interrupted by owner process exit", + }) + .where("receipt_id", "=", current.receipt_id) + .where("status", "=", "running"), + ); + } else { + throw new CronRunReceiptConflictError(receiptFromRow(current)); + } + } + validateCurrentJob({ database, handle, resolveAgentId: params.resolveAgentId }); + executeSqliteQuerySync( + database, + query(database) + .insertInto("cron_run_receipts") + .values({ + receipt_id: receiptId, + store_key: storeKey, + job_id: params.job.id, + config_revision: handle.configRevision, + agent_id: params.agentId, + request_run_id: params.requestRunId ?? null, + status: "running", + owner_pid: handle.ownerPid, + owner_start_time: handle.ownerStartTime, + started_at_ms: params.startedAtMs, + finished_at_ms: null, + error_text: null, + }), + ); + return receiptHandle(receiptFromRow(activeRow(database, storeKey, params.job.id)!)); + }); + locallyOwnedReceipts.add(claimed.receiptId); + return claimed; +} + +/** Synchronous transaction guard used immediately before a run side effect or state write. */ +export function assertCronRunReceiptCurrentInDatabase(params: { + database: DatabaseSync; + handle: CronRunReceiptHandle; + resolveAgentId: ResolveReceiptAgentId; +}): void { + const current = activeRow(params.database, params.handle.storeKey, params.handle.jobId); + if ( + !current || + current.receipt_id !== params.handle.receiptId || + current.owner_pid !== params.handle.ownerPid || + current.owner_start_time !== params.handle.ownerStartTime + ) { + throw new CronRunReceiptRevisionError( + params.handle.receiptId, + "cron run fence is no longer current", + ); + } + validateCurrentJob({ + database: params.database, + handle: params.handle, + resolveAgentId: params.resolveAgentId, + }); +} + +export function assertCronRunReceiptCurrent(params: { + handle: CronRunReceiptHandle; + resolveAgentId: ResolveReceiptAgentId; + isAgentAvailable?: (agentId: string) => boolean; + env?: NodeJS.ProcessEnv; +}): void { + if (params.isAgentAvailable && !params.isAgentAvailable(params.handle.agentId)) { + throw new CronRunReceiptRevisionError( + params.handle.receiptId, + `cron run owner ${params.handle.agentId} is no longer configured`, + ); + } + withReceiptWrite( + "cron.run-receipt.assert-current", + params.env ? { env: params.env } : {}, + (database) => + assertCronRunReceiptCurrentInDatabase({ + database, + handle: params.handle, + resolveAgentId: params.resolveAgentId, + }), + ); +} + +export function finishCronRunReceipt(params: { + handle: CronRunReceiptHandle; + status: Exclude; + finishedAtMs: number; + error?: string; + env?: NodeJS.ProcessEnv; +}): CronRunReceipt | undefined { + const finished = withReceiptWrite( + "cron.run-receipt.finish", + params.env ? { env: params.env } : {}, + (database) => finishCronRunReceiptInDatabase({ database, ...params }), + ); + locallyOwnedReceipts.delete(params.handle.receiptId); + return finished; +} + +/** Releases only this process's liveness proof after terminal persistence fails. */ +export function releaseLocalCronRunReceiptOwnership(handle: CronRunReceiptHandle): void { + locallyOwnedReceipts.delete(handle.receiptId); +} + +/** Completes the exact active receipt inside its caller's cron-state transaction. */ +export function finishCronRunReceiptInDatabase(params: { + database: DatabaseSync; + handle: CronRunReceiptHandle; + status: Exclude; + finishedAtMs: number; + error?: string; +}): CronRunReceipt | undefined { + executeSqliteQuerySync( + params.database, + query(params.database) + .updateTable("cron_run_receipts") + .set({ + status: params.status, + finished_at_ms: params.finishedAtMs, + error_text: params.error ?? null, + }) + .where("receipt_id", "=", params.handle.receiptId) + .where("status", "=", "running") + .where("owner_pid", "=", params.handle.ownerPid), + ); + const row = executeSqliteQueryTakeFirstSync( + params.database, + query(params.database) + .selectFrom("cron_run_receipts") + .selectAll() + .where("receipt_id", "=", params.handle.receiptId), + ); + return row ? receiptFromRow(row) : undefined; +} + +/** Returns a still-live foreign claim, or retires a provably stale owner. */ +export function reconcileCronRunReceiptForStartup(params: { + storePath: string; + jobId: string; + startedAtMs: number; + nowMs: number; + env?: NodeJS.ProcessEnv; +}): CronRunReceipt | undefined { + const storeKey = cronStoreKey(params.storePath); + const options = params.env ? { env: params.env } : {}; + const observed = withReceiptWrite("cron.run-receipt.startup-inspect", options, (database) => + activeRow(database, storeKey, params.jobId), + ); + if (!observed || observed.started_at_ms !== params.startedAtMs) { + return undefined; + } + const stale = ownerDefinitelyStale(observed); + return withReceiptWrite("cron.run-receipt.startup-reconcile", options, (database) => { + const current = activeRow(database, storeKey, params.jobId); + if (!current || !sameOwner(current, observed) || current.started_at_ms !== params.startedAtMs) { + return current ? receiptFromRow(current) : undefined; + } + if (!stale) { + return receiptFromRow(current); + } + executeSqliteQuerySync( + database, + query(database) + .updateTable("cron_run_receipts") + .set({ + status: "interrupted", + finished_at_ms: params.nowMs, + error_text: "cron: job interrupted by owner process exit", + }) + .where("receipt_id", "=", current.receipt_id) + .where("status", "=", "running"), + ); + return undefined; + }); +} + +/** Stable history order for operator and recovery reads. */ +export function listCronRunReceipts( + storePath: string, + jobId?: string, + env?: NodeJS.ProcessEnv, +): CronRunReceipt[] { + return withReceiptWrite("cron.run-receipt.list", env ? { env } : {}, (database) => { + let builder = query(database) + .selectFrom("cron_run_receipts") + .selectAll() + .where("store_key", "=", cronStoreKey(storePath)); + if (jobId) { + builder = builder.where("job_id", "=", jobId); + } + return executeSqliteQuerySync( + database, + builder.orderBy("started_at_ms", "desc").orderBy("receipt_id", "desc"), + ).rows.map(receiptFromRow); + }); +} diff --git a/src/state/openclaw-state-db-contract.ts b/src/state/openclaw-state-db-contract.ts index 0a13b3d34785..8488c319647c 100644 --- a/src/state/openclaw-state-db-contract.ts +++ b/src/state/openclaw-state-db-contract.ts @@ -25,6 +25,7 @@ export const FIRST_USE_STATE_INDEXES = [ // lazy ensures run; fold them into the next natural schema-version bump. export const LAZY_ADDITIVE_STATE_TABLES = [ ...FIRST_USE_STATE_TABLES, + "cron_run_receipts", "cron_store_epochs", "model_catalog_remote", "secret_store_entries", @@ -41,6 +42,8 @@ export const LAZY_ADDITIVE_STATE_TABLES = [ ] as const; export const LAZY_ADDITIVE_STATE_INDEXES = [ ...FIRST_USE_STATE_INDEXES, + "idx_cron_run_receipts_active_job", + "idx_cron_run_receipts_job_history", "secret_store_entries_live_idx", ] as const; /** Maximum time one synchronous SQLite call may wait for a lock. */ diff --git a/src/state/openclaw-state-db.generated.d.ts b/src/state/openclaw-state-db.generated.d.ts index 3e7fd2d9e236..5e85ec1bbb90 100644 --- a/src/state/openclaw-state-db.generated.d.ts +++ b/src/state/openclaw-state-db.generated.d.ts @@ -478,6 +478,21 @@ export interface CronJobs { wake_mode: string; } +export interface CronRunReceipts { + agent_id: string; + config_revision: string; + error_text: string | null; + finished_at_ms: number | null; + job_id: string; + owner_pid: number; + owner_start_time: number | null; + receipt_id: string; + request_run_id: string | null; + started_at_ms: number; + status: string; + store_key: string; +} + export interface CronStoreEpochs { store_epoch: Generated; store_key: string; @@ -1708,6 +1723,7 @@ export interface DB { cron_job_runtime_authorities: CronJobRuntimeAuthorities; cron_job_scratch: CronJobScratch; cron_jobs: CronJobs; + cron_run_receipts: CronRunReceipts; cron_store_epochs: CronStoreEpochs; current_conversation_bindings: CurrentConversationBindings; delivery_queue_entries: DeliveryQueueEntries; diff --git a/src/state/openclaw-state-schema-compatibility.test.ts b/src/state/openclaw-state-schema-compatibility.test.ts new file mode 100644 index 000000000000..2b3194afa8a0 --- /dev/null +++ b/src/state/openclaw-state-schema-compatibility.test.ts @@ -0,0 +1,12 @@ +import { describe, expect, it } from "vitest"; +import { getOpenClawStateRuntimeSchema } from "./openclaw-state-schema-compatibility.js"; + +describe("OpenClaw state runtime schema projection", () => { + it("omits lazy additive tables and their unique indexes before first use", () => { + const schema = getOpenClawStateRuntimeSchema({ includeVersionLazyAdditiveTables: false }); + + expect(schema).not.toContain("CREATE TABLE IF NOT EXISTS cron_run_receipts"); + expect(schema).not.toContain("idx_cron_run_receipts_active_job"); + expect(schema).not.toContain("idx_cron_run_receipts_job_history"); + }); +}); diff --git a/src/state/openclaw-state-schema-compatibility.ts b/src/state/openclaw-state-schema-compatibility.ts index 02c383b406da..c2b3965f1474 100644 --- a/src/state/openclaw-state-schema-compatibility.ts +++ b/src/state/openclaw-state-schema-compatibility.ts @@ -68,7 +68,9 @@ export function getOpenClawStateRuntimeSchema(options: { schema = `${schema.slice(0, start)}${schema.slice(end + endMarker.length)}`; } for (const indexName of omittedIndexes) { - const start = schema.indexOf(`CREATE INDEX IF NOT EXISTS ${indexName}`); + const plainStart = schema.indexOf(`CREATE INDEX IF NOT EXISTS ${indexName}`); + const uniqueStart = schema.indexOf(`CREATE UNIQUE INDEX IF NOT EXISTS ${indexName}`); + const start = plainStart >= 0 ? plainStart : uniqueStart; const end = start >= 0 ? schema.indexOf(";", start) : -1; if (start < 0 || end < 0) { throw new Error(`lazy additive state schema index is missing for ${indexName}`); diff --git a/src/state/openclaw-state-schema.sql b/src/state/openclaw-state-schema.sql index 8e9dcd7c6438..46b4fd330fdb 100644 --- a/src/state/openclaw-state-schema.sql +++ b/src/state/openclaw-state-schema.sql @@ -1496,6 +1496,36 @@ CREATE INDEX IF NOT EXISTS idx_cron_jobs_agent_session ON cron_jobs(agent_id, session_key, updated_at DESC, job_id) WHERE agent_id IS NOT NULL OR session_key IS NOT NULL; +-- One owner-native receipt is also the durable execution fence. Receipts +-- survive job deletion so operators can distinguish a run from log inference. +CREATE TABLE IF NOT EXISTS cron_run_receipts ( + receipt_id TEXT PRIMARY KEY, + store_key TEXT NOT NULL, + job_id TEXT NOT NULL, + config_revision TEXT NOT NULL, + agent_id TEXT NOT NULL, + request_run_id TEXT, + status TEXT NOT NULL, + owner_pid INTEGER NOT NULL, + owner_start_time INTEGER, + started_at_ms INTEGER NOT NULL, + finished_at_ms INTEGER, + error_text TEXT, + CHECK (status IN ('running', 'ok', 'error', 'skipped', 'interrupted', 'superseded')), + CHECK ( + (status = 'running' AND finished_at_ms IS NULL) + OR + (status != 'running' AND finished_at_ms IS NOT NULL) + ) +) STRICT; + +CREATE UNIQUE INDEX IF NOT EXISTS idx_cron_run_receipts_active_job + ON cron_run_receipts(store_key, job_id) + WHERE status = 'running'; + +CREATE INDEX IF NOT EXISTS idx_cron_run_receipts_job_history + ON cron_run_receipts(store_key, job_id, started_at_ms DESC, receipt_id DESC); + -- Runtime-private authority is independent of job_json so downgraded writers -- can rewrite recognized job config without erasing or silently widening it. CREATE TABLE IF NOT EXISTS cron_job_runtime_authorities (