fix(sdk): make pending-approval consumption atomic

.changeset/approval-atomicity.md

@@ -0,0 +1,20 @@
+---
+"@executor-js/sdk": patch
+---
+
+fix: make pending-approval consumption atomic
+
+`PendingApprovalStore.consume` previously read the record and deleted it as
+two separate operations. Two concurrent resumes (a double-click, a client
+retry, or two hosts racing the same approval) could both read the record
+before either deleted it, and both would execute the approved tool call —
+duplicated side effects from a single approval.
+
+Consumption now goes through a new `BlobStore.compareAndDelete` primitive
+with a single-winner guarantee: exactly one concurrent consumer observes the
+record as present-and-removed; everyone else observes it as absent. The
+in-memory store implements it as a synchronous Map operation (atomic in JS's
+single-threaded model); the FumaDB-backed store implements it as
+get+delete inside the serializing transaction the driver already provides
+(libSQL/Postgres BEGIN/COMMIT). The approval's expiry and corrupt-record
+semantics are unchanged.

packages/core/sdk/src/approval-atomicity.test.ts

@@ -0,0 +1,128 @@
+import { describe, expect, it } from "@effect/vitest";
+import { Effect, Predicate } from "effect";
+
+import { makeInMemoryBlobStore, type BlobStore } from "./blob";
+import {
+  makePendingApprovalStore,
+  PENDING_APPROVAL_TTL_MS,
+  type PendingApproval,
+} from "./pending-approval";
+
+// ---------------------------------------------------------------------------
+// Focused tests — pending-approval consume atomicity, deterministic layer.
+//
+// These tests pin the exactly-once consume invariant deterministically,
+// plus the expiry/corrupt semantics required to survive the consume
+// restructure. A companion fast-check property (concurrent consumes ⇒
+// exactly one non-null) lives elsewhere.
+//
+// NOTE on test infrastructure:
+// - it.effect runs under @effect/vitest's TestContext scheduler, where
+//   Effect.sleep never advances — any async boundary deadlocks the test.
+//   Sync-only effects work; async ones must go through Effect.runPromise in
+//   a plain vitest test.
+// - The concurrency proof needs the real scheduler AND an explicit
+//   read-barrier: without one, the synchronous Map store completes each
+//   consume before the next fiber starts, so the double-resume race can
+//   never manifest and the test would be vacuous as a concurrency proof.
+// ---------------------------------------------------------------------------
+
+const approval = (overrides?: Partial<PendingApproval>): PendingApproval => ({
+  executionId: "exec_1",
+  artifactId: "art_1",
+  code: 'return await tools.github.user.main.issues.create({"title":"x"})',
+  address: "github.user.main.issues.create",
+  expiresAt: Date.now() + PENDING_APPROVAL_TTL_MS,
+  ...overrides,
+});
+
+describe("approval consume atomicity (compareAndDelete gate)", () => {
+  it("exactly one of N concurrent consumes wins (single-winner invariant)", async () => {
+    const N = 8;
+    const backing = makeInMemoryBlobStore();
+    // Promise-latch barrier: every fiber increments the arrival count after
+    // its read and parks on the shared promise, which resolves only when
+    // all N have arrived — every read happens before any delete. Pure JS
+    // promise semantics, immune to Effect scheduler/runtime differences.
+    let arrived = 0;
+    let releaseAll!: () => void;
+    const allArrived = new Promise<void>((resolve) => {
+      releaseAll = resolve;
+    });
+    const barrier: BlobStore = {
+      ...backing,
+      get: (ns, key) =>
+        backing.get(ns, key).pipe(
+          Effect.tap(() =>
+            Effect.sync(() => {
+              arrived++;
+              if (arrived === N) releaseAll();
+            }),
+          ),
+          // tap (not andThen — that would discard the payload): park until
+          // every fiber has read, then pass the payload through untouched.
+          Effect.tap(() => Effect.promise(() => allArrived)),
+        ),
+    };
+
+    const outcome = await Effect.runPromise(
+      Effect.gen(function* () {
+        const store = makePendingApprovalStore(barrier, "u:t:s");
+        yield* store.put(approval());
+
+        const results = yield* Effect.all(
+          Array.from({ length: N }, () => store.consume("exec_1")),
+          { concurrency: "unbounded" },
+        );
+        const winners = results.filter(Predicate.isNotNull);
+
+        // Post-condition: the record is gone for everyone.
+        const after = yield* store.consume("exec_1");
+        return { winners: winners.map((w) => w.address), after };
+      }),
+    );
+
+    expect(outcome.winners.length).toBe(1);
+    expect(outcome.winners[0]).toBe("github.user.main.issues.create");
+    expect(outcome.after).toBeNull();
+  });
+
+  it.effect("a replayed consume after a win reads absent", () =>
+    Effect.gen(function* () {
+      const store = makePendingApprovalStore(makeInMemoryBlobStore(), "u:t:s");
+      yield* store.put(approval());
+      expect(yield* store.consume("exec_1")).not.toBeNull();
+      expect(yield* store.consume("exec_1")).toBeNull();
+      expect(yield* store.consume("exec_1")).toBeNull();
+    }),
+  );
+
+  it.effect("expired records are consumed-and-dropped (never retried)", () =>
+    Effect.gen(function* () {
+      const blobs = makeInMemoryBlobStore();
+      let now = 1_000_000;
+      const store = makePendingApprovalStore(blobs, "u:t:s", () => now);
+      yield* store.put(approval({ expiresAt: now + 10 }));
+
+      now += 100; // expires
+      expect(yield* store.consume("exec_1")).toBeNull();
+
+      // The record is gone — rolling the clock back does not resurrect it.
+      now = 1_000_000;
+      expect(yield* store.consume("exec_1")).toBeNull();
+    }),
+  );
+
+  it.effect("corrupt records are consumed-and-dropped (never surfaced)", () =>
+    Effect.gen(function* () {
+      const blobs = makeInMemoryBlobStore();
+      yield* blobs.put("u:t:s/@pending-approval", "exec_1", "not-json{{");
+
+      const store = makePendingApprovalStore(blobs, "u:t:s");
+      expect(yield* store.consume("exec_1")).toBeNull();
+
+      // Gone for good — a second consume finds nothing.
+      expect(yield* store.consume("exec_1")).toBeNull();
+    }),
+  );
+});

