test(cron): inspect receipts through SQLite

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",
     });

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));
 }
 

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",

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);
-  });
-}