fix(webapp): thread per-claim ownership token through publish/release

The buffer's claim API now requires a caller-supplied ownership token
so compare-and-act protects the slot against a stale predecessor.
Wires the token end-to-end:

- `claimOrAwait` generates the token (UUID) up front and reuses it
  across the retry path; returns it on the `claimed` outcome.
- `publishClaim` and `releaseClaim` wrappers accept and forward the
  token to the buffer.
- `ClaimedIdempotency` carries the token so the trigger pipeline can
  publish or release with the same token it claimed under.
- `triggerTask.server.ts` threads the token into the publish call.

Tests pin token round-trip: claimOrAwait → claimIdempotency, plus the
publish and release pass-through.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: d-cs

apps/webapp/app/runEngine/concerns/idempotencyKeys.server.ts

@@ -18,6 +18,10 @@ export type ClaimedIdempotency = {
   envId: string;
   taskIdentifier: string;
   idempotencyKey: string;
+  // Ownership token from `claimOrAwait`. The caller's trigger pipeline
+  // MUST thread this into publishClaim/releaseClaim so the buffer's
+  // compare-and-act protects the slot against a stale predecessor.
+  token: string;
 };
 
 export type IdempotencyKeyConcernResult =
@@ -279,7 +283,8 @@ export class IdempotencyKeyConcern {
       }
       if (outcome.kind === "claimed") {
         // Caller MUST publish/release. Signalled via the result's
-        // `claim` field.
+        // `claim` field, including the ownership token so the buffer
+        // can compare-and-act on the slot we now own.
         return {
           isCached: false,
           idempotencyKey,
@@ -288,6 +293,7 @@ export class IdempotencyKeyConcern {
             envId: request.environment.id,
             taskIdentifier: request.taskId,
             idempotencyKey,
+            token: outcome.token,
           },
         };
       }

apps/webapp/app/runEngine/services/triggerTask.server.ts

@@ -627,6 +627,7 @@ export class RunEngineTriggerTaskService {
           envId: idempotencyClaim.envId,
           taskIdentifier: idempotencyClaim.taskIdentifier,
           idempotencyKey: idempotencyClaim.idempotencyKey,
+          token: idempotencyClaim.token,
           runId: result.run.friendlyId,
         });
       }

apps/webapp/app/v3/mollifier/idempotencyClaim.server.ts

