From 016ae038c31ef074691741bc330909de995d0690 Mon Sep 17 00:00:00 2001 From: pt-act <211776491+pt-act@users.noreply.github.com> Date: Sun, 30 Aug 2026 16:01:45 +0100 Subject: [PATCH 1/2] fix(sdk): wrap tool-policy writes in a transaction --- .changeset/policy-transactional-visibility.md | 19 +++ packages/core/sdk/src/executor.ts | 94 ++++++++----- .../policy-transactional-visibility.test.ts | 133 ++++++++++++++++++ 3 files changed, 209 insertions(+), 37 deletions(-) create mode 100644 .changeset/policy-transactional-visibility.md create mode 100644 packages/core/sdk/src/policy-transactional-visibility.test.ts diff --git a/.changeset/policy-transactional-visibility.md b/.changeset/policy-transactional-visibility.md new file mode 100644 index 0000000000..1391d77aa3 --- /dev/null +++ b/.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. diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index aba6b674c2..0592a6ad3d 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -5396,30 +5396,41 @@ export const createExecutor = 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 = 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 = { 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 = { 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 => diff --git a/packages/core/sdk/src/policy-transactional-visibility.test.ts b/packages/core/sdk/src/policy-transactional-visibility.test.ts new file mode 100644 index 0000000000..525813fc54 --- /dev/null +++ b/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"); + }); + }), + ), + ); +}); From f6cc5dd4bbbe4709c6cb3c2ce162cc018cae2377 Mon Sep 17 00:00:00 2001 From: pt-act Date: Wed, 2 Sep 2026 09:02:12 +0100 Subject: [PATCH 2/2] apply review: discriminating test, one-line changeset, trimmed comments - delete policy-transactional-visibility.test.ts (passed with main's executor.ts swapped in - not discriminating; sequential awaits are no concurrency proof) - add a real concurrent-creates case to policies.test.ts (verified to fail on main, pass here) - changeset to one sentence per repo norm - shorten the two executor.ts comment blocks to one line each --- .changeset/policy-transactional-visibility.md | 16 +-- packages/core/sdk/src/executor.ts | 14 +- packages/core/sdk/src/policies.test.ts | 17 +++ .../policy-transactional-visibility.test.ts | 133 ------------------ 4 files changed, 20 insertions(+), 160 deletions(-) delete mode 100644 packages/core/sdk/src/policy-transactional-visibility.test.ts diff --git a/.changeset/policy-transactional-visibility.md b/.changeset/policy-transactional-visibility.md index 1391d77aa3..d238d09683 100644 --- a/.changeset/policy-transactional-visibility.md +++ b/.changeset/policy-transactional-visibility.md @@ -2,18 +2,4 @@ "@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. +Wrap tool-policy create and update in a transaction so concurrent edits can no longer read the same snapshot and commit duplicate positions or overwrite each other. diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 0592a6ad3d..51c58b5cd5 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -5396,13 +5396,7 @@ export const createExecutor = ownedKeys(input.owner), catch: (cause) => storageFailureFromUnknown("invalid owner", cause), }); - // 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. + // Scan → position → insert runs atomically so concurrent creates cannot commit duplicate positions. return yield* transaction( Effect.gen(function* () { const existing = yield* core.findMany("tool_policy", { @@ -5444,11 +5438,7 @@ export const createExecutor = b.and(byOwner(input.owner)(b), b("id", "=", input.id)); - // 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. + // Existence check, write, and re-read commit together. return yield* transaction( Effect.gen(function* () { const existing = yield* core.findFirst("tool_policy", { where }); diff --git a/packages/core/sdk/src/policies.test.ts b/packages/core/sdk/src/policies.test.ts index beb9703c49..c05e634842 100644 --- a/packages/core/sdk/src/policies.test.ts +++ b/packages/core/sdk/src/policies.test.ts @@ -428,6 +428,23 @@ describe("executor.policies", () => { }), ); + it.live("concurrent creates of equally specific rules get distinct positions", () => + Effect.gen(function* () { + const executor = yield* setupExecutor(); + yield* Effect.all( + [ + executor.policies.create({ owner: "org", pattern: "vercel.dns.create", action: "block" }), + executor.policies.create({ owner: "org", pattern: "vercel.dns.delete", action: "block" }), + ], + { concurrency: "unbounded" }, + ); + + const rules = yield* executor.policies.list(); + expect(rules).toHaveLength(2); + expect(new Set(rules.map((r) => r.position)).size).toBe(2); + }), + ); + it.effect("create stores rules at the requested owner", () => Effect.gen(function* () { const executor = yield* setupExecutor(); diff --git a/packages/core/sdk/src/policy-transactional-visibility.test.ts b/packages/core/sdk/src/policy-transactional-visibility.test.ts deleted file mode 100644 index 525813fc54..0000000000 --- a/packages/core/sdk/src/policy-transactional-visibility.test.ts +++ /dev/null @@ -1,133 +0,0 @@ -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"); - }); - }), - ), - ); -});