fix(selfhost): use ownership tokens for transient PR actuation locks

src/env.d.ts

@@ -32,8 +32,11 @@ declare global {
        *  already held by someone else. Unlike a get-then-set pair, there is no window where two concurrent
        *  callers can both observe an absent key and both claim it — the store (e.g. Redis SET NX) performs the
        *  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). */
+       *  type-checks; callers fail open (proceed without exclusivity) when absent — no get-then-set fallback (#2129). */
       claim?(key: string, value: string, ttlSeconds: number): Promise<boolean>;
+      /** Delete `key` only when its current value equals `value` (compare-and-delete). Returns true when the
+       *  key was removed. Optional; lock release skips when absent and relies on TTL instead of blind `del()`. */
+      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

@@ -315,6 +315,7 @@ import {
 import { isDuplicateClusterWinnerByClaim } from "../signals/duplicate-winner";
 import { buildUnifiedReviewDiff } from "../review/review-diff";
 import { buildUnifiedCommentBody } from "../review/unified-comment-bridge";
+import { randomUUID } from "node:crypto";
 import { isRetryableJobError, RetryableJobError } from "./retryable";
 import { screenshotsAllowed } from "../review/visual-wire";
 import { isVisualPath } from "../review/visual/paths";
@@ -2059,7 +2060,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 actuationLock = await claimPrActuationLock(env, repoFullName, pr.number);
+  if (!actuationLock.acquired) return;
   try {
     await runAgentMaintenancePlanAndExecute(env, {
       installationId,
@@ -2073,7 +2075,7 @@ async function maybeRunAgentMaintenance(
       liveFacts: args.liveFacts,
     });
   } finally {
-    await releasePrActuationLock(env, repoFullName, pr.number);
+    await releasePrActuationLock(env, repoFullName, pr.number, actuationLock.ownerToken);
   }
 }
 
@@ -3005,23 +3007,24 @@ async function putTransientKey(
 // 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.
+// Ownership tokens + releaseIfValue (compare-and-delete) prevent a stale holder's finally block from
+// deleting a successor's live lock after TTL expiry (#2129/#2135).
 const PR_ACTUATION_LOCK_TTL_SECONDS = 600;
 function prActuationLockKey(repoFullName: string, prNumber: number): string {
   return `pr-actuation-lock:${repoFullName.toLowerCase()}#${prNumber}`;
 }
+
+/** Result of claiming a transient-cache mutex. `ownerToken` is null on fail-open paths (no cache / claim error). */
+export type TransientLockClaim = {
+  acquired: boolean;
+  ownerToken: string | null;
+};
+
 export async function claimPrActuationLock(
   env: Env,
   repoFullName: string,
   prNumber: number,
-): Promise<boolean> {
+): Promise<TransientLockClaim> {
   return claimTransientLock(
     env,
     prActuationLockKey(repoFullName, prNumber),
@@ -3032,12 +3035,9 @@ export async function releasePrActuationLock(
   env: Env,
   repoFullName: string,
   prNumber: number,
+  ownerToken: string | null,
 ): Promise<void> {
-  try {
-    await env.SELFHOST_TRANSIENT_CACHE?.del?.(prActuationLockKey(repoFullName, prNumber));
-  } catch {
-    // best-effort
-  }
+  await releaseTransientLockIfOwner(env, prActuationLockKey(repoFullName, prNumber), ownerToken);
 }
 
 // A plain thrown Error still reaches the queue's retry path (this call site is deliberately uncaught, same as
@@ -3375,12 +3375,32 @@ async function claimTransientLock(
   env: Env,
   key: string,
   ttlSeconds: number,
-): Promise<boolean> {
-  if (!env.SELFHOST_TRANSIENT_CACHE?.claim) return true; // no atomic primitive — nothing to serialize against.
+): Promise<TransientLockClaim> {
+  if (!env.SELFHOST_TRANSIENT_CACHE?.claim) return { acquired: true, ownerToken: null };
+  const ownerToken = randomUUID();
+  try {
+    const acquired = await env.SELFHOST_TRANSIENT_CACHE.claim(key, ownerToken, ttlSeconds);
+    return { acquired, ownerToken: acquired ? ownerToken : null };
+  } catch {
+    return { acquired: true, ownerToken: null };
+  }
+}
+
+async function releaseTransientLockIfOwner(
+  env: Env,
+  key: string,
+  ownerToken: string | null,
+): Promise<void> {
+  if (!ownerToken) return;
   try {
-    return await env.SELFHOST_TRANSIENT_CACHE.claim(key, "1", ttlSeconds);
+    const cache = env.SELFHOST_TRANSIENT_CACHE;
+    if (cache?.releaseIfValue) {
+      await cache.releaseIfValue(key, ownerToken);
+      return;
+    }
+    // Without compare-and-delete, skip release and let TTL expire — blind del() would reopen #2129/#2135.
   } catch {
-    return true; // fail open — see the doc comment above.
+    // best-effort; TTL is the backstop if release fails
   }
 }
 
@@ -3427,7 +3447,7 @@ export async function claimAiReviewLock(
   prNumber: number,
   headSha: string,
   mode: string,
-): Promise<boolean> {
+): Promise<TransientLockClaim> {
   return claimTransientLock(
     env,
     aiReviewLockKey(repoFullName, prNumber, headSha, mode),
@@ -3442,12 +3462,13 @@ export async function releaseAiReviewLock(
   prNumber: number,
   headSha: string,
   mode: string,
+  ownerToken: string | null,
 ): Promise<void> {
-  try {
-    await env.SELFHOST_TRANSIENT_CACHE?.del?.(aiReviewLockKey(repoFullName, prNumber, headSha, mode));
-  } catch {
-    // best-effort; the TTL is the backstop if release fails
-  }
+  await releaseTransientLockIfOwner(
+    env,
+    aiReviewLockKey(repoFullName, prNumber, headSha, mode),
+    ownerToken,
+  );
 }
 
 /** Read the CI head SHA off a `check_suite`/`check_run` `completed` payload (the event node carries `head_sha`;
@@ -5892,15 +5913,14 @@ export async function runAiReviewForAdvisory(
   // return different verdicts. Claim before the expensive section below; a pass that loses the race returns the
   // same inconclusive-hold shape the "AI produced no usable verdict" path already returns, so the gate is held
   // (neutral) for a human rather than either pass's independently-decided verdict racing the other's cache write.
-  if (
-    !(await claimAiReviewLock(
-      env,
-      args.repoFullName,
-      args.pr.number,
-      args.advisory.headSha,
-      args.settings.aiReviewMode,
-    ))
-  ) {
+  const aiReviewLock = await claimAiReviewLock(
+    env,
+    args.repoFullName,
+    args.pr.number,
+    args.advisory.headSha,
+    args.settings.aiReviewMode,
+  );
+  if (!aiReviewLock.acquired) {
     const findings: AdvisoryFinding[] = [
       {
         code: "ai_review_inconclusive",
@@ -6234,6 +6254,7 @@ export async function runAiReviewForAdvisory(
       args.pr.number,
       args.advisory.headSha,
       args.settings.aiReviewMode,
+      aiReviewLock.ownerToken,
     );
   }
 }
@@ -9462,7 +9483,8 @@ async function maybeCloseDraftDodgeAttempt(
   pr: PullRequestRecord,
   settings: RepositorySettings,
 ): Promise<void> {
-  if (!(await claimPrActuationLock(env, repoFullName, pr.number))) {
+  const actuationLock = await claimPrActuationLock(env, repoFullName, pr.number);
+  if (!actuationLock.acquired) {
     throw new PrActuationLockContendedError(repoFullName, pr.number, "draft-dodge");
   }
   try {
@@ -9475,7 +9497,7 @@ async function maybeCloseDraftDodgeAttempt(
       settings,
     );
   } finally {
-    await releasePrActuationLock(env, repoFullName, pr.number);
+    await releasePrActuationLock(env, repoFullName, pr.number, actuationLock.ownerToken);
   }
 }
 
@@ -9668,7 +9690,8 @@ async function maybeRecloseDisallowedReopen(
   pr: PullRequestRecord,
   payload: GitHubWebhookPayload,
 ): Promise<ReopenRecloseOutcome> {
-  if (!(await claimPrActuationLock(env, repoFullName, pr.number))) {
+  const actuationLock = await claimPrActuationLock(env, repoFullName, pr.number);
+  if (!actuationLock.acquired) {
     throw new PrActuationLockContendedError(repoFullName, pr.number, "reopen-reclose");
   }
   try {
@@ -9682,7 +9705,7 @@ async function maybeRecloseDisallowedReopen(
     );
     return reclosed ? "reclosed" : "allowed";
   } finally {
-    await releasePrActuationLock(env, repoFullName, pr.number);
+    await releasePrActuationLock(env, repoFullName, pr.number, actuationLock.ownerToken);
   }
 }
 

src/selfhost/redis-cache.ts

@@ -23,6 +23,16 @@ export function createRedisCache(redis: Redis) {
       const result = await redis.set(key, value, "EX", ttlSeconds, "NX");
       return result === "OK";
     },
+    // Atomic compare-and-delete: only the holder whose token still matches may release the key.
+    async releaseIfValue(key: string, value: string): Promise<boolean> {
+      const result = await redis.eval(
+        "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end",
+        1,
+        key,
+        value,
+      );
+      return result === 1;
+    },
   };
 }
 

test/helpers/d1.ts

@@ -102,7 +102,7 @@ export function createTestEnv(overrides: Partial<Env> = {}): Env {
       async get(key: string) {
         return transientCache.get(key) ?? null;
       },
-      async set(key: string, value: string) {
+      async set(key: string, value: string, _ttlSeconds: number) {
         transientCache.set(key, value);
       },
       async del(key: string) {
@@ -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/ai-review-advisory.test.ts

@@ -582,7 +582,7 @@ describe("runAiReviewForAdvisory", () => {
     // Simulate a webhook pass already in-flight for this exact (repo, PR, head, mode) tuple — the caller under
     // test (a sweep-shaped pass, say) must defer instead of racing it with a second, independently-decided
     // LLM call.
-    expect(await claimAiReviewLock(env, "acme/widgets", 3, "sha3", "block")).toBe(true);
+    expect((await claimAiReviewLock(env, "acme/widgets", 3, "sha3", "block")).acquired).toBe(true);
 
     const result = await runAiReviewForAdvisory(env, {
       settings: { aiReviewMode: "block" } as RepositorySettings,