fix: reclaim stale automation actions

cloudflare/automation/src/index.js

@@ -236,6 +236,7 @@ function safeActionError(error) {
 
 export async function claimAction(env, eventId, actionType, actionKey) {
   const timestamp = now();
+  const staleBefore = new Date(Date.now() - 15 * 60 * 1000).toISOString();
   const result = await env.STATE_DB.prepare(
     `INSERT INTO action_log (action_key, event_id, action_type, status, created_at, updated_at)
      VALUES (?, ?, ?, 'running', ?, ?)
@@ -244,8 +245,11 @@ export async function claimAction(env, eventId, actionType, actionKey) {
   if (result.meta?.changes === 1) return true;
   const existing = await env.STATE_DB.prepare("SELECT status, updated_at FROM action_log WHERE action_key = ?").bind(actionKey).first();
   if (existing?.status === "completed") return false;
-  if (existing?.status === "running" && existing.updated_at > new Date(Date.now() - 15 * 60 * 1000).toISOString()) return false;
-  const reclaimed = await env.STATE_DB.prepare("UPDATE action_log SET status = 'running', error = NULL, updated_at = ? WHERE action_key = ? AND status = 'failed'").bind(timestamp, actionKey).run();
+  if (existing?.status === "running" && existing.updated_at > staleBefore) return false;
+  const reclaimed = await env.STATE_DB.prepare(
+    "UPDATE action_log SET status = 'running', error = NULL, updated_at = ? " +
+      "WHERE action_key = ? AND (status = 'failed' OR (status = 'running' AND updated_at <= ?))",
+  ).bind(timestamp, actionKey, staleBefore).run();
   return reclaimed.meta?.changes === 1;
 }
 

cloudflare/automation/test/review-gates.test.js

@@ -1,7 +1,48 @@
 import test from "node:test";
 import assert from "node:assert/strict";
 import { evaluateMergeGate, latestCheckRuns, normalizeReviewResult } from "../src/github.js";
-import { callLuoxin, shouldInvokeLuoxin } from "../src/index.js";
+import { callLuoxin, claimAction, shouldInvokeLuoxin } from "../src/index.js";
+
+function actionLogDb(
+  existing,
+  { insertChanges = 0, updateChanges = 1, stateful = false } = {},
+) {
+  let row = existing ? { ...existing } : null;
+  const statements = [];
+  return {
+    statements,
+    getRow() {
+      return row;
+    },
+    prepare(sql) {
+      statements.push(sql);
+      return {
+        bind(...args) {
+          return {
+            async run() {
+              if (sql.startsWith("INSERT")) return { meta: { changes: insertChanges } };
+              if (sql.startsWith("UPDATE")) {
+                if (!stateful) return { meta: { changes: updateChanges } };
+                const [timestamp, , staleBefore] = args;
+                const reclaimable = row && (
+                  row.status === "failed" ||
+                  (row.status === "running" && row.updated_at <= staleBefore)
+                );
+                if (!reclaimable) return { meta: { changes: 0 } };
+                row = { ...row, status: "running", updated_at: timestamp };
+                return { meta: { changes: 1 } };
+              }
+              throw new Error(`unexpected run: ${sql}`);
+            },
+            async first() {
+              return row;
+            },
+          };
+        },
+      };
+    },
+  };
+}
 
 test("an approved clean PR with passing checks can pass the merge gate", () => {
   const result = evaluateMergeGate(
@@ -100,3 +141,94 @@ test("old failed reruns do not keep a newer successful check red", () => {
   ]);
   assert.deepEqual(current, [{ id: 2, name: "build", status: "completed", conclusion: "success", completed_at: "2026-09-17T02:00:00Z" }]);
 });
+
+test("a stale running action can be reclaimed after a worker interruption", async () => {
+  const db = actionLogDb({
+    status: "running",
+    updated_at: new Date(Date.now() - 16 * 60 * 1000).toISOString(),
+  }, { stateful: true });
+
+  assert.equal(
+    await claimAction(
+      { STATE_DB: db },
+      "event-stale",
+      "review_engine",
+      "review:repo#1:sha-stale",
+    ),
+    true,
+  );
+  assert.match(db.statements[2], /status = 'running'/);
+  assert.match(db.statements[2], /updated_at <= \?/);
+});
+
+test("only the first retry can reclaim the same stale action", async () => {
+  const db = actionLogDb({
+    status: "running",
+    updated_at: new Date(Date.now() - 16 * 60 * 1000).toISOString(),
+  }, { stateful: true });
+
+  const firstClaim = await claimAction(
+    { STATE_DB: db },
+    "event-race",
+    "review_engine",
+    "review:repo#1:sha-race",
+  );
+  const secondClaim = await claimAction(
+    { STATE_DB: db },
+    "event-race-retry",
+    "review_engine",
+    "review:repo#1:sha-race",
+  );
+
+  assert.equal(firstClaim, true);
+  assert.equal(secondClaim, false);
+  assert.equal(db.getRow().status, "running");
+});
+
+test("a recent running action is still owned by the active worker", async () => {
+  const db = actionLogDb({
+    status: "running",
+    updated_at: new Date(Date.now() - 14 * 60 * 1000).toISOString(),
+  });
+
+  assert.equal(
+    await claimAction(
+      { STATE_DB: db },
+      "event-recent",
+      "review_engine",
+      "review:repo#1:sha-recent",
+    ),
+    false,
+  );
+  assert.equal(db.statements.length, 2);
+});
+
+test("completed actions remain idempotent and are not reclaimed", async () => {
+  const db = actionLogDb({ status: "completed", updated_at: new Date().toISOString() });
+
+  assert.equal(
+    await claimAction(
+      { STATE_DB: db },
+      "event-completed",
+      "review_engine",
+      "review:repo#1:sha-completed",
+    ),
+    false,
+  );
+  assert.equal(db.statements.length, 2);
+});
+
+test("failed actions keep the existing retry path", async () => {
+  const db = actionLogDb({ status: "failed", updated_at: new Date().toISOString() });
+
+  assert.equal(
+    await claimAction(
+      { STATE_DB: db },
+      "event-failed",
+      "review_engine",
+      "review:repo#1:sha-failed",
+    ),
+    true,
+  );
+  assert.match(db.statements[2], /status = 'failed'/);
+});