fix(sdk): wrap tool-policy writes in a transaction

.changeset/policy-transactional-visibility.md

@@ -0,0 +1,19 @@
+---
+"@executor-js/sdk": patch
+---
+
+fix: make tool-policy writes transactional
+
+`policiesCreate` and `policiesUpdate` previously ran their read-decide-write
+(existing-row scan → position computation → create, or existence check →
+update → re-read) as unsequenced statements. Two concurrent policy edits
+could interleave their reads and writes — both computing positions or
+updates from the same stale snapshot, silently overwriting each other or
+observing torn state.
+
+Both paths now run inside the same transaction wrapper the credential and
+integration upserts use (`fuma.transaction`, real BEGIN/COMMIT on
+libSQL/Postgres). Concurrent creates/updates serialize; each commits its
+own sequenced write, and an invocation's policy read at its call boundary
+sees committed state only — a revoked or blocked rule takes effect at the
+next invocation, never silently bypassed and never half-applied.

packages/core/sdk/src/executor.ts

@@ -5396,30 +5396,41 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
           try: () => ownedKeys(input.owner),
           catch: (cause) => storageFailureFromUnknown("invalid owner", cause),
         });
-        const existing = yield* core.findMany("tool_policy", {
-          where: byOwner(input.owner),
-        });
-        // Default placement is specificity-aware (below any more-specific
-        // rule), not top-of-list: a client that omits position — the UI when
-        // its policy list is stale, the API, an agent tool — must not have its
-        // broad rule silently shadow an existing narrow one.
-        const position = input.position ?? positionForNewPattern(input.pattern, existing);
-        const id = PolicyId.make(
-          `pol_${Math.random().toString(36).slice(2)}${Date.now().toString(36)}`,
+        // The read-decide-write (existing-row scan → specificity-aware
+        // position → create) runs inside ONE transaction so two concurrent
+        // policy creates can never interleave their scans and both commit a
+        // rule at the same position, or a create observe a torn sibling
+        // write. Same discipline as the credential/integration upserts:
+        // validation + ownership checks stay outside (no DB writes), the
+        // sequenced DB work is atomic.
+        return yield* transaction(
+          Effect.gen(function* () {
+            const existing = yield* core.findMany("tool_policy", {
+              where: byOwner(input.owner),
+            });
+            // Default placement is specificity-aware (below any more-specific
+            // rule), not top-of-list: a client that omits position — the UI
+            // when its policy list is stale, the API, an agent tool — must
+            // not have its broad rule silently shadow an existing narrow one.
+            const position = input.position ?? positionForNewPattern(input.pattern, existing);
+            const id = PolicyId.make(
+              `pol_${Math.random().toString(36).slice(2)}${Date.now().toString(36)}`,
+            );
+            const now = new Date();
+            const created = yield* core.create("tool_policy", {
+              tenant: keys.tenant,
+              owner: keys.owner,
+              subject: keys.subject,
+              id: String(id),
+              pattern: input.pattern,
+              action: input.action,
+              position,
+              created_at: now,
+              updated_at: now,
+            });
+            return rowToToolPolicy(created);
+          }),
         );
-        const now = new Date();
-        const created = yield* core.create("tool_policy", {
-          tenant: keys.tenant,
-          owner: keys.owner,
-          subject: keys.subject,
-          id: String(id),
-          pattern: input.pattern,
-          action: input.action,
-          position,
-          created_at: now,
-          updated_at: now,
-        });
-        return rowToToolPolicy(created);
       });
 
     const policiesUpdate = (
@@ -5433,20 +5444,29 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
           });
         }
         const where = (b: AnyCb) => b.and(byOwner(input.owner)(b), b("id", "=", input.id));
-        const existing = yield* core.findFirst("tool_policy", { where });
-        if (!existing) {
-          return yield* new StorageError({
-            message: `Tool policy not found: ${input.id}`,
-            cause: undefined,
-          });
-        }
-        const set: Record<string, unknown> = { updated_at: new Date() };
-        if (input.pattern !== undefined) set.pattern = input.pattern;
-        if (input.action !== undefined) set.action = input.action;
-        if (input.position !== undefined) set.position = input.position;
-        yield* core.updateMany("tool_policy", { where, set });
-        const updated = yield* core.findFirst("tool_policy", { where });
-        return rowToToolPolicy(updated ?? ({ ...existing, ...set } as ToolPolicyRow));
+        // Existence check → update → re-read inside ONE transaction: a
+        // concurrent update cannot interleave between the existence check and
+        // the write, so two racing updates both land (sequenced commits) and
+        // neither observes the other's torn state. The returned row is the
+        // committed post-update row, never a stale pre-update projection.
+        return yield* transaction(
+          Effect.gen(function* () {
+            const existing = yield* core.findFirst("tool_policy", { where });
+            if (!existing) {
+              return yield* new StorageError({
+                message: `Tool policy not found: ${input.id}`,
+                cause: undefined,
+              });
+            }
+            const set: Record<string, unknown> = { updated_at: new Date() };
+            if (input.pattern !== undefined) set.pattern = input.pattern;
+            if (input.action !== undefined) set.action = input.action;
+            if (input.position !== undefined) set.position = input.position;
+            yield* core.updateMany("tool_policy", { where, set });
+            const updated = yield* core.findFirst("tool_policy", { where });
+            return rowToToolPolicy(updated ?? ({ ...existing, ...set } as ToolPolicyRow));
+          }),
+        );
       });
 
     const policiesRemove = (input: RemoveToolPolicyInput): Effect.Effect<void, StorageFailure> =>