packages/core/sdk/src/blob.ts

@@ -38,6 +38,26 @@ export interface BlobStore {
   ) => Effect.Effect<void, StorageError>;
   readonly delete: (namespace: string, key: string) => Effect.Effect<void, StorageError>;
   readonly has: (namespace: string, key: string) => Effect.Effect<boolean, StorageError>;
+  /**
+   * Atomically delete a record IF it exists. Returns true iff this caller's
+   * delete removed an existing record; false iff it was already absent.
+   *
+   * The single-winner invariant: exactly one concurrent caller observes true
+   * for a given (namespace, key); everyone else observes false, and the
+   * post-condition is that the record is absent for all callers. There is no
+   * read-your-undefined window — a caller that observed true is guaranteed
+   * the record was present at deletion time, and no other caller can observe
+   * true for the same key afterwards.
+   *
+   * Implementations MUST NOT do check-then-act across two statements:
+   * either a single atomic statement with a rows-affected/count check, or
+   * get+delete inside a serializing transaction (libSQL/Postgres
+   * `fuma.transaction`), or a single synchronous Map op (in-memory).
+   */
+  readonly compareAndDelete: (
+    namespace: string,
+    key: string,
+  ) => Effect.Effect<boolean, StorageError>;
 }
 
 export interface PluginBlobStore {
@@ -154,6 +174,16 @@ export const makeInMemoryBlobStore = (): BlobStore => {
         store.delete(k(ns, key));
       }),
     has: (ns, key) => Effect.sync(() => store.has(k(ns, key))),
+    // Atomic by construction: a synchronous Map has+delete runs to completion
+    // without yielding, so no other fiber can interleave between the check
+    // and the delete in JS's single-threaded model.
+    compareAndDelete: (ns, key) =>
+      Effect.sync(() => {
+        const id = k(ns, key);
+        if (!store.has(id)) return false;
+        store.delete(id);
+        return true;
+      }),
   };
 };
 
@@ -243,6 +273,37 @@ export const makeFumaBlobStore = (fuma: IFumaClient): BlobStore => ({
           (cause) => new StorageError({ message: "FumaDB blob operation failed", cause }),
         ),
       ),
+  compareAndDelete: (namespace, key) =>
+    fuma
+      .transaction(
+        Effect.gen(function* () {
+          const id = blobId(namespace, key);
+          // Read inside the transaction: on libSQL/Postgres, `fuma.transaction`
+          // runs real BEGIN/COMMIT, so concurrent transactions serialize and
+          // no other fiber can interleave between this get and the delete —
+          // exactly one caller observes a present row, everyone else sees
+          // absent-after-commit. FumaDB's query builder discards rows-affected
+          // counts (deleteMany -> Promise<void>) and exposes no raw driver
+          // handle, so a single `DELETE ... RETURNING` statement is not
+          // reachable through this abstraction without a cross-host driver
+          // change; the serializing transaction is the equivalent guarantee
+          // here. (The in-memory store's synchronous Map op is the atomic
+          // counterpart.)
+          const row = (yield* fuma.use("blob.cad.find", (db) =>
+            db.findFirst("blob", { where: (b) => b("id", "=", id) }),
+          )) as BlobRow | null;
+          if (row === null) return false;
+          yield* fuma.use("blob.cad.delete", (db) =>
+            db.deleteMany("blob", { where: (b) => b("id", "=", id) }),
+          );
+          return true;
+        }),
+      )
+      .pipe(
+        Effect.mapError(
+          (cause) => new StorageError({ message: "FumaDB blob operation failed", cause }),
+        ),
+      ),
   has: (namespace, key) =>
     fuma
       .use("blob.has", (db) =>

packages/core/sdk/src/pending-approval.ts

@@ -18,8 +18,9 @@
 // to resuming the fiber, without requiring the fiber to still exist.
 //
 // Records live in the existing owner-scoped `blob` table (no new table, no
-// migration). They are single-use: consumed on resume, so one approval authorizes
-// exactly one invocation and a replayed resume cannot re-run the call.
+// migration). They are single-use: consumed atomically (compareAndDelete) on
+// resume, so one approval authorizes exactly one invocation and NEITHER a
+// replayed resume NOR two concurrent resumes can re-run the call.
 // ---------------------------------------------------------------------------
 
 import { Effect, Option, Schema } from "effect";
@@ -93,11 +94,24 @@ export const makePendingApprovalStore = (
 
     consume: (executionId) =>
       Effect.gen(function* () {
+        // Read the payload first — compareAndDelete returns only a boolean,
+        // so the winner needs the record's value to validate and return.
+        // Reading before the gate is safe: a concurrent loser may read the
+        // same payload but will fail the compareAndDelete gate below and
+        // never return it. The ordering that matters is the DELETE's
+        // single-winner guarantee, not the read's.
         const raw = yield* blobs.get(namespace, executionId);
         if (raw === null) return null;
-        // Consume before validating: a record we are about to reject is a record
-        // nobody should be able to retry against.
-        yield* blobs.delete(namespace, executionId);
+        // Atomic single-winner delete: exactly one concurrent consumer
+        // observes true (the record existed and was removed); everyone else
+        // observes false. This closes the get→delete race that previously
+        // let two concurrent resumes both read a record before either
+        // deleted it — duplicated side effects.
+        const removed = yield* blobs.compareAndDelete(namespace, executionId);
+        if (!removed) return null;
+        // Consume before validating (preserved): a record we are about to
+        // reject is a record nobody should be able to retry against — the
+        // atomic delete already removed it, so no other consumer can see it.
         const decoded = decodePendingApproval(raw);
         if (Option.isNone(decoded)) return null;
         const approval = decoded.value;