From 8bbfdfec3f40653e87f669018c084013eee5345c Mon Sep 17 00:00:00 2001 From: Nathan Disidore Date: Mon, 17 Aug 2026 16:13:55 -0500 Subject: [PATCH 1/3] Migrate Confluence and Notion action resolution --- .../__tests__/gatekeeper-action.test.ts | 79 ++++++- .../backend-utils/src/gatekeeper-action.ts | 77 ++++++- packages/backend-utils/tsconfig.json | 2 +- .../__tests__/apply.test.ts | 201 ++++++++++++++++++ packages/gatekeeper-confluence/package.json | 1 + .../src/confluence-actions.ts | 151 ++++++++++--- .../gatekeeper-confluence/src/confluence.ts | 50 ++++- .../__tests__/notion-actions.test.ts | 106 +++++++++ packages/gatekeeper-notion/package.json | 1 + .../gatekeeper-notion/src/notion-actions.ts | 166 +++++++++++---- packages/gatekeeper-notion/src/notion.ts | 36 ++-- pnpm-lock.yaml | 28 ++- 12 files changed, 782 insertions(+), 116 deletions(-) create mode 100644 packages/gatekeeper-notion/__tests__/notion-actions.test.ts diff --git a/packages/backend-utils/__tests__/gatekeeper-action.test.ts b/packages/backend-utils/__tests__/gatekeeper-action.test.ts index a4b4f53e2..92bbf6e30 100644 --- a/packages/backend-utils/__tests__/gatekeeper-action.test.ts +++ b/packages/backend-utils/__tests__/gatekeeper-action.test.ts @@ -1,5 +1,82 @@ import { describe, expect, it } from "vitest"; -import { SerialTaskQueue } from "../src/gatekeeper-action"; +import { + InvalidationLog, + SerialTaskQueue, + displayReason, + validateApplyThroughArgs, +} from "../src/gatekeeper-action"; + +type Kv = ConstructorParameters[0]; + +// Minimal in-memory KV matching the slice of the DO storage API the log uses. +function makeKv(): Kv { + const map = new Map(); + return { + put: (k: string, v: unknown) => void map.set(k, v), + delete: (k: string) => void map.delete(k), + list: ({ prefix }: { prefix: string }) => + [...map.entries()].filter(([k]) => k.startsWith(prefix)) as [string, T][], + } as unknown as Kv; +} + +describe("validateApplyThroughArgs", () => { + it("deduplicates in-range vetoes", () => { + expect(validateApplyThroughArgs(5, [2, 2, 5])).toEqual(new Set([2, 5])); + }); + + it("rejects a non-positive or non-integer frontier", () => { + expect(() => validateApplyThroughArgs(0, [])).toThrow("Invalid action ID"); + expect(() => validateApplyThroughArgs(1.5, [])).toThrow("Invalid action ID"); + }); + + it("rejects vetoes above the frontier or out of the integer range", () => { + expect(() => validateApplyThroughArgs(3, [4])).toThrow("Invalid veto action ID"); + expect(() => validateApplyThroughArgs(3, [0])).toThrow("Invalid veto action ID"); + }); +}); + +describe("InvalidationLog", () => { + it("reports only entries attributed to the requested vetoes, ascending", () => { + const log = new InvalidationLog(makeKv()); + log.record(10, 3); + log.record(4, 3); + log.record(7, 5); + + expect(log.attributedTo(new Set([3]))).toEqual([ + { action: 4, invalidatedBy: 3 }, + { action: 10, invalidatedBy: 3 }, + ]); + }); + + it("prunes entries below the pending floor but keeps the current request's vetoes", () => { + const log = new InvalidationLog(makeKv()); + log.record(4, 3); + log.record(7, 5); + + log.prune(new Set([3]), 4); + + expect(log.attributedTo(new Set([3, 5]))).toEqual([ + { action: 4, invalidatedBy: 3 }, + { action: 7, invalidatedBy: 5 }, + ]); + + log.prune(new Set(), Infinity); + + expect(log.attributedTo(new Set([3, 5]))).toEqual([]); + }); +}); + +describe("displayReason", () => { + it("passes through an Error with a message", () => { + const error = new Error("page was deleted upstream"); + expect(displayReason(error, "fallback")).toBe(error); + }); + + it("wraps non-Error throws in the fallback text", () => { + expect(displayReason("oops", "Vendor could not apply this action").message) + .toBe("Vendor could not apply this action: oops"); + }); +}); // This package deliberately avoids the full Node type environment (see node-async-hooks.d.ts). // Tests run under vitest on Node, so type the small `process` surface used here locally. diff --git a/packages/backend-utils/src/gatekeeper-action.ts b/packages/backend-utils/src/gatekeeper-action.ts index 922bbd6a5..45fc3ed36 100644 --- a/packages/backend-utils/src/gatekeeper-action.ts +++ b/packages/backend-utils/src/gatekeeper-action.ts @@ -1,4 +1,9 @@ -// Shared helpers for gatekeepers implementing the `Gatekeeper` action contract. +// Shared helpers for gatekeepers implementing the `Gatekeeper` action contract. They cover the +// obligations every action-queueing gatekeeper repeats: serializing resolution methods, argument +// validation, durable attribution of veto-cascade invalidations (so repeated requests can +// re-report them), and the display-safe `stopped.reason` error. + +type Kv = DurableObjectStorage["kv"]; /** Runs asynchronous operations sequentially in submission order. */ export class SerialTaskQueue { @@ -11,3 +16,73 @@ export class SerialTaskQueue { return result; } } + +/** One `ApplyActionsThroughResult.invalidatedByVeto` entry. */ +export type VetoInvalidation = { action: number, invalidatedBy: number }; + +/** + * Validate an `applyActionsThrough(actionId, vetoes)` request before touching any state: the + * frontier must be a positive integer and every veto must be a positive integer at or below it. + * Returns the deduplicated veto set. + */ +export function validateApplyThroughArgs(actionId: number, vetoes: number[]): Set { + if (!Number.isSafeInteger(actionId) || actionId < 1) throw new TypeError("Invalid action ID."); + const result = new Set(); + for (const veto of vetoes) { + if (!Number.isSafeInteger(veto) || veto < 1 || veto > actionId) { + throw new TypeError("Invalid veto action ID."); + } + result.add(veto); + } + return result; +} + +/** + * Durable record of which veto invalidated which cascade-deleted action, kept in its own KV + * keyspace so a gatekeeper that deletes rejected records can still satisfy the contract's + * requirement that a repeated request re-report invalidations attributable to its vetoes. + */ +export class InvalidationLog { + #kv: Kv; + + constructor(kv: Kv) { + this.#kv = kv; + } + + /** Record that the veto of `vetoedBy` invalidated `invalidatedId`, for later re-reporting. */ + record(invalidatedId: number, vetoedBy: number): void { + this.#kv.put(`invalidation:${invalidatedId}`, vetoedBy); + } + + /** Persisted invalidations attributed to any of the given vetoed action IDs, ascending. */ + attributedTo(vetoes: Set): VetoInvalidation[] { + return [...this.#kv.list({ prefix: "invalidation:" })] + .map(([key, invalidatedBy]) => ({ action: Number(key.slice("invalidation:".length)), invalidatedBy })) + .filter(entry => vetoes.has(entry.invalidatedBy)) + .toSorted((a, b) => a.action - b.action); + } + + /** + * Drop entries no future request can attribute: a veto is only ever re-sent while some action at + * or below it is still undecided, so entries whose veto precedes every remaining undecided + * action (`pendingFloor`, `Infinity` when none remain) are unreachable. `keep` protects the + * current request's vetoes. + */ + prune(keep: Set, pendingFloor: number): void { + for (const [key, vetoedBy] of this.#kv.list({ prefix: "invalidation:" })) { + if (vetoedBy < pendingFloor && !keep.has(vetoedBy)) this.#kv.delete(key); + } + } +} + +/** + * The error a gatekeeper should report in `stopped.reason`. Only its `message` survives the RPC + * hop to the overseer, so it must stand alone as text the user can act on; apply errors already + * carry a specific, display-safe message, exactly as the legacy single-action path surfaced them. + * `fallback` describes the failure generically for non-Error throws with no message of their own. + */ +export function displayReason(error: unknown, fallback: string): Error { + return error instanceof Error && error.message + ? error + : new Error(`${fallback}: ${String(error)}`); +} diff --git a/packages/backend-utils/tsconfig.json b/packages/backend-utils/tsconfig.json index ae8d4640b..3b380fab4 100644 --- a/packages/backend-utils/tsconfig.json +++ b/packages/backend-utils/tsconfig.json @@ -2,7 +2,7 @@ "extends": "../../tsconfig.json", "compilerOptions": { "target": "ES2022", - "lib": ["ES2022"], + "lib": ["ES2023"], "module": "ESNext", "moduleResolution": "bundler", "types": ["@cloudflare/workers-types/experimental", "./src/node-async-hooks.d.ts"] diff --git a/packages/gatekeeper-confluence/__tests__/apply.test.ts b/packages/gatekeeper-confluence/__tests__/apply.test.ts index 7b0385dac..e670ce956 100644 --- a/packages/gatekeeper-confluence/__tests__/apply.test.ts +++ b/packages/gatekeeper-confluence/__tests__/apply.test.ts @@ -1,8 +1,10 @@ import { describe, expect, it } from "vitest"; import { ConfluenceStore, + applyStoredActionsThrough, applyStoredAction, revertStoredAction, + stageAction, type ConfluenceAction, } from "../src/confluence-actions"; import type { ConfluenceApi } from "../src/confluence-api"; @@ -114,6 +116,205 @@ describe("applyStoredAction", () => { }); }); +describe("applyStoredActionsThrough", () => { + it("skips sparse IDs and stops at the first failed action", async () => { + const { api, calls } = makeApi(); + const store = storeWith(api); + const first = stage(store, { type: "addComment", contentId: "first", text: "one" }); + const hole = stage(store, { type: "addComment", contentId: "hole", text: "two" }); + const failed = stage(store, { type: "addComment", contentId: "failed", text: "three" }); + const later = stage(store, { type: "addComment", contentId: "later", text: "four" }); + store.deleteAction(hole); + api.addComment = async (id: string, _storage: string, type: string) => { + calls.addComment.push({ id, type }); + if (id === "failed") throw new Error("safe failure"); + return { id: `comment-${id}` }; + }; + + const result = await applyStoredActionsThrough(store, later, []); + + expect(calls.addComment.map(call => call.id)).toEqual(["first", "failed"]); + expect(store.getAction(first)?.state).toBe("applied"); + expect(store.getAction(failed)?.state).toBe("pending"); + expect(store.getAction(later)?.state).toBe("pending"); + // The specific apply error is passed through so the user can resolve the problem. + expect(result.stopped).toMatchObject({ at: failed, reason: expect.any(Error) }); + expect(result.stopped?.reason.message).toBe("safe failure"); + }); + + it("does not re-apply earlier actions when retried after a stop", async () => { + const { api, calls } = makeApi(); + const store = storeWith(api); + const first = stage(store, { type: "addComment", contentId: "first", text: "one" }); + const flaky = stage(store, { type: "addComment", contentId: "flaky", text: "two" }); + let failOnce = true; + api.addComment = async (id: string, _storage: string, type: string) => { + calls.addComment.push({ id, type }); + if (id === "flaky" && failOnce) { + failOnce = false; + throw new Error("temporarily unavailable"); + } + return { id: `comment-${id}` }; + }; + + const stopped = await applyStoredActionsThrough(store, flaky, []); + const retry = await applyStoredActionsThrough(store, flaky, []); + + expect(stopped.stopped).toMatchObject({ at: flaky }); + expect(retry).toEqual({}); + expect(calls.addComment.map(call => call.id)).toEqual(["first", "flaky", "flaky"]); + expect(store.getAction(first)?.state).toBe("applied"); + expect(store.getAction(flaky)?.state).toBe("applied"); + }); + + it("never applies a staged action whose submission has not completed", async () => { + const { api, calls } = makeApi(); + const store = storeWith(api); + const staged = store.nextActionId(); + store.putAction({ + id: staged, action: { type: "addComment", contentId: "staged", text: "early" }, + state: "staged", submittedAt: staged, + }); + const pending = stage(store, { type: "addComment", contentId: "later", text: "late" }); + + const result = await applyStoredActionsThrough(store, pending, []); + + expect(result).toEqual({}); + expect(calls.addComment.map(call => call.id)).toEqual(["later"]); + expect(store.getAction(staged)?.state).toBe("staged"); + }); + + it("persists transitive invalidations and reports them again on retry", async () => { + const { api, calls } = makeApi(); + const store = storeWith(api); + const root = stage(store, { + type: "createContent", provisionalId: "~root", kind: "page", + parent: { type: "space", spaceKey: "ENG" }, title: "Root", status: "current", + }); + const edit = stage(store, { + type: "setTitle", contentId: "~root", title: "Edited", previousTitle: "Root", + }); + const child = stage(store, { + type: "createContent", provisionalId: "~child", kind: "page", + parent: { type: "page", parentId: "~root", spaceKey: "ENG" }, title: "Child", status: "current", + }); + const childEdit = stage(store, { + type: "addComment", contentId: "~child", text: "Comment", + }); + + const first = await applyStoredActionsThrough(store, root, [root]); + const retry = await applyStoredActionsThrough(store, root, [root]); + + expect(first.invalidatedByVeto).toEqual([ + { action: edit, invalidatedBy: root }, + { action: child, invalidatedBy: root }, + { action: childEdit, invalidatedBy: root }, + ]); + expect(retry.invalidatedByVeto).toEqual(first.invalidatedByVeto); + // Vetoed and invalidated records are deleted so read overlays recompute without them. + expect(store.getAction(root)).toBeUndefined(); + expect(store.getAction(childEdit)).toBeUndefined(); + expect(store.knowsProvisional("~root")).toBe(false); + expect(calls.addComment).toHaveLength(0); + }); + + it("makes the legacy single-action path throw for a cascade-invalidated action", async () => { + const { api } = makeApi(); + const store = storeWith(api); + const root = stage(store, { + type: "createContent", provisionalId: "~root", kind: "page", + parent: { type: "space", spaceKey: "ENG" }, title: "Root", status: "current", + }); + const edit = stage(store, { + type: "setTitle", contentId: "~root", title: "Edited", previousTitle: "Root", + }); + + await applyStoredActionsThrough(store, root, [root]); + + // An un-migrated overseer applying the orphan must see a failure, not a silent success. + await expect(applyStoredAction(store, edit)).rejects.toThrow(`Unknown action: ${edit}`); + }); + + it("ignores a veto of an already-applied action", async () => { + const { api } = makeApi(); + const store = storeWith(api); + const id = stage(store, { type: "addComment", contentId: "page", text: "Comment" }); + await applyStoredActionsThrough(store, id, []); + + const result = await applyStoredActionsThrough(store, id, [id]); + + expect(result).toEqual({}); + expect(store.getAction(id)?.state).toBe("applied"); + }); + + it("rejects an out-of-range veto before changing state", async () => { + const { api, calls } = makeApi(); + const store = storeWith(api); + const id = stage(store, { type: "addComment", contentId: "page", text: "Comment" }); + + await expect(applyStoredActionsThrough(store, id, [id + 1])) + .rejects.toThrow("Invalid veto action ID"); + + expect(store.getAction(id)?.state).toBe("pending"); + expect(calls.addComment).toHaveLength(0); + }); + + it("persists later vetoes before an earlier action fails", async () => { + const { api } = makeApi(); + const store = storeWith(api); + const failed = stage(store, { type: "addComment", contentId: "failed", text: "Comment" }); + const vetoed = stage(store, { + type: "createContent", provisionalId: "~root", kind: "page", + parent: { type: "space", spaceKey: "ENG" }, title: "Root", status: "current", + }); + const invalidated = stage(store, { + type: "setTitle", contentId: "~root", title: "Edited", previousTitle: "Root", + }); + api.addComment = async () => { throw new Error("safe failure"); }; + + const result = await applyStoredActionsThrough(store, invalidated, [vetoed]); + + expect(result.stopped?.at).toBe(failed); + expect(result.invalidatedByVeto).toEqual([{ action: invalidated, invalidatedBy: vetoed }]); + expect(store.getAction(vetoed)).toBeUndefined(); + expect(store.getAction(invalidated)).toBeUndefined(); + }); +}); + +describe("stageAction", () => { + it("keeps the record staged until submitAction completes", async () => { + const { api } = makeApi(); + const store = storeWith(api); + let stateDuringSubmit: string | undefined; + const approvalQueue = { + submitAction: async (id: number) => { + stateDuringSubmit = store.getAction(id)?.state; + }, + } as unknown as Parameters[1]; + + const id = await stageAction(store, approvalQueue, { + type: "addComment", contentId: "page", text: "Comment", + }); + + expect(stateDuringSubmit).toBe("staged"); + expect(store.getAction(id)?.state).toBe("pending"); + }); + + it("rolls the record back when submitAction fails", async () => { + const { api } = makeApi(); + const store = storeWith(api); + const approvalQueue = { + submitAction: async () => { throw new Error("submit failed"); }, + } as unknown as Parameters[1]; + + await expect(stageAction(store, approvalQueue, { + type: "addComment", contentId: "page", text: "Comment", + })).rejects.toThrow("submit failed"); + + expect(store.allActions()).toHaveLength(0); + }); +}); + describe("revertStoredAction", () => { it("marks the action reverted on success", async () => { const { api } = makeApi(); diff --git a/packages/gatekeeper-confluence/package.json b/packages/gatekeeper-confluence/package.json index 9bb0bbb41..015e7eb9e 100644 --- a/packages/gatekeeper-confluence/package.json +++ b/packages/gatekeeper-confluence/package.json @@ -10,6 +10,7 @@ "test:run": "vitest run" }, "dependencies": { + "@gadgets/backend-utils": "workspace:*", "@gadgets/configurator-ui": "workspace:*", "@gadgets/workshop-shared": "workspace:*", "capnweb": "catalog:", diff --git a/packages/gatekeeper-confluence/src/confluence-actions.ts b/packages/gatekeeper-confluence/src/confluence-actions.ts index db6dc8953..5a4866ca1 100644 --- a/packages/gatekeeper-confluence/src/confluence-actions.ts +++ b/packages/gatekeeper-confluence/src/confluence-actions.ts @@ -9,7 +9,18 @@ // without it. import type { RpcStub } from "cloudflare:workers"; -import type { ActionDescription, ApprovalQueue, ObservationDescription } from "@gadgets/workshop-shared/gatekeeper"; +import { + InvalidationLog, + displayReason, + validateApplyThroughArgs, +} from "@gadgets/backend-utils/gatekeeper-action"; +import { createLogger } from "@gadgets/backend-utils/logger"; +import type { + ActionDescription, + ApplyActionsThroughResult, + ApprovalQueue, + ObservationDescription, +} from "@gadgets/workshop-shared/gatekeeper"; import { ConfluenceApi, contentBodyMarkdown, @@ -19,6 +30,15 @@ import { import { markdownToStorage, storageToMarkdown } from "./confluence-markdown"; import type { Comment, ContentSummary, ContentType } from "./types"; +/** Observability fields emitted by Confluence action resolution. */ +type ConfluenceActionLogFields = { actionId: number; vendorId: string }; + +const VENDOR_ID = "confluence"; + +const logger = createLogger({ + component: "gatekeeper.confluence.actions", vendorId: VENDOR_ID, +}); + // --------------------------------------------------------------------------------------------- // Action model @@ -56,7 +76,11 @@ export type ConfluenceAction = export type StoredActionRecord = { id: number; action: ConfluenceAction; - state: "pending" | "applied" | "reverted"; + /** + * "staged" means submitAction() has not completed yet: the record overlays reads like a pending + * one, but applyStoredActionsThrough() must not apply it. + */ + state: "staged" | "pending" | "applied" | "reverted"; submittedAt: number; /** For createContent / addComment: the real content ID assigned on apply. */ createdContentId?: string; @@ -81,9 +105,13 @@ export class ConfluenceStore { #kv: Kv; #api: ConfluenceApi; + /** Durable attribution of veto-cascade invalidations, re-reported on repeated requests. */ + readonly invalidations: InvalidationLog; + constructor(kv: Kv, api: ConfluenceApi) { this.#kv = kv; this.#api = api; + this.invalidations = new InvalidationLog(kv); } get api(): ConfluenceApi { @@ -124,8 +152,9 @@ export class ConfluenceStore { .toSorted((a, b) => a.id - b.id); } + /** Not-yet-applied actions, including staged ones (read overlays must reflect both). */ pendingActions(): StoredActionRecord[] { - return this.allActions().filter(r => r.state === "pending"); + return this.allActions().filter(r => r.state === "pending" || r.state === "staged"); } /** Pending actions targeting a piece of content (addressed by either provisional or real ID). */ @@ -417,18 +446,25 @@ function truncate(text: string, max = 2000): string { // --------------------------------------------------------------------------------------------- // Staging -/** Record a pending action and submit it for approval. Rolls back the record if submit fails. */ +/** Record a staged action and submit it for approval. Rolls back the record if submit fails. */ export async function stageAction( store: ConfluenceStore, approvalQueue: RpcStub, action: ConfluenceAction, ): Promise { const id = store.nextActionId(); - store.putAction({ id, action, state: "pending", submittedAt: Date.now() }); + store.putAction({ id, action, state: "staged", submittedAt: Date.now() }); try { await approvalQueue.submitAction(id, describeAction(action)); } catch (err) { store.deleteAction(id); throw err; } + // Only now may the action be applied: the overseer has accepted it, so a decision frontier can + // legitimately cover it. A concurrent veto cascade may have deleted the record meanwhile. + const record = store.getAction(id); + if (record?.state === "staged") { + record.state = "pending"; + store.putAction(record); + } return id; } @@ -568,43 +604,88 @@ export async function applyStoredAction(store: ConfluenceStore, id: number): Pro await applyAction(store, record); } -export function rejectStoredAction(store: ConfluenceStore, id: number): void | { restart?: boolean } { - const record = store.getAction(id); - if (!record) return; - store.deleteAction(id); - - // Rejecting a creation invalidates any pending actions on that (now-nonexistent) content, - // including child pages created under it, transitively. Cascade-delete and request a restart. - if (record.action.type === "createContent") { - const pending = store.pendingActions(); - const purge = new Set([record.action.provisionalId]); - for (;;) { - let added = false; - for (const r of pending) { - if (r.action.type === "createContent" && r.action.parent.type === "page" && - purge.has(r.action.parent.parentId) && !purge.has(r.action.provisionalId)) { - purge.add(r.action.provisionalId); - added = true; - } +/** + * Rejecting a creation invalidates any pending actions on that (now-nonexistent) content, + * including child pages created under it, transitively. Cascade-delete them all, recording which + * veto invalidated each so a repeated request can re-report the attribution. + */ +function cascadeRejectedCreation(store: ConfluenceStore, record: StoredActionRecord): void { + if (record.action.type !== "createContent") return; + + // Snapshot the pending set once — deleting actions below would otherwise change it under us. + const pending = store.pendingActions(); + const purge = new Set([record.action.provisionalId]); + for (;;) { + let added = false; + for (const candidate of pending) { + if (candidate.action.type === "createContent" && candidate.action.parent.type === "page" && + purge.has(candidate.action.parent.parentId) && !purge.has(candidate.action.provisionalId)) { + purge.add(candidate.action.provisionalId); + added = true; } - if (!added) break; } - let deleted = false; - for (const r of pending) { - const t = actionContentId(r.action); - if (t !== null && purge.has(t)) { - store.deleteAction(r.id); - deleted = true; - } + if (!added) break; + } + + for (const candidate of pending) { + const target = actionContentId(candidate.action); + if (target !== null && purge.has(target)) { + store.deleteAction(candidate.id); + store.invalidations.record(candidate.id, record.id); } - return deleted ? { restart: true } : undefined; } +} + +/** Delete vetoed records and cascade to actions they invalidate. Settled records are left alone. */ +function rejectRecords(store: ConfluenceStore, vetoes: Set): void { + const rejected: StoredActionRecord[] = []; + for (const id of vetoes) { + const record = store.getAction(id); + if (!record || record.state === "applied" || record.state === "reverted") continue; + store.deleteAction(id); + rejected.push(record); + } + for (const record of rejected) cascadeRejectedCreation(store, record); +} + +/** Resolve all stored actions through a Gatekeeper-local action ID. */ +export async function applyStoredActionsThrough( + store: ConfluenceStore, actionId: number, vetoes: number[], +): Promise { + const vetoSet = validateApplyThroughArgs(actionId, vetoes); + rejectRecords(store, vetoSet); + store.invalidations.prune(vetoSet, store.pendingActions()[0]?.id ?? Infinity); + + const invalidatedByVeto = store.invalidations.attributedTo(vetoSet); + const invalidations = invalidatedByVeto.length > 0 ? { invalidatedByVeto } : {}; + for (const record of store.pendingActions()) { + if (record.id > actionId) break; + if (record.state === "staged") continue; // submitAction() has not completed; not coverable yet + try { + await applyAction(store, record); + } catch (error) { + logger.warn("failed to apply action", { + event: "action.apply.failed", + actionId: record.id, + error, + }); + return { + ...invalidations, + stopped: { + at: record.id, + reason: displayReason(error, "Confluence could not apply this action"), + }, + }; + } + } + return invalidations; +} - const target = actionContentId(record.action); - if (target && store.pendingForContent(target).length > 0) return { restart: true }; +export function rejectStoredAction(store: ConfluenceStore, id: number): void { + rejectRecords(store, new Set([id])); } -type RevertResult = void | { message?: string; canRetry?: boolean; restart?: boolean }; +type RevertResult = void | { message?: string; canRetry?: boolean }; export async function revertStoredAction(store: ConfluenceStore, id: number): Promise { const record = store.getAction(id); diff --git a/packages/gatekeeper-confluence/src/confluence.ts b/packages/gatekeeper-confluence/src/confluence.ts index a2a033f01..b98431ff4 100644 --- a/packages/gatekeeper-confluence/src/confluence.ts +++ b/packages/gatekeeper-confluence/src/confluence.ts @@ -15,6 +15,7 @@ // 7. Observer verification — all bindings track independently restricted spaces and content. import { DurableObject, RpcStub, RpcTarget, WorkerEntrypoint } from "cloudflare:workers"; +import { SerialTaskQueue } from "@gadgets/backend-utils/gatekeeper-action"; import { skipRpcValidation, validateRpc } from "capnweb-validate"; import { type AccountDescription, @@ -58,6 +59,7 @@ import { } from "./confluence-api"; import { ConfluenceStore, + applyStoredActionsThrough, applyStoredAction, observation, overlayChildPages, @@ -568,6 +570,7 @@ function makeApi(ctx: { exports: Cloudflare.Env }, props: BaseProps): Confluence @validateRpc() export class ConfluenceSiteGatekeeperImpl extends DurableObject implements Gatekeeper { + #actionResolution = new SerialTaskQueue(); #store() { return new ConfluenceStore(this.ctx.storage.kv, makeApi(this.ctx, this.ctx.props)); } #tracker() { return new ConfluenceObserverTracker(this.ctx.storage.kv, this.ctx.props.cloudId); } @@ -609,14 +612,24 @@ export class ConfluenceSiteGatekeeperImpl extends DurableObject { this.#tracker().removeObserver(id); } - async applyAction(action: number): Promise { await applyStoredAction(this.#store(), action); } - async rejectAction(action: number) { return rejectStoredAction(this.#store(), action); } - async revertAction(action: number) { return await revertStoredAction(this.#store(), action); } + applyActionsThrough(actionId: number, vetoes: number[]) { + return this.#actionResolution.run(() => applyStoredActionsThrough(this.#store(), actionId, vetoes)); + } + applyAction(action: number): Promise { + return this.#actionResolution.run(() => applyStoredAction(this.#store(), action)); + } + rejectAction(action: number) { + return this.#actionResolution.run(() => rejectStoredAction(this.#store(), action)); + } + revertAction(action: number) { + return this.#actionResolution.run(() => revertStoredAction(this.#store(), action)); + } } @validateRpc() export class ConfluenceSpaceGatekeeperImpl extends DurableObject implements Gatekeeper { + #actionResolution = new SerialTaskQueue(); #store() { return new ConfluenceStore(this.ctx.storage.kv, makeApi(this.ctx, this.ctx.props)); } #tracker() { return new ConfluenceObserverTracker(this.ctx.storage.kv, this.ctx.props.cloudId); } @@ -655,14 +668,24 @@ export class ConfluenceSpaceGatekeeperImpl extends DurableObject { this.#tracker().removeObserver(id); } - async applyAction(action: number): Promise { await applyStoredAction(this.#store(), action); } - async rejectAction(action: number) { return rejectStoredAction(this.#store(), action); } - async revertAction(action: number) { return await revertStoredAction(this.#store(), action); } + applyActionsThrough(actionId: number, vetoes: number[]) { + return this.#actionResolution.run(() => applyStoredActionsThrough(this.#store(), actionId, vetoes)); + } + applyAction(action: number): Promise { + return this.#actionResolution.run(() => applyStoredAction(this.#store(), action)); + } + rejectAction(action: number) { + return this.#actionResolution.run(() => rejectStoredAction(this.#store(), action)); + } + revertAction(action: number) { + return this.#actionResolution.run(() => revertStoredAction(this.#store(), action)); + } } @validateRpc() export class ConfluenceContentGatekeeperImpl extends DurableObject implements Gatekeeper { + #actionResolution = new SerialTaskQueue(); #store() { return new ConfluenceStore(this.ctx.storage.kv, makeApi(this.ctx, this.ctx.props)); } #tracker() { return new ConfluenceObserverTracker(this.ctx.storage.kv, this.ctx.props.cloudId); } @@ -702,9 +725,18 @@ export class ConfluenceContentGatekeeperImpl extends DurableObject { this.#tracker().removeObserver(id); } - async applyAction(action: number): Promise { await applyStoredAction(this.#store(), action); } - async rejectAction(action: number) { return rejectStoredAction(this.#store(), action); } - async revertAction(action: number) { return await revertStoredAction(this.#store(), action); } + applyActionsThrough(actionId: number, vetoes: number[]) { + return this.#actionResolution.run(() => applyStoredActionsThrough(this.#store(), actionId, vetoes)); + } + applyAction(action: number): Promise { + return this.#actionResolution.run(() => applyStoredAction(this.#store(), action)); + } + rejectAction(action: number) { + return this.#actionResolution.run(() => rejectStoredAction(this.#store(), action)); + } + revertAction(action: number) { + return this.#actionResolution.run(() => revertStoredAction(this.#store(), action)); + } } function accountFor(ctx: { exports: Cloudflare.Env }, userObjectId: string) { diff --git a/packages/gatekeeper-notion/__tests__/notion-actions.test.ts b/packages/gatekeeper-notion/__tests__/notion-actions.test.ts new file mode 100644 index 000000000..be4e80dc8 --- /dev/null +++ b/packages/gatekeeper-notion/__tests__/notion-actions.test.ts @@ -0,0 +1,106 @@ +import { describe, expect, it } from "vitest"; +import { + NotionStore, + applyStoredActionsThrough, + type NotionAction, +} from "../src/notion-actions"; +import type { NotionApi } from "../src/notion-api"; + +type Kv = ConstructorParameters[0]; + +function makeKv(): Kv { + const map = new Map(); + return { + get: (key: string) => map.get(key) as T | undefined, + put: (key: string, value: unknown) => void map.set(key, value), + delete: (key: string) => void map.delete(key), + list: ({ prefix }: { prefix: string }) => + [...map.entries()].filter(([key]) => key.startsWith(prefix)) as [string, T][], + } as unknown as Kv; +} + +function makeStore() { + const comments: string[] = []; + const api = { + createComment: async ({ parent }: { parent: { page_id: string } }) => { + comments.push(parent.page_id); + if (parent.page_id === "failed") throw new Error("safe failure"); + return {}; + }, + } as unknown as NotionApi; + return { store: new NotionStore(makeKv(), api), comments }; +} + +function stage(store: NotionStore, action: NotionAction): number { + const id = store.nextActionId(); + store.putAction({ id, action, state: "pending", submittedAt: id }); + return id; +} + +describe("applyStoredActionsThrough", () => { + it("skips sparse IDs and stops at the first failed action", async () => { + const { store, comments } = makeStore(); + const first = stage(store, { type: "addComment", pageId: "first", text: "one" }); + const hole = stage(store, { type: "addComment", pageId: "hole", text: "two" }); + const failed = stage(store, { type: "addComment", pageId: "failed", text: "three" }); + const later = stage(store, { type: "addComment", pageId: "later", text: "four" }); + store.deleteAction(hole); + + const result = await applyStoredActionsThrough(store, later, []); + + expect(comments).toEqual(["first", "failed"]); + expect(store.getAction(first)?.state).toBe("applied"); + expect(store.getAction(failed)?.state).toBe("pending"); + expect(store.getAction(later)?.state).toBe("pending"); + expect(result.stopped).toMatchObject({ at: failed, reason: expect.any(Error) }); + }); + + it("persists transitive invalidations and reports them again on retry", async () => { + const { store, comments } = makeStore(); + const root = stage(store, { + type: "createPage", provisionalId: "~root", parent: { kind: "workspace" }, title: "Root", + }); + const child = stage(store, { + type: "createPage", provisionalId: "~child", parent: { kind: "page", pageId: "~root" }, + title: "Child", + }); + const edit = stage(store, { type: "addComment", pageId: "~child", text: "Comment" }); + + const first = await applyStoredActionsThrough(store, root, [root]); + const retry = await applyStoredActionsThrough(store, root, [root]); + + expect(first.invalidatedByVeto).toEqual([ + { action: child, invalidatedBy: root }, + { action: edit, invalidatedBy: root }, + ]); + expect(retry.invalidatedByVeto).toEqual(first.invalidatedByVeto); + // Vetoed and invalidated records are deleted so read overlays recompute without them. + expect(store.getAction(root)).toBeUndefined(); + expect(store.getAction(edit)).toBeUndefined(); + expect(store.knowsProvisional("~root")).toBe(false); + expect(comments).toHaveLength(0); + }); + + it("ignores a veto of an already-applied action", async () => { + const { store, comments } = makeStore(); + const id = stage(store, { type: "addComment", pageId: "page", text: "Comment" }); + await applyStoredActionsThrough(store, id, []); + + const result = await applyStoredActionsThrough(store, id, [id]); + + expect(result).toEqual({}); + expect(store.getAction(id)?.state).toBe("applied"); + expect(comments).toEqual(["page"]); + }); + + it("rejects an out-of-range veto before changing state", async () => { + const { store, comments } = makeStore(); + const id = stage(store, { type: "addComment", pageId: "page", text: "Comment" }); + + await expect(applyStoredActionsThrough(store, id, [id + 1])) + .rejects.toThrow("Invalid veto action ID"); + + expect(store.getAction(id)?.state).toBe("pending"); + expect(comments).toHaveLength(0); + }); +}); diff --git a/packages/gatekeeper-notion/package.json b/packages/gatekeeper-notion/package.json index 8dd05ac74..14f600534 100644 --- a/packages/gatekeeper-notion/package.json +++ b/packages/gatekeeper-notion/package.json @@ -10,6 +10,7 @@ "test:run": "vitest run" }, "dependencies": { + "@gadgets/backend-utils": "workspace:*", "@gadgets/configurator-ui": "workspace:*", "@gadgets/workshop-shared": "workspace:*", "capnweb": "catalog:", diff --git a/packages/gatekeeper-notion/src/notion-actions.ts b/packages/gatekeeper-notion/src/notion-actions.ts index a43248503..05719781d 100644 --- a/packages/gatekeeper-notion/src/notion-actions.ts +++ b/packages/gatekeeper-notion/src/notion-actions.ts @@ -27,7 +27,18 @@ import { type NotionPageResponse, } from "./notion-api"; import type { RpcStub } from "cloudflare:workers"; -import type { ActionDescription, ApprovalQueue, ObservationDescription } from "@gadgets/workshop-shared/gatekeeper"; +import { + InvalidationLog, + displayReason, + validateApplyThroughArgs, +} from "@gadgets/backend-utils/gatekeeper-action"; +import { createLogger } from "@gadgets/backend-utils/logger"; +import type { + ActionDescription, + ApplyActionsThroughResult, + ApprovalQueue, + ObservationDescription, +} from "@gadgets/workshop-shared/gatekeeper"; import type { NotionComment, NotionDatabaseSchema, @@ -40,6 +51,15 @@ import type { NotionUser, } from "./types"; +/** Observability fields emitted by Notion action resolution. */ +type NotionActionLogFields = { actionId: number; vendorId: string }; + +const VENDOR_ID = "notion"; + +const logger = createLogger({ + component: "gatekeeper.notion.actions", vendorId: VENDOR_ID, +}); + // --------------------------------------------------------------------------------------------- // Action model @@ -77,7 +97,11 @@ export type NotionAction = export type StoredActionRecord = { id: number; action: NotionAction; - state: "pending" | "applied" | "reverted"; + /** + * "staged" means submitAction() has not completed yet: the record overlays reads like a pending + * one, but applyStoredActionsThrough() must not apply it. + */ + state: "staged" | "pending" | "applied" | "reverted"; submittedAt: number; /** For appendContent revert: the IDs of the blocks created on apply. */ appendedBlockIds?: string[]; @@ -110,9 +134,13 @@ export class NotionStore { #kv: Kv; #api: NotionApi; + /** Durable attribution of veto-cascade invalidations, re-reported on repeated requests. */ + readonly invalidations: InvalidationLog; + constructor(kv: Kv, api: NotionApi) { this.#kv = kv; this.#api = api; + this.invalidations = new InvalidationLog(kv); } get api(): NotionApi { @@ -153,8 +181,9 @@ export class NotionStore { .toSorted((a, b) => a.id - b.id); } + /** Not-yet-applied actions, including staged ones (read overlays must reflect both). */ pendingActions(): StoredActionRecord[] { - return this.allActions().filter(r => r.state === "pending"); + return this.allActions().filter(r => r.state === "pending" || r.state === "staged"); } /** @@ -820,7 +849,7 @@ export async function applyNotionAction(store: NotionStore, record: StoredAction export async function revertNotionAction( store: NotionStore, record: StoredActionRecord, -): Promise { +): Promise { const api = store.api; const action = record.action; @@ -1004,7 +1033,7 @@ export function buildCreateBody( } /** - * Record a pending action and submit it to the approval queue for later approval. If the submit + * Record a staged action and submit it to the approval queue for later approval. If the submit * fails, the stored record is rolled back so it doesn't pollute simulation. Returns the action ID. */ export async function stageAction( @@ -1013,13 +1042,20 @@ export async function stageAction( action: NotionAction, ): Promise { const id = store.nextActionId(); - store.putAction({ id, action, state: "pending", submittedAt: Date.now() }); + store.putAction({ id, action, state: "staged", submittedAt: Date.now() }); try { await approvalQueue.submitAction(id, describeAction(action)); } catch (err) { store.deleteAction(id); throw err; } + // Only now may the action be applied: the overseer has accepted it, so a decision frontier can + // legitimately cover it. A concurrent veto cascade may have deleted the record meanwhile. + const record = store.getAction(id); + if (record?.state === "staged") { + record.state = "pending"; + store.putAction(record); + } return id; } @@ -1032,57 +1068,95 @@ export async function applyStoredAction(store: NotionStore, id: number): Promise await applyNotionAction(store, record); } -export function rejectStoredAction(store: NotionStore, id: number): void | { restart?: boolean } { - const record = store.getAction(id); - if (!record) return; - store.deleteAction(id); - - // Rejecting a page creation invalidates every pending action that targeted that provisional page - // (they could never be applied — the page won't exist), including sub-pages created under it and - // their edits, transitively. Cascade-delete them all so the overseer never tries to apply an - // orphan, and request a restart since the Gadget already observed simulated state built on them. - if (record.action.type === "createPage") { - // Snapshot the pending set once — deleting actions below would otherwise change it under us. - const pending = store.pendingActions(); - const purge = new Set([record.action.provisionalId]); - // Expand to transitively-nested sub-page creations. Only `createSubPage` nests a creation under - // a provisional page (parent.kind === "page"); database/workspace creates never have a - // provisional parent, so they don't need handling here. - for (;;) { - let added = false; - for (const r of pending) { - if (r.action.type === "createPage" && r.action.parent.kind === "page" && - purge.has(r.action.parent.pageId) && !purge.has(r.action.provisionalId)) { - purge.add(r.action.provisionalId); - added = true; - } +/** + * Rejecting a page creation invalidates every pending action that targeted that provisional page + * (they could never be applied — the page won't exist), including sub-pages created under it and + * their edits, transitively. Cascade-delete them all, recording which veto invalidated each so a + * repeated request can re-report the attribution. + */ +function cascadeRejectedCreation(store: NotionStore, record: StoredActionRecord): void { + if (record.action.type !== "createPage") return; + + // Snapshot the pending set once — deleting actions below would otherwise change it under us. + const pending = store.pendingActions(); + const purge = new Set([record.action.provisionalId]); + // Expand to transitively-nested sub-page creations. Only `createSubPage` nests a creation under + // a provisional page (parent.kind === "page"); database/workspace creates never have a + // provisional parent, so they don't need handling here. + for (;;) { + let added = false; + for (const candidate of pending) { + if (candidate.action.type === "createPage" && candidate.action.parent.kind === "page" && + purge.has(candidate.action.parent.pageId) && !purge.has(candidate.action.provisionalId)) { + purge.add(candidate.action.provisionalId); + added = true; } - if (!added) break; } - let deleted = false; - for (const r of pending) { - const t = actionPageId(r.action); - if (t !== null && purge.has(t)) { - store.deleteAction(r.id); - deleted = true; - } + if (!added) break; + } + + for (const candidate of pending) { + const target = actionPageId(candidate.action); + if (target !== null && purge.has(target)) { + store.deleteAction(candidate.id); + store.invalidations.record(candidate.id, record.id); } - if (deleted) return { restart: true }; - return; } +} - // Rejecting a mid-stack edit leaves the simulated overlay the Gadget already observed - // inconsistent; ask for a restart if other pending actions still target the same page. - const target = actionPageId(record.action); - if (target && store.pendingForPage(target).length > 0) { - return { restart: true }; +/** Delete vetoed records and cascade to actions they invalidate. Settled records are left alone. */ +function rejectRecords(store: NotionStore, vetoes: Set): void { + const rejected: StoredActionRecord[] = []; + for (const id of vetoes) { + const record = store.getAction(id); + if (!record || record.state === "applied" || record.state === "reverted") continue; + store.deleteAction(id); + rejected.push(record); } + for (const record of rejected) cascadeRejectedCreation(store, record); +} + +/** Resolve all stored actions through a Gatekeeper-local action ID. */ +export async function applyStoredActionsThrough( + store: NotionStore, actionId: number, vetoes: number[], +): Promise { + const vetoSet = validateApplyThroughArgs(actionId, vetoes); + rejectRecords(store, vetoSet); + store.invalidations.prune(vetoSet, store.pendingActions()[0]?.id ?? Infinity); + + const invalidatedByVeto = store.invalidations.attributedTo(vetoSet); + const invalidations = invalidatedByVeto.length > 0 ? { invalidatedByVeto } : {}; + for (const record of store.pendingActions()) { + if (record.id > actionId) break; + if (record.state === "staged") continue; // submitAction() has not completed; not coverable yet + try { + await applyNotionAction(store, record); + } catch (error) { + logger.warn("failed to apply action", { + event: "action.apply.failed", + actionId: record.id, + error, + }); + return { + ...invalidations, + stopped: { + at: record.id, + reason: displayReason(error, "Notion could not apply this action"), + }, + }; + } + } + return invalidations; +} + +export function rejectStoredAction(store: NotionStore, id: number): void { + rejectRecords(store, new Set([id])); } export async function revertStoredAction( store: NotionStore, id: number, -): Promise { +): Promise { const record = store.getAction(id); if (!record) throw new Error(`Unknown action: ${id}`); return await revertNotionAction(store, record); diff --git a/packages/gatekeeper-notion/src/notion.ts b/packages/gatekeeper-notion/src/notion.ts index af56af206..ed6f32f53 100644 --- a/packages/gatekeeper-notion/src/notion.ts +++ b/packages/gatekeeper-notion/src/notion.ts @@ -14,6 +14,7 @@ // writes immediately. List simulation has documented limitations (see types.d.ts). import { DurableObject, RpcStub, RpcTarget, WorkerEntrypoint } from "cloudflare:workers"; +import { SerialTaskQueue } from "@gadgets/backend-utils/gatekeeper-action"; import { skipRpcValidation, validateRpc } from "capnweb-validate"; import { stripTrailingSlashes, @@ -48,6 +49,7 @@ import { } from "./notion-api"; import { NotionStore, + applyStoredActionsThrough, applyStoredAction, defaultPropertiesFromSchema, observation, @@ -694,6 +696,7 @@ type NotionItemGatekeeperImplProps = { @validateRpc() export class NotionItemGatekeeperImpl extends DurableObject implements Gatekeeper { + #actionResolution = new SerialTaskQueue(); #api(): NotionApi { const userObjectId = this.ctx.props.userObjectId; const account = () => @@ -778,16 +781,20 @@ export class NotionItemGatekeeperImpl extends DurableObject {} - async applyAction(action: number): Promise { - await applyStoredAction(this.#store(), action); + applyActionsThrough(actionId: number, vetoes: number[]) { + return this.#actionResolution.run(() => applyStoredActionsThrough(this.#store(), actionId, vetoes)); } - async rejectAction(action: number): Promise { - return rejectStoredAction(this.#store(), action); + applyAction(action: number): Promise { + return this.#actionResolution.run(() => applyStoredAction(this.#store(), action)); } - async revertAction(action: number) { - return await revertStoredAction(this.#store(), action); + rejectAction(action: number): Promise { + return this.#actionResolution.run(() => rejectStoredAction(this.#store(), action)); + } + + revertAction(action: number) { + return this.#actionResolution.run(() => revertStoredAction(this.#store(), action)); } } @@ -799,6 +806,7 @@ type NotionWorkspaceGatekeeperImplProps = { export class NotionWorkspaceGatekeeperImpl extends DurableObject implements Gatekeeper { + #actionResolution = new SerialTaskQueue(); #api(): NotionApi { const userObjectId = this.ctx.props.userObjectId; const account = () => @@ -940,16 +948,20 @@ export class NotionWorkspaceGatekeeperImpl this.ctx.storage.kv.delete(this.#observerKey(id)); } - async applyAction(action: number): Promise { - await applyStoredAction(this.#store(), action); + applyActionsThrough(actionId: number, vetoes: number[]) { + return this.#actionResolution.run(() => applyStoredActionsThrough(this.#store(), actionId, vetoes)); + } + + applyAction(action: number): Promise { + return this.#actionResolution.run(() => applyStoredAction(this.#store(), action)); } - async rejectAction(action: number): Promise { - return rejectStoredAction(this.#store(), action); + rejectAction(action: number): Promise { + return this.#actionResolution.run(() => rejectStoredAction(this.#store(), action)); } - async revertAction(action: number) { - return await revertStoredAction(this.#store(), action); + revertAction(action: number) { + return this.#actionResolution.run(() => revertStoredAction(this.#store(), action)); } } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index f572acfe0..14fc76222 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -117,6 +117,9 @@ importers: packages/gatekeeper-confluence: dependencies: + '@gadgets/backend-utils': + specifier: workspace:* + version: link:../backend-utils '@gadgets/configurator-ui': specifier: workspace:* version: link:../configurator-ui @@ -465,6 +468,9 @@ importers: packages/gatekeeper-notion: dependencies: + '@gadgets/backend-utils': + specifier: workspace:* + version: link:../backend-utils '@gadgets/configurator-ui': specifier: workspace:* version: link:../configurator-ui @@ -1148,7 +1154,7 @@ packages: resolution: {integrity: sha512-ehg3jIkYKulZh+8om/O25vkvSsXXwC+skXmyA87FFx6A/45eqOkZsBltMw/TVteb0mloiGT8oGRTcjRAz66zaA==} '@cloudflare/kumo@2.9.2': - resolution: {integrity: sha512-c3RZBmx0TqxTKAPT4PWyTgVwPcDVW+KrFmf4mKCnwWBe6OIc0vWn+wMhnaARarJz/2kvsx87tMGmNRBsCn7pUA==} + resolution: {integrity: sha512-c3RZBmx0TqxTKAPT4PWyTgVwPcDVW+KrFmf4mKCnwWBe6OIc0vWn+wMhnaARarJz/2kvsx87tMGmNRBsCn7pUA==, tarball: https://registry.npmjs.org/@cloudflare/kumo/-/kumo-2.9.2.tgz} hasBin: true peerDependencies: '@phosphor-icons/react': ^2.1.10 @@ -1163,15 +1169,15 @@ packages: optional: true '@cloudflare/kv-asset-handler@0.5.0': - resolution: {integrity: sha512-jxQYkj8dSIzc0cD6cMMNdOc1UVjqSqu8BZdor5s8cGjW2I8BjODt/kWPVdY+u9zj3ms75Q5qaZgnxUad83+eAg==} + resolution: {integrity: sha512-jxQYkj8dSIzc0cD6cMMNdOc1UVjqSqu8BZdor5s8cGjW2I8BjODt/kWPVdY+u9zj3ms75Q5qaZgnxUad83+eAg==, tarball: https://registry.npmjs.org/@cloudflare/kv-asset-handler/-/kv-asset-handler-0.5.0.tgz} engines: {node: '>=22.0.0'} '@cloudflare/puppeteer@1.3.0': - resolution: {integrity: sha512-NBrJEUnqe082nopLh0eqnTXK4DjwsTsZGzoAcs71NFnBgzWU6Yb/ibUJHveCHV4AyAkM+mE/DChFev5gwaKZEg==} + resolution: {integrity: sha512-NBrJEUnqe082nopLh0eqnTXK4DjwsTsZGzoAcs71NFnBgzWU6Yb/ibUJHveCHV4AyAkM+mE/DChFev5gwaKZEg==, tarball: https://registry.npmjs.org/@cloudflare/puppeteer/-/puppeteer-1.3.0.tgz} engines: {node: '>=18'} '@cloudflare/unenv-preset@2.16.1': - resolution: {integrity: sha512-ECxObrMfyTl5bhQf/lZCXwo5G6xX9IAUo+nDMKK4SZ8m4Jvvxp52vilxyySSWh2YTZz8+HQ07qGH/2rEom1vDw==} + resolution: {integrity: sha512-ECxObrMfyTl5bhQf/lZCXwo5G6xX9IAUo+nDMKK4SZ8m4Jvvxp52vilxyySSWh2YTZz8+HQ07qGH/2rEom1vDw==, tarball: https://registry.npmjs.org/@cloudflare/unenv-preset/-/unenv-preset-2.16.1.tgz} peerDependencies: unenv: 2.0.0-rc.24 workerd: '>1.20260305.0 <2.0.0-0' @@ -1180,44 +1186,44 @@ packages: optional: true '@cloudflare/vitest-pool-workers@0.20.3': - resolution: {integrity: sha512-aCMvM5zQ3MTz8SorSZB6ZxjVK2Gof6UhIc2Z1z9zbX/obucSYwDUkx8A0MUOod4I7clfB0QLarKBjRdIczK3vg==} + resolution: {integrity: sha512-aCMvM5zQ3MTz8SorSZB6ZxjVK2Gof6UhIc2Z1z9zbX/obucSYwDUkx8A0MUOod4I7clfB0QLarKBjRdIczK3vg==, tarball: https://registry.npmjs.org/@cloudflare/vitest-pool-workers/-/vitest-pool-workers-0.20.3.tgz} peerDependencies: '@vitest/runner': ^4.1.0 '@vitest/snapshot': ^4.1.0 vitest: ^4.1.0 '@cloudflare/workerd-darwin-64@1.20260801.1': - resolution: {integrity: sha512-wuJWbXpKvncJi1P0GKS+iYpN5tHdb7JPJJ/+6ZQe8zzovHHVMkLJPNBsgWpqeUhpM3g9qTwEKd2rglNKejuh5A==} + resolution: {integrity: sha512-wuJWbXpKvncJi1P0GKS+iYpN5tHdb7JPJJ/+6ZQe8zzovHHVMkLJPNBsgWpqeUhpM3g9qTwEKd2rglNKejuh5A==, tarball: https://registry.npmjs.org/@cloudflare/workerd-darwin-64/-/workerd-darwin-64-1.20260801.1.tgz} engines: {node: '>=16'} cpu: [x64] os: [darwin] '@cloudflare/workerd-darwin-arm64@1.20260801.1': - resolution: {integrity: sha512-kwoZiTpnhNrF3+APx84Q/oAqvJ3sU9yefGagwm/ASaH/2W19x0vghkW/r4qCoHCK0WW7EPugZ+aXjgPMRtlq1Q==} + resolution: {integrity: sha512-kwoZiTpnhNrF3+APx84Q/oAqvJ3sU9yefGagwm/ASaH/2W19x0vghkW/r4qCoHCK0WW7EPugZ+aXjgPMRtlq1Q==, tarball: https://registry.npmjs.org/@cloudflare/workerd-darwin-arm64/-/workerd-darwin-arm64-1.20260801.1.tgz} engines: {node: '>=16'} cpu: [arm64] os: [darwin] '@cloudflare/workerd-linux-64@1.20260801.1': - resolution: {integrity: sha512-r0vAxCZH+Jih9Unm1yoyiByPNWNgawcKciOHDm5Q37ZVGOkKLsT9AtLe3yLSaul76WrKqtf+JP2n0WW32VBLJg==} + resolution: {integrity: sha512-r0vAxCZH+Jih9Unm1yoyiByPNWNgawcKciOHDm5Q37ZVGOkKLsT9AtLe3yLSaul76WrKqtf+JP2n0WW32VBLJg==, tarball: https://registry.npmjs.org/@cloudflare/workerd-linux-64/-/workerd-linux-64-1.20260801.1.tgz} engines: {node: '>=16'} cpu: [x64] os: [linux] '@cloudflare/workerd-linux-arm64@1.20260801.1': - resolution: {integrity: sha512-zWgpdZtSozvIgzQNmQiDSF8yEOQJUkRAWNsDXXzAAoy+fCn8YUoSibj3mpFSbZRvbUldeBEaW6SCdC2VEMkhNQ==} + resolution: {integrity: sha512-zWgpdZtSozvIgzQNmQiDSF8yEOQJUkRAWNsDXXzAAoy+fCn8YUoSibj3mpFSbZRvbUldeBEaW6SCdC2VEMkhNQ==, tarball: https://registry.npmjs.org/@cloudflare/workerd-linux-arm64/-/workerd-linux-arm64-1.20260801.1.tgz} engines: {node: '>=16'} cpu: [arm64] os: [linux] '@cloudflare/workerd-windows-64@1.20260801.1': - resolution: {integrity: sha512-2oQz+Ksu4ji6e/+ZoYX+tWQEcxAii2p7l+iR8kx48W1llMalaufAsmVxTlk+3/vrM7D3/2c0iK448e0UQTcIMg==} + resolution: {integrity: sha512-2oQz+Ksu4ji6e/+ZoYX+tWQEcxAii2p7l+iR8kx48W1llMalaufAsmVxTlk+3/vrM7D3/2c0iK448e0UQTcIMg==, tarball: https://registry.npmjs.org/@cloudflare/workerd-windows-64/-/workerd-windows-64-1.20260801.1.tgz} engines: {node: '>=16'} cpu: [x64] os: [win32] '@cloudflare/workers-types@5.20260808.1': - resolution: {integrity: sha512-DN7G9SMyeOq031YhQexoExFAK78ms74cFiFF1teDlTK4+LjHgIc5Z8VbvXQtcCIP//o38btxzxW0b9MGPA83CA==} + resolution: {integrity: sha512-DN7G9SMyeOq031YhQexoExFAK78ms74cFiFF1teDlTK4+LjHgIc5Z8VbvXQtcCIP//o38btxzxW0b9MGPA83CA==, tarball: https://registry.npmjs.org/@cloudflare/workers-types/-/workers-types-5.20260808.1.tgz} '@codemirror/autocomplete@6.20.3': resolution: {integrity: sha512-tlosUqb+3BbxCxZdu4tKeRghPFC+QM7q4X5YhKV2eCmPG+1r2F3f4AaSz5sCrFqUtX4Jh20VFTKecl16MgiV9g==} From bad2ee20f53083d5a01a7bfaf4a13b302d717fcb Mon Sep 17 00:00:00 2001 From: Nathan Disidore Date: Mon, 17 Aug 2026 15:58:55 -0500 Subject: [PATCH 2/3] Drive action resolution through batch sync passes --- .../__tests__/action-sync.test.ts | 442 ++++++++++++++++++ .../__tests__/auto-approval.test.ts | 214 --------- packages/workshop-backend/src/action-sync.ts | 303 ++++++++++++ .../workshop-backend/src/auto-approval.ts | 97 ---- packages/workshop-backend/src/overseer.ts | 133 +++--- packages/workshop-shared/src/api.ts | 18 +- 6 files changed, 841 insertions(+), 366 deletions(-) create mode 100644 packages/workshop-backend/__tests__/action-sync.test.ts delete mode 100644 packages/workshop-backend/__tests__/auto-approval.test.ts create mode 100644 packages/workshop-backend/src/action-sync.ts delete mode 100644 packages/workshop-backend/src/auto-approval.ts diff --git a/packages/workshop-backend/__tests__/action-sync.test.ts b/packages/workshop-backend/__tests__/action-sync.test.ts new file mode 100644 index 000000000..a61342bee --- /dev/null +++ b/packages/workshop-backend/__tests__/action-sync.test.ts @@ -0,0 +1,442 @@ +import { describe, it, expect } from "vitest"; +import { createTypedStorage, collection } from "@gadgets/typed-storage"; +import { ActionSyncDriver, ActionSyncStorage, GatekeeperActionTarget } from "../src/action-sync.js"; +import type { ActionRecord, AutoApproveTagRecord } from "../src/overseer.js"; +import type { AiChatAuthorInfo } from "@gadgets/workshop-shared/api"; +import type { ApplyActionsThroughResult } from "@gadgets/workshop-shared/gatekeeper"; +import { makeMockStorage } from "./mock-storage.js"; + +function makeStorage(): ActionSyncStorage { + return createTypedStorage(makeMockStorage(), { + collections: { + actions: collection()({ primaryKey: "id" }), + autoApproveTags: collection()({ + primaryKey: (r: AutoApproveTagRecord) => `${r.gatekeeperId}:${r.actionKind.tag}`, + }), + }, + }); +} + +const GK = 1; +const ENABLER: AiChatAuthorInfo = { type: "user", id: "enabler@example.com", name: "Enabler" }; +const APPROVER: AiChatAuthorInfo = { type: "user", id: "approver@example.com", name: "Approver" }; +const REJECTER: AiChatAuthorInfo = { type: "user", id: "rejecter@example.com", name: "Rejecter" }; + +function enableRule(storage: ActionSyncStorage, actionTag = "edit", gatekeeperId = GK) { + storage.autoApproveTags.put({ + gatekeeperId, actionKind: { tag: actionTag, label: "Edits" }, enabledBy: ENABLER }); +} + +// Workspace record ids are deliberately offset from gatekeeper-local action ids (`id = action*10`) +// so a test that confuses the two ID spaces fails loudly. +function putAction( + storage: ActionSyncStorage, action: number, + opts: { gatekeeperId?: number; actionTag?: string; autoApprovable?: boolean; + state?: ActionRecord["state"]; chatId?: number; awaitDecision?: boolean; + vetoPending?: true; resolvedBy?: AiChatAuthorInfo; failure?: string } = {}): number { + let id = action * 10; + storage.actions.put({ + id, + gatekeeperId: opts.gatekeeperId ?? GK, + caller: { from: "agent", chatId: opts.chatId ?? 1 }, + createdAt: new Date(), + state: opts.state ?? "pending", + type: "action", + action, + ...(opts.vetoPending ? { vetoPending: true } : {}), + ...(opts.resolvedBy ? { resolvedBy: opts.resolvedBy } : {}), + ...(opts.failure !== undefined ? { failure: opts.failure } : {}), + description: { + title: `Action ${action}`, + description: `Action ${action} description`, + implementsRevert: true, + actionKind: { tag: opts.actionTag ?? "edit", label: "Edits" }, + autoApprovable: opts.autoApprovable ?? true, + ...(opts.awaitDecision ? { awaitDecision: true } : {}), + }, + }); + return id; +} + +function getAction(storage: ActionSyncStorage, action: number): ActionRecord & {type: "action"} { + let record = storage.actions.get(action * 10); + if (!record || record.type !== "action") throw new Error(`No action ${action}`); + return record; +} + +// A migrated gatekeeper stub: records every batch call and answers from a scripted queue (or {}). +function makeBatchGatekeeper() { + let calls: Array<{actionId: number, vetoes: number[]}> = []; + let results: Array = []; + let target = { + async applyActionsThrough(actionId: number, vetoes: number[]) { + calls.push({ actionId, vetoes }); + let next = results.shift() ?? {}; + if (next instanceof Error) throw next; + return next; + }, + async applyAction() { throw new Error("legacy applyAction must not be called"); }, + async rejectAction() { throw new Error("legacy rejectAction must not be called"); }, + } as unknown as GatekeeperActionTarget; + return { target, calls, results }; +} + +// A pre-migration gatekeeper stub: applyActionsThrough is missing (locally undefined, or throwing +// workerd's method-missing TypeError when `remote` mimics a live stub), so the driver must fall +// back to per-action legacy calls. +function makeLegacyGatekeeper(opts: {remote?: boolean, failApply?: number[]} = {}) { + let probes = 0; + let calls: string[] = []; + let target = { + ...(opts.remote ? { + async applyActionsThrough() { + probes++; + throw new TypeError( + 'The RPC receiver does not implement the method "applyActionsThrough".'); + }, + } : {}), + async applyAction(action: number) { + calls.push(`apply:${action}`); + if (opts.failApply?.includes(action)) throw new Error(`apply ${action} failed`); + }, + async rejectAction(action: number) { + calls.push(`reject:${action}`); + return { restart: true }; // must be discarded + }, + } as unknown as GatekeeperActionTarget; + return { target, calls, probeCount: () => probes }; +} + +function makeDriver(storage: ActionSyncStorage, target: GatekeeperActionTarget) { + return new ActionSyncDriver(storage, () => target); +} + +// Drain the microtask queue (and one macrotask) so parked continuations reach their next await. +function flush(): Promise { + return new Promise((resolve) => setTimeout(resolve, 0)); +} + +describe("ActionSyncDriver.sync", () => { + it("applies through a manual frontier, attributing covered actions to the approver and " + + "auto-extended ones to the rule enabler", async () => { + let storage = makeStorage(); + enableRule(storage); + let a1 = putAction(storage, 1, { autoApprovable: false }); + let a2 = putAction(storage, 2, { autoApprovable: false }); + let a3 = putAction(storage, 3); // auto-eligible beyond the manual frontier + + let { target, calls } = makeBatchGatekeeper(); + let decided = await makeDriver(storage, target) + .sync(GK, { frontier: 2, resolvedBy: APPROVER }); + + expect(calls).toEqual([{ actionId: 3, vetoes: [] }]); + expect(decided.toSorted((a, b) => a - b)).toEqual([a1, a2, a3]); + for (let action of [1, 2]) { + let record = getAction(storage, action); + expect(record.state).toBe("approved"); + expect(record.autoApproved).toBe(false); + expect(record.resolvedBy?.id).toBe(APPROVER.id); + } + let auto = getAction(storage, 3); + expect(auto.state).toBe("approved"); + expect(auto.autoApproved).toBe(true); + expect(auto.resolvedBy?.id).toBe(ENABLER.id); + }); + + it("never auto-approves past a manual gate", async () => { + let storage = makeStorage(); + enableRule(storage); + putAction(storage, 1); + putAction(storage, 2, { autoApprovable: false }); // manual gate + putAction(storage, 3); + + let { target, calls } = makeBatchGatekeeper(); + await makeDriver(storage, target).sync(GK); + + expect(calls).toEqual([{ actionId: 1, vetoes: [] }]); + expect(getAction(storage, 1).state).toBe("approved"); + expect(getAction(storage, 2).state).toBe("pending"); + expect(getAction(storage, 3).state).toBe("pending"); + }); + + it("makes no call when nothing is eligible", async () => { + let storage = makeStorage(); + putAction(storage, 1, { autoApprovable: false }); + + let { target, calls } = makeBatchGatekeeper(); + let decided = await makeDriver(storage, target).sync(GK); + + expect(decided).toEqual([]); + expect(calls).toEqual([]); + }); + + it("records a display-safe failure on the stopped action and clears it on a later success", + async () => { + let storage = makeStorage(); + let a1 = putAction(storage, 1, { autoApprovable: false }); + putAction(storage, 2, { autoApprovable: false }); + + let { target, calls, results } = makeBatchGatekeeper(); + results.push({ stopped: { at: 2, reason: new Error("page was deleted upstream") } }); + let driver = makeDriver(storage, target); + + let first = await driver.sync(GK, { frontier: 2, resolvedBy: APPROVER }); + + expect(first).toEqual([a1]); + expect(getAction(storage, 1).state).toBe("approved"); + let stopped = getAction(storage, 2); + expect(stopped.state).toBe("pending"); + expect(stopped.failure).toBe("page was deleted upstream"); + + // Retry after the user resolves the problem: only the stopped action remains pending, and its + // failure is cleared. The already-applied action is never re-sent (idempotent contract), and + // the gatekeeper sees a second call at the same frontier. + let retry = await driver.sync(GK, { frontier: 2, resolvedBy: APPROVER }); + + expect(retry).toEqual([getAction(storage, 2).id]); + expect(calls).toEqual([{ actionId: 2, vetoes: [] }, { actionId: 2, vetoes: [] }]); + let retried = getAction(storage, 2); + expect(retried.state).toBe("approved"); + expect(retried.failure).toBeUndefined(); + }); + + it("keeps a veto staged while an earlier action is undecided", async () => { + let storage = makeStorage(); + putAction(storage, 1, { autoApprovable: false }); + putAction(storage, 2, { state: "rejected", vetoPending: true, resolvedBy: REJECTER }); + + let { target, calls } = makeBatchGatekeeper(); + await makeDriver(storage, target).sync(GK); + + expect(calls).toEqual([]); + expect(getAction(storage, 2).vetoPending).toBe(true); + }); + + it("delivers a staged veto at the current frontier once everything below is decided, even " + + "from a fresh driver", async () => { + let storage = makeStorage(); + putAction(storage, 1, { state: "approved" }); + putAction(storage, 2, { state: "rejected", vetoPending: true, resolvedBy: REJECTER }); + + // A fresh driver over the same storage (e.g. after DO hibernation) must still see the staged + // veto -- it is durable state, not driver memory. + let { target, calls } = makeBatchGatekeeper(); + await makeDriver(storage, target).sync(GK); + + expect(calls).toEqual([{ actionId: 2, vetoes: [2] }]); + expect(getAction(storage, 2).vetoPending).toBeUndefined(); + }); + + it("rides staged vetoes along with a covering approval", async () => { + let storage = makeStorage(); + let a1 = putAction(storage, 1, { autoApprovable: false }); + putAction(storage, 2, { state: "rejected", vetoPending: true, resolvedBy: REJECTER }); + let a3 = putAction(storage, 3, { autoApprovable: false }); + + let { target, calls } = makeBatchGatekeeper(); + let decided = await makeDriver(storage, target) + .sync(GK, { frontier: 3, resolvedBy: APPROVER }); + + expect(calls).toEqual([{ actionId: 3, vetoes: [2] }]); + expect(decided.toSorted((a, b) => a - b)).toEqual([a1, a3]); + expect(getAction(storage, 2).vetoPending).toBeUndefined(); + }); + + it("marks cascade-invalidated actions rejected with the vetoing record's attribution", + async () => { + let storage = makeStorage(); + putAction(storage, 1, { state: "approved" }); + let vetoId = putAction(storage, 2, + { state: "rejected", vetoPending: true, resolvedBy: REJECTER }); + let a3 = putAction(storage, 3, { autoApprovable: false }); + + let { target, results } = makeBatchGatekeeper(); + results.push({ invalidatedByVeto: [{ action: 3, invalidatedBy: 2 }] }); + let decided = await makeDriver(storage, target).sync(GK); + + expect(decided).toEqual([a3]); + let invalidated = getAction(storage, 3); + expect(invalidated.state).toBe("rejected"); + expect(invalidated.cascadedFrom).toBe(vetoId); + expect(invalidated.resolvedBy?.id).toBe(REJECTER.id); + }); + + it("ignores invalidations for unknown or already-decided actions", async () => { + let storage = makeStorage(); + putAction(storage, 1, { state: "approved" }); + putAction(storage, 2, { state: "rejected", vetoPending: true, resolvedBy: REJECTER }); + + let { target, results } = makeBatchGatekeeper(); + results.push({ invalidatedByVeto: [ + { action: 1, invalidatedBy: 2 }, // already applied + { action: 99, invalidatedBy: 2 }, // unknown + ]}); + let decided = await makeDriver(storage, target).sync(GK); + + expect(decided).toEqual([]); + expect(getAction(storage, 1).state).toBe("approved"); + }); + + it("coalesces concurrent approvals into one follow-up pass at the highest frontier", async () => { + let storage = makeStorage(); + putAction(storage, 1, { autoApprovable: false }); + putAction(storage, 2, { autoApprovable: false }); + putAction(storage, 3, { autoApprovable: false }); + + let calls: Array<{actionId: number, vetoes: number[]}> = []; + let gates: Array<() => void> = []; + let target = { + applyActionsThrough(actionId: number, vetoes: number[]) { + calls.push({ actionId, vetoes }); + return new Promise(resolve => { + gates.push(() => resolve({})); + }); + }, + } as unknown as GatekeeperActionTarget; + let driver = makeDriver(storage, target); + + let first = driver.sync(GK, { frontier: 1, resolvedBy: APPROVER }); // parks mid-RPC + await flush(); + let second = driver.sync(GK, { frontier: 3, resolvedBy: APPROVER }); // staged + let third = driver.sync(GK, { frontier: 2, resolvedBy: APPROVER }); // merged with second + expect(calls).toEqual([{ actionId: 1, vetoes: [] }]); + + gates.shift()!(); // finish pass 1 + await flush(); + expect(calls).toEqual([{ actionId: 1, vetoes: [] }, { actionId: 3, vetoes: [] }]); + + gates.shift()!(); // finish pass 2 + let [a, b, c] = await Promise.all([first, second, third]); + expect(a).toEqual([10]); + // The coalesced requests share the pass and its decided set. + expect(b.toSorted((x, y) => x - y)).toEqual([20, 30]); + expect(c).toBe(b); + for (let action of [1, 2, 3]) expect(getAction(storage, action).state).toBe("approved"); + }); + + it("settled() resolves only after the in-flight pass (and its reruns) complete", async () => { + let storage = makeStorage(); + putAction(storage, 1, { autoApprovable: false }); + + let gates: Array<() => void> = []; + let target = { + applyActionsThrough() { + return new Promise(resolve => { + gates.push(() => resolve({})); + }); + }, + } as unknown as GatekeeperActionTarget; + let driver = makeDriver(storage, target); + + let pass = driver.sync(GK, { frontier: 1, resolvedBy: APPROVER }); + await flush(); + let settledDone = false; + let settled = driver.settled(GK).then(() => { settledDone = true; }); + await flush(); + expect(settledDone).toBe(false); + + gates.shift()!(); + await Promise.all([pass, settled]); + expect(settledDone).toBe(true); + }); + + it("propagates a transport failure to the awaiting caller and recovers on the next sync", + async () => { + let storage = makeStorage(); + putAction(storage, 1, { autoApprovable: false }); + + let { target, results } = makeBatchGatekeeper(); + results.push(new Error("network unreachable")); + let driver = makeDriver(storage, target); + + await expect(driver.sync(GK, { frontier: 1, resolvedBy: APPROVER })) + .rejects.toThrow("network unreachable"); + expect(getAction(storage, 1).state).toBe("pending"); + + await driver.sync(GK, { frontier: 1, resolvedBy: APPROVER }); + expect(getAction(storage, 1).state).toBe("approved"); + }); + + it("returns decided awaited records across chats so every affected turn can resume", async () => { + let storage = makeStorage(); + let a1 = putAction(storage, 1, { autoApprovable: false, chatId: 7, awaitDecision: true }); + let a2 = putAction(storage, 2, { autoApprovable: false, chatId: 8, awaitDecision: true }); + + let { target } = makeBatchGatekeeper(); + let decided = await makeDriver(storage, target) + .sync(GK, { frontier: 2, resolvedBy: APPROVER }); + + expect(decided.toSorted((a, b) => a - b)).toEqual([a1, a2]); + }); +}); + +describe("ActionSyncDriver legacy fallback", () => { + it("falls back on workerd's method-missing TypeError, delivering vetoes then applies in " + + "ascending order, and probes only once", async () => { + let storage = makeStorage(); + putAction(storage, 1, { autoApprovable: false }); + putAction(storage, 2, { state: "rejected", vetoPending: true, resolvedBy: REJECTER }); + putAction(storage, 3, { autoApprovable: false }); + + let legacy = makeLegacyGatekeeper({ remote: true }); + let driver = makeDriver(storage, legacy.target); + + await driver.sync(GK, { frontier: 3, resolvedBy: APPROVER }); + + // Vetoes first (the {restart} return is discarded), then pending actions ascending. + expect(legacy.calls).toEqual(["reject:2", "apply:1", "apply:3"]); + expect(legacy.probeCount()).toBe(1); + expect(getAction(storage, 1).state).toBe("approved"); + expect(getAction(storage, 2).vetoPending).toBeUndefined(); + expect(getAction(storage, 3).state).toBe("approved"); + + // The legacy verdict is cached: a later pass goes straight to per-action calls. + putAction(storage, 4, { autoApprovable: false }); + await driver.sync(GK, { frontier: 4, resolvedBy: APPROVER }); + expect(legacy.probeCount()).toBe(1); + expect(legacy.calls).toEqual(["reject:2", "apply:1", "apply:3", "apply:4"]); + }); + + it("handles a target with no applyActionsThrough at all", async () => { + let storage = makeStorage(); + putAction(storage, 1, { autoApprovable: false }); + + let legacy = makeLegacyGatekeeper(); + await makeDriver(storage, legacy.target).sync(GK, { frontier: 1, resolvedBy: APPROVER }); + + expect(legacy.calls).toEqual(["apply:1"]); + expect(getAction(storage, 1).state).toBe("approved"); + }); + + it("synthesizes {stopped} from the first legacy apply failure", async () => { + let storage = makeStorage(); + putAction(storage, 1, { autoApprovable: false }); + putAction(storage, 2, { autoApprovable: false }); + putAction(storage, 3, { autoApprovable: false }); + + let legacy = makeLegacyGatekeeper({ failApply: [2] }); + await makeDriver(storage, legacy.target).sync(GK, { frontier: 3, resolvedBy: APPROVER }); + + expect(legacy.calls).toEqual(["apply:1", "apply:2"]); // never skips ahead of the failure + expect(getAction(storage, 1).state).toBe("approved"); + let stopped = getAction(storage, 2); + expect(stopped.state).toBe("pending"); + expect(stopped.failure).toBe("apply 2 failed"); + expect(getAction(storage, 3).state).toBe("pending"); + }); + + it("keeps delivering vetoes even when a legacy reject throws", async () => { + let storage = makeStorage(); + putAction(storage, 1, { state: "approved" }); + putAction(storage, 2, { state: "rejected", vetoPending: true, resolvedBy: REJECTER }); + + let legacy = makeLegacyGatekeeper(); + legacy.target.rejectAction = + (async () => { throw new Error("already settled"); }) as typeof legacy.target.rejectAction; + await makeDriver(storage, legacy.target).sync(GK); + + // The reject was attempted once and is not re-staged: legacy gatekeepers throw forever on + // settled actions, so retrying would wedge the queue. + expect(getAction(storage, 2).vetoPending).toBeUndefined(); + }); +}); diff --git a/packages/workshop-backend/__tests__/auto-approval.test.ts b/packages/workshop-backend/__tests__/auto-approval.test.ts deleted file mode 100644 index 0b4ce7389..000000000 --- a/packages/workshop-backend/__tests__/auto-approval.test.ts +++ /dev/null @@ -1,214 +0,0 @@ -import { describe, it, expect } from "vitest"; -import { createTypedStorage, collection } from "@gadgets/typed-storage"; -import { AutoApprovalDrainer, AutoApprovalStorage, ApplyPendingActionFn } from "../src/auto-approval.js"; -import type { ActionRecord, AutoApproveTagRecord } from "../src/overseer.js"; -import type { AiChatAuthorInfo } from "@gadgets/workshop-shared/api"; -import { makeMockStorage } from "./mock-storage.js"; - -function makeStorage(): AutoApprovalStorage { - return createTypedStorage(makeMockStorage(), { - collections: { - actions: collection()({ primaryKey: "id" }), - autoApproveTags: collection()({ - primaryKey: (r: AutoApproveTagRecord) => `${r.gatekeeperId}:${r.actionKind.tag}`, - }), - }, - }); -} - -const GK = 1; -const ENABLER: AiChatAuthorInfo = { type: "user", id: "enabler@example.com", name: "Enabler" }; - -function enableRule(storage: AutoApprovalStorage, actionTag = "edit", gatekeeperId = GK) { - storage.autoApproveTags.put({ - gatekeeperId, actionKind: { tag: actionTag, label: "Edits" }, enabledBy: ENABLER }); -} - -function putAction( - storage: AutoApprovalStorage, id: number, - opts: { gatekeeperId?: number; actionTag?: string; autoApprovable?: boolean; - state?: ActionRecord["state"] } = {}) { - storage.actions.put({ - id, - gatekeeperId: opts.gatekeeperId ?? GK, - caller: { from: "agent", chatId: 1 }, - createdAt: new Date(), - state: opts.state ?? "pending", - type: "action", - action: id, - description: { - title: `Action ${id}`, - description: `Action ${id} description`, - implementsRevert: true, - actionKind: { tag: opts.actionTag ?? "edit", label: "Edits" }, - autoApprovable: opts.autoApprovable ?? true, - }, - }); -} - -function getAction(storage: AutoApprovalStorage, id: number): ActionRecord & {type: "action"} { - let record = storage.actions.get(id); - if (!record || record.type !== "action") throw new Error(`No action ${id}`); - return record; -} - -// An apply fn that resolves immediately, mirroring OverseerImpl.applyPendingAction's effect: -// mark the record approved and persist. Records the order of applied action ids. -function makeImmediateApply(storage: AutoApprovalStorage) { - let calls: number[] = []; - let applyFn: ApplyPendingActionFn = async (record, resolvedBy, autoApproved) => { - calls.push(record.id); - let fresh = storage.actions.get(record.id); - if (fresh && fresh.type === "action") { - fresh.state = "approved"; - fresh.appliedAt = new Date(); - fresh.resolvedBy = resolvedBy; - fresh.autoApproved = autoApproved; - storage.actions.put(fresh); - } - }; - return { applyFn, calls }; -} - -// An apply fn whose every invocation parks on a test-held promise until released. Lets a test hold -// an apply mid-flight (input gate open) while launching a second concurrent drain. On release it -// performs the same approve+persist effect as the real apply. -function makeControlledApply(storage: AutoApprovalStorage) { - let calls: number[] = []; - let gates: Array<() => void> = []; - let applyFn: ApplyPendingActionFn = (record, resolvedBy, autoApproved) => { - calls.push(record.id); - return new Promise((resolve) => { - gates.push(() => { - let fresh = storage.actions.get(record.id); - if (fresh && fresh.type === "action") { - fresh.state = "approved"; - fresh.appliedAt = new Date(); - fresh.resolvedBy = resolvedBy; - fresh.autoApproved = autoApproved; - storage.actions.put(fresh); - } - resolve(); - }); - }); - }; - return { - applyFn, - calls, - inFlight: () => gates.length, - releaseNext() { - let gate = gates.shift(); - if (!gate) throw new Error("no apply in flight to release"); - gate(); - }, - }; -} - -// Drain all microtasks (and the macrotask queue) so suspended drain continuations run to their next -// park point. -function flush(): Promise { - return new Promise((resolve) => setTimeout(resolve, 0)); -} - -describe("AutoApprovalDrainer.drain", () => { - it("applies all eligible pending actions in ascending id order", async () => { - let storage = makeStorage(); - enableRule(storage); - putAction(storage, 1); - putAction(storage, 2); - putAction(storage, 3); - - let { applyFn, calls } = makeImmediateApply(storage); - await new AutoApprovalDrainer(storage, applyFn).drain(GK); - - expect(calls).toEqual([1, 2, 3]); - for (let id of [1, 2, 3]) { - let record = getAction(storage, id); - expect(record.state).toBe("approved"); - expect(record.autoApproved).toBe(true); - expect(record.resolvedBy?.id).toBe(ENABLER.id); - } - }); - - it("stops at a manual gate without skipping ahead, then resumes once it clears", async () => { - let storage = makeStorage(); - enableRule(storage); - putAction(storage, 1); - putAction(storage, 2, { autoApprovable: false }); // manual gate - putAction(storage, 3); - - let { applyFn, calls } = makeImmediateApply(storage); - let drainer = new AutoApprovalDrainer(storage, applyFn); - await drainer.drain(GK); - - // Only the action before the gate is applied; the gate and everything behind it stay pending. - expect(calls).toEqual([1]); - expect(getAction(storage, 2).state).toBe("pending"); - expect(getAction(storage, 3).state).toBe("pending"); - - // Clear the gate (as a manual approval would) and re-drain: the rest applies, still in order. - let gate = getAction(storage, 2); - gate.state = "approved"; - storage.actions.put(gate); - await drainer.drain(GK); - - expect(calls).toEqual([1, 3]); - expect(getAction(storage, 3).state).toBe("approved"); - }); - - // Two concurrent drains for the same gatekeeper must not double-apply. The input gate is open - // across the apply await, so without the single-flight guard the second drain's pending re-check - // would see the still-"pending" record and apply it again. - it("never applies an action more than once under concurrent drains", async () => { - let storage = makeStorage(); - enableRule(storage); - putAction(storage, 1); - - let apply = makeControlledApply(storage); - let drainer = new AutoApprovalDrainer(storage, apply.applyFn); - - let first = drainer.drain(GK); // starts, calls apply(1), parks mid-apply - let second = drainer.drain(GK); // must coalesce, not start a second apply - await second; - - expect(apply.calls).toEqual([1]); - expect(apply.inFlight()).toBe(1); - - apply.releaseNext(); // resolve apply(1); record becomes approved - await first; // rerun pass re-lists: action 1 no longer pending -> no re-apply - - expect(apply.calls).toEqual([1]); - expect(getAction(storage, 1).state).toBe("approved"); - }); - - // Work that arrives while a drain is parked must still be applied -- the coalescing - // "rerun" flag must not drop the wakeup. - it("applies work submitted while a drain is parked mid-apply", async () => { - let storage = makeStorage(); - enableRule(storage); - putAction(storage, 1); - - let apply = makeControlledApply(storage); - let drainer = new AutoApprovalDrainer(storage, apply.applyFn); - - let first = drainer.drain(GK); // parks mid-apply on action 1 - - putAction(storage, 2); // new eligible action arrives mid-drain - let second = drainer.drain(GK); // coalesces -> sets the rerun flag - await second; - expect(apply.calls).toEqual([1]); - - apply.releaseNext(); // finish action 1; rerun pass should pick up action 2 - await flush(); - - expect(apply.calls).toEqual([1, 2]); - expect(apply.inFlight()).toBe(1); - - apply.releaseNext(); // finish action 2 - await first; - - expect(apply.calls).toEqual([1, 2]); - expect(getAction(storage, 1).state).toBe("approved"); - expect(getAction(storage, 2).state).toBe("approved"); - }); -}); diff --git a/packages/workshop-backend/src/action-sync.ts b/packages/workshop-backend/src/action-sync.ts new file mode 100644 index 000000000..1d06637ec --- /dev/null +++ b/packages/workshop-backend/src/action-sync.ts @@ -0,0 +1,303 @@ +// Action-sync core: reconciles this workspace's pending action records with a gatekeeper through +// one batch `applyActionsThrough(actionId, vetoes)` call per pass. A pass computes the decision +// frontier (manual approvals staged by the caller, then auto-approval rules, then deliverable +// vetoes), makes the call, and translates the result back onto the records: everything at or below +// the applied frontier becomes "approved" with the right attribution, a `stopped` action keeps its +// pending state plus a display-safe `failure`, and veto-cascade invalidations become "rejected" +// with `cascadedFrom` attribution. +// +// A per-gatekeeper single-flight guard (the DO's input gate is open across the RPC await) +// coalesces concurrent requests into the next pass, so two approvals arriving together produce one +// call at the higher frontier. The gatekeeper accessor is injected, keeping the driver +// constructible over a mock storage in tests. + +import type { Collection } from "@gadgets/typed-storage"; +import type { AiChatAuthorInfo } from "@gadgets/workshop-shared/api"; +import type { ApplyActionsThroughResult, Gatekeeper } from "@gadgets/workshop-shared/gatekeeper"; +import { createWorkshopLogger } from "./observability"; +import type { ActionRecord, AutoApproveTagRecord } from "./overseer.js"; + +const logger = createWorkshopLogger("workshop.action.sync"); + +export interface ActionSyncStorage { + actions: Collection; + autoApproveTags: Collection; +} + +/** + * The slice of the gatekeeper stub surface the driver drives, derived from the RPC contract. + * `applyActionsThrough` is optional during the migration; on a live stub the property is always a + * callable proxy and an un-migrated gatekeeper throws when it is invoked (see isMethodMissing). + */ +export type GatekeeperActionTarget = + Pick>, "applyActionsThrough" | "applyAction" | "rejectAction">; + +export type GetGatekeeperFn = (gatekeeperId: number) => GatekeeperActionTarget; + +/** + * A staged manual approval: apply every undecided action through `frontier` (a gatekeeper-local + * action ID) under `resolvedBy`'s authority. + */ +export type ManualApproval = { frontier: number, resolvedBy: AiChatAuthorInfo }; + +type SyncIntent = { manualApprovals: ManualApproval[] }; + +type StagedSync = { + intent: SyncIntent; + resolve: (decided: number[]) => void; + reject: (error: unknown) => void; + promise: Promise; +}; + +// workerd raises `TypeError: The RPC receiver does not implement the method +// "applyActionsThrough".` for an un-migrated gatekeeper. The error is untyped after the RPC hop +// (only the message survives), so this matches the message text. +function isMethodMissing(error: unknown): boolean { + return error instanceof Error && + error.message.includes('does not implement the method "applyActionsThrough"'); +} + +export class ActionSyncDriver { + // Per-gatekeeper intent for the NEXT pass. A key is present while a request waits to be picked + // up; requests arriving mid-pass merge here, so work submitted during a pass isn't lost. + #staged = new Map(); + + // Per-gatekeeper single-flight guard. Key present => a run loop is active for that gatekeeper. + #running = new Map>(); + + // Gatekeepers observed to lack applyActionsThrough. In-memory only: a fresh isolate re-probes, + // which is what lets a migrated deploy shed the fallback without bookkeeping. + #legacy = new Set(); + + constructor( + private storage: ActionSyncStorage, + private getGatekeeper: GetGatekeeperFn) {} + + /** + * Reconcile the gatekeeper's queue, optionally staging a manual approval. Resolves with the + * workspace record IDs decided (approved or cascade-rejected) by the pass that carried this + * request's intent. Concurrent calls for the same gatekeeper coalesce into one pass. + */ + sync(gatekeeperId: number, manualApproval?: ManualApproval): Promise { + let slot = this.#staged.get(gatekeeperId); + if (!slot) { + let resolve!: (decided: number[]) => void; + let reject!: (error: unknown) => void; + let promise = new Promise((res, rej) => { resolve = res; reject = rej; }); + slot = { intent: { manualApprovals: [] }, resolve, reject, promise }; + this.#staged.set(gatekeeperId, slot); + } + if (manualApproval) slot.intent.manualApprovals.push(manualApproval); + + if (!this.#running.has(gatekeeperId)) { + this.#running.set(gatekeeperId, this.#run(gatekeeperId)); + } + return slot.promise; + } + + /** + * Resolves once no sync pass is in flight for the gatekeeper. Used by rejection: a veto must + * never be staged while a pass that might apply the same action is mid-RPC. + */ + async settled(gatekeeperId: number): Promise { + for (;;) { + let running = this.#running.get(gatekeeperId); + if (!running) return; + await running.catch(() => {}); + } + } + + async #run(gatekeeperId: number): Promise { + try { + for (;;) { + let slot = this.#staged.get(gatekeeperId); + if (!slot) break; + this.#staged.delete(gatekeeperId); + try { + slot.resolve(await this.#syncOnce(gatekeeperId, slot.intent)); + } catch (error) { + slot.reject(error); + } + } + } finally { + // Synchronous with the loop's empty-staged check above, so a request staged mid-pass either + // was picked up by the loop or sees #running empty and starts a fresh one. + this.#running.delete(gatekeeperId); + } + } + + async #syncOnce(gatekeeperId: number, intent: SyncIntent): Promise { + // Materialize a snapshot first: list() is a lazy generator over storage, and we mutate the + // actions collection as we reconcile. Keyed and ordered by `record.action` (the + // gatekeeper-local ID space) -- `record.id` shares a counter with observations and hooks. + let records = [...this.storage.actions.list()].filter( + (rec): rec is ActionRecord & {type: "action"} => + rec.gatekeeperId === gatekeeperId && rec.type === "action"); + let byAction = new Map(records.map(record => [record.action, record])); + let pending = records.filter(record => record.state === "pending") + .toSorted((a, b) => a.action - b.action); + let stagedVetoes = records.filter(record => record.state === "rejected" && record.vetoPending) + .toSorted((a, b) => a.action - b.action); + + // Decide the frontier and, for every pending action it covers, the attribution to record if + // the gatekeeper applies it. Attribution is captured now, before the RPC, so a rule removed + // mid-call can't leave an applied action unattributed: this is the single pending->approved + // chokepoint, and every transition must record the resolving user and whether it was + // automatic. + let manualAscending = intent.manualApprovals.toSorted((a, b) => a.frontier - b.frontier); + let frontier = manualAscending.at(-1)?.frontier ?? 0; + let attribution = new Map(); + for (let record of pending) { + if (record.action <= frontier) { + // Covered by a manual approval; the smallest covering frontier's user takes + // responsibility for this earlier action riding along. + let covering = manualAscending.find(manual => manual.frontier >= record.action); + if (covering) { + attribution.set(record.action, {resolvedBy: covering.resolvedBy, autoApproved: false}); + continue; + } + } + // Above every manual frontier: extend while auto-eligible, exactly like the old drain. + // Eligibility requires BOTH signals: the author's `autoApprovable` verdict on the action AND + // a user-enabled rule for the action's kind. Stop at the first manual gate -- nothing is + // ever applied past one. + let tag = record.description.actionKind?.tag; + let rule = tag !== undefined + ? this.storage.autoApproveTags.get(`${gatekeeperId}:${tag}`) + : undefined; + if (record.description.autoApprovable !== true || rule === undefined) break; + attribution.set(record.action, {resolvedBy: rule.enabledBy, autoApproved: true}); + frontier = record.action; + } + + // Vetoes ride along up to the frontier. Beyond it, a staged veto is deliverable only when + // every action below it is already decided (the frontier may equal the current one for + // veto-only delivery) -- a veto must never drag undecided actions into application. + let firstUndecided = pending.find(record => record.action > frontier)?.action ?? Infinity; + for (let veto of stagedVetoes) { + if (veto.action < firstUndecided && veto.action > frontier) frontier = veto.action; + } + let sendVetoes = stagedVetoes.filter(veto => veto.action <= frontier); + + if (attribution.size === 0 && sendVetoes.length === 0) return []; + + let result = await this.#applyThrough( + gatekeeperId, frontier, sendVetoes.map(veto => veto.action), [...attribution.keys()]); + + // Reconcile. The contract makes `appliedThrough` sound despite ID holes: a gatekeeper never + // silently skips a pending in-range action -- it applies it or reports it via `stopped`. + let appliedThrough = result.stopped ? result.stopped.at - 1 : frontier; + let decided: number[] = []; + + for (let [actionId, attr] of attribution) { + if (actionId > appliedThrough) continue; + let fresh = this.#freshAction(byAction, actionId); + if (!fresh || fresh.state !== "pending") continue; + fresh.state = "approved"; + fresh.appliedAt = new Date(); + fresh.resolvedBy = attr.resolvedBy; + fresh.autoApproved = attr.autoApproved; + delete fresh.failure; + this.storage.actions.put(fresh); + decided.push(fresh.id); + } + + // Cascade invalidations: display-attributed to the veto that caused them, resolved by the + // user whose rejection it was. + for (let entry of result.invalidatedByVeto ?? []) { + let fresh = this.#freshAction(byAction, entry.action); + if (!fresh || fresh.state !== "pending") continue; + let vetoer = byAction.get(entry.invalidatedBy); + fresh.state = "rejected"; + fresh.appliedAt = new Date(); + if (vetoer?.resolvedBy) fresh.resolvedBy = vetoer.resolvedBy; + fresh.cascadedFrom = vetoer?.id; + delete fresh.failure; + this.storage.actions.put(fresh); + decided.push(fresh.id); + } + + // The stopping action stays pending, carrying a display-safe reason the user can act on. + if (result.stopped) { + let fresh = this.#freshAction(byAction, result.stopped.at); + if (fresh?.state === "pending") { + fresh.failure = + result.stopped.reason?.message || "The gatekeeper could not apply this action."; + this.storage.actions.put(fresh); + logger.warn("apply stopped", { + event: "action.sync.stopped", actionId: fresh.id, error: result.stopped.reason, + }); + } + } + + // Sent vetoes are delivered even on a `stopped` result (gatekeepers process vetoes before + // applying), so clear their staging flag. + for (let veto of sendVetoes) { + let fresh = this.#freshAction(byAction, veto.action); + if (fresh?.vetoPending) { + delete fresh.vetoPending; + this.storage.actions.put(fresh); + } + } + + return decided; + } + + // Re-read a record immediately before mutating it, guarding against concurrent decisions made + // while the pass's RPC await held the input gate open. + #freshAction(byAction: Map, actionId: number) + : (ActionRecord & {type: "action"}) | undefined { + let record = byAction.get(actionId); + if (!record) return undefined; + let fresh = this.storage.actions.get(record.id); + return fresh?.type === "action" ? fresh : undefined; + } + + // Batch call with a legacy fallback for gatekeepers that predate applyActionsThrough + // (gadgets-internal). Delete this whole method body's fallback half -- and the #legacy cache -- + // once the fallback warning stops appearing in logs and the method becomes required. + async #applyThrough(gatekeeperId: number, actionId: number, vetoes: number[], + pendingPlan: number[]): Promise { + let gatekeeper = this.getGatekeeper(gatekeeperId); + + if (!this.#legacy.has(gatekeeperId)) { + try { + if (typeof gatekeeper.applyActionsThrough === "function") { + return await gatekeeper.applyActionsThrough(actionId, vetoes); + } + } catch (error) { + if (!isMethodMissing(error)) throw error; + } + this.#legacy.add(gatekeeperId); + logger.warn("gatekeeper does not implement applyActionsThrough; using per-action fallback", { + event: "action.sync.legacy", gatekeeperId, + }); + } + + // Legacy path: per-action calls in the same order the batch would use -- vetoes first, then + // pending actions ascending. Rejects are individually best-effort (some gatekeepers throw on + // already-settled actions, and a veto can arrive long after the fact); `{restart}` returns + // are discarded, as the overseer always has. Never reports `invalidatedByVeto` (display-only, + // so an un-migrated gatekeeper's cascades simply go unattributed). + for (let veto of vetoes) { + try { + await gatekeeper.rejectAction(veto); + } catch (error) { + logger.warn("legacy rejectAction failed", { + event: "action.sync.legacy.reject.failed", gatekeeperId, error, + }); + } + } + for (let action of pendingPlan) { + try { + await gatekeeper.applyAction(action); + } catch (error) { + return {stopped: { + at: action, + reason: error instanceof Error ? error : new Error(String(error)), + }}; + } + } + return {}; + } +} diff --git a/packages/workshop-backend/src/auto-approval.ts b/packages/workshop-backend/src/auto-approval.ts deleted file mode 100644 index 0c1fa032f..000000000 --- a/packages/workshop-backend/src/auto-approval.ts +++ /dev/null @@ -1,97 +0,0 @@ -// Auto-approval drain core: applies eligible pending actions in id order, with a per-gatekeeper -// single-flight guard so two concurrent drains (the DO's input gate is open across the apply await) -// can't double-apply the same action. The apply is injected, keeping this constructible over a -// mock storage in tests. - -import type { Collection } from "@gadgets/typed-storage"; -import type { AiChatAuthorInfo } from "@gadgets/workshop-shared/api"; -import { createWorkshopLogger } from "./observability"; -import type { ActionRecord, AutoApproveTagRecord } from "./overseer.js"; - -const logger = createWorkshopLogger("workshop.auto.approval"); - -export interface AutoApprovalStorage { - actions: Collection; - autoApproveTags: Collection; -} - -/** - * Applies a single eligible pending action: invoke the gatekeeper, mark it approved, persist. The - * caller has already validated that the record is still pending. - */ -export type ApplyPendingActionFn = ( - record: ActionRecord & {type: "action"}, - resolvedBy: AiChatAuthorInfo, - autoApproved: boolean) => Promise; - -export class AutoApprovalDrainer { - // Per-gatekeeper single-flight state. Key present => a drain is running for that gatekeeper; the - // value is a "rerun" flag, set when another drain is requested while one is in flight, so work - // submitted during a drain isn't lost. - #draining = new Map(); - - constructor( - private storage: AutoApprovalStorage, - private applyPendingAction: ApplyPendingActionFn) {} - - async drain(gatekeeperId: number): Promise { - if (this.#draining.has(gatekeeperId)) { - this.#draining.set(gatekeeperId, true); // ask the running drain to loop again - return; - } - this.#draining.set(gatekeeperId, false); - try { - do { - this.#draining.set(gatekeeperId, false); - await this.#drainOnce(gatekeeperId); - } while (this.#draining.get(gatekeeperId)); - } finally { - this.#draining.delete(gatekeeperId); - } - } - - // Apply all currently-eligible pending actions of the gatekeeper, in ascending id order. Stops at - // the first pending action that is NOT auto-eligible (a manual gate) or that throws while applying - // -- it is never skipped ahead of. This preserves in-order application and the invariant that - // nothing is silently applied past a human gate. - // - // Eligibility requires BOTH signals: the author's `autoApprovable` verdict on the action AND a - // user-enabled rule for the action's type on this gatekeeper. - async #drainOnce(gatekeeperId: number): Promise { - // Materialize a snapshot first: list() is a lazy generator over storage, and we mutate the - // actions collection (via applyPendingAction) as we go. - let pending = [...this.storage.actions.list()].filter( - (rec): rec is ActionRecord & {type: "action"} => - rec.gatekeeperId === gatekeeperId && rec.type === "action" && rec.state === "pending"); - - for (let record of pending) { - let tag = record.description.actionKind?.tag; - let rule = tag !== undefined - ? this.storage.autoApproveTags.get(`${gatekeeperId}:${tag}`) - : undefined; - if (record.description.autoApprovable !== true || rule === undefined) { - // A manual gate. Stop rather than skipping ahead to any later auto-eligible action. - break; - } - - // Re-check immediately before applying, to guard against a concurrent drain having already - // taken this one. - let fresh = this.storage.actions.get(record.id); - if (!fresh || fresh.type !== "action" || fresh.state !== "pending") { - continue; - } - - try { - // Attribute the auto-approval to the user who enabled the rule -- it runs under their - // authority. - await this.applyPendingAction(fresh, rule.enabledBy, true); - } catch (err) { - // Leave the action pending for manual handling and stop the drain (never skip ahead). - logger.error("auto-approval failed", { - event: "auto.approval.failed", actionId: fresh.id, error: err, - }); - break; - } - } - } -} diff --git a/packages/workshop-backend/src/overseer.ts b/packages/workshop-backend/src/overseer.ts index e5fed5bb2..b1be3e72d 100644 --- a/packages/workshop-backend/src/overseer.ts +++ b/packages/workshop-backend/src/overseer.ts @@ -35,7 +35,7 @@ import { checkUsageAndBalance } from "./ai-gateway-billing/limits/usage-checker" import { completeAgentCatalogSnapshot, normalizeAgentCatalog } from "./agent-catalog"; import { refreshCachedBalance } from "./ai-gateway-billing/cloudflare/connection-service"; import { SharingManager, SharingCaller, CollaboratorRecord, ShareKeyRecord } from "./sharing"; -import { AutoApprovalDrainer } from "./auto-approval"; +import { ActionSyncDriver, ManualApproval } from "./action-sync"; import { collectSlashCommands, invokeSlashCommand } from "./slash-commands"; import { createWorkshopLogger, obsContext, traced } from "./observability"; import { wrapDoStubForTelemetry } from "./do-telemetry"; @@ -522,6 +522,25 @@ export type ActionRecord = { description: ActionDescription; resolvedBy?: AiChatAuthorInfo; // set when resolved (approved/rejected); absent while pending (or legacy) autoApproved?: boolean; // set when applied by an auto-approval rule rather than a human + + /** + * Display-safe reason the most recent application attempt stopped at this action. Only present + * while the action remains pending; cleared when it applies or is rejected. + */ + failure?: string; + + /** + * Workspace action ID of the rejected action whose veto invalidated this one. Only set when + * `state` is "rejected" and the rejection came from a gatekeeper dependency cascade. + */ + cascadedFrom?: number; + + /** + * Present while the user's rejection has not yet been delivered to the gatekeeper as a veto. + * The sync driver clears it on delivery. Absent on records rejected before this field existed, + * so legacy rejections are never re-delivered. + */ + vetoPending?: true; } | { type: "observation"; description: ObservationDescription; @@ -730,6 +749,8 @@ function actionRecordToLog(record: ActionRecord): ActionLogEntry { description: record.description, resolvedBy: record.resolvedBy, autoApproved: record.autoApproved, + cascadedFrom: record.cascadedFrom, + failure: record.failure, }; case "bindHook": return { @@ -1117,7 +1138,7 @@ class OverseerImpl implements AgentHooks { #liveChats = new Map(); #chatSubscribers: Set> = new Set(); - #autoApprovalDrainer: AutoApprovalDrainer; + #actionSync: ActionSyncDriver; #preparingChatMessages = new Map>(); @@ -1403,10 +1424,8 @@ class OverseerImpl implements AgentHooks { this.#migrateStorage(); this.defaultGadgetId = this.storage.defaultGadgetId.get(); - this.#autoApprovalDrainer = new AutoApprovalDrainer( - this.storage, - (record, resolvedBy, autoApproved) => - this.applyPendingAction(record, resolvedBy, autoApproved)); + this.#actionSync = new ActionSyncDriver( + this.storage, gatekeeperId => this.getGatekeeperFacet(gatekeeperId)); // Mirror every gadget-registry change into the owner's outputs index. Subscribing here makes // the registry the single chokepoint, so creation, acceptance, renaming, reverting and @@ -2567,35 +2586,18 @@ class OverseerImpl implements AgentHooks { }); } - // Apply a single pending action: invoke the gatekeeper, mark it approved, and persist (the put - // auto-notifies subscribeToActions). Shared by manual approval (`approveAction`) and the - // auto-approval drain (`drainAutoApprovals`). The caller is responsible for validating that the - // record is still pending before calling. - // - // `resolvedBy`/`autoApproved` are required (not defaulted) so that no apply path can omit how the - // gate was cleared: this is the single chokepoint where an action transitions to "approved", so - // requiring them here guarantees the audit log always records the resolving user and whether it - // was applied automatically. For an auto-approval, `resolvedBy` is the user who enabled the rule. - async applyPendingAction(record: ActionRecord & {type: "action"}, - resolvedBy: AiChatAuthorInfo, autoApproved: boolean): Promise { - let gatekeeper = this.getGatekeeperFacet(record.gatekeeperId); - await gatekeeper.applyAction(record.action); - record.state = "approved"; - record.appliedAt = new Date(); - record.resolvedBy = resolvedBy; - record.autoApproved = autoApproved; - this.storage.actions.put(record); + // Reconcile the gatekeeper's pending actions through one batch applyActionsThrough call: staged + // manual approvals, then auto-eligible actions (stopping at the first manual gate -- nothing is + // silently applied past one), then any deliverable staged vetoes. Resolves with the workspace + // record ids the pass decided. Delegates to the single-flight driver, which coalesces concurrent + // requests for the same gatekeeper (the DO's input gate is open across the RPC await). + syncActions(gatekeeperId: number, manualApproval?: ManualApproval): Promise { + return this.#actionSync.sync(gatekeeperId, manualApproval); } - // Apply all currently-eligible pending actions of the given gatekeeper, in ascending id order. - // Stops at the first pending action that is NOT auto-eligible (i.e. a manual gate) or that throws - // while applying -- it is never skipped ahead of. This preserves in-order application and the - // invariant that nothing is silently applied past a human gate. - // - // Delegates to the single-flight drainer, which guards against concurrent drains for the same - // gatekeeper double-applying an action (the DO's input gate is open across the apply await). - drainAutoApprovals(gatekeeperId: number): Promise { - return this.#autoApprovalDrainer.drain(gatekeeperId); + // Resolves once no sync pass is in flight for the gatekeeper (see ActionSyncDriver.settled). + settledActionSync(gatekeeperId: number): Promise { + return this.#actionSync.settled(gatekeeperId); } // Blocks other messages and agent turns for this chat until the returned object is disposed. @@ -2996,7 +2998,7 @@ class OverseerImpl implements AgentHooks { } if (willAutoApprove) { - this.ctx.waitUntil(this.drainAutoApprovals(gatekeeperId)); + this.ctx.waitUntil(this.syncActions(gatekeeperId)); } } @@ -7749,17 +7751,33 @@ class OverseerClientInterface extends RpcTarget implements Overseer { // Resolve the approver's identity before applying, so a failed profile fetch can't leave the // action applied in the world but still "pending" in storage. let profile = await this.#getClientProfile(); - await this.impl.applyPendingAction(action, profile, false); - // If this was an awaited agent action, resume only after all awaited actions in the turn are - // approved. If applyPendingAction throws, the action stays pending and the turn stays suspended. - if (action.caller.from === "agent" && action.description.awaitDecision) { - await this.#maybeResumeAfterActionDecision(action.caller.chatId); + // Approving is a decision frontier: the sync pass applies this action AND every earlier + // undecided action from the same gatekeeper (attributed to this approver), then continues + // through any auto-eligible actions the cleared gate unblocked. + let decided = await this.impl.syncActions( + action.gatekeeperId, {frontier: action.action, resolvedBy: profile}); + + let fresh = this.impl.storage.actions.get(id); + if (fresh?.type === "action" && fresh.state === "pending") { + // The gatekeeper stopped at (or before) this action; surface the display-safe reason so the + // user can resolve the problem and retry. The action stays pending. + throw new Error(fresh.failure ?? `Failed to apply action: ${id}`); } - // Clearing this manual gate may unblock later auto-eligible pending actions on the same - // gatekeeper, so cascade a drain (in-order) once this one is applied. - this.impl.ctx.waitUntil(this.impl.drainAutoApprovals(action.gatekeeperId)); + // Resume turns suspended on awaitDecision whose awaited actions this pass decided -- the batch + // may have covered earlier actions from other chats, not just the approved one. + let chatIds = new Set(); + for (let recordId of decided) { + let record = this.impl.storage.actions.get(recordId); + if (record?.type === "action" && record.caller.from === "agent" && + record.description.awaitDecision) { + chatIds.add(record.caller.chatId); + } + } + for (let chatId of chatIds) { + await this.#maybeResumeAfterActionDecision(chatId); + } } async listHooks(): Promise { @@ -7894,18 +7912,29 @@ class OverseerClientInterface extends RpcTarget implements Overseer { throw new Error(`Can't reject an observation: ${id}`); } - let gatekeeper = this.impl.getGatekeeperFacet(action.gatekeeperId); - - // Resolve the rejecter's identity before notifying the gatekeeper, so a failed profile fetch - // can't leave the action rejected with the gatekeeper but still "pending" in storage. + // Resolve the rejecter's identity first, so a failed profile fetch can't leave the action + // half-rejected. let profile = await this.#getClientProfile(); - await gatekeeper.rejectAction(action.action); + // A rejection must never interleave with an in-flight sync pass that might be applying this + // very action; wait it out, then re-check. + await this.impl.settledActionSync(action.gatekeeperId); + let fresh = this.impl.storage.actions.get(id); + if (fresh?.type !== "action" || fresh.state !== "pending") { + throw new Error(`Action is not pending: ${id}`); + } + + // The rejection is decided here and now; delivery to the gatekeeper is a staged veto. It goes + // out with the next sync pass whose frontier covers it -- immediately below, if every earlier + // action is already decided, otherwise once the actions below it are. + fresh.state = "rejected"; + fresh.appliedAt = new Date(); + fresh.resolvedBy = profile; + fresh.vetoPending = true; + delete fresh.failure; + this.impl.storage.actions.put(fresh); - action.state = "rejected"; - action.appliedAt = new Date(); - action.resolvedBy = profile; - this.impl.storage.actions.put(action); + this.impl.ctx.waitUntil(this.impl.syncActions(action.gatekeeperId)); // Deny leaves the turn ended, like denyConnectionRequest. The rejected record also prevents a // sibling approval from resuming this turn. @@ -7929,7 +7958,7 @@ class OverseerClientInterface extends RpcTarget implements Overseer { enabledBy: profile, }); // Apply the currently-visible pending action(s) with this tag right away. - this.impl.ctx.waitUntil(this.impl.drainAutoApprovals(gatekeeperId)); + this.impl.ctx.waitUntil(this.impl.syncActions(gatekeeperId)); } // Remove the auto-approval rule for `tag` on the given gatekeeper, so future matching actions diff --git a/packages/workshop-shared/src/api.ts b/packages/workshop-shared/src/api.ts index 8fb641693..ac26644fa 100644 --- a/packages/workshop-shared/src/api.ts +++ b/packages/workshop-shared/src/api.ts @@ -1467,7 +1467,7 @@ export interface CodeSubscriber { * Specifies the state of an action in the action log: * * pending: Action has not been applied yet. It is waiting for approval. * * approved: Action was approved and applied. - * * rejected: Action was rejected by the user. + * * rejected: Action was rejected by the user or invalidated by another rejected action. */ export type ActionState = "pending" | "approved" | "rejected"; @@ -1504,6 +1504,18 @@ export type ActionLogEntry = { * clicking Approve. Only ever set alongside state "approved" (there is no automatic rejection). */ autoApproved?: boolean; + + /** + * Workspace action ID whose rejection invalidated this action. Only set when `state` is + * "rejected" and the action was rejected as part of a dependency cascade. + */ + cascadedFrom?: number; + + /** + * Display-safe reason the most recent application attempt stopped at this action. Only set while + * the action remains pending. The action may be retried or rejected. + */ + failure?: string; } | { type: "observation"; description: ObservationDescription; @@ -1710,8 +1722,8 @@ export interface Overseer extends RpcTarget { listActions(): Promise; /** - * Approve an action that is currently in the "pending" state. The action will be performed on - * approval. + * Approve an action that is currently in the "pending" state. This performs the action and may + * also perform earlier pending actions from the same Gatekeeper connection. */ approveAction(id: number): Promise; From 374ed24adae3ae0f46d4e640e246b86a7bd043c5 Mon Sep 17 00:00:00 2001 From: Nathan Disidore Date: Mon, 17 Aug 2026 16:05:54 -0500 Subject: [PATCH 3/3] Surface batch-apply consequences in the action UI --- packages/workshop-frontend/src/Activity.tsx | 22 +++++++++- .../src/ActivityNotifications.tsx | 2 + .../workshop-frontend/src/ChatInterface.tsx | 19 ++++++++- .../src/components/ResolveButton.tsx | 10 +++++ packages/workshop-frontend/src/useActions.ts | 40 +++++++++++++++++++ 5 files changed, 90 insertions(+), 3 deletions(-) diff --git a/packages/workshop-frontend/src/Activity.tsx b/packages/workshop-frontend/src/Activity.tsx index f4ab21e53..66d93fdcc 100644 --- a/packages/workshop-frontend/src/Activity.tsx +++ b/packages/workshop-frontend/src/Activity.tsx @@ -8,7 +8,7 @@ import { GatekeeperIcon } from './components/GatekeeperIcon' import { HookToggle } from './components/HookToggle' import { AlwaysApproveButton, ResolveButton } from './components/ResolveButton' import { WorkshopButton } from './components/WorkshopControls' -import { useActions } from './useActions' +import { countEarlierPending, invalidationNote, useActions } from './useActions' import { useAutoApproval, autoApprovalKey, type AutoApprovalEntry } from './useAutoApproval' import { useAlwaysApproveTag } from './useAlwaysApproveTag' import { useAuthenticatedApi } from './AuthContext' @@ -233,6 +233,7 @@ export default function Activity({ toggleExpanded(record.id)} @@ -310,6 +311,7 @@ export default function Activity({ toggleExpanded(record.id)} togglingHook={record.type === 'bindHook' && record.hookId !== undefined @@ -489,6 +491,7 @@ function AutoApprovalPanel({ function ReviewRequest({ record, + earlierPendingCount, expanded, processing, onToggle, @@ -497,6 +500,7 @@ function ReviewRequest({ onAlwaysApprove, }: { record: ActionLogEntry + earlierPendingCount: number expanded: boolean processing: boolean onToggle: () => void @@ -505,6 +509,7 @@ function ReviewRequest({ onAlwaysApprove?: () => void }) { const resourceUrl = safeExternalUrl(record.resourceUrl) + const failure = record.type === 'action' ? record.failure : undefined return (
@@ -543,7 +548,12 @@ function ReviewRequest({ )} - +
@@ -552,18 +562,25 @@ function ReviewRequest({ {record.description.description}

)} + {failure && ( +

+ {failure} +

+ )}
) } function HistoryRow({ record, + invalidation, expanded, onToggle, togglingHook, onToggleHook, }: { record: ActionLogEntry + invalidation?: string expanded: boolean onToggle: () => void togglingHook: boolean @@ -620,6 +637,7 @@ function HistoryRow({ {autoApproved ? `Auto-approved (${resolvedBy.name}'s rule)` : `By ${resolvedBy.name}`} )} + {invalidation && {invalidation}} {resourceUrl && ( @@ -108,6 +109,7 @@ export default function ActivityNotifications({ tone="approve" disabled={isProcessing} onClick={() => void resolveAction(action.id, 'approve')} + earlierCount={countEarlierPending(pending, action)} /> diff --git a/packages/workshop-frontend/src/ChatInterface.tsx b/packages/workshop-frontend/src/ChatInterface.tsx index 23988f3e2..8e6b6965a 100644 --- a/packages/workshop-frontend/src/ChatInterface.tsx +++ b/packages/workshop-frontend/src/ChatInterface.tsx @@ -113,7 +113,7 @@ import DeleteConfirmationDialog from "./components/DeleteConfirmationDialog"; import AutoApproveConfirmDialog from "./components/AutoApproveConfirmDialog"; import { AlwaysApproveButton, ResolveButton } from "./components/ResolveButton"; import { WorkshopButton, WorkshopIconButton, WorkshopInput } from "./components/WorkshopControls"; -import { useActionEntries } from "./useActions"; +import { countEarlierPending, invalidationNote, useActionEntries, useActions } from "./useActions"; import { useAlwaysApproveTag } from "./useAlwaysApproveTag"; import { useResolveAction } from "./useResolveAction"; import { safeExternalUrl } from "./utils/safeExternalUrl"; @@ -5911,6 +5911,10 @@ function ChatInterface({ if (applyOptimisticActionState(actionId, state)) forceUpdate(); }); + // Full action list, for the approve buttons' "+N earlier" hint (approving applies every earlier + // pending action from the same connection) and for naming cascade-invalidation sources. + const { actionsById } = useActions(overseer); + // Handle enabling/disabling a bound hook from the chat thread. const handleToggleHook = async (actionId: number, hookId: number, enabled: boolean) => { setProcessingActions((prev) => new Set(prev).add(actionId)); @@ -6463,6 +6467,9 @@ function ChatInterface({ const isPending = state === "pending"; const isApproved = state === "approved"; const isRejected = state === "rejected"; + const earlierCount = isPending ? countEarlierPending(actionsById.values(), log) : 0; + const failure = isAct && isPending ? log.failure : undefined; + const invalidation = isAct ? invalidationNote(log, actionsById) : undefined; // A blocking (awaitDecision) pending action suspends the agent turn and blocks the composer, so // present it as a prominent callout with its details expanded by default. const isBlocking = isPending && log.description.awaitDecision === true; @@ -6518,6 +6525,7 @@ function ChatInterface({ variant={isBlocking ? "filled" : "quiet"} onClick={() => void resolveAction(msg.actionId, "approve")} disabled={isProc} + earlierCount={earlierCount} /> ) : null; @@ -6567,6 +6575,9 @@ function ChatInterface({
+ {failure && ( +

{failure}

+ )}
{actionControls} @@ -6627,6 +6638,12 @@ function ChatInterface({
+ {failure && ( +

{failure}

+ )} + {invalidation && ( +

{invalidation}

+ )} {resourceMeta}
)} diff --git a/packages/workshop-frontend/src/components/ResolveButton.tsx b/packages/workshop-frontend/src/components/ResolveButton.tsx index 26b51cfa6..669fc7373 100644 --- a/packages/workshop-frontend/src/components/ResolveButton.tsx +++ b/packages/workshop-frontend/src/components/ResolveButton.tsx @@ -5,12 +5,16 @@ export function ResolveButton({ variant = 'quiet', disabled, onClick, + earlierCount = 0, }: { tone: 'approve' | 'deny' variant?: 'quiet' | 'filled' disabled: boolean onClick: MouseEventHandler + /** Approve only: earlier pending actions from the same connection this approval also applies. */ + earlierCount?: number }) { + const showEarlier = tone === 'approve' && earlierCount > 0 const toneClassName = variant === 'filled' ? 'h-7 bg-kumo-brand px-3 text-white enabled:hover:opacity-90' : tone === 'approve' @@ -22,9 +26,15 @@ export function ResolveButton({ type="button" onClick={onClick} disabled={disabled} + title={showEarlier + ? `Also applies ${earlierCount} earlier pending ${earlierCount === 1 ? 'action' : 'actions'} from this connection.` + : undefined} className={`flex cursor-pointer items-center rounded-md text-[12px] font-medium tracking-[-0.15px] transition-colors disabled:cursor-not-allowed disabled:opacity-40 ${toneClassName}`} > {tone === 'approve' ? 'Approve' : 'Deny'} + {showEarlier && ( + +{earlierCount} earlier + )} ) } diff --git a/packages/workshop-frontend/src/useActions.ts b/packages/workshop-frontend/src/useActions.ts index 81d8acaad..1df476c3a 100644 --- a/packages/workshop-frontend/src/useActions.ts +++ b/packages/workshop-frontend/src/useActions.ts @@ -125,6 +125,46 @@ function release(overseer: RpcStub) { } } +/** + * Earlier pending actions from the same gatekeeper connection that approving `record` would also + * apply: the backend treats an approval as a decision frontier covering everything before it. + */ +export function countEarlierPending( + entries: Iterable, + record: ActionLogEntry, +): number { + if (record.type !== 'action' || record.gatekeeperId === undefined) return 0 + let count = 0 + for (const other of entries) { + if ( + other.type === 'action' && + other.state === 'pending' && + other.gatekeeperId === record.gatekeeperId && + other.id < record.id + ) { + count++ + } + } + return count +} + +/** + * Display note for a rejection that cascaded from another action's veto, naming the vetoed action + * when it is still in the local list. + */ +export function invalidationNote( + record: ActionLogEntry, + actionsById: Map, +): string | undefined { + if (record.type !== 'action' || record.state !== 'rejected' || record.cascadedFrom === undefined) { + return undefined + } + const source = actionsById.get(record.cascadedFrom) + return source?.type === 'action' + ? `Invalidated by rejection of “${source.description.title}”` + : 'Invalidated by a rejected action' +} + export type UseActionsResult = { actionsById: Map isReady: boolean