@@ -1,3 +1,4 @@
+import { randomUUID } from "node:crypto";
 import type {
   IdempotencyClaimResult,
   IdempotencyLookupInput,
@@ -17,7 +18,11 @@ export const DEFAULT_CLAIM_WAIT_MS = 5_000;
 export const DEFAULT_CLAIM_POLL_MS = 25;
 
 export type ClaimOrAwaitOutcome =
-  | { kind: "claimed" } // we own the claim; caller proceeds with the trigger pipeline
+  // We own the claim. `token` MUST be passed to publishClaim/releaseClaim
+  // so the buffer can compare-and-act against our ownership marker — a
+  // late release from a previous claimant whose TTL expired cannot
+  // erase our slot.
+  | { kind: "claimed"; token: string }
   | { kind: "resolved"; runId: string } // someone else's runId; caller returns isCached:true
   | { kind: "timed_out" };
 
@@ -30,6 +35,10 @@ export type ClaimOrAwaitInput = IdempotencyLookupInput & {
   buffer?: MollifierBuffer | null;
   now?: () => number;
   sleep?: (ms: number) => Promise<void>;
+  // Test override for the ownership-token generator. Defaults to
+  // `crypto.randomUUID()`. Tests pass a deterministic value so they
+  // can assert publish/release pass-through.
+  generateToken?: () => string;
 };
 
 // Pre-gate Redis claim. All same-key triggers serialise through here
@@ -50,12 +59,19 @@ export type ClaimOrAwaitInput = IdempotencyLookupInput & {
 //   IdempotencyKeyConcern PG-first lookup.
 export async function claimOrAwait(input: ClaimOrAwaitInput): Promise<ClaimOrAwaitOutcome> {
   const buffer = input.buffer === undefined ? getMollifierBuffer() : input.buffer;
+  const generateToken = input.generateToken ?? randomUUID;
+  // Generate the ownership token up front so the retry loop reuses it
+  // — we're the same logical claimant across attempts; only the slot
+  // owner changes between releases.
+  const token = generateToken();
   if (!buffer) {
     // Mollifier disabled / buffer construction failed. Fall open —
     // caller proceeds with the trigger pipeline (PG unique constraint
     // backstop). Without the claim machinery the race-window scenarios
-    // from the plan doc revert to today's behaviour.
-    return { kind: "claimed" };
+    // from the plan doc revert to today's behaviour. Token is still
+    // generated so callers don't have to branch on the "no buffer" case
+    // — publish/release become buffer-null no-ops downstream.
+    return { kind: "claimed", token };
   }
   const ttlSeconds = input.ttlSeconds ?? DEFAULT_CLAIM_TTL_SECONDS;
   const safetyNetMs = input.safetyNetMs ?? DEFAULT_CLAIM_WAIT_MS;
@@ -74,17 +90,17 @@ export async function claimOrAwait(input: ClaimOrAwaitInput): Promise<ClaimOrAwa
   // a prior burst).
   let result: IdempotencyClaimResult;
   try {
-    result = await buffer.claimIdempotency({ ...lookupInput, ttlSeconds });
+    result = await buffer.claimIdempotency({ ...lookupInput, token, ttlSeconds });
   } catch (err) {
     logger.warn("idempotency claim failed (fail-open)", {
       envId: input.envId,
       taskIdentifier: input.taskIdentifier,
       err: err instanceof Error ? err.message : String(err),
     });
-    return { kind: "claimed" };
+    return { kind: "claimed", token };
   }
 
-  if (result.kind === "claimed") return { kind: "claimed" };
+  if (result.kind === "claimed") return { kind: "claimed", token };
   if (result.kind === "resolved") return result;
 
   // result.kind === "pending" — wait/poll loop. May see the value flip
@@ -108,17 +124,19 @@ export async function claimOrAwait(input: ClaimOrAwaitInput): Promise<ClaimOrAwa
 
     if (current === null) {
       // Claimant released on error. Re-attempt the claim — one of the
-      // waiters will win, the rest see "pending" again.
+      // waiters will win, the rest see "pending" again. Reuse our token:
+      // we're still the same logical claimant, just contending for a
+      // freshly empty slot.
       try {
-        const retry = await buffer.claimIdempotency({ ...lookupInput, ttlSeconds });
-        if (retry.kind === "claimed") return { kind: "claimed" };
+        const retry = await buffer.claimIdempotency({ ...lookupInput, token, ttlSeconds });
+        if (retry.kind === "claimed") return { kind: "claimed", token };
         if (retry.kind === "resolved") return retry;
         // "pending" again → keep polling.
       } catch (err) {
         logger.warn("idempotency claim retry failed", {
           err: err instanceof Error ? err.message : String(err),
         });
-        return { kind: "claimed" };
+        return { kind: "claimed", token };
       }
       continue;
     }
@@ -136,6 +154,10 @@ export async function publishClaim(input: {
   envId: string;
   taskIdentifier: string;
   idempotencyKey: string;
+  // Ownership token from the `claimed` outcome. Buffer compare-and-sets
+  // on this so a publish from a stale claimant (TTL expired, another
+  // claimant moved in) is a no-op rather than overwriting their claim.
+  token: string;
   runId: string;
   ttlSeconds?: number;
   buffer?: MollifierBuffer | null;
@@ -148,6 +170,7 @@ export async function publishClaim(input: {
       envId: input.envId,
       taskIdentifier: input.taskIdentifier,
       idempotencyKey: input.idempotencyKey,
+      token: input.token,
       runId: input.runId,
       ttlSeconds,
     });
@@ -166,6 +189,10 @@ export async function releaseClaim(input: {
   envId: string;
   taskIdentifier: string;
   idempotencyKey: string;
+  // Ownership token from the `claimed` outcome. Buffer compare-and-
+  // deletes on this so a release from a stale claimant whose TTL
+  // expired can't wipe a new owner's claim.
+  token: string;
   buffer?: MollifierBuffer | null;
 }): Promise<void> {
   const buffer = input.buffer === undefined ? getMollifierBuffer() : input.buffer;
@@ -175,6 +202,7 @@ export async function releaseClaim(input: {
       envId: input.envId,
       taskIdentifier: input.taskIdentifier,
       idempotencyKey: input.idempotencyKey,
+      token: input.token,
     });
   } catch (err) {
     logger.warn("idempotency claim release failed", {

apps/webapp/test/mollifierIdempotencyClaim.test.ts

@@ -60,8 +60,12 @@ const baseInput = {
 describe("claimOrAwait", () => {
   it("returns 'claimed' for the first caller — empty key wins SETNX", async () => {
     const { buffer } = makeBuffer({ value: null });
-    const outcome = await claimOrAwait({ ...baseInput, buffer });
-    expect(outcome).toEqual({ kind: "claimed" });
+    const outcome = await claimOrAwait({
+      ...baseInput,
+      buffer,
+      generateToken: () => "token-1",
+    });
+    expect(outcome).toEqual({ kind: "claimed", token: "token-1" });
   });
 
   it("returns 'resolved' immediately when the key already holds a runId", async () => {
@@ -117,6 +121,7 @@ describe("claimOrAwait", () => {
     const outcome = await claimOrAwait({
       ...baseInput,
       buffer,
+      generateToken: () => "token-retry",
       now: () => nowValue,
       sleep: async (ms) => {
         nowValue += ms;
@@ -127,12 +132,16 @@ describe("claimOrAwait", () => {
       safetyNetMs: 1000,
       pollStepMs: 25,
     });
-    expect(outcome).toEqual({ kind: "claimed" });
+    expect(outcome).toEqual({ kind: "claimed", token: "token-retry" });
   });
 
   it("fails open with 'claimed' when buffer is null (mollifier disabled)", async () => {
-    const outcome = await claimOrAwait({ ...baseInput, buffer: null });
-    expect(outcome).toEqual({ kind: "claimed" });
+    const outcome = await claimOrAwait({
+      ...baseInput,
+      buffer: null,
+      generateToken: () => "token-fallopen-null",
+    });
+    expect(outcome).toEqual({ kind: "claimed", token: "token-fallopen-null" });
   });
 
   it("fails open with 'claimed' if buffer.claimIdempotency throws (Redis down)", async () => {
@@ -141,8 +150,12 @@ describe("claimOrAwait", () => {
         throw new Error("ECONNREFUSED");
       }),
     } as unknown as MollifierBuffer;
-    const outcome = await claimOrAwait({ ...baseInput, buffer });
-    expect(outcome).toEqual({ kind: "claimed" });
+    const outcome = await claimOrAwait({
+      ...baseInput,
+      buffer,
+      generateToken: () => "token-fallopen-throw",
+    });
+    expect(outcome).toEqual({ kind: "claimed", token: "token-fallopen-throw" });
   });
 
   it("respects an aborted signal during the wait loop", async () => {
@@ -170,14 +183,14 @@ describe("claimOrAwait", () => {
 describe("publishClaim", () => {
   it("writes the runId to the claim key", async () => {
     const { buffer, state } = makeBuffer({ value: "pending" });
-    await publishClaim({ ...baseInput, runId: "run_X", buffer });
+    await publishClaim({ ...baseInput, token: "owner-token", runId: "run_X", buffer });
     expect(state.value).toBe("run_X");
     expect(buffer.publishClaim).toHaveBeenCalledOnce();
   });
 
   it("no-op when buffer is null", async () => {
     await expect(
-      publishClaim({ ...baseInput, runId: "run_X", buffer: null }),
+      publishClaim({ ...baseInput, token: "owner-token", runId: "run_X", buffer: null }),
     ).resolves.toBeUndefined();
   });
 
@@ -188,19 +201,68 @@ describe("publishClaim", () => {
       }),
     } as unknown as MollifierBuffer;
     await expect(
-      publishClaim({ ...baseInput, runId: "run_X", buffer }),
+      publishClaim({ ...baseInput, token: "owner-token", runId: "run_X", buffer }),
     ).resolves.toBeUndefined();
   });
 });
 
 describe("releaseClaim", () => {
   it("DELs the claim so waiters can re-acquire", async () => {
     const { buffer, state } = makeBuffer({ value: "pending" });
-    await releaseClaim({ ...baseInput, buffer });
+    await releaseClaim({ ...baseInput, token: "owner-token", buffer });
     expect(state.value).toBeNull();
   });
 
   it("no-op when buffer is null", async () => {
-    await expect(releaseClaim({ ...baseInput, buffer: null })).resolves.toBeUndefined();
+    await expect(releaseClaim({ ...baseInput, token: "owner-token", buffer: null })).resolves.toBeUndefined();
+  });
+});
+
+// End-to-end: the token from `claimOrAwait`'s `claimed` outcome must
+// reach `buffer.claimIdempotency` and round-trip through publishClaim /
+// releaseClaim. Without this the compare-and-act ownership protection
+// in the buffer is bypassed and the stale-claimant hazard returns.
+describe("claim ownership token wiring", () => {
+  it("threads the token from claimOrAwait into buffer.claimIdempotency", async () => {
+    const { buffer } = makeBuffer({ value: null });
+    const outcome = await claimOrAwait({
+      ...baseInput,
+      buffer,
+      generateToken: () => "owner-token-xyz",
+    });
+    expect(outcome).toEqual({ kind: "claimed", token: "owner-token-xyz" });
+    expect(buffer.claimIdempotency).toHaveBeenCalledWith({
+      ...baseInput,
+      token: "owner-token-xyz",
+      ttlSeconds: 30,
+    });
+  });
+
+  it("threads the token from publishClaim into buffer.publishClaim", async () => {
+    const { buffer } = makeBuffer({ value: "pending" });
+    await publishClaim({
+      ...baseInput,
+      token: "owner-token-xyz",
+      runId: "run_X",
+      buffer,
+    });
+    expect(buffer.publishClaim).toHaveBeenCalledWith(
+      expect.objectContaining({
+        token: "owner-token-xyz",
+        runId: "run_X",
+      }),
+    );
+  });
+
+  it("threads the token from releaseClaim into buffer.releaseClaim", async () => {
+    const { buffer } = makeBuffer({ value: "pending" });
+    await releaseClaim({
+      ...baseInput,
+      token: "owner-token-xyz",
+      buffer,
+    });
+    expect(buffer.releaseClaim).toHaveBeenCalledWith(
+      expect.objectContaining({ token: "owner-token-xyz" }),
+    );
   });
 });