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"); + }); + }), + ), + ); +});