From 97877385bced49edb05d2670c9a45a38d5ba7d9c Mon Sep 17 00:00:00 2001 From: Nathan Disidore Date: Mon, 17 Aug 2026 15:38:07 -0500 Subject: [PATCH 1/5] Add apply-through gatekeeper contract --- packages/workshop-shared/src/gatekeeper.ts | 81 ++++++++++++++++------ 1 file changed, 60 insertions(+), 21 deletions(-) diff --git a/packages/workshop-shared/src/gatekeeper.ts b/packages/workshop-shared/src/gatekeeper.ts index 48bcdd729..a2883552d 100644 --- a/packages/workshop-shared/src/gatekeeper.ts +++ b/packages/workshop-shared/src/gatekeeper.ts @@ -688,6 +688,42 @@ export interface GatekeeperUser extends WorkerEntrypoint { */ export interface GatekeeperUserVerifier extends WorkerEntrypoint {} +/** Result of applying a Gatekeeper's queued actions through a decision frontier. */ +export interface ApplyActionsThroughResult { + /** + * Something went unexpectedly wrong at the given action number; remaining actions were not + * applied. The user may retry after resolving the problem or vetoing. + */ + stopped?: { + /** Gatekeeper-local action ID that could not be applied. */ + at: number; + + /** + * Explanation of why application stopped. Only the error's `message` survives the RPC hop, + * so it must stand alone as display-safe text, specific enough for the user to resolve the + * problem. + */ + reason: Error; + }; + + /** + * Indicates actions which were invalidated as a result of vetoes. Each entry's `action` is an + * action number which has been invalidated (these may be action numbers within the range just + * applied, as well as future action numbers not yet applied), and its `invalidatedBy` is the + * vetoed action number that invalidated it (always an action listed in `vetoes`). These actions + * will have no effect when applied (but will not produce an error either). The + * `invalidatedByVeto` list is provided for display purposes only, so that the UI may indicate the + * invalidated actions. + * + * Note that the Gatekeeper is not necessarily obliged to track when a veto may invalidate a + * future action. A Gatekeeper implementation may instead choose not to track dependencies, and + * instead let the future action fail with an error (producing `stopped`), leaving it up to the + * user to figure out the conflict and veto the dependent action manually. It is up to each + * Gatekeeper to decide the right trade-off between implementation complexity and UX. + */ + invalidatedByVeto?: Array<{action: number, invalidatedBy: number}>; +} + /** * Interface exposed by a Gatekeeper instance implementing a specific resource binding on a * specific Gadget. @@ -798,33 +834,39 @@ export interface Gatekeeper extends DurableObject { getSlashCommandProvider?(): Promise; // --------------------------------------------------------------------------- - // Callbacks invoked by the overseer to apply (or reject) actions that were previously queued - // for approval via the ApprovalQueue. + // Callback invoked by the overseer to resolve actions that were previously queued for approval + // via the ApprovalQueue. // // Each action is identified by a sequential integer action ID, assigned by the gatekeeper when - // it submits the action for approval. The action ID is passed back to these methods so the + // it submits the action for approval. The action ID is passed back to this method so the // gatekeeper can look up the action details in its own storage. /** - * Action was approved. This call should apply the action (or schedule it to be applied). + * Applies all actions through the given action ID (includes all previous actions that are not + * yet applied). Action IDs listed in `vetoes` are actions the user has rejected. + * + * Actions are applied in ascending ID order. Vetoed actions and actions invalidated by a veto + * become terminal no-ops. Processing stops at the first application failure; a pending in-range + * action the gatekeeper still holds must never be silently skipped — it is either applied or + * reported via `stopped`. An action whose `submitAction()` call has not yet completed must not + * be applied. * - * If this throws an exception, the user will be informed that the action failed and given the - * opportunity to retry or discard. + * `actionId` is the decision frontier and may equal the current frontier to deliver vetoes only. + * Every action ID in `vetoes` must be less than or equal to `actionId`. A veto may arrive long + * after the user rejected the action; delivery is opportunistic, not prompt. * - * Depending on policy conditions, an action may be approved and applied automatically. However, - * the gatekeeper is nevertheless expected to submit all actions for approval; there is no mode - * in which it's OK to skip the check. + * Calls must be idempotent. Missing IDs and vetoes of unknown or already-applied actions are + * ignored. A repeated request must re-report persisted invalidations attributable to its vetoes. */ + applyActionsThrough?(actionId: number, vetoes: number[]): Promise; + + /** @deprecated Implement `applyActionsThrough()` instead. */ applyAction(action: number): Promise; /** - * Indicates that an action was rejected by the user. The gatekeeper should clean up any - * associated storage. + * The returned `restart` flag is ignored; the overseer discards it. * - * If the returned `restart` flag is true, rejecting this action requires restarting the Gadget. - * This is sometimes needed by gatekeepers that simulate actions as if they had been approved -- - * the session may be in a state that is difficult to roll back without confusing the Gadget. - * The Overseer will take care of the restart, possibly after rejecting other actions. + * @deprecated Implement `applyActionsThrough()` instead. */ rejectAction(action: number): Promise; @@ -845,11 +887,9 @@ export interface Gatekeeper extends DurableObject { * `canRetry` should be true if the revert failed (for a reason described in `message`), but * it could make sense to retry later. In this case the UI will continue to give the user the * option to revert. - * - * `restart` has the same meaning as for `rejectAction()`. */ revertAction(action: number): - Promise; + Promise; } export interface ObservationAuthorizer extends RpcTarget { @@ -944,9 +984,8 @@ export interface ApprovalQueue extends ObservationAuthorizer { * be carried out until much later. It's intended that the user might not approve actions until * hours or days later, but this shouldn't cause any problems. * - * `action` is a sequential integer action ID assigned by the gatekeeper. It will be passed back - * to the Gatekeeper's applyAction() or rejectAction() when the action is later approved or - * rejected. + * `action` is a sequential integer action ID assigned by the gatekeeper. It will later be used as + * a decision frontier or veto in the Gatekeeper's `applyActionsThrough()` method. * * `description` describes the action in a way that can direct UI representation and policy * enforcement details. From aca13bcce92097e440ae07604e765426940d4161 Mon Sep 17 00:00:00 2001 From: Nathan Disidore Date: Mon, 17 Aug 2026 15:58:55 -0500 Subject: [PATCH 2/5] Drive action resolution through batch sync passes --- .../__tests__/actions.test.ts | 473 ++++++++++++++++++ .../__tests__/auto-approval.test.ts | 283 ----------- .../workshop-backend/__tests__/fixtures.ts | 5 +- packages/workshop-backend/src/actions.ts | 303 +++++++++++ .../workshop-backend/src/auto-approval.ts | 99 ---- packages/workshop-backend/src/overseer.ts | 148 ++++-- packages/workshop-shared/src/api.ts | 18 +- 7 files changed, 886 insertions(+), 443 deletions(-) create mode 100644 packages/workshop-backend/__tests__/actions.test.ts delete mode 100644 packages/workshop-backend/__tests__/auto-approval.test.ts create mode 100644 packages/workshop-backend/src/actions.ts delete mode 100644 packages/workshop-backend/src/auto-approval.ts diff --git a/packages/workshop-backend/__tests__/actions.test.ts b/packages/workshop-backend/__tests__/actions.test.ts new file mode 100644 index 000000000..08d515974 --- /dev/null +++ b/packages/workshop-backend/__tests__/actions.test.ts @@ -0,0 +1,473 @@ +import { describe, it, expect, vi } from "vitest"; +import { ActionSyncDriver, ActionSyncStorage, GatekeeperActionTarget } from "../src/actions.js"; +import type { ActionRecord } from "../src/overseer.js"; +import type { AiChatAuthorInfo } from "@gadgets/workshop-shared/api"; +import type { ApplyActionsThroughResult } from "@gadgets/workshop-shared/gatekeeper"; +import { makeActionStorage } from "./fixtures.js"; + +const makeStorage = makeActionStorage; + +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("does not scan resolved or unrelated action history", async () => { + let storage = makeStorage(); + enableRule(storage); + for (let action = 1; action <= 500; action++) { + putAction(storage, action, { state: "approved", gatekeeperId: GK + 1 }); + } + putAction(storage, 501); + putAction(storage, 502, { state: "rejected", vetoPending: true, resolvedBy: REJECTER }); + let fullScan = vi.spyOn(storage.actions, "list"); + + let { target } = makeBatchGatekeeper(); + await makeDriver(storage, target).sync(GK); + + expect(fullScan).not.toHaveBeenCalled(); + expect(getAction(storage, 501).state).toBe("approved"); + expect(getAction(storage, 502).vetoPending).toBeUndefined(); + }); + + 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("marks an action rejected, not approved, when the frontier covers it but the same pass's " + + "veto cascade-invalidates it", async () => { + let storage = makeStorage(); + let a1 = putAction(storage, 1, { autoApprovable: false }); + let vetoId = putAction(storage, 2, + { state: "rejected", vetoPending: true, resolvedBy: REJECTER }); + let a3 = putAction(storage, 3, { autoApprovable: false }); // depends on the vetoed action 2 + + // Approving 3 rides veto 2 along; the gatekeeper applies 1, deletes 3 as a cascade of 2. + let { target, calls, results } = makeBatchGatekeeper(); + results.push({ invalidatedByVeto: [{ action: 3, invalidatedBy: 2 }] }); + 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, 1).state).toBe("approved"); + 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 a2e0a42f2..000000000 --- a/packages/workshop-backend/__tests__/auto-approval.test.ts +++ /dev/null @@ -1,283 +0,0 @@ -import { describe, it, expect } from "vitest"; -import { AutoApprovalDrainer, AutoApprovalStorage, ApplyPendingActionFn } - from "../src/auto-approval.js"; -import type { ActionRecord } from "../src/overseer.js"; -import type { AiChatAuthorInfo } from "@gadgets/workshop-shared/api"; -import { makeMockStorage } from "./mock-storage.js"; -import { makeActionStorage, makePreIndexActionStorage, putAction } from "./fixtures.js"; - -const makeStorage = makeActionStorage; - -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 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"); - }); - - it("drains a large log, applying eligible actions in ascending order", async () => { - let storage = makeStorage(); - enableRule(storage); - let eligible: number[] = []; - for (let id = 0; id < 230; id++) { - if (id % 5 === 0) { - putAction(storage, id, { gatekeeperId: GK + 1 }); // other gatekeeper: skipped, not a gate - } else if (id % 5 === 1) { - putAction(storage, id, { state: "approved" }); // already resolved - } else { - putAction(storage, id); - eligible.push(id); - } - } - - let { applyFn, calls } = makeImmediateApply(storage); - await new AutoApprovalDrainer(storage, applyFn).drain(GK); - - expect(calls).toEqual(eligible); - }); - - it("halts at a manual gate deep in the log", async () => { - let storage = makeStorage(); - enableRule(storage); - let gateId = 105; - for (let id = 0; id < 120; id++) { - putAction(storage, id, { autoApprovable: id !== gateId }); - } - - let { applyFn, calls } = makeImmediateApply(storage); - await new AutoApprovalDrainer(storage, applyFn).drain(GK); - - expect(calls).toEqual(Array.from({ length: gateId }, (_, i) => i)); - expect(getAction(storage, gateId).state).toBe("pending"); - expect(getAction(storage, gateId + 1).state).toBe("pending"); - }); - - it("drains pendings written before the index existed once a rebuild backfills it", async () => { - // Mirrors the version-3 migration: the records predate the action-index declarations. - let mock = makeMockStorage(); - let legacy = makePreIndexActionStorage(mock); - putAction(legacy, 1); - putAction(legacy, 2, { state: "approved" }); - putAction(legacy, 3); - - let storage = makeStorage(mock); - storage.actions.pendingByGatekeeper.rebuild(); - storage.actions.byHistoryFilter.rebuild(); - storage.actions.byLastChanged.rebuild(); - enableRule(storage); - - // The apply persists a resolved state, which must not throw on the backfilled index. - let { applyFn, calls } = makeImmediateApply(storage); - await new AutoApprovalDrainer(storage, applyFn).drain(GK); - - expect(calls).toEqual([1, 3]); - expect(getAction(storage, 1).state).toBe("approved"); - expect(getAction(storage, 3).state).toBe("approved"); - }); - - it("halts when an apply fails, leaving it and everything after pending", async () => { - let storage = makeStorage(); - enableRule(storage); - putAction(storage, 1); - putAction(storage, 2); - putAction(storage, 3); - - let inner = makeImmediateApply(storage); - let applyFn: ApplyPendingActionFn = (record, resolvedBy, autoApproved) => { - if (record.id === 2) throw new Error("apply failed"); - return inner.applyFn(record, resolvedBy, autoApproved); - }; - await new AutoApprovalDrainer(storage, applyFn).drain(GK); - - expect(inner.calls).toEqual([1]); - expect(getAction(storage, 2).state).toBe("pending"); - expect(getAction(storage, 3).state).toBe("pending"); - }); - - // An action created after a drain snapshotted the pending index is out of that drain's scope; - // the creation path is responsible for its own drain() call (which the rerun flag folds in -- - // see the parked-mid-apply test above). - it("leaves actions created after the drain's snapshot for their own drain call", 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); // snapshots pending = [1] - - putAction(storage, 2); // arrives mid-drain, with no accompanying drain() call - apply.releaseNext(); - await first; - - expect(apply.calls).toEqual([1]); - expect(getAction(storage, 2).state).toBe("pending"); - }); -}); diff --git a/packages/workshop-backend/__tests__/fixtures.ts b/packages/workshop-backend/__tests__/fixtures.ts index 9afac58d0..14349c419 100644 --- a/packages/workshop-backend/__tests__/fixtures.ts +++ b/packages/workshop-backend/__tests__/fixtures.ts @@ -11,9 +11,8 @@ import type { ActionRecord } from "../src/overseer.js"; import { makeMockStorage } from "./mock-storage.js"; /** - * The production schema over mock storage, so the action suites (auto-approval drain, pending - * history query) exercise the shipped actions collection and pendingByGatekeeper index rather - * than a copy. + * The production schema over mock storage, so action-sync and history-query tests exercise the + * shipped actions collection and its indexes rather than a copy. */ export function makeActionStorage(mockStorage = makeMockStorage()) { return makeOverseerStorage(mockStorage); diff --git a/packages/workshop-backend/src/actions.ts b/packages/workshop-backend/src/actions.ts new file mode 100644 index 000000000..5ef71a172 --- /dev/null +++ b/packages/workshop-backend/src/actions.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, NonUniqueIndex } 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 & { + pendingByGatekeeper: NonUniqueIndex; + vetoPendingByGatekeeper: NonUniqueIndex; + }; + 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 StagedSync = { + manualApprovals: ManualApproval[]; + 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) { + slot = { manualApprovals: [], ...Promise.withResolvers() }; + this.#staged.set(gatekeeperId, slot); + } + if (manualApproval) slot.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.manualApprovals)); + } 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, manualApprovals: ManualApproval[]): Promise { + // Materialize before reconciling: index reads are lazy, and the pass mutates both indexes. The + // pending index was backfilled by the action-index migration; vetoPending only exists on records + // written after its index was introduced, so it needs no legacy backfill. Both are keyed by the + // workspace's gatekeeper ID, then ordered below by `record.action` (the gatekeeper-local ID). + let pending = [...this.storage.actions.pendingByGatekeeper.get(gatekeeperId)] + .filter((rec): rec is ActionRecord & {type: "action"} => rec.type === "action") + .toSorted((a, b) => a.action - b.action); + let stagedVetoes = [...this.storage.actions.vetoPendingByGatekeeper.get(gatekeeperId)] + .filter((rec): rec is ActionRecord & {type: "action"} => + rec.type === "action" && rec.state === "rejected" && rec.vetoPending === true) + .toSorted((a, b) => a.action - b.action); + let byAction = new Map([...pending, ...stagedVetoes].map(record => [record.action, record])); + + // 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 = manualApprovals.toSorted((a, b) => a.frontier - b.frontier); + let frontier = manualAscending.at(-1)?.frontier ?? 0; + let attribution = new Map(); + for (let record of pending) { + // 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[] = []; + + // Cascade invalidations first: an action inside the frontier can also be cascade-invalidated + // by a veto delivered in this same pass, and then it was deleted, not applied -- marking it + // rejected here keeps the approval loop below (which only touches pending records) from + // mislabeling it approved. Display-attributed to the veto that caused it, 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); + } + + 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); + } + + // 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 ba3e1078d..000000000 --- a/packages/workshop-backend/src/auto-approval.ts +++ /dev/null @@ -1,99 +0,0 @@ -// Auto-approval drain core: applies the gatekeeper's eligible pending actions (read off the sparse -// pendingByGatekeeper index) 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, NonUniqueIndex } 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 - & { pendingByGatekeeper: NonUniqueIndex }; - 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 before applying: the index yields lazily in ascending id order, and applying - // mutates it mid-iteration. Actions created after this snapshot trigger their own drain(), - // which drain()'s rerun flag folds into this run if it's still in flight. - let pending = [...this.storage.actions.pendingByGatekeeper.get(gatekeeperId)]; - - for (let record of pending) { - if (record.type !== "action") continue; - - 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. - return; - } - - // 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, - }); - return; - } - } - } -} diff --git a/packages/workshop-backend/src/overseer.ts b/packages/workshop-backend/src/overseer.ts index 889573a16..770533b69 100644 --- a/packages/workshop-backend/src/overseer.ts +++ b/packages/workshop-backend/src/overseer.ts @@ -43,7 +43,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 "./actions"; import { collectSlashCommands, invokeSlashCommand } from "./slash-commands"; import { createWorkshopLogger, obsContext, traced } from "./observability"; import { retryOnDoReset, wrapDoStubForTelemetry } from "./do-retry"; @@ -568,6 +568,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; @@ -893,6 +912,8 @@ function actionRecordToLog(record: ActionRecord): ActionLogEntry { description: record.description, resolvedBy: record.resolvedBy, autoApproved: record.autoApproved, + cascadedFrom: record.cascadedFrom, + failure: record.failure, }; case "bindHook": return { @@ -1094,7 +1115,8 @@ export function makeOverseerStorage(storage: DurableObjectStorage) { actions: collection()({ primaryKey: "id", - // All three indexes are backfilled by the version-3 migration. + // The three pre-veto indexes are backfilled by the version-3 migration. The sparse + // vetoPending index needs no backfill: that flag and index are introduced together. uniqueIndexes: { // Resume-replay index (see subscribeToActions): keyed by last state-change time so a // reconnect replays only the records changed during the gap. @@ -1102,12 +1124,20 @@ export function makeOverseerStorage(storage: DurableObjectStorage) { }, nonUniqueIndexes: { - // Sparse index over just the pending records, keyed by gatekeeper, so the auto-approval - // drain is O(pending on that gatekeeper) rather than a full-log scan. + // Sparse index over just the pending records, keyed by gatekeeper, so action sync is + // O(pending on that gatekeeper) rather than a full-log scan. pendingByGatekeeper(record: ActionRecord) { return record.state === "pending" ? record.gatekeeperId : null; }, + // Rejected actions awaiting veto delivery. Kept separate from pendingByGatekeeper because + // the states are disjoint and both sets are sparse. + vetoPendingByGatekeeper(record: ActionRecord) { + return record.type === "action" && record.state === "rejected" && record.vetoPending + ? record.gatekeeperId + : null; + }, + // Keyed by the wire ActionHistoryFilter values, in lockstep with // matchesActionHistoryFilter (api.ts), so every listActions() filter is one ranged // read. The "all" filter has no key: it reads the collection itself. @@ -1425,7 +1455,7 @@ class OverseerImpl implements AgentHooks { #liveChats = new Map(); #chatSubscribers: Set> = new Set(); - #autoApprovalDrainer: AutoApprovalDrainer; + #actionSync: ActionSyncDriver; #preparingChatMessages = new Map>(); @@ -1714,10 +1744,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 @@ -4268,35 +4296,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. @@ -4704,7 +4715,7 @@ class OverseerImpl implements AgentHooks { } if (willAutoApprove) { - this.ctx.waitUntil(this.drainAutoApprovals(gatekeeperId)); + this.ctx.waitUntil(this.syncActions(gatekeeperId)); } } @@ -9492,17 +9503,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 { @@ -9627,18 +9654,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. @@ -9662,7 +9700,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 3dfe0bad5..efb38a819 100644 --- a/packages/workshop-shared/src/api.ts +++ b/packages/workshop-shared/src/api.ts @@ -1475,7 +1475,7 @@ export type CommitInfo = { * 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"; @@ -1512,6 +1512,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; @@ -1780,8 +1792,8 @@ export interface Overseer extends RpcTarget { : 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 d377e6c81ccc8254252d6e29f259aea0a53dbe32 Mon Sep 17 00:00:00 2001 From: Nathan Disidore Date: Fri, 28 Aug 2026 11:07:21 -0500 Subject: [PATCH 3/5] Pin apply-through missing-method fallback --- .../__tests__/actions.test.ts | 28 +++++++++++++++++-- packages/workshop-backend/src/actions.ts | 17 +++++++---- 2 files changed, 37 insertions(+), 8 deletions(-) diff --git a/packages/workshop-backend/__tests__/actions.test.ts b/packages/workshop-backend/__tests__/actions.test.ts index 08d515974..fe1c1bc82 100644 --- a/packages/workshop-backend/__tests__/actions.test.ts +++ b/packages/workshop-backend/__tests__/actions.test.ts @@ -1,10 +1,18 @@ +import { env } from "cloudflare:workers"; import { describe, it, expect, vi } from "vitest"; -import { ActionSyncDriver, ActionSyncStorage, GatekeeperActionTarget } from "../src/actions.js"; -import type { ActionRecord } from "../src/overseer.js"; +import { + ActionSyncDriver, ActionSyncStorage, GatekeeperActionTarget, isMethodMissing, +} from "../src/actions.js"; +import type { ActionRecord, OverseerDurableObject } from "../src/overseer.js"; import type { AiChatAuthorInfo } from "@gadgets/workshop-shared/api"; import type { ApplyActionsThroughResult } from "@gadgets/workshop-shared/gatekeeper"; import { makeActionStorage } from "./fixtures.js"; +declare module "cloudflare:workers" { + interface ProvidedEnv { + TEST_OVERSEER: DurableObjectNamespace; + } +} const makeStorage = makeActionStorage; const GK = 1; @@ -402,6 +410,22 @@ describe("ActionSyncDriver.sync", () => { }); describe("ActionSyncDriver legacy fallback", () => { + it("recognizes workerd's real missing-method error", async () => { + let stub = env.TEST_OVERSEER.get(env.TEST_OVERSEER.newUniqueId()); + let call = (stub as any).applyActionsThrough(1, []); + let error: unknown; + try { + await call; + } catch (caught) { + error = caught; + } finally { + call[Symbol.dispose](); + } + + expect(error).toBeInstanceOf(Error); + expect((error as Error).message).toContain('does not implement "applyActionsThrough"'); + expect(isMethodMissing(error)).toBe(true); + }); it("falls back on workerd's method-missing TypeError, delivering vetoes then applies in " + "ascending order, and probes only once", async () => { let storage = makeStorage(); diff --git a/packages/workshop-backend/src/actions.ts b/packages/workshop-backend/src/actions.ts index 5ef71a172..dfffb4b41 100644 --- a/packages/workshop-backend/src/actions.ts +++ b/packages/workshop-backend/src/actions.ts @@ -50,12 +50,17 @@ type StagedSync = { 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"'); +/** + * Returns whether `error` is workerd's missing-`applyActionsThrough` RPC error. + * + * Production workerd includes `the method` in this error; Miniflare's real DO stub omits it. The + * error is untyped after the RPC hop (only the message survives), so both runtime variants are + * matched narrowly and retain the method name. + */ +export function isMethodMissing(error: unknown): boolean { + return error instanceof Error && ( + error.message.includes('does not implement the method "applyActionsThrough"') || + error.message.includes('does not implement "applyActionsThrough"')); } export class ActionSyncDriver { From 38120a1e7682d6f10ba6506a1c28f5aade407855 Mon Sep 17 00:00:00 2001 From: Nathan Disidore Date: Fri, 28 Aug 2026 14:50:27 -0500 Subject: [PATCH 4/5] Harden action sync against partial application Four fixes in the batch sync driver and its approval entry point: - The legacy per-action fallback (still the only path any gatekeeper implements) applied every action before persisting any approval. A failure after an apply landed left the record pending, and a replayed legacy applyAction throws on an already-applied action, wedging it forever. Approvals are now persisted as each one lands, through one idempotent approve() the batch reconcile replays harmlessly. - A legacy rejectAction failure cleared vetoPending unconditionally. Only a DO reset -- rolled back, hence provably undelivered -- now keeps the veto staged for the next pass; a settled/unknown action still clears, since it would throw forever. - A cascade invalidation naming an action submitted during the RPC await was discarded, leaving it pending to be recorded approved later though the gatekeeper had deleted it. The action map is refreshed before reconciling when invalidations are reported. - approveAction() returned success whenever the action was no longer pending, including when the same pass cascade-rejected it, which the client displays optimistically as approved. It now throws unless the record ended approved, and resumes awaitDecision turns first so a failure here can't strand another chat. --- .../__tests__/actions.test.ts | 50 +++++++++++ packages/workshop-backend/src/actions.ts | 86 ++++++++++++------- packages/workshop-backend/src/overseer.ts | 25 ++++-- 3 files changed, 123 insertions(+), 38 deletions(-) diff --git a/packages/workshop-backend/__tests__/actions.test.ts b/packages/workshop-backend/__tests__/actions.test.ts index fe1c1bc82..ab5c1681e 100644 --- a/packages/workshop-backend/__tests__/actions.test.ts +++ b/packages/workshop-backend/__tests__/actions.test.ts @@ -407,6 +407,23 @@ describe("ActionSyncDriver.sync", () => { expect(decided.toSorted((a, b) => a - b)).toEqual([a1, a2]); }); + + it("rejects a cascade-invalidated action that was submitted during the pass", async () => { + let storage = makeStorage(); + let vetoId = putAction(storage, 2, + { state: "rejected", vetoPending: true, resolvedBy: REJECTER }); + + let { target, results } = makeBatchGatekeeper(); + results.push({ invalidatedByVeto: [{ action: 3, invalidatedBy: 2 }] }); + let pass = makeDriver(storage, target).sync(GK); + let a3 = putAction(storage, 3, { autoApprovable: false }); // arrives while the RPC is in + let decided = await pass; // flight, so it misses the snapshot + + expect(decided).toContain(a3); + let invalidated = getAction(storage, 3); + expect(invalidated.state).toBe("rejected"); + expect(invalidated.cascadedFrom).toBe(vetoId); + }); }); describe("ActionSyncDriver legacy fallback", () => { @@ -494,4 +511,37 @@ describe("ActionSyncDriver legacy fallback", () => { // settled actions, so retrying would wedge the queue. expect(getAction(storage, 2).vetoPending).toBeUndefined(); }); + + it("records each legacy approval before issuing the next external call", async () => { + let storage = makeStorage(); + putAction(storage, 1, { autoApprovable: false }); + putAction(storage, 2, { autoApprovable: false }); + + // What action 1's record looks like at the moment each apply is issued: a crash (or an + // outcome-unknown failure) after the first one must not lose it, since a replayed legacy + // applyAction throws on an already-applied action. + let seen: string[] = []; + let target = { + async applyAction() { seen.push(getAction(storage, 1).state); }, + } as unknown as GatekeeperActionTarget; + await makeDriver(storage, target).sync(GK, { frontier: 2, resolvedBy: APPROVER }); + + expect(seen).toEqual(["pending", "approved"]); + expect(getAction(storage, 2).state).toBe("approved"); + }); + + it("keeps a veto staged when a legacy reject was rolled back by a DO reset", 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 Object.assign(new Error("Durable Object reset."), { durableObjectReset: true }); + }) as typeof legacy.target.rejectAction; + await makeDriver(storage, legacy.target).sync(GK); + + // Undelivered, not dropped: the next pass on this gatekeeper re-sends it. + expect(getAction(storage, 2).vetoPending).toBe(true); + }); }); diff --git a/packages/workshop-backend/src/actions.ts b/packages/workshop-backend/src/actions.ts index dfffb4b41..67ad1d012 100644 --- a/packages/workshop-backend/src/actions.ts +++ b/packages/workshop-backend/src/actions.ts @@ -14,6 +14,7 @@ import type { Collection, NonUniqueIndex } from "@gadgets/typed-storage"; import type { AiChatAuthorInfo } from "@gadgets/workshop-shared/api"; import type { ApplyActionsThroughResult, Gatekeeper } from "@gadgets/workshop-shared/gatekeeper"; +import { isDoResetError } from "./do-retry"; import { createWorkshopLogger } from "./observability"; import type { ActionRecord, AutoApproveTagRecord } from "./overseer.js"; @@ -183,19 +184,40 @@ export class ActionSyncDriver { 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[] = []; + // The single pending->approved chokepoint. Idempotent, so the legacy path can persist an + // approval the moment it lands and the reconcile loop below can replay it harmlessly. + let approve = (action: number) => { + let attr = attribution.get(action); + let fresh = this.#freshAction(byAction, action); + if (!attr || fresh?.state !== "pending") return; + 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); + }; + + let {result, undelivered} = await this.#applyThrough( + gatekeeperId, frontier, sendVetoes.map(veto => veto.action), [...attribution.keys()], + approve); + // Cascade invalidations first: an action inside the frontier can also be cascade-invalidated // by a veto delivered in this same pass, and then it was deleted, not applied -- marking it // rejected here keeps the approval loop below (which only touches pending records) from // mislabeling it approved. Display-attributed to the veto that caused it, resolved by the user // whose rejection it was. + if (result.invalidatedByVeto?.length) { + // A cascade may name an action submitted during the RPC await, which the pre-call snapshot + // can't contain; left pending it would later be recorded approved though the gatekeeper had + // deleted it. + for (let record of this.storage.actions.pendingByGatekeeper.get(gatekeeperId)) { + if (record.type === "action") byAction.set(record.action, record); + } + } for (let entry of result.invalidatedByVeto ?? []) { let fresh = this.#freshAction(byAction, entry.action); if (!fresh || fresh.state !== "pending") continue; @@ -209,17 +231,11 @@ export class ActionSyncDriver { decided.push(fresh.id); } - 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); + // 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; + for (let action of attribution.keys()) { + if (action <= appliedThrough) approve(action); } // The stopping action stays pending, carrying a display-safe reason the user can act on. @@ -236,8 +252,9 @@ export class ActionSyncDriver { } // Sent vetoes are delivered even on a `stopped` result (gatekeepers process vetoes before - // applying), so clear their staging flag. + // applying), so clear the staging flag on every one that landed. for (let veto of sendVetoes) { + if (undelivered?.includes(veto.action)) continue; let fresh = this.#freshAction(byAction, veto.action); if (fresh?.vetoPending) { delete fresh.vetoPending; @@ -258,17 +275,19 @@ export class ActionSyncDriver { 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. + // Batch call with a legacy fallback for gatekeepers that predate applyActionsThrough -- which + // is still all of them. Returns the pass result plus any vetoes that provably never reached the + // gatekeeper. 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 { + pendingPlan: number[], approve: (action: number) => void) + : Promise<{result: ApplyActionsThroughResult, undelivered?: number[]}> { let gatekeeper = this.getGatekeeper(gatekeeperId); if (!this.#legacy.has(gatekeeperId)) { try { if (typeof gatekeeper.applyActionsThrough === "function") { - return await gatekeeper.applyActionsThrough(actionId, vetoes); + return {result: await gatekeeper.applyActionsThrough(actionId, vetoes)}; } } catch (error) { if (!isMethodMissing(error)) throw error; @@ -280,29 +299,36 @@ export class ActionSyncDriver { } // 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). + // pending actions ascending. `{restart}` returns are discarded, as the overseer always has, + // and this path never reports `invalidatedByVeto` (display-only, so an un-migrated + // gatekeeper's cascades simply go unattributed). + let undelivered: number[] = []; for (let veto of vetoes) { try { await gatekeeper.rejectAction(veto); } catch (error) { + // A settled or unknown action throws forever, so the veto is dropped rather than + // re-staged; a DO reset rolled the call back, so that one is kept for a later pass. + if (isDoResetError(error)) undelivered.push(veto); logger.warn("legacy rejectAction failed", { event: "action.sync.legacy.reject.failed", gatekeeperId, error, }); } } + // Each approval is persisted as it lands: unlike a replayed frontier, a replayed per-action + // call throws on an already-applied action, so an unrecorded apply would wedge the record as + // pending forever. for (let action of pendingPlan) { try { await gatekeeper.applyAction(action); } catch (error) { - return {stopped: { + return {result: {stopped: { at: action, reason: error instanceof Error ? error : new Error(String(error)), - }}; + }}, undelivered}; } + approve(action); } - return {}; + return {result: {}, undelivered}; } } diff --git a/packages/workshop-backend/src/overseer.ts b/packages/workshop-backend/src/overseer.ts index 770533b69..758f198c1 100644 --- a/packages/workshop-backend/src/overseer.ts +++ b/packages/workshop-backend/src/overseer.ts @@ -9510,15 +9510,9 @@ class OverseerClientInterface extends RpcTarget implements Overseer { 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}`); - } - // 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. + // may have covered earlier actions from other chats, not just the approved one. Done before + // the outcome check below, so a failure on this action doesn't strand another chat's turn. let chatIds = new Set(); for (let recordId of decided) { let record = this.impl.storage.actions.get(recordId); @@ -9530,6 +9524,21 @@ class OverseerClientInterface extends RpcTarget implements Overseer { for (let chatId of chatIds) { await this.#maybeResumeAfterActionDecision(chatId); } + + let fresh = this.impl.storage.actions.get(id); + if (fresh?.type === "action" && fresh.state !== "approved") { + // Rejected: a veto this pass delivered cascade-invalidated it, or another client rejected it + // while the pass was in flight. Either way it was not applied, so this must not report + // success -- the client displays a resolved approval optimistically. + if (fresh.state === "rejected") { + throw new Error(fresh.cascadedFrom !== undefined + ? `Action was invalidated by a rejected earlier action: ${id}` + : `Action was rejected: ${id}`); + } + // Still pending: the gatekeeper stopped at (or before) this action; surface the + // display-safe reason so the user can resolve the problem and retry. + throw new Error(fresh.failure ?? `Failed to apply action: ${id}`); + } } async listHooks(): Promise { From c2151bfb3dd1b2005e4e0ef3a6a3e3c9dae6b4d5 Mon Sep 17 00:00:00 2001 From: Nathan Disidore Date: Fri, 28 Aug 2026 15:07:26 -0500 Subject: [PATCH 5/5] Name the action-sync entry points after their effect `syncActions` read as a read/mirror operation, but the call sends irreversible external writes. Renamed to `applyDecidedActions`, which also names the union the pass delivers: approvals (manual, then auto-eligible) applied, rejections delivered as vetoes. `settledActionSync` -> `awaitActionsSettled`, and the driver follows so the vocabulary doesn't split at the boundary: `sync` -> `apply`, `#syncOnce` -> `#applyOnce`, `settled` -> `awaitSettled`, `StagedSync` -> `StagedPass`. The class, log component, and event prefix keep "sync": the subsystem is action-sync, its operation is applying decisions. Also log a failed pass. Two callers deliver one through `waitUntil` (rejection, auto-approve opt-in) and never observe the rejection, so a background veto or auto-approval that failed to reach the gatekeeper left no structured trace. --- .../__tests__/actions.test.ts | 62 +++++++++---------- packages/workshop-backend/src/actions.ts | 19 +++--- packages/workshop-backend/src/overseer.ts | 26 ++++---- 3 files changed, 54 insertions(+), 53 deletions(-) diff --git a/packages/workshop-backend/__tests__/actions.test.ts b/packages/workshop-backend/__tests__/actions.test.ts index ab5c1681e..b42b03bf7 100644 --- a/packages/workshop-backend/__tests__/actions.test.ts +++ b/packages/workshop-backend/__tests__/actions.test.ts @@ -114,7 +114,7 @@ function flush(): Promise { return new Promise((resolve) => setTimeout(resolve, 0)); } -describe("ActionSyncDriver.sync", () => { +describe("ActionSyncDriver.apply", () => { it("applies through a manual frontier, attributing covered actions to the approver and " + "auto-extended ones to the rule enabler", async () => { let storage = makeStorage(); @@ -124,8 +124,7 @@ describe("ActionSyncDriver.sync", () => { 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 }); + let decided = await makeDriver(storage, target).apply(GK, { frontier: 2, resolvedBy: APPROVER }); expect(calls).toEqual([{ actionId: 3, vetoes: [] }]); expect(decided.toSorted((a, b) => a - b)).toEqual([a1, a2, a3]); @@ -149,7 +148,7 @@ describe("ActionSyncDriver.sync", () => { putAction(storage, 3); let { target, calls } = makeBatchGatekeeper(); - await makeDriver(storage, target).sync(GK); + await makeDriver(storage, target).apply(GK); expect(calls).toEqual([{ actionId: 1, vetoes: [] }]); expect(getAction(storage, 1).state).toBe("approved"); @@ -168,7 +167,7 @@ describe("ActionSyncDriver.sync", () => { let fullScan = vi.spyOn(storage.actions, "list"); let { target } = makeBatchGatekeeper(); - await makeDriver(storage, target).sync(GK); + await makeDriver(storage, target).apply(GK); expect(fullScan).not.toHaveBeenCalled(); expect(getAction(storage, 501).state).toBe("approved"); @@ -180,7 +179,7 @@ describe("ActionSyncDriver.sync", () => { putAction(storage, 1, { autoApprovable: false }); let { target, calls } = makeBatchGatekeeper(); - let decided = await makeDriver(storage, target).sync(GK); + let decided = await makeDriver(storage, target).apply(GK); expect(decided).toEqual([]); expect(calls).toEqual([]); @@ -196,7 +195,7 @@ describe("ActionSyncDriver.sync", () => { 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 }); + let first = await driver.apply(GK, { frontier: 2, resolvedBy: APPROVER }); expect(first).toEqual([a1]); expect(getAction(storage, 1).state).toBe("approved"); @@ -207,7 +206,7 @@ describe("ActionSyncDriver.sync", () => { // 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 }); + let retry = await driver.apply(GK, { frontier: 2, resolvedBy: APPROVER }); expect(retry).toEqual([getAction(storage, 2).id]); expect(calls).toEqual([{ actionId: 2, vetoes: [] }, { actionId: 2, vetoes: [] }]); @@ -222,7 +221,7 @@ describe("ActionSyncDriver.sync", () => { putAction(storage, 2, { state: "rejected", vetoPending: true, resolvedBy: REJECTER }); let { target, calls } = makeBatchGatekeeper(); - await makeDriver(storage, target).sync(GK); + await makeDriver(storage, target).apply(GK); expect(calls).toEqual([]); expect(getAction(storage, 2).vetoPending).toBe(true); @@ -237,7 +236,7 @@ describe("ActionSyncDriver.sync", () => { // 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); + await makeDriver(storage, target).apply(GK); expect(calls).toEqual([{ actionId: 2, vetoes: [2] }]); expect(getAction(storage, 2).vetoPending).toBeUndefined(); @@ -250,8 +249,7 @@ describe("ActionSyncDriver.sync", () => { let a3 = putAction(storage, 3, { autoApprovable: false }); let { target, calls } = makeBatchGatekeeper(); - let decided = await makeDriver(storage, target) - .sync(GK, { frontier: 3, resolvedBy: APPROVER }); + let decided = await makeDriver(storage, target).apply(GK, { frontier: 3, resolvedBy: APPROVER }); expect(calls).toEqual([{ actionId: 3, vetoes: [2] }]); expect(decided.toSorted((a, b) => a - b)).toEqual([a1, a3]); @@ -268,7 +266,7 @@ describe("ActionSyncDriver.sync", () => { let { target, results } = makeBatchGatekeeper(); results.push({ invalidatedByVeto: [{ action: 3, invalidatedBy: 2 }] }); - let decided = await makeDriver(storage, target).sync(GK); + let decided = await makeDriver(storage, target).apply(GK); expect(decided).toEqual([a3]); let invalidated = getAction(storage, 3); @@ -288,8 +286,7 @@ describe("ActionSyncDriver.sync", () => { // Approving 3 rides veto 2 along; the gatekeeper applies 1, deletes 3 as a cascade of 2. let { target, calls, results } = makeBatchGatekeeper(); results.push({ invalidatedByVeto: [{ action: 3, invalidatedBy: 2 }] }); - let decided = await makeDriver(storage, target) - .sync(GK, { frontier: 3, resolvedBy: APPROVER }); + let decided = await makeDriver(storage, target).apply(GK, { frontier: 3, resolvedBy: APPROVER }); expect(calls).toEqual([{ actionId: 3, vetoes: [2] }]); expect(decided.toSorted((a, b) => a - b)).toEqual([a1, a3]); @@ -310,7 +307,7 @@ describe("ActionSyncDriver.sync", () => { { action: 1, invalidatedBy: 2 }, // already applied { action: 99, invalidatedBy: 2 }, // unknown ]}); - let decided = await makeDriver(storage, target).sync(GK); + let decided = await makeDriver(storage, target).apply(GK); expect(decided).toEqual([]); expect(getAction(storage, 1).state).toBe("approved"); @@ -334,10 +331,10 @@ describe("ActionSyncDriver.sync", () => { } as unknown as GatekeeperActionTarget; let driver = makeDriver(storage, target); - let first = driver.sync(GK, { frontier: 1, resolvedBy: APPROVER }); // parks mid-RPC + let first = driver.apply(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 + let second = driver.apply(GK, { frontier: 3, resolvedBy: APPROVER }); // staged + let third = driver.apply(GK, { frontier: 2, resolvedBy: APPROVER }); // merged with second expect(calls).toEqual([{ actionId: 1, vetoes: [] }]); gates.shift()!(); // finish pass 1 @@ -367,10 +364,10 @@ describe("ActionSyncDriver.sync", () => { } as unknown as GatekeeperActionTarget; let driver = makeDriver(storage, target); - let pass = driver.sync(GK, { frontier: 1, resolvedBy: APPROVER }); + let pass = driver.apply(GK, { frontier: 1, resolvedBy: APPROVER }); await flush(); let settledDone = false; - let settled = driver.settled(GK).then(() => { settledDone = true; }); + let settled = driver.awaitSettled(GK).then(() => { settledDone = true; }); await flush(); expect(settledDone).toBe(false); @@ -388,11 +385,11 @@ describe("ActionSyncDriver.sync", () => { results.push(new Error("network unreachable")); let driver = makeDriver(storage, target); - await expect(driver.sync(GK, { frontier: 1, resolvedBy: APPROVER })) + await expect(driver.apply(GK, { frontier: 1, resolvedBy: APPROVER })) .rejects.toThrow("network unreachable"); expect(getAction(storage, 1).state).toBe("pending"); - await driver.sync(GK, { frontier: 1, resolvedBy: APPROVER }); + await driver.apply(GK, { frontier: 1, resolvedBy: APPROVER }); expect(getAction(storage, 1).state).toBe("approved"); }); @@ -402,8 +399,7 @@ describe("ActionSyncDriver.sync", () => { 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 }); + let decided = await makeDriver(storage, target).apply(GK, { frontier: 2, resolvedBy: APPROVER }); expect(decided.toSorted((a, b) => a - b)).toEqual([a1, a2]); }); @@ -415,7 +411,7 @@ describe("ActionSyncDriver.sync", () => { let { target, results } = makeBatchGatekeeper(); results.push({ invalidatedByVeto: [{ action: 3, invalidatedBy: 2 }] }); - let pass = makeDriver(storage, target).sync(GK); + let pass = makeDriver(storage, target).apply(GK); let a3 = putAction(storage, 3, { autoApprovable: false }); // arrives while the RPC is in let decided = await pass; // flight, so it misses the snapshot @@ -453,7 +449,7 @@ describe("ActionSyncDriver legacy fallback", () => { let legacy = makeLegacyGatekeeper({ remote: true }); let driver = makeDriver(storage, legacy.target); - await driver.sync(GK, { frontier: 3, resolvedBy: APPROVER }); + await driver.apply(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"]); @@ -464,7 +460,7 @@ describe("ActionSyncDriver legacy fallback", () => { // 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 }); + await driver.apply(GK, { frontier: 4, resolvedBy: APPROVER }); expect(legacy.probeCount()).toBe(1); expect(legacy.calls).toEqual(["reject:2", "apply:1", "apply:3", "apply:4"]); }); @@ -474,7 +470,7 @@ describe("ActionSyncDriver legacy fallback", () => { putAction(storage, 1, { autoApprovable: false }); let legacy = makeLegacyGatekeeper(); - await makeDriver(storage, legacy.target).sync(GK, { frontier: 1, resolvedBy: APPROVER }); + await makeDriver(storage, legacy.target).apply(GK, { frontier: 1, resolvedBy: APPROVER }); expect(legacy.calls).toEqual(["apply:1"]); expect(getAction(storage, 1).state).toBe("approved"); @@ -487,7 +483,7 @@ describe("ActionSyncDriver legacy fallback", () => { putAction(storage, 3, { autoApprovable: false }); let legacy = makeLegacyGatekeeper({ failApply: [2] }); - await makeDriver(storage, legacy.target).sync(GK, { frontier: 3, resolvedBy: APPROVER }); + await makeDriver(storage, legacy.target).apply(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"); @@ -505,7 +501,7 @@ describe("ActionSyncDriver legacy fallback", () => { let legacy = makeLegacyGatekeeper(); legacy.target.rejectAction = (async () => { throw new Error("already settled"); }) as typeof legacy.target.rejectAction; - await makeDriver(storage, legacy.target).sync(GK); + await makeDriver(storage, legacy.target).apply(GK); // The reject was attempted once and is not re-staged: legacy gatekeepers throw forever on // settled actions, so retrying would wedge the queue. @@ -524,7 +520,7 @@ describe("ActionSyncDriver legacy fallback", () => { let target = { async applyAction() { seen.push(getAction(storage, 1).state); }, } as unknown as GatekeeperActionTarget; - await makeDriver(storage, target).sync(GK, { frontier: 2, resolvedBy: APPROVER }); + await makeDriver(storage, target).apply(GK, { frontier: 2, resolvedBy: APPROVER }); expect(seen).toEqual(["pending", "approved"]); expect(getAction(storage, 2).state).toBe("approved"); @@ -539,7 +535,7 @@ describe("ActionSyncDriver legacy fallback", () => { legacy.target.rejectAction = (async () => { throw Object.assign(new Error("Durable Object reset."), { durableObjectReset: true }); }) as typeof legacy.target.rejectAction; - await makeDriver(storage, legacy.target).sync(GK); + await makeDriver(storage, legacy.target).apply(GK); // Undelivered, not dropped: the next pass on this gatekeeper re-sends it. expect(getAction(storage, 2).vetoPending).toBe(true); diff --git a/packages/workshop-backend/src/actions.ts b/packages/workshop-backend/src/actions.ts index 67ad1d012..7be68cbf8 100644 --- a/packages/workshop-backend/src/actions.ts +++ b/packages/workshop-backend/src/actions.ts @@ -44,7 +44,7 @@ export type GetGatekeeperFn = (gatekeeperId: number) => GatekeeperActionTarget; */ export type ManualApproval = { frontier: number, resolvedBy: AiChatAuthorInfo }; -type StagedSync = { +type StagedPass = { manualApprovals: ManualApproval[]; resolve: (decided: number[]) => void; reject: (error: unknown) => void; @@ -67,7 +67,7 @@ export function isMethodMissing(error: unknown): boolean { 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(); + #staged = new Map(); // Per-gatekeeper single-flight guard. Key present => a run loop is active for that gatekeeper. #running = new Map>(); @@ -85,7 +85,7 @@ export class ActionSyncDriver { * 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 { + apply(gatekeeperId: number, manualApproval?: ManualApproval): Promise { let slot = this.#staged.get(gatekeeperId); if (!slot) { slot = { manualApprovals: [], ...Promise.withResolvers() }; @@ -100,10 +100,10 @@ export class ActionSyncDriver { } /** - * Resolves once no sync pass is in flight for the gatekeeper. Used by rejection: a veto must + * Resolves once no apply 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 { + async awaitSettled(gatekeeperId: number): Promise { for (;;) { let running = this.#running.get(gatekeeperId); if (!running) return; @@ -118,8 +118,13 @@ export class ActionSyncDriver { if (!slot) break; this.#staged.delete(gatekeeperId); try { - slot.resolve(await this.#syncOnce(gatekeeperId, slot.manualApprovals)); + slot.resolve(await this.#applyOnce(gatekeeperId, slot.manualApprovals)); } catch (error) { + // Two callers deliver a pass through waitUntil (rejection, auto-approve opt-in) and + // never see this rejection, so log it here; awaiting callers still get the error. + logger.warn("action sync pass failed", { + event: "action.sync.failed", gatekeeperId, error, + }); slot.reject(error); } } @@ -130,7 +135,7 @@ export class ActionSyncDriver { } } - async #syncOnce(gatekeeperId: number, manualApprovals: ManualApproval[]): Promise { + async #applyOnce(gatekeeperId: number, manualApprovals: ManualApproval[]): Promise { // Materialize before reconciling: index reads are lazy, and the pass mutates both indexes. The // pending index was backfilled by the action-index migration; vetoPending only exists on records // written after its index was introduced, so it needs no legacy backfill. Both are keyed by the diff --git a/packages/workshop-backend/src/overseer.ts b/packages/workshop-backend/src/overseer.ts index 758f198c1..fc3d6a711 100644 --- a/packages/workshop-backend/src/overseer.ts +++ b/packages/workshop-backend/src/overseer.ts @@ -4301,13 +4301,13 @@ class OverseerImpl implements AgentHooks { // 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); + applyDecidedActions(gatekeeperId: number, manualApproval?: ManualApproval): Promise { + return this.#actionSync.apply(gatekeeperId, manualApproval); } - // Resolves once no sync pass is in flight for the gatekeeper (see ActionSyncDriver.settled). - settledActionSync(gatekeeperId: number): Promise { - return this.#actionSync.settled(gatekeeperId); + // Resolves once no apply pass is in flight (see ActionSyncDriver.awaitSettled). + awaitActionsSettled(gatekeeperId: number): Promise { + return this.#actionSync.awaitSettled(gatekeeperId); } // Blocks other messages and agent turns for this chat until the returned object is disposed. @@ -4715,7 +4715,7 @@ class OverseerImpl implements AgentHooks { } if (willAutoApprove) { - this.ctx.waitUntil(this.syncActions(gatekeeperId)); + this.ctx.waitUntil(this.applyDecidedActions(gatekeeperId)); } } @@ -9504,10 +9504,10 @@ class OverseerClientInterface extends RpcTarget implements Overseer { // action applied in the world but still "pending" in storage. let profile = await this.#getClientProfile(); - // Approving is a decision frontier: the sync pass applies this action AND every earlier + // Approving is a decision frontier: the 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( + let decided = await this.impl.applyDecidedActions( action.gatekeeperId, {frontier: action.action, resolvedBy: profile}); // Resume turns suspended on awaitDecision whose awaited actions this pass decided -- the batch @@ -9667,16 +9667,16 @@ class OverseerClientInterface extends RpcTarget implements Overseer { // half-rejected. let profile = await this.#getClientProfile(); - // A rejection must never interleave with an in-flight sync pass that might be applying this + // A rejection must never interleave with an in-flight apply pass that might be applying this // very action; wait it out, then re-check. - await this.impl.settledActionSync(action.gatekeeperId); + await this.impl.awaitActionsSettled(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 + // out with the next apply 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(); @@ -9685,7 +9685,7 @@ class OverseerClientInterface extends RpcTarget implements Overseer { delete fresh.failure; this.impl.storage.actions.put(fresh); - this.impl.ctx.waitUntil(this.impl.syncActions(action.gatekeeperId)); + this.impl.ctx.waitUntil(this.impl.applyDecidedActions(action.gatekeeperId)); // Deny leaves the turn ended, like denyConnectionRequest. The rejected record also prevents a // sibling approval from resuming this turn. @@ -9709,7 +9709,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.syncActions(gatekeeperId)); + this.impl.ctx.waitUntil(this.impl.applyDecidedActions(gatekeeperId)); } // Remove the auto-approval rule for `tag` on the given gatekeeper, so future matching actions