diff --git a/src/cron/service/owner-hardening.test.ts b/src/cron/service/owner-hardening.test.ts index ebf64216ee58..34d1ea155f54 100644 --- a/src/cron/service/owner-hardening.test.ts +++ b/src/cron/service/owner-hardening.test.ts @@ -10,7 +10,8 @@ import { resolveOpenClawStateDirForDatabasePath } from "../../state/openclaw-sta 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 { cronStoreKey } from "../store/key.js"; +import { reconcileCronRunReceiptForStartup } from "../store/run-receipt-store.js"; import type { CronJob } from "../types.js"; const { makeStorePath } = createCronStoreHarness({ prefix: "cron-owner-hardening-" }); @@ -155,6 +156,21 @@ function makeParentService(storePath: string, runCommandJob = vi.fn()) { }); } +function receipts(storePath: string, jobId: string) { + return openOpenClawStateDatabase() + .db.prepare( + `SELECT receipt_id AS receiptId, status, agent_id AS agentId + FROM cron_run_receipts + WHERE store_key = ? AND job_id = ? + ORDER BY started_at_ms DESC, receipt_id DESC`, + ) + .all(cronStoreKey(storePath), jobId) as Array<{ + receiptId: string; + status: string; + agentId: string; + }>; +} + describe("cron durable run ownership", () => { it("does not execute when the durable receipt cannot be recorded", async () => { vi.useRealTimers(); @@ -162,7 +178,12 @@ describe("cron durable run ownership", () => { const now = Date.now(); const job = makeCommandJob("receipt-required", now + 60_000); await saveCronStore(storePath, { version: 1, jobs: [job] }); - listCronRunReceipts(storePath, job.id); + reconcileCronRunReceiptForStartup({ + storePath, + jobId: job.id, + startedAtMs: 0, + nowMs: now, + }); const database = openOpenClawStateDatabase().db; database.exec(` CREATE TRIGGER reject_cron_run_receipt @@ -202,7 +223,7 @@ describe("cron durable run ownership", () => { reason: "already-running", }); expect(replacementRunner).not.toHaveBeenCalled(); - expect(listCronRunReceipts(storePath, job.id)).toMatchObject([{ status: "running" }]); + expect(receipts(storePath, job.id)).toMatchObject([{ status: "running" }]); replacement.stop(); owner.kill("SIGKILL"); @@ -211,7 +232,7 @@ describe("cron durable run ownership", () => { await recovered.start(); recovered.stop(); - expect(listCronRunReceipts(storePath, job.id)[0]).toMatchObject({ status: "interrupted" }); + expect(receipts(storePath, job.id)[0]).toMatchObject({ status: "interrupted" }); expect((await loadCronStore(storePath)).jobs[0]?.state.lastError).toContain( "interrupted by gateway restart", ); @@ -233,7 +254,7 @@ describe("cron durable run ownership", () => { ? fs.readFileSync(outputPath, "utf8").trim().split("\n").filter(Boolean) : []; expect(invocations).toHaveLength(1); - expect(listCronRunReceipts(storePath, job.id)).toMatchObject([{ status: "ok" }]); + expect(receipts(storePath, job.id)).toMatchObject([{ status: "ok" }]); }); it("supersedes a live run before payload effects after its owner changes", async () => { @@ -260,7 +281,7 @@ describe("cron durable run ownership", () => { await waitForExit(owner); expect(fs.existsSync(outputPath)).toBe(false); - expect(listCronRunReceipts(storePath, job.id)[0]).toMatchObject({ + expect(receipts(storePath, job.id)[0]).toMatchObject({ agentId: "alpha", status: "superseded", }); diff --git a/src/cron/service/run-receipts.ts b/src/cron/service/run-receipts.ts index fa9cfe34450c..1be74562537c 100644 --- a/src/cron/service/run-receipts.ts +++ b/src/cron/service/run-receipts.ts @@ -17,7 +17,7 @@ function currentDefaultAgentId(state: CronServiceState): string | undefined { return state.deps.resolveDefaultAgentId?.() ?? state.deps.defaultAgentId; } -export function resolveCronRunReceiptAgentId(state: CronServiceState, job: CronJob): string { +function resolveCronRunReceiptAgentId(state: CronServiceState, job: CronJob): string { return resolveEffectiveJobAgentId(job, currentDefaultAgentId(state)); } diff --git a/src/cron/store/run-receipt-store.test.ts b/src/cron/store/run-receipt-store.test.ts index a6ca681276c8..606d30100225 100644 --- a/src/cron/store/run-receipt-store.test.ts +++ b/src/cron/store/run-receipt-store.test.ts @@ -3,13 +3,13 @@ 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 { cronStoreKey } from "./key.js"; import { assertCronRunReceiptCurrent, claimCronRunReceipt, CronRunReceiptConflictError, CronRunReceiptRevisionError, finishCronRunReceipt, - listCronRunReceipts, } from "./run-receipt-store.js"; const { makeStorePath } = setupCronServiceSuite({ prefix: "cron-run-receipt-" }); @@ -40,6 +40,24 @@ function claim(storePath: string, job: CronJob, startedAtMs: number) { }); } +function receipts(storePath: string, jobId: string) { + return openOpenClawStateDatabase() + .db.prepare( + `SELECT receipt_id AS receiptId, status, agent_id AS agentId, + started_at_ms AS startedAtMs, error_text AS error + FROM cron_run_receipts + WHERE store_key = ? AND job_id = ? + ORDER BY started_at_ms DESC, receipt_id DESC`, + ) + .all(cronStoreKey(storePath), jobId) as Array<{ + receiptId: string; + status: string; + agentId: string; + startedAtMs: number; + error: string | null; + }>; +} + describe("cron run receipt store", () => { it("records one durable active run and rejects an overlapping claimant", async () => { const { storePath } = await makeStorePath(); @@ -49,7 +67,7 @@ describe("cron run receipt store", () => { const first = claim(storePath, job, 100); expect(() => claim(storePath, job, 101)).toThrow(CronRunReceiptConflictError); - expect(listCronRunReceipts(storePath, job.id)).toMatchObject([ + expect(receipts(storePath, job.id)).toMatchObject([ { receiptId: first.receiptId, status: "running", startedAtMs: 100 }, ]); @@ -57,10 +75,7 @@ describe("cron run receipt store", () => { 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", - ]); + expect(receipts(storePath, job.id).map((receipt) => receipt.status)).toEqual(["skipped", "ok"]); }); it("retires a provably dead process claim before admitting its successor", async () => { @@ -75,7 +90,7 @@ describe("cron run receipt store", () => { const replacement = claim(storePath, job, 220); expect(replacement.receiptId).not.toBe(abandoned.receiptId); - expect(listCronRunReceipts(storePath, job.id)).toMatchObject([ + expect(receipts(storePath, job.id)).toMatchObject([ { receiptId: replacement.receiptId, status: "running" }, { receiptId: abandoned.receiptId, status: "interrupted" }, ]); @@ -102,7 +117,7 @@ describe("cron run receipt store", () => { finishedAtMs: 310, error: "owner changed", }); - expect(listCronRunReceipts(storePath, admitted.id)[0]).toMatchObject({ + expect(receipts(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 index 796f7d5ae362..3ee2c37cbde4 100644 --- a/src/cron/store/run-receipt-store.ts +++ b/src/cron/store/run-receipt-store.ts @@ -29,7 +29,7 @@ export type CronRunReceiptStatus = | "interrupted" | "superseded"; -export type CronRunReceipt = { +type CronRunReceipt = { receiptId: string; storeKey: string; jobId: string; @@ -440,24 +440,3 @@ export function reconcileCronRunReceiptForStartup(params: { 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); - }); -}