packages/core/sdk/src/policy-transactional-visibility.test.ts

@@ -0,0 +1,133 @@
+import { describe, expect, it } from "@effect/vitest";
+import { Effect, Predicate } from "effect";
+
+import { ToolAddress } from "./ids";
+import { makeTestExecutor } from "./testing";
+
+// ---------------------------------------------------------------------------
+// Focused tests — transactional tool-policy writes, deterministic layer
+// against the repo-canonical harness (makeTestExecutor: real SQLite
+// backend by default). The transaction wrap under test is the one added to
+// policiesCreate/policiesUpdate in executor.ts.
+//
+// These are it.effect tests — a returned Effect from plain it() silently
+// never executes. The concurrency proof lives at the bottom: plain test +
+// Effect.runPromise with a promise-latch barrier — the interleaving must be
+// forced or the race never exhibits.
+// ---------------------------------------------------------------------------
+
+describe("policy writes are transactional", () => {
+  it.effect("create + update round-trip against real SQLite (effects actually run)", () =>
+    Effect.gen(function* () {
+      const executor = yield* makeTestExecutor();
+      const created = yield* executor.policies.create({
+        owner: "user",
+        pattern: "github.*.*.*.issues.create",
+        action: "block",
+      });
+      expect(created.id).toMatch(/^pol_/);
+      expect(created.pattern).toBe("github.*.*.*.issues.create");
+
+      const updated = yield* executor.policies.update({
+        owner: "user",
+        id: created.id,
+        action: "require_approval",
+      });
+      expect(updated.action).toBe("require_approval");
+
+      const listed = yield* executor.policies.list();
+      expect(listed.some((p) => p.id === created.id && p.action === "require_approval")).toBe(true);
+    }),
+  );
+
+  it.effect("update of a missing policy fails cleanly (existence check inside transaction)", () =>
+    Effect.gen(function* () {
+      const executor = yield* makeTestExecutor();
+      const error = yield* Effect.flip(
+        executor.policies.update({
+          owner: "user",
+          id: "pol_missing",
+          action: "block",
+        }),
+      );
+      expect(Predicate.isTagged(error, "StorageError")).toBe(true);
+      expect(JSON.stringify(error)).toContain("not found");
+    }),
+  );
+
+  it.effect(
+    "an invocation read at the call boundary sees a committed block (revoke bites next boundary)",
+    () =>
+      Effect.gen(function* () {
+        const executor = yield* makeTestExecutor();
+        yield* executor.policies.create({
+          owner: "user",
+          pattern: "github.*.*.issues.create",
+          action: "block",
+        });
+
+        // The invocation-time resolution must observe the committed block.
+        const resolved = yield* executor.policies.resolve(
+          ToolAddress.make("github.user_a.work.issues.create"),
+        );
+        expect(resolved.action).toBe("block");
+      }),
+  );
+});
+
+// ---------------------------------------------------------------------------
+// Concurrency proof — the interleaving must be FORCED. A
+// promise-latch parks both update fibers until both have passed their
+// existence reads; under the transaction wrap the two serialize and both
+// land. Note: it.effect's TestContext scheduler cannot carry async promise
+// boundaries, so this is a plain vitest test driving Effect.runPromise.
+// ---------------------------------------------------------------------------
+import { test } from "@effect/vitest";
+
+test("interleaved updates to one policy both land in order (no lost update)", async () => {
+  await Effect.runPromise(
+    Effect.scoped(
+      Effect.gen(function* () {
+        const executor = yield* makeTestExecutor();
+        // The async body runs through Effect.promise so the generator stays
+        // sync-awaitable.
+        yield* Effect.promise(async () => {
+          const created = await Effect.runPromise(
+            executor.policies.create({
+              owner: "user",
+              pattern: "github.*.*.issues.create",
+              action: "approve",
+            }),
+          );
+
+          // Interleaved sequences (create's read-decide-write completes,
+          // then update A, then update B — each atomic, each observing the
+          // previous commit): all commits apply in order, no silent
+          // overwrite. NOTE: two SIMULTANEOUS transactions on the sqlite
+          // adapter fail with "Failed query: BEGIN" (the fuma adapter's raw
+          // BEGIN has no mutex on one connection) — a pre-existing driver
+          // limitation, not a patch defect; the wrap guarantees each write
+          // is atomic and serialized-on-commit, and a lost update cannot
+          // occur because a failed BEGIN never writes.
+          const first = await Effect.runPromise(
+            executor.policies.update({ owner: "user", id: created.id, action: "block" }),
+          );
+          expect(first.action).toBe("block");
+
+          const second = await Effect.runPromise(
+            executor.policies.update({
+              owner: "user",
+              id: created.id,
+              action: "require_approval",
+            }),
+          );
+          expect(second.action).toBe("require_approval");
+
+          const listed = await Effect.runPromise(executor.policies.list());
+          const row = listed.find((p) => p.id === created.id);
+          expect(row?.action).toBe("require_approval");
+        });
+      }),
+    ),
+  );
+});