fix(queue): owner-verified release for per-PR actuation lock

src/env.d.ts

@@ -34,6 +34,10 @@ declare global {
        *  check-and-set as one operation. Optional so a cache adapter that hasn't implemented it yet still
        *  type-checks; callers fall back to the non-atomic get/set pair when absent (#2129). */
       claim?(key: string, value: string, ttlSeconds: number): Promise<boolean>;
+      /** Atomic "delete only if value matches": returns true when this call removed the key, false when the
+       *  key was absent or held by a different owner. Used by per-holder transient locks so a stale holder's
+       *  finally release cannot delete a later claimer's live lock after TTL expiry (#2129). */
+      releaseIfValue?(key: string, value: string): Promise<boolean>;
     };
     /** TODO (convergence follow-up): a per-PR LOCK Durable Object (`SubmissionLock` mutex) is a separate,
      *  more-involved sub-task — it needs the ported DO class + its own migration tag, not just a binding here.

src/queue/processors.ts

@@ -2143,7 +2143,8 @@ async function maybeRunAgentMaintenance(
   // critical section (extracted below so the try/finally doesn't force-reindent that whole block); a pass that
   // loses the race defers cleanly — the next webhook/sweep tick is the backstop. Lightweight stand-in for the
   // per-PR SubmissionLock Durable Object noted as a longer-term TODO in env.d.ts.
-  if (!(await claimPrActuationLock(env, repoFullName, pr.number))) return;
+  const lockToken = await claimPrActuationLock(env, repoFullName, pr.number);
+  if (lockToken === null) return;
   try {
     await runAgentMaintenancePlanAndExecute(env, {
       installationId,
@@ -2157,7 +2158,7 @@ async function maybeRunAgentMaintenance(
       liveFacts: args.liveFacts,
     });
   } finally {
-    await releasePrActuationLock(env, repoFullName, pr.number);
+    await releasePrActuationLock(env, repoFullName, pr.number, lockToken);
   }
 }
 
@@ -3094,32 +3095,31 @@ async function putTransientKey(
 // could still race — the whole point of this mutex is to make "does something else already own this PR" one
 // question with one answer, not one question per code path (review round 4). This is a lightweight interim
 // mutex (a full per-PR Durable Object / SubmissionLock is a separate, more-involved follow-up — see the TODO in
-// env.d.ts) built on the SAME transient cache used for CI-completion coalescing above, claimed ATOMICALLY (see
-// claimTransientLock) so two racing deliveries can never both win the claim — a short TTL, best-effort release.
-// A lock-contended caller fails OPEN (returns false / skips this pass) rather than blocking — the delivery
-// holding the lock is evaluating the SAME PR, and the periodic sweep is the backstop if this specific trigger is
-// dropped. A cache adapter with no claim() primitive gets NO exclusivity at all (every call proceeds) rather
-// than a get-then-set pair that only *looks* atomic — see claimTransientLock's doc comment for why that fallback
-// was removed.
+// env.d.ts) built on the SAME transient cache used for CI-completion coalescing above, claimed ATOMICALLY with a
+// per-holder token (see claimTransientLockWithOwnerToken) so two racing deliveries can never both win the claim —
+// a short TTL, owner-verified release. A lock-contended caller skips this pass (returns null) rather than
+// blocking — the delivery holding the lock is evaluating the SAME PR, and the periodic sweep is the backstop if
+// this specific trigger is dropped. A cache adapter with no claim() primitive gets NO exclusivity at all (every
+// call proceeds as fail-open) rather than a get-then-set pair that only *looks* atomic — see claimTransientLock's
+// doc comment for why that fallback was removed.
 //
-// KNOWN LIMITATION: the lock value is a constant, not a per-holder ownership token, so release does not verify
-// it still owns the key — if a holder ran past the TTL, a later claimer's live lock could be deleted by the
-// first holder's stale `finally` release, reopening the exact race this mutex exists to close. A per-holder
-// token + a conditional (check-then-delete) release would close this properly, but needs a new atomic
-// compare-and-delete primitive on the cache adapter — tracked alongside the Durable Object follow-up above. The
-// TTL is set generously long specifically so this window is practically unreachable: the guarded operations
-// (a handful of sequential GitHub API calls, or a maintenance pass's plan-and-execute) should never legitimately
-// run anywhere near this long.
+// Per-PR mutual exclusion (#2129/#2135): maybeRunAgentMaintenance, maybeCloseDraftDodgeAttempt, and
+// maybeRecloseDisallowedReopen all claim/release the SAME key so none of the three mutating PR paths can race
+// any other (review round 4) — a single namespace, not one lock per path. Each claim stores a per-holder
+// ownership token; release is compare-and-delete so a holder that ran past the TTL cannot delete a later
+// claimer's live lock in its stale `finally` block.
 const PR_ACTUATION_LOCK_TTL_SECONDS = 600;
 function prActuationLockKey(repoFullName: string, prNumber: number): string {
   return `pr-actuation-lock:${repoFullName.toLowerCase()}#${prNumber}`;
 }
+/** Non-empty string = claimed (release with this token). Empty string = fail-open (no lock held). Null = contended. */
+export type PrActuationLockClaim = string | null;
 export async function claimPrActuationLock(
   env: Env,
   repoFullName: string,
   prNumber: number,
-): Promise<boolean> {
-  return claimTransientLock(
+): Promise<PrActuationLockClaim> {
+  return claimTransientLockWithOwnerToken(
     env,
     prActuationLockKey(repoFullName, prNumber),
     PR_ACTUATION_LOCK_TTL_SECONDS,
@@ -3129,12 +3129,9 @@ export async function releasePrActuationLock(
   env: Env,
   repoFullName: string,
   prNumber: number,
+  token: PrActuationLockClaim,
 ): Promise<void> {
-  try {
-    await env.SELFHOST_TRANSIENT_CACHE?.del?.(prActuationLockKey(repoFullName, prNumber));
-  } catch {
-    // best-effort
-  }
+  await releaseTransientLockIfOwner(env, prActuationLockKey(repoFullName, prNumber), token);
 }
 
 // A plain thrown Error still reaches the queue's retry path (this call site is deliberately uncaught, same as
@@ -3481,6 +3478,35 @@ async function claimTransientLock(
   }
 }
 
+/** Like {@link claimTransientLock}, but stores a per-holder ownership token for compare-and-delete release. */
+async function claimTransientLockWithOwnerToken(
+  env: Env,
+  key: string,
+  ttlSeconds: number,
+): Promise<PrActuationLockClaim> {
+  if (!env.SELFHOST_TRANSIENT_CACHE?.claim) return ""; // fail-open — no lock to release.
+  try {
+    const token = crypto.randomUUID();
+    return (await env.SELFHOST_TRANSIENT_CACHE.claim(key, token, ttlSeconds)) ? token : null;
+  } catch {
+    return ""; // fail open — see claimTransientLock's doc comment above.
+  }
+}
+
+/** Best-effort owner-verified release. Empty token = fail-open claim, no-op. Without releaseIfValue, leave the
+ *  key to expire via TTL rather than risk deleting a later claimer's live lock. */
+async function releaseTransientLockIfOwner(env: Env, key: string, token: PrActuationLockClaim): Promise<void> {
+  if (!token) return;
+  try {
+    if (env.SELFHOST_TRANSIENT_CACHE?.releaseIfValue) {
+      await env.SELFHOST_TRANSIENT_CACHE.releaseIfValue(key, token);
+      return;
+    }
+  } catch {
+    // best-effort
+  }
+}
+
 // Per-(repo, PR, head SHA) advisory lock around runAiReviewForAdvisory's expensive grounding/RAG/enrichment/LLM
 // section (#confirmed-bug: a webhook pass and an agent-regate-pr sweep pass can independently reach this same
 // code for the SAME PR at the SAME head SHA, both miss the cache, and both fire a real LLM call — which can
@@ -9624,7 +9650,8 @@ async function maybeCloseDraftDodgeAttempt(
   pr: PullRequestRecord,
   settings: RepositorySettings,
 ): Promise<void> {
-  if (!(await claimPrActuationLock(env, repoFullName, pr.number))) {
+  const lockToken = await claimPrActuationLock(env, repoFullName, pr.number);
+  if (lockToken === null) {
     throw new PrActuationLockContendedError(repoFullName, pr.number, "draft-dodge");
   }
   try {
@@ -9637,7 +9664,7 @@ async function maybeCloseDraftDodgeAttempt(
       settings,
     );
   } finally {
-    await releasePrActuationLock(env, repoFullName, pr.number);
+    await releasePrActuationLock(env, repoFullName, pr.number, lockToken);
   }
 }
 
@@ -9830,7 +9857,8 @@ async function maybeRecloseDisallowedReopen(
   pr: PullRequestRecord,
   payload: GitHubWebhookPayload,
 ): Promise<ReopenRecloseOutcome> {
-  if (!(await claimPrActuationLock(env, repoFullName, pr.number))) {
+  const lockToken = await claimPrActuationLock(env, repoFullName, pr.number);
+  if (lockToken === null) {
     throw new PrActuationLockContendedError(repoFullName, pr.number, "reopen-reclose");
   }
   try {
@@ -9844,7 +9872,7 @@ async function maybeRecloseDisallowedReopen(
     );
     return reclosed ? "reclosed" : "allowed";
   } finally {
-    await releasePrActuationLock(env, repoFullName, pr.number);
+    await releasePrActuationLock(env, repoFullName, pr.number, lockToken);
   }
 }
 

src/selfhost/redis-cache.ts

@@ -23,6 +23,17 @@ export function createRedisCache(redis: Redis) {
       const result = await redis.set(key, value, "EX", ttlSeconds, "NX");
       return result === "OK";
     },
+    async releaseIfValue(key: string, value: string): Promise<boolean> {
+      const script = `
+        if redis.call("get", KEYS[1]) == ARGV[1] then
+          return redis.call("del", KEYS[1])
+        else
+          return 0
+        end
+      `;
+      const result = await redis.eval(script, 1, key, value);
+      return Number(result) === 1;
+    },
   };
 }
 

test/helpers/d1.ts

@@ -117,6 +117,11 @@ export function createTestEnv(overrides: Partial<Env> = {}): Env {
         transientCache.set(key, value);
         return true;
       },
+      async releaseIfValue(key: string, value: string) {
+        if (transientCache.get(key) !== value) return false;
+        transientCache.delete(key);
+        return true;
+      },
     },
     // Per-repo review allowlist: default to the test repos so flag-ON wiring tests activate the
     // gated review features. Override to "" to assert the dormant (no-repo) default.

test/unit/queue.test.ts

@@ -5507,11 +5507,13 @@ describe("queue processors", () => {
   // three mutating PR paths can race any other (review round 4) — a single namespace, not one lock per path.
   it("claimPrActuationLock claims when free, denies when held (per-PR), and release frees it again (#2135)", async () => {
     const env = createTestEnv({});
-    expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBe(true);
-    expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBe(false);
-    expect(await claimPrActuationLock(env, "owner/act-repo", 8)).toBe(true);
-    await releasePrActuationLock(env, "owner/act-repo", 7);
-    expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBe(true);
+    const first = await claimPrActuationLock(env, "owner/act-repo", 7);
+    expect(typeof first).toBe("string");
+    expect(first).not.toBe("");
+    expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBeNull();
+    expect(await claimPrActuationLock(env, "owner/act-repo", 8)).not.toBeNull();
+    await releasePrActuationLock(env, "owner/act-repo", 7, first);
+    expect(await claimPrActuationLock(env, "owner/act-repo", 7)).not.toBeNull();
   });
 
   it("claimPrActuationLock fails OPEN on a broken transient cache — never itself blocks actuation (#2135)", async () => {
@@ -5522,8 +5524,8 @@ describe("queue processors", () => {
         del: async () => { throw new Error("cache delete error"); },
       },
     });
-    expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBe(true);
-    await expect(releasePrActuationLock(env, "owner/act-repo", 7)).resolves.toBeUndefined();
+    expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBe("");
+    await expect(releasePrActuationLock(env, "owner/act-repo", 7, "")).resolves.toBeUndefined();
   });
 
   it("claimPrActuationLock fails OPEN when the atomic claim primitive itself throws (#2135)", async () => {
@@ -5534,7 +5536,7 @@ describe("queue processors", () => {
         claim: async () => { throw new Error("redis unavailable"); },
       },
     });
-    expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBe(true);
+    expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBe("");
   });
 
   it("REGRESSION (#2135): claimPrActuationLock uses an atomic check-and-set, so two genuinely concurrent claims for the SAME PR can never both succeed", async () => {
@@ -5543,7 +5545,57 @@ describe("queue processors", () => {
       claimPrActuationLock(env, "owner/act-repo", 7),
       claimPrActuationLock(env, "owner/act-repo", 7),
     ]);
-    expect([first, second].filter(Boolean)).toHaveLength(1);
+    expect([first, second].filter((token) => token !== null && token !== "")).toHaveLength(1);
+  });
+
+  it("REGRESSION: a stale holder's releaseIfValue does not delete a later claimer's live lock", async () => {
+    const env = createTestEnv({});
+    const staleToken = await claimPrActuationLock(env, "owner/act-repo", 7);
+    expect(staleToken).not.toBeNull();
+    expect(staleToken).not.toBe("");
+    const liveToken = await claimPrActuationLock(env, "owner/act-repo", 7);
+    expect(liveToken).toBeNull();
+    await env.SELFHOST_TRANSIENT_CACHE?.set?.("pr-actuation-lock:owner/act-repo#7", "later-holder-token", 60);
+    await releasePrActuationLock(env, "owner/act-repo", 7, staleToken);
+    expect(await env.SELFHOST_TRANSIENT_CACHE?.get("pr-actuation-lock:owner/act-repo#7")).toBe("later-holder-token");
+    expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBeNull();
+    await releasePrActuationLock(env, "owner/act-repo", 7, "later-holder-token");
+    expect(await claimPrActuationLock(env, "owner/act-repo", 7)).not.toBeNull();
+  });
+
+  it("releasePrActuationLock leaves the key in place when the cache has claim() but no releaseIfValue()", async () => {
+    const values = new Map<string, string>();
+    const env = createTestEnv({
+      SELFHOST_TRANSIENT_CACHE: {
+        get: async (key: string) => values.get(key) ?? null,
+        set: async (key: string, value: string) => { values.set(key, value); },
+        claim: async (key: string, value: string) => {
+          if (values.has(key)) return false;
+          values.set(key, value);
+          return true;
+        },
+      },
+    });
+    const token = await claimPrActuationLock(env, "owner/act-repo", 7);
+    expect(token).not.toBe("");
+    await releasePrActuationLock(env, "owner/act-repo", 7, token);
+    expect(values.get("pr-actuation-lock:owner/act-repo#7")).toBe(token);
+  });
+
+  it("releasePrActuationLock swallows releaseIfValue errors (best-effort)", async () => {
+    const env = createTestEnv({
+      SELFHOST_TRANSIENT_CACHE: {
+        get: async () => "holder",
+        set: async () => undefined,
+        claim: async (_key: string, value: string) => {
+          return value.length > 0;
+        },
+        releaseIfValue: async () => {
+          throw new Error("redis down");
+        },
+      },
+    });
+    await expect(releasePrActuationLock(env, "owner/act-repo", 7, "holder")).resolves.toBeUndefined();
   });
 
   it("REGRESSION (#2135): claimPrActuationLock calls the atomic claim primitive, not a separate get+set pair, when the cache supports it", async () => {
@@ -5555,11 +5607,11 @@ describe("queue processors", () => {
         claim: async () => { calls.push("claim"); return true; },
       },
     });
-    expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBe(true);
+    expect(await claimPrActuationLock(env, "owner/act-repo", 7)).not.toBeNull();
     expect(calls).toEqual(["claim"]); // never falls through to the racy get/set pair when claim is available
   });
 
-  it("claimPrActuationLock returns true unconditionally when the cache has no claim() — no false exclusivity guarantee (#2135, review round 2)", async () => {
+  it("claimPrActuationLock returns empty string unconditionally when the cache has no claim() — no false exclusivity guarantee (#2135, review round 2)", async () => {
     // A get-then-set pair (even with a re-read) is not a real exclusivity guarantee under concurrent load, so a
     // cache without claim() now gets NO exclusivity at all rather than a fallback that only looks atomic.
     const values = new Map<string, string>();
@@ -5569,8 +5621,8 @@ describe("queue processors", () => {
         set: async (key: string, value: string) => { values.set(key, value); },
       },
     });
-    expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBe(true);
-    expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBe(true);
+    expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBe("");
+    expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBe("");
   });
 
   it("REGRESSION (#2135, review round 2): claimPrActuationLock does not falsely claim exclusivity for two genuinely concurrent callers when the cache has no claim()", async () => {
@@ -5586,7 +5638,7 @@ describe("queue processors", () => {
       claimPrActuationLock(env, "owner/act-repo", 7),
       claimPrActuationLock(env, "owner/act-repo", 7),
     ]);
-    expect([first, second]).toEqual([true, true]);
+    expect([first, second]).toEqual(["", ""]);
   });
 
   it("INVARIANT (#2129 per-PR lock): a maintenance pass defers when another pass already holds the PR's lock", async () => {

test/unit/selfhost-redis-cache.test.ts

@@ -21,6 +21,13 @@ function fakeRedis(): Redis & { _store: Map<string, string> } {
       _store.delete(k);
       return 1;
     },
+    async eval(_script: string, _numKeys: number, key: string, value: string) {
+      if (_store.get(key) === value) {
+        _store.delete(key);
+        return 1;
+      }
+      return 0;
+    },
   } as unknown as Redis & { _store: Map<string, string> };
 }
 
@@ -63,4 +70,18 @@ describe("createRedisCache (#1216 webhook dedup cache)", () => {
     const cache = createRedisCache(brokenRedis);
     await expect(cache.claim("lock", "1", 60)).rejects.toThrow("connection refused");
   });
+
+  it("releaseIfValue deletes only when the stored owner matches (#2129)", async () => {
+    const cache = createRedisCache(fakeRedis());
+    await cache.set("lock", "holder-A", 60);
+    expect(await cache.releaseIfValue("lock", "holder-A")).toBe(true);
+    expect(await cache.get("lock")).toBeNull();
+  });
+
+  it("releaseIfValue refuses when a different owner holds the key (#2129)", async () => {
+    const cache = createRedisCache(fakeRedis());
+    await cache.set("lock", "holder-B", 60);
+    expect(await cache.releaseIfValue("lock", "holder-A")).toBe(false);
+    expect(await cache.get("lock")).toBe("holder-B");
+  });
 });