fix(cron): bind recovery to exact run receipts

Amp-Thread-ID: https://ampcode.com/threads/T-01a0220e-6236-745d-8b14-759d26595b12

src/cron/service/ops.test.ts

@@ -1060,20 +1060,21 @@ describe("cron service ops seam coverage", () => {
         job.payload = { kind: "script", script: "return { state: { cursor: 'payload' } }" };
         job.state.triggerState = { cursor: "old" };
         await writeCronStoreSnapshot({ storePath, jobs: [job] });
-        const preparedReceipt = runReceiptStore.prepareCronRunReceiptClaim({
-          storePath,
-          job,
-          agentId: "main",
-          startedAtMs: startedAt,
-        });
-        const receipt = runOpenClawStateWriteTransaction(({ db }) =>
-          runReceiptStore.claimCronRunReceiptInDatabase({
-            database: db,
-            prepared: preparedReceipt,
-            resolveAgentId: (current) => current.agentId ?? "main",
-          }),
-        );
-        runReceiptStore.releaseLocalCronRunReceiptOwnership(receipt);
+        const receipt =
+          reservationOffsetMs === undefined
+            ? runOpenClawStateWriteTransaction(({ db }) =>
+                runReceiptStore.claimCronRunReceiptInDatabase({
+                  database: db,
+                  prepared: runReceiptStore.prepareCronRunReceiptClaim({
+                    storePath,
+                    job,
+                    agentId: "main",
+                    startedAtMs: startedAt,
+                  }),
+                  resolveAgentId: (current) => current.agentId ?? "main",
+                }),
+              )
+            : undefined;
         const events: CronEvent[] = [];
         const state = createCronServiceState({
           storePath,
@@ -1085,23 +1086,27 @@ describe("cron service ops seam coverage", () => {
           runIsolatedAgentJob: vi.fn(async () => ({ status: "ok" as const })),
           onEvent: (event) => events.push(structuredClone(event)),
         });
-        const taskRunId =
-          reservationOffsetMs === undefined
-            ? taskRuns.tryCreateCronTaskRun({ state, job, startedAt })
-            : taskExecutor.createRunningTaskRunCore({
-                runtime: "cron",
-                sourceId: job.id,
-                ownerKey: "",
-                scopeKind: "system",
-                runId: `${createCronExecutionId(job.id, startedAt - reservationOffsetMs)}:legacy-upgrade`,
-                agentId: "main",
-                task: job.name,
-                deliveryStatus: "not_applicable",
-                notifyPolicy: "silent",
-                startedAt,
-                lastEventAt: startedAt,
-                detail: { storeKey: cronStoreKey(storePath) },
-              })?.runId;
+        const taskRunId = receipt
+          ? taskRuns.tryCreateCronTaskRun({
+              state,
+              job,
+              startedAt,
+              publicRunId: receipt.receiptId,
+            })
+          : taskExecutor.createRunningTaskRunCore({
+              runtime: "cron",
+              sourceId: job.id,
+              ownerKey: "",
+              scopeKind: "system",
+              runId: `${createCronExecutionId(job.id, startedAt - (reservationOffsetMs ?? 0))}:legacy-upgrade`,
+              agentId: "main",
+              task: job.name,
+              deliveryStatus: "not_applicable",
+              notifyPolicy: "silent",
+              startedAt,
+              lastEventAt: startedAt,
+              detail: { storeKey: cronStoreKey(storePath) },
+            })?.runId;
         if (!taskRunId) {
           throw new Error("expected reserved cron task run");
         }
@@ -1125,6 +1130,9 @@ describe("cron service ops seam coverage", () => {
             triggerFired: true,
           },
         });
+        if (receipt) {
+          runReceiptStore.releaseLocalCronRunReceiptOwnership(receipt);
+        }
 
         await start(state);
 
@@ -1159,14 +1167,16 @@ describe("cron service ops seam coverage", () => {
         expect(persisted.jobs[0]?.state.runningAtMs).toBeUndefined();
         expect(persisted.jobs[0]?.state.lastError).toBeUndefined();
         expect(persisted.jobs[0]?.state.nextRunAtMs).toBeUndefined();
-        const receiptRow = runOpenClawStateWriteTransaction(({ db }) =>
-          db
-            .prepare(
-              "SELECT status, finished_at_ms AS finishedAtMs, error_text AS error FROM cron_run_receipts WHERE receipt_id = ?",
-            )
-            .get(receipt.receiptId),
-        ) as { status: string; finishedAtMs: number; error: string | null };
-        expect(receiptRow).toEqual({ status: "ok", finishedAtMs: endedAt, error: null });
+        if (receipt) {
+          const receiptRow = runOpenClawStateWriteTransaction(({ db }) =>
+            db
+              .prepare(
+                "SELECT status, finished_at_ms AS finishedAtMs, error_text AS error FROM cron_run_receipts WHERE receipt_id = ?",
+              )
+              .get(receipt.receiptId),
+          ) as { status: string; finishedAtMs: number; error: string | null };
+          expect(receiptRow).toEqual({ status: "ok", finishedAtMs: endedAt, error: null });
+        }
         expect(events.filter((event) => event.action === "finished")).toEqual([]);
         stop(state);
       });
@@ -1195,7 +1205,6 @@ describe("cron service ops seam coverage", () => {
           resolveAgentId: (current) => current.agentId ?? "main",
         }),
       );
-      runReceiptStore.releaseLocalCronRunReceiptOwnership(receipt);
       const state = createCronServiceState({
         storePath,
         cronEnabled: true,
@@ -1205,7 +1214,12 @@ describe("cron service ops seam coverage", () => {
         requestHeartbeat: vi.fn(),
         runIsolatedAgentJob: vi.fn(async () => ({ status: "ok" as const })),
       });
-      const taskRunId = taskRuns.tryCreateCronTaskRun({ state, job, startedAt });
+      const taskRunId = taskRuns.tryCreateCronTaskRun({
+        state,
+        job,
+        startedAt,
+        publicRunId: receipt.receiptId,
+      });
       if (!taskRunId) {
         throw new Error("expected invalid finalized cron task run");
       }
@@ -1225,6 +1239,7 @@ describe("cron service ops seam coverage", () => {
       runOpenClawStateWriteTransaction(({ db }) => {
         db.prepare("UPDATE task_runs SET ended_at = -1 WHERE run_id = ?").run(taskRunId);
       });
+      runReceiptStore.releaseLocalCronRunReceiptOwnership(receipt);
 
       await start(state);
 
@@ -1573,13 +1588,16 @@ describe("cron service ops seam coverage", () => {
         ran: true,
       });
 
-      expectTaskRun({
-        runId: `cron:isolated-timeout:${now}:${manualRunId}`,
+      const task = findCronTaskByBaseRunId(`cron:isolated-timeout:${now}`);
+      expect(task).toMatchObject({
         runtime: "cron",
         status: "succeeded",
         sourceId: "isolated-timeout",
         progressSummary: "Running automation.",
       });
+      expect(task?.runId).toMatch(
+        new RegExp(`^cron:isolated-timeout:${now}:[^:]+:${manualRunId}$`),
+      );
       expect(findTaskByRunId(manualRunId)).toBeUndefined();
     });
   });
@@ -1753,6 +1771,56 @@ describe("cron service ops seam coverage", () => {
     createTaskRecordSpy.mockRestore();
   });
 
+  it("keeps terminal fallback bound to its receipt before a same-time successor", async () => {
+    const { storePath } = await makeStorePath();
+    const now = Date.parse("2026-03-23T12:00:00.000Z");
+
+    await withStateDirForStorePath(storePath, async () => {
+      await writeDueIsolatedJobSnapshot(storePath, now);
+      const state = createOkIsolatedCronState({ storePath, now, summary: "done" });
+      const createTaskRecord = taskExecutor.createRunningTaskRunCore;
+      const createTaskRecordSpy = vi
+        .spyOn(taskExecutor, "createRunningTaskRunCore")
+        .mockImplementationOnce(() => {
+          throw new Error("transient ledger failure");
+        })
+        .mockImplementation((params) => createTaskRecord(params));
+
+      try {
+        await expect(run(state, "isolated-timeout", "force")).resolves.toEqual({
+          ok: true,
+          ran: true,
+        });
+        await expect(run(state, "isolated-timeout", "force")).resolves.toEqual({
+          ok: true,
+          ran: true,
+        });
+
+        const receiptIds = openOpenClawStateDatabase()
+          .db.prepare(
+            `SELECT receipt_id AS receiptId
+               FROM cron_run_receipts
+              WHERE store_key = ? AND job_id = ?
+              ORDER BY receipt_id`,
+          )
+          .all(cronStoreKey(storePath), "isolated-timeout")
+          .map((row) => (row as { receiptId: string }).receiptId);
+        const taskRunIds = listTaskRecordsUnsorted()
+          .filter((task) => task.runtime === "cron" && task.sourceId === "isolated-timeout")
+          .map((task) => task.runId);
+
+        expect(receiptIds).toHaveLength(2);
+        expect(taskRunIds).toHaveLength(2);
+        for (const receiptId of receiptIds) {
+          expect(taskRunIds).toContain(`cron:isolated-timeout:${now}:${receiptId}`);
+        }
+      } finally {
+        createTaskRecordSpy.mockRestore();
+        stop(state);
+      }
+    });
+  });
+
   it("keeps manual cron cleanup progressing when task ledger updates fail", async () => {
     const { storePath } = await makeStorePath();
     const now = Date.parse("2026-03-23T12:00:00.000Z");

src/cron/service/run-recovery.test.ts

@@ -1,7 +1,10 @@
 import { describe, expect, it, vi } from "vitest";
 import { runOpenClawStateWriteTransaction } from "../../state/openclaw-state-db.js";
+import { createRunningTaskRunCore } from "../../tasks/task-executor.js";
+import { createCronExecutionId } from "../run-id.js";
 import { setupCronServiceSuite, writeCronStoreSnapshot } from "../service.test-harness.js";
 import { loadCronStore } from "../store.js";
+import { cronStoreKey } from "../store/key.js";
 import {
   claimCronRunReceiptInDatabase,
   finishCronRunReceipt,
@@ -15,7 +18,11 @@ import type { CronJob } from "../types.js";
 import { proposeCronRunRecovery, recoverCronRunProposal } from "./run-recovery.js";
 import { createCronServiceState } from "./state.js";
 import { runPostPersistCronNotifications } from "./store.js";
-import { tryCreateCronTaskRun, tryFinishCronTaskRunWithoutHistory } from "./task-runs.js";
+import {
+  tryCreateCronTaskRun,
+  tryFinishCronTaskRun,
+  tryFinishCronTaskRunWithoutHistory,
+} from "./task-runs.js";
 
 const { logger, makeStorePath } = setupCronServiceSuite({ prefix: "cron-run-recovery-" });
 
@@ -71,6 +78,32 @@ function claimReceipt(storePath: string, job: CronJob, startedAtMs: number) {
   );
 }
 
+function createLegacyCronTaskRun(params: {
+  storePath: string;
+  job: CronJob;
+  startedAtMs: number;
+  runId: string;
+}): string {
+  const task = createRunningTaskRunCore({
+    runtime: "cron",
+    sourceId: params.job.id,
+    ownerKey: "",
+    scopeKind: "system",
+    runId: params.runId,
+    agentId: params.job.agentId,
+    task: params.job.name,
+    deliveryStatus: "not_applicable",
+    notifyPolicy: "silent",
+    startedAt: params.startedAtMs,
+    lastEventAt: params.startedAtMs,
+    detail: { storeKey: cronStoreKey(params.storePath) },
+  });
+  if (!task) {
+    throw new Error("expected legacy cron task run");
+  }
+  return params.runId;
+}
+
 async function commitCompletedJob(params: {
   storePath: string;
   jobs: CronJob[];
@@ -243,20 +276,32 @@ describe("atomic cron run recovery", () => {
     expect(sendCronFailureAlert).not.toHaveBeenCalled();
   });
 
-  it("restores a finalized quiet trigger with a skipped receipt", async () => {
+  it("restores a pre-upgrade receipt-keyed task over a prior legacy manual task", async () => {
     const { storePath } = await makeStorePath();
     const startedAtMs = Date.parse("2026-08-13T10:45:00.000Z");
     const job = makeJob("quiet-trigger-recovery", startedAtMs);
     job.trigger = { script: "return false" };
     await writeCronStoreSnapshot({ storePath, jobs: [job] });
     const state = makeState(storePath, startedAtMs + 30_000);
+    const priorReceipt = claimReceipt(storePath, job, startedAtMs);
+    createLegacyCronTaskRun({
+      storePath,
+      job,
+      startedAtMs,
+      runId: `${createCronExecutionId(job.id, startedAtMs)}:prior-manual-run`,
+    });
+    finishCronRunReceipt({
+      handle: priorReceipt,
+      status: "interrupted",
+      finishedAtMs: startedAtMs + 1,
+    });
     const receipt = claimReceipt(storePath, job, startedAtMs);
     const proposal = proposeCronRunRecovery(state, job.id, undefined, startedAtMs);
-    const taskRunId = tryCreateCronTaskRun({
-      state,
+    const taskRunId = createLegacyCronTaskRun({
+      storePath,
       job,
-      startedAt: startedAtMs,
-      publicRunId: receipt.receiptId,
+      startedAtMs,
+      runId: `${createCronExecutionId(job.id, startedAtMs)}:${receipt.receiptId}`,
     });
     tryFinishCronTaskRunWithoutHistory(state, {
       taskRunId,
@@ -277,6 +322,165 @@ describe("atomic cron run recovery", () => {
         .get(receipt.receiptId),
     ) as { status: string };
     expect(receiptRow.status).toBe("skipped");
+    const requestRow = runOpenClawStateWriteTransaction(({ db }) =>
+      db
+        .prepare(
+          "SELECT request_run_id AS requestRunId FROM cron_run_receipts WHERE receipt_id = ?",
+        )
+        .get(receipt.receiptId),
+    ) as { requestRunId: string | null };
+    expect(requestRow.requestRunId).toBeNull();
+  });
+
+  it("does not restore a prior same-millisecond task for a different receipt", async () => {
+    const { storePath } = await makeStorePath();
+    const startedAtMs = Date.parse("2026-08-13T10:50:00.000Z");
+    const job = makeJob("same-millisecond-task-recovery", startedAtMs);
+    await writeCronStoreSnapshot({ storePath, jobs: [job] });
+    const state = makeState(storePath, startedAtMs + 30_000);
+    const priorReceipt = claimReceipt(storePath, job, startedAtMs);
+    const priorTaskRunId = tryCreateCronTaskRun({
+      state,
+      job,
+      startedAt: startedAtMs,
+      publicRunId: priorReceipt.receiptId,
+    });
+    if (!priorTaskRunId) {
+      throw new Error("expected prior cron task run");
+    }
+    tryFinishCronTaskRun(state, {
+      taskRunId: priorTaskRunId,
+      job,
+      event: {
+        jobId: job.id,
+        action: "finished",
+        job,
+        status: "ok",
+        summary: "prior receipt completed",
+        runAtMs: startedAtMs,
+        durationMs: 1,
+      },
+    });
+    finishCronRunReceipt({
+      handle: priorReceipt,
+      status: "ok",
+      finishedAtMs: startedAtMs + 1,
+    });
+    const receipt = claimReceipt(storePath, job, startedAtMs);
+    const proposal = proposeCronRunRecovery(state, job.id, undefined, startedAtMs);
+    releaseLocalCronRunReceiptOwnership(receipt);
+
+    expect(recoverCronRunProposal(state, proposal)).toMatchObject({ kind: "repaired" });
+    expect((await loadCronStore(storePath)).jobs[0]?.state).toMatchObject({
+      lastRunStatus: "error",
+      lastError: expect.stringContaining("interrupted by gateway restart"),
+    });
+    const receiptRow = runOpenClawStateWriteTransaction(({ db }) =>
+      db
+        .prepare("SELECT status FROM cron_run_receipts WHERE receipt_id = ?")
+        .get(receipt.receiptId),
+    ) as { status: string };
+    expect(receiptRow.status).toBe("interrupted");
+  });
+
+  it("does not restore a sole legacy caller-ID task after its prior receipt was pruned", async () => {
+    const { storePath } = await makeStorePath();
+    const startedAtMs = Date.parse("2026-08-13T10:52:00.000Z");
+    const job = makeJob("pruned-manual-task-recovery", startedAtMs);
+    await writeCronStoreSnapshot({ storePath, jobs: [job] });
+    const state = makeState(storePath, startedAtMs + 30_000);
+    const priorReceipt = claimReceipt(storePath, job, startedAtMs);
+    const priorTaskRunId = createLegacyCronTaskRun({
+      storePath,
+      job,
+      startedAtMs,
+      runId: `${createCronExecutionId(job.id, startedAtMs)}:prior-manual-run`,
+    });
+    tryFinishCronTaskRun(state, {
+      taskRunId: priorTaskRunId,
+      job,
+      event: {
+        jobId: job.id,
+        action: "finished",
+        job,
+        status: "ok",
+        runId: "prior-manual-run",
+        runAtMs: startedAtMs,
+        durationMs: 1,
+      },
+    });
+    finishCronRunReceipt({
+      handle: priorReceipt,
+      status: "ok",
+      finishedAtMs: startedAtMs + 1,
+    });
+    // Receipt retention is shorter than task history retention.
+    runOpenClawStateWriteTransaction(({ db }) => {
+      db.prepare("DELETE FROM cron_run_receipts WHERE receipt_id = ?").run(priorReceipt.receiptId);
+    });
+    const receipt = claimReceipt(storePath, job, startedAtMs);
+    const proposal = proposeCronRunRecovery(state, job.id, undefined, startedAtMs);
+    // This final shape (one active receipt plus one same-ms legacy task) is
+    // indistinguishable from a pre-upgrade caller-ID task for the active run.
+    releaseLocalCronRunReceiptOwnership(receipt);
+
+    expect(recoverCronRunProposal(state, proposal)).toMatchObject({ kind: "repaired" });
+    expect((await loadCronStore(storePath)).jobs[0]?.state).toMatchObject({
+      lastRunStatus: "error",
+      lastError: expect.stringContaining("interrupted by gateway restart"),
+    });
+    const receiptRow = runOpenClawStateWriteTransaction(({ db }) =>
+      db
+        .prepare("SELECT status FROM cron_run_receipts WHERE receipt_id = ?")
+        .get(receipt.receiptId),
+    ) as { status: string };
+    expect(receiptRow.status).toBe("interrupted");
+  });
+
+  it.each([
+    {
+      identity: "pre-upgrade caller-supplied manual",
+      reservationOffsetMs: 0,
+      discriminator: "legacy-manual",
+    },
+    { identity: "reservation-keyed", reservationOffsetMs: 250, discriminator: "legacy-upgrade" },
+  ])("fails closed for a $identity task without exact receipt identity", async (testCase) => {
+    const { storePath } = await makeStorePath();
+    const startedAtMs = Date.parse("2026-08-13T10:54:00.000Z");
+    const job = makeJob("reservation-task-recovery", startedAtMs);
+    await writeCronStoreSnapshot({ storePath, jobs: [job] });
+    const state = makeState(storePath, startedAtMs + 30_000);
+    const receipt = claimReceipt(storePath, job, startedAtMs);
+    const proposal = proposeCronRunRecovery(state, job.id, undefined, startedAtMs);
+    const taskRunId = `${createCronExecutionId(job.id, startedAtMs - testCase.reservationOffsetMs)}:${testCase.discriminator}`;
+    createLegacyCronTaskRun({ storePath, job, startedAtMs, runId: taskRunId });
+    tryFinishCronTaskRun(state, {
+      taskRunId,
+      job,
+      event: {
+        jobId: job.id,
+        action: "finished",
+        job,
+        status: "ok",
+        runAtMs: startedAtMs,
+        durationMs: 1,
+      },
+    });
+    releaseLocalCronRunReceiptOwnership(receipt);
+
+    expect(recoverCronRunProposal(state, proposal)).toMatchObject({ kind: "repaired" });
+    expect((await loadCronStore(storePath)).jobs[0]?.state).toMatchObject({
+      lastRunStatus: "error",
+      lastError: expect.stringContaining("interrupted by gateway restart"),
+    });
+    const receiptRow = runOpenClawStateWriteTransaction(({ db }) =>
+      db
+        .prepare(
+          "SELECT status, request_run_id AS requestRunId FROM cron_run_receipts WHERE receipt_id = ?",
+        )
+        .get(receipt.receiptId),
+    ) as { status: string; requestRunId: string | null };
+    expect(receiptRow).toEqual({ status: "interrupted", requestRunId: null });
   });
 
   it("retires a dead owner receipt after timeout state already finalized", async () => {

src/cron/service/run-recovery.ts

@@ -143,6 +143,7 @@ function repairInDatabase(params: {
       jobId: proposal.jobId,
       startedAt: proposal.runningAtMs,
       storeKey,
+      ...(proposal.receipt ? { receiptId: proposal.receipt.receiptId } : {}),
     });
     const finalized = task.finalized;
     const restored = finalized

src/cron/service/task-runs.ts

@@ -20,6 +20,7 @@ import {
 import { createCronExecutionId } from "../run-id.js";
 import type { CronRunLogEntry } from "../run-log-types.js";
 import { cronStoreKey } from "../store/key.js";
+import { bindCronRunReceiptTaskIdentity } from "../store/run-receipt-store.js";
 import {
   cronRunLogEntryToTaskDetail,
   cronRunStatusToTaskStatus,
@@ -95,19 +96,52 @@ export function tryCreateCronTaskRun(params: {
   startedAt: number;
   publicRunId?: string;
 }): string | undefined {
-  const runId = createCronTaskRunId(params.job.id, params.startedAt, params.publicRunId);
-  return tryCreateCronTaskRunRecord({
+  let runId: string;
+  try {
+    runId = resolveCronTaskRunId({
+      state: params.state,
+      jobId: params.job.id,
+      startedAt: params.startedAt,
+      publicRunId: params.publicRunId,
+    });
+  } catch (error) {
+    params.state.deps.log.warn(
+      { jobId: params.job.id, error },
+      "cron: failed to bind task ledger record to its run receipt",
+    );
+    return undefined;
+  }
+  tryCreateCronTaskRunRecord({
     state: params.state,
     job: params.job,
     jobId: params.job.id,
     startedAt: params.startedAt,
     runId,
   });
+  return runId;
 }
 
-function createCronTaskRunId(jobId: string, startedAt: number, publicRunId?: string): string {
-  const discriminator = publicRunId?.trim() || randomUUID();
-  return `${createCronExecutionId(jobId, startedAt)}:${discriminator}`;
+function resolveCronTaskRunId(params: {
+  state: CronServiceState;
+  jobId: string;
+  startedAt: number;
+  publicRunId?: string;
+}): string {
+  const publicRunId = params.publicRunId?.trim() || undefined;
+  const receiptId = bindCronRunReceiptTaskIdentity({
+    storePath: params.state.deps.storePath,
+    jobId: params.jobId,
+    startedAtMs: params.startedAt,
+    requestRunId: publicRunId,
+  });
+  const executionRunId = createCronExecutionId(params.jobId, params.startedAt);
+  if (receiptId) {
+    const receiptRunId = `${executionRunId}:${receiptId}`;
+    return publicRunId && publicRunId !== receiptId
+      ? `${receiptRunId}:${publicRunId}`
+      : receiptRunId;
+  }
+  return `${executionRunId}:${publicRunId ?? randomUUID()}`;
 }
 
 function findLatestCronTaskRunForRecoveryFromRecords(
@@ -115,34 +149,51 @@ function findLatestCronTaskRunForRecoveryFromRecords(
   jobId: string,
   startedAt: number,
   storeKey: string,
+  receiptId?: string,
 ): TaskRecord | undefined {
   const executionRunId = createCronExecutionId(jobId, startedAt);
   const prefix = `${executionRunId}:`;
+  const receiptRunId = receiptId ? `${prefix}${receiptId}` : undefined;
   return records
     .filter((task) => {
       if (task.runtime !== "cron" || task.sourceId !== jobId) {
         return false;
       }
       const taskStoreKey = cronTaskRecordStoreKey(task);
+      if (receiptRunId) {
+        // Sequential receipts can share the timestamp and public id, so only
+        // the receipt-prefixed task identity can own receipt-based recovery.
+        return (
+          taskStoreKey === storeKey &&
+          (task.runId === receiptRunId || task.runId?.startsWith(`${receiptRunId}:`))
+        );
+      }
       if (taskStoreKey === undefined) {
         // Exact match covers detail-less pre-discriminator rows from older releases.
         return task.runId === executionRunId;
       }
-      return (
-        taskStoreKey === storeKey &&
-        (task.runId === executionRunId ||
-          task.runId?.startsWith(prefix) ||
-          // Released reservation-keyed rows still record the authoritative execution start.
-          task.startedAt === startedAt)
-      );
+      if (taskStoreKey !== storeKey) {
+        return false;
+      }
+      if (task.runId === executionRunId) {
+        return true;
+      }
+      if (task.runId?.startsWith(prefix)) {
+        return true;
+      }
+      // Released reservation-keyed rows still record the authoritative execution start.
+      return task.startedAt === startedAt;
     })
-    .toSorted(
-      (left, right) =>
-        Number(left.endedAt !== undefined) - Number(right.endedAt !== undefined) ||
+    .toSorted((left, right) => {
+      const finalizedOrder =
+        Number(right.endedAt !== undefined) - Number(left.endedAt !== undefined);
+      return (
+        (receiptRunId ? finalizedOrder : -finalizedOrder) ||
         resolveCronTaskRecordTimestamp(right) - resolveCronTaskRecordTimestamp(left) ||
         right.createdAt - left.createdAt ||
-        right.taskId.localeCompare(left.taskId),
-    )[0];
+        right.taskId.localeCompare(left.taskId)
+      );
+    })[0];
 }
 
 type FinalizedCronTaskRun = {
@@ -193,12 +244,14 @@ export function findCronTaskRunRecoveryInDatabase(params: {
   jobId: string;
   startedAt: number;
   storeKey: string;
+  receiptId?: string;
 }): { taskRunId?: string; finalized?: FinalizedCronTaskRun } {
   const task = findLatestCronTaskRunForRecoveryFromRecords(
     listTaskRecordsByRuntimeSourceIdInDatabase(params.database, "cron", params.jobId),
     params.jobId,
     params.startedAt,
     params.storeKey,
+    params.receiptId,
   );
   const finalized = finalizedCronTaskRun(task, params.jobId);
   return {
@@ -337,9 +390,14 @@ export function tryFinishCronTaskRun(
     result.errorClassification,
   );
   const startedAt = entry.runAtMs ?? entry.ts;
-  const candidateRunId =
-    result.taskRunId ?? createCronTaskRunId(entry.jobId, startedAt, entry.runId);
+  let candidateRunId = result.taskRunId;
   try {
+    candidateRunId ??= resolveCronTaskRunId({
+      state,
+      jobId: entry.jobId,
+      startedAt,
+      publicRunId: entry.runId,
+    });
     const existingCandidate = findTaskByRunId(candidateRunId);
     const taskRunId =
       existingCandidate?.runtime === "cron"
@@ -432,7 +490,7 @@ export function tryFinishCronTaskRun(
     }
   } catch (error) {
     state.deps.log.warn(
-      { runId: candidateRunId, jobStatus: entry.status, error },
+      { runId: candidateRunId ?? entry.runId, jobStatus: entry.status, error },
       "cron: failed to update task ledger record",
     );
   }

src/cron/store/run-receipt-store.test.ts

@@ -9,12 +9,14 @@ import type { CronJob } from "../types.js";
 import { cronStoreKey } from "./key.js";
 import {
   assertCronRunReceiptCurrent,
+  bindCronRunReceiptTaskIdentity,
   claimCronRunReceiptInDatabase,
   CronRunReceiptConflictError,
   CronRunReceiptRevisionError,
   findActiveCronRunReceiptInDatabase,
   finishCronRunReceipt,
   prepareCronRunReceiptClaim,
+  releaseLocalCronRunReceiptOwnership,
 } from "./run-receipt-store.js";
 
 const { makeStorePath } = setupCronServiceSuite({ prefix: "cron-run-receipt-" });
@@ -55,7 +57,8 @@ 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
+              request_run_id AS requestRunId, 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`,
@@ -64,6 +67,7 @@ function receipts(storePath: string, jobId: string) {
     receiptId: string;
     status: string;
     agentId: string;
+    requestRunId: string | null;
     startedAtMs: number;
     error: string | null;
   }>;
@@ -96,10 +100,31 @@ describe("cron run receipt store", () => {
     await saveCronStore(storePath, { version: 1, jobs: [job] });
 
     const first = claim(storePath, job, 100);
+    expect(
+      bindCronRunReceiptTaskIdentity({
+        storePath,
+        jobId: job.id,
+        startedAtMs: 100,
+        requestRunId: "manual-overlap-1",
+      }),
+    ).toBe(first.receiptId);
+    expect(() =>
+      bindCronRunReceiptTaskIdentity({
+        storePath,
+        jobId: job.id,
+        startedAtMs: 100,
+        requestRunId: "manual-overlap-2",
+      }),
+    ).toThrow(CronRunReceiptRevisionError);
 
     expect(() => claim(storePath, job, 101)).toThrow(CronRunReceiptConflictError);
     expect(receipts(storePath, job.id)).toMatchObject([
-      { receiptId: first.receiptId, status: "running", startedAtMs: 100 },
+      {
+        receiptId: first.receiptId,
+        requestRunId: "manual-overlap-1",
+        status: "running",
+        startedAtMs: 100,
+      },
     ]);
 
     finishCronRunReceipt({ handle: first, status: "ok", finishedAtMs: 110 });
@@ -109,6 +134,35 @@ describe("cron run receipt store", () => {
     expect(receipts(storePath, job.id).map((receipt) => receipt.status)).toEqual(["skipped", "ok"]);
   });
 
+  it.each(["process-start identity", "process-local ownership"])(
+    "rejects task binding without matching %s",
+    async (fence) => {
+      const { storePath } = await makeStorePath();
+      const job = makeJob(`task-bind-${fence}`);
+      await saveCronStore(storePath, { version: 1, jobs: [job] });
+      const receipt = claim(storePath, job, 150);
+      if (fence === "process-start identity") {
+        openOpenClawStateDatabase()
+          .db.prepare(
+            "UPDATE cron_run_receipts SET owner_start_time = owner_start_time + 1 WHERE receipt_id = ?",
+          )
+          .run(receipt.receiptId);
+      } else {
+        releaseLocalCronRunReceiptOwnership(receipt);
+      }
+
+      expect(() =>
+        bindCronRunReceiptTaskIdentity({
+          storePath,
+          jobId: job.id,
+          startedAtMs: 150,
+          requestRunId: "manual-fenced",
+        }),
+      ).toThrow(CronRunReceiptRevisionError);
+      releaseLocalCronRunReceiptOwnership(receipt);
+    },
+  );
+
   it("retires a provably dead process claim before admitting its successor", async () => {
     const { storePath } = await makeStorePath();
     const job = makeJob("restart");

src/cron/store/run-receipt-store.ts

@@ -559,6 +559,61 @@ export function activateCronRunReceiptInDatabase(params: {
   return { ...params.handle, startedAtMs: params.startedAtMs };
 }
 
+/** Binds a task/public identity to the exact locally owned active receipt. */
+export function bindCronRunReceiptTaskIdentity(params: {
+  storePath: string;
+  jobId: string;
+  startedAtMs: number;
+  requestRunId?: string;
+}): string | undefined {
+  const storeKey = cronStoreKey(params.storePath);
+  const requestRunId = params.requestRunId?.trim() || undefined;
+  return withReceiptWrite("cron.run-receipt.bind-task", {}, (database) => {
+    const current = activeRow(database, storeKey, params.jobId);
+    if (!current) {
+      return undefined;
+    }
+    const ownerStartTime = getFileLockProcessStartTime(process.pid);
+    if (
+      current.started_at_ms !== params.startedAtMs ||
+      current.owner_pid !== process.pid ||
+      ownerStartTime === null ||
+      current.owner_start_time !== ownerStartTime ||
+      !locallyOwnedReceipts.has(current.receipt_id)
+    ) {
+      throw new CronRunReceiptRevisionError(
+        current.receipt_id,
+        "cron run task identity no longer belongs to the active receipt",
+      );
+    }
+    if (
+      requestRunId &&
+      current.request_run_id !== null &&
+      current.request_run_id !== requestRunId
+    ) {
+      throw new CronRunReceiptRevisionError(
+        current.receipt_id,
+        "cron run request identity changed after admission",
+      );
+    }
+    if (requestRunId && current.request_run_id === null) {
+      executeSqliteQuerySync(
+        database,
+        query(database)
+          .updateTable("cron_run_receipts")
+          .set({ request_run_id: requestRunId })
+          .where("receipt_id", "=", current.receipt_id)
+          .where("status", "=", "running")
+          .where("owner_pid", "=", current.owner_pid)
+          .where("owner_start_time", "=", current.owner_start_time)
+          .where("started_at_ms", "=", current.started_at_ms)
+          .where("request_run_id", "is", null),
+      );
+    }
+    return current.receipt_id;
+  });
+}
+
 export function assertCronRunReceiptCurrent(params: {
   handle: CronRunReceiptHandle;
   resolveAgentId: ResolveReceiptAgentId;