diff --git a/packages/workshop-backend/__tests__/action-sync.test.ts b/packages/workshop-backend/__tests__/action-sync.test.ts new file mode 100644 index 000000000..121873d1e --- /dev/null +++ b/packages/workshop-backend/__tests__/action-sync.test.ts @@ -0,0 +1,465 @@ +import { describe, it, expect } from "vitest"; +import { createTypedStorage, collection } from "@gadgets/typed-storage"; +import { ActionSyncDriver, ActionSyncStorage, GatekeeperActionTarget } from "../src/action-sync.js"; +import type { ActionRecord, AutoApproveTagRecord } from "../src/overseer.js"; +import type { AiChatAuthorInfo } from "@gadgets/workshop-shared/api"; +import type { ApplyActionsThroughResult } from "@gadgets/workshop-shared/gatekeeper"; +import { makeMockStorage } from "./mock-storage.js"; + +function makeStorage(): ActionSyncStorage { + return createTypedStorage(makeMockStorage(), { + collections: { + actions: collection()({ primaryKey: "id" }), + autoApproveTags: collection()({ + primaryKey: (r: AutoApproveTagRecord) => `${r.gatekeeperId}:${r.actionKind.tag}`, + }), + }, + }); +} + +const GK = 1; +const ENABLER: AiChatAuthorInfo = { type: "user", id: "enabler@example.com", name: "Enabler" }; +const APPROVER: AiChatAuthorInfo = { type: "user", id: "approver@example.com", name: "Approver" }; +const REJECTER: AiChatAuthorInfo = { type: "user", id: "rejecter@example.com", name: "Rejecter" }; + +function enableRule(storage: ActionSyncStorage, actionTag = "edit", gatekeeperId = GK) { + storage.autoApproveTags.put({ + gatekeeperId, actionKind: { tag: actionTag, label: "Edits" }, enabledBy: ENABLER }); +} + +// Workspace record ids are deliberately offset from gatekeeper-local action ids (`id = action*10`) +// so a test that confuses the two ID spaces fails loudly. +function putAction( + storage: ActionSyncStorage, action: number, + opts: { gatekeeperId?: number; actionTag?: string; autoApprovable?: boolean; + state?: ActionRecord["state"]; chatId?: number; awaitDecision?: boolean; + vetoPending?: true; resolvedBy?: AiChatAuthorInfo; failure?: string } = {}): number { + let id = action * 10; + storage.actions.put({ + id, + gatekeeperId: opts.gatekeeperId ?? GK, + caller: { from: "agent", chatId: opts.chatId ?? 1 }, + createdAt: new Date(), + state: opts.state ?? "pending", + type: "action", + action, + ...(opts.vetoPending ? { vetoPending: true } : {}), + ...(opts.resolvedBy ? { resolvedBy: opts.resolvedBy } : {}), + ...(opts.failure !== undefined ? { failure: opts.failure } : {}), + description: { + title: `Action ${action}`, + description: `Action ${action} description`, + implementsRevert: true, + actionKind: { tag: opts.actionTag ?? "edit", label: "Edits" }, + autoApprovable: opts.autoApprovable ?? true, + ...(opts.awaitDecision ? { awaitDecision: true } : {}), + }, + }); + return id; +} + +function getAction(storage: ActionSyncStorage, action: number): ActionRecord & {type: "action"} { + let record = storage.actions.get(action * 10); + if (!record || record.type !== "action") throw new Error(`No action ${action}`); + return record; +} + +// A migrated gatekeeper stub: records every batch call and answers from a scripted queue (or {}). +function makeBatchGatekeeper() { + let calls: Array<{actionId: number, vetoes: number[]}> = []; + let results: Array = []; + let target = { + async applyActionsThrough(actionId: number, vetoes: number[]) { + calls.push({ actionId, vetoes }); + let next = results.shift() ?? {}; + if (next instanceof Error) throw next; + return next; + }, + async applyAction() { throw new Error("legacy applyAction must not be called"); }, + async rejectAction() { throw new Error("legacy rejectAction must not be called"); }, + } as unknown as GatekeeperActionTarget; + return { target, calls, results }; +} + +// A pre-migration gatekeeper stub: applyActionsThrough is missing (locally undefined, or throwing +// workerd's method-missing TypeError when `remote` mimics a live stub), so the driver must fall +// back to per-action legacy calls. +function makeLegacyGatekeeper(opts: {remote?: boolean, failApply?: number[]} = {}) { + let probes = 0; + let calls: string[] = []; + let target = { + ...(opts.remote ? { + async applyActionsThrough() { + probes++; + throw new TypeError( + 'The RPC receiver does not implement the method "applyActionsThrough".'); + }, + } : {}), + async applyAction(action: number) { + calls.push(`apply:${action}`); + if (opts.failApply?.includes(action)) throw new Error(`apply ${action} failed`); + }, + async rejectAction(action: number) { + calls.push(`reject:${action}`); + return { restart: true }; // must be discarded + }, + } as unknown as GatekeeperActionTarget; + return { target, calls, probeCount: () => probes }; +} + +function makeDriver(storage: ActionSyncStorage, target: GatekeeperActionTarget) { + return new ActionSyncDriver(storage, () => target); +} + +// Drain the microtask queue (and one macrotask) so parked continuations reach their next await. +function flush(): Promise { + return new Promise((resolve) => setTimeout(resolve, 0)); +} + +describe("ActionSyncDriver.sync", () => { + it("applies through a manual frontier, attributing covered actions to the approver and " + + "auto-extended ones to the rule enabler", async () => { + let storage = makeStorage(); + enableRule(storage); + let a1 = putAction(storage, 1, { autoApprovable: false }); + let a2 = putAction(storage, 2, { autoApprovable: false }); + let a3 = putAction(storage, 3); // auto-eligible beyond the manual frontier + + let { target, calls } = makeBatchGatekeeper(); + let decided = await makeDriver(storage, target) + .sync(GK, { frontier: 2, resolvedBy: APPROVER }); + + expect(calls).toEqual([{ actionId: 3, vetoes: [] }]); + expect(decided.toSorted((a, b) => a - b)).toEqual([a1, a2, a3]); + for (let action of [1, 2]) { + let record = getAction(storage, action); + expect(record.state).toBe("approved"); + expect(record.autoApproved).toBe(false); + expect(record.resolvedBy?.id).toBe(APPROVER.id); + } + let auto = getAction(storage, 3); + expect(auto.state).toBe("approved"); + expect(auto.autoApproved).toBe(true); + expect(auto.resolvedBy?.id).toBe(ENABLER.id); + }); + + it("never auto-approves past a manual gate", async () => { + let storage = makeStorage(); + enableRule(storage); + putAction(storage, 1); + putAction(storage, 2, { autoApprovable: false }); // manual gate + putAction(storage, 3); + + let { target, calls } = makeBatchGatekeeper(); + await makeDriver(storage, target).sync(GK); + + expect(calls).toEqual([{ actionId: 1, vetoes: [] }]); + expect(getAction(storage, 1).state).toBe("approved"); + expect(getAction(storage, 2).state).toBe("pending"); + expect(getAction(storage, 3).state).toBe("pending"); + }); + + it("makes no call when nothing is eligible", async () => { + let storage = makeStorage(); + putAction(storage, 1, { autoApprovable: false }); + + let { target, calls } = makeBatchGatekeeper(); + let decided = await makeDriver(storage, target).sync(GK); + + expect(decided).toEqual([]); + expect(calls).toEqual([]); + }); + + it("records a display-safe failure on the stopped action and clears it on a later success", + async () => { + let storage = makeStorage(); + let a1 = putAction(storage, 1, { autoApprovable: false }); + putAction(storage, 2, { autoApprovable: false }); + + let { target, calls, results } = makeBatchGatekeeper(); + results.push({ stopped: { at: 2, reason: new Error("page was deleted upstream") } }); + let driver = makeDriver(storage, target); + + let first = await driver.sync(GK, { frontier: 2, resolvedBy: APPROVER }); + + expect(first).toEqual([a1]); + expect(getAction(storage, 1).state).toBe("approved"); + let stopped = getAction(storage, 2); + expect(stopped.state).toBe("pending"); + expect(stopped.failure).toBe("page was deleted upstream"); + + // Retry after the user resolves the problem: only the stopped action remains pending, and its + // failure is cleared. The already-applied action is never re-sent (idempotent contract), and + // the gatekeeper sees a second call at the same frontier. + let retry = await driver.sync(GK, { frontier: 2, resolvedBy: APPROVER }); + + expect(retry).toEqual([getAction(storage, 2).id]); + expect(calls).toEqual([{ actionId: 2, vetoes: [] }, { actionId: 2, vetoes: [] }]); + let retried = getAction(storage, 2); + expect(retried.state).toBe("approved"); + expect(retried.failure).toBeUndefined(); + }); + + it("keeps a veto staged while an earlier action is undecided", async () => { + let storage = makeStorage(); + putAction(storage, 1, { autoApprovable: false }); + putAction(storage, 2, { state: "rejected", vetoPending: true, resolvedBy: REJECTER }); + + let { target, calls } = makeBatchGatekeeper(); + await makeDriver(storage, target).sync(GK); + + expect(calls).toEqual([]); + expect(getAction(storage, 2).vetoPending).toBe(true); + }); + + it("delivers a staged veto at the current frontier once everything below is decided, even " + + "from a fresh driver", async () => { + let storage = makeStorage(); + putAction(storage, 1, { state: "approved" }); + putAction(storage, 2, { state: "rejected", vetoPending: true, resolvedBy: REJECTER }); + + // A fresh driver over the same storage (e.g. after DO hibernation) must still see the staged + // veto -- it is durable state, not driver memory. + let { target, calls } = makeBatchGatekeeper(); + await makeDriver(storage, target).sync(GK); + + expect(calls).toEqual([{ actionId: 2, vetoes: [2] }]); + expect(getAction(storage, 2).vetoPending).toBeUndefined(); + }); + + it("rides staged vetoes along with a covering approval", async () => { + let storage = makeStorage(); + let a1 = putAction(storage, 1, { autoApprovable: false }); + putAction(storage, 2, { state: "rejected", vetoPending: true, resolvedBy: REJECTER }); + let a3 = putAction(storage, 3, { autoApprovable: false }); + + let { target, calls } = makeBatchGatekeeper(); + let decided = await makeDriver(storage, target) + .sync(GK, { frontier: 3, resolvedBy: APPROVER }); + + expect(calls).toEqual([{ actionId: 3, vetoes: [2] }]); + expect(decided.toSorted((a, b) => a - b)).toEqual([a1, a3]); + expect(getAction(storage, 2).vetoPending).toBeUndefined(); + }); + + it("marks cascade-invalidated actions rejected with the vetoing record's attribution", + async () => { + let storage = makeStorage(); + putAction(storage, 1, { state: "approved" }); + let vetoId = putAction(storage, 2, + { state: "rejected", vetoPending: true, resolvedBy: REJECTER }); + let a3 = putAction(storage, 3, { autoApprovable: false }); + + let { target, results } = makeBatchGatekeeper(); + results.push({ invalidatedByVeto: [{ action: 3, invalidatedBy: 2 }] }); + let decided = await makeDriver(storage, target).sync(GK); + + expect(decided).toEqual([a3]); + let invalidated = getAction(storage, 3); + expect(invalidated.state).toBe("rejected"); + expect(invalidated.cascadedFrom).toBe(vetoId); + expect(invalidated.resolvedBy?.id).toBe(REJECTER.id); + }); + + it("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 0b4ce7389..000000000 --- a/packages/workshop-backend/__tests__/auto-approval.test.ts +++ /dev/null @@ -1,214 +0,0 @@ -import { describe, it, expect } from "vitest"; -import { createTypedStorage, collection } from "@gadgets/typed-storage"; -import { AutoApprovalDrainer, AutoApprovalStorage, ApplyPendingActionFn } from "../src/auto-approval.js"; -import type { ActionRecord, AutoApproveTagRecord } from "../src/overseer.js"; -import type { AiChatAuthorInfo } from "@gadgets/workshop-shared/api"; -import { makeMockStorage } from "./mock-storage.js"; - -function makeStorage(): AutoApprovalStorage { - return createTypedStorage(makeMockStorage(), { - collections: { - actions: collection()({ primaryKey: "id" }), - autoApproveTags: collection()({ - primaryKey: (r: AutoApproveTagRecord) => `${r.gatekeeperId}:${r.actionKind.tag}`, - }), - }, - }); -} - -const GK = 1; -const ENABLER: AiChatAuthorInfo = { type: "user", id: "enabler@example.com", name: "Enabler" }; - -function enableRule(storage: AutoApprovalStorage, actionTag = "edit", gatekeeperId = GK) { - storage.autoApproveTags.put({ - gatekeeperId, actionKind: { tag: actionTag, label: "Edits" }, enabledBy: ENABLER }); -} - -function putAction( - storage: AutoApprovalStorage, id: number, - opts: { gatekeeperId?: number; actionTag?: string; autoApprovable?: boolean; - state?: ActionRecord["state"] } = {}) { - storage.actions.put({ - id, - gatekeeperId: opts.gatekeeperId ?? GK, - caller: { from: "agent", chatId: 1 }, - createdAt: new Date(), - state: opts.state ?? "pending", - type: "action", - action: id, - description: { - title: `Action ${id}`, - description: `Action ${id} description`, - implementsRevert: true, - actionKind: { tag: opts.actionTag ?? "edit", label: "Edits" }, - autoApprovable: opts.autoApprovable ?? true, - }, - }); -} - -function getAction(storage: AutoApprovalStorage, id: number): ActionRecord & {type: "action"} { - let record = storage.actions.get(id); - if (!record || record.type !== "action") throw new Error(`No action ${id}`); - return record; -} - -// An apply fn that resolves immediately, mirroring OverseerImpl.applyPendingAction's effect: -// mark the record approved and persist. Records the order of applied action ids. -function makeImmediateApply(storage: AutoApprovalStorage) { - let calls: number[] = []; - let applyFn: ApplyPendingActionFn = async (record, resolvedBy, autoApproved) => { - calls.push(record.id); - let fresh = storage.actions.get(record.id); - if (fresh && fresh.type === "action") { - fresh.state = "approved"; - fresh.appliedAt = new Date(); - fresh.resolvedBy = resolvedBy; - fresh.autoApproved = autoApproved; - storage.actions.put(fresh); - } - }; - return { applyFn, calls }; -} - -// An apply fn whose every invocation parks on a test-held promise until released. Lets a test hold -// an apply mid-flight (input gate open) while launching a second concurrent drain. On release it -// performs the same approve+persist effect as the real apply. -function makeControlledApply(storage: AutoApprovalStorage) { - let calls: number[] = []; - let gates: Array<() => void> = []; - let applyFn: ApplyPendingActionFn = (record, resolvedBy, autoApproved) => { - calls.push(record.id); - return new Promise((resolve) => { - gates.push(() => { - let fresh = storage.actions.get(record.id); - if (fresh && fresh.type === "action") { - fresh.state = "approved"; - fresh.appliedAt = new Date(); - fresh.resolvedBy = resolvedBy; - fresh.autoApproved = autoApproved; - storage.actions.put(fresh); - } - resolve(); - }); - }); - }; - return { - applyFn, - calls, - inFlight: () => gates.length, - releaseNext() { - let gate = gates.shift(); - if (!gate) throw new Error("no apply in flight to release"); - gate(); - }, - }; -} - -// Drain all microtasks (and the macrotask queue) so suspended drain continuations run to their next -// park point. -function flush(): Promise { - return new Promise((resolve) => setTimeout(resolve, 0)); -} - -describe("AutoApprovalDrainer.drain", () => { - it("applies all eligible pending actions in ascending id order", async () => { - let storage = makeStorage(); - enableRule(storage); - putAction(storage, 1); - putAction(storage, 2); - putAction(storage, 3); - - let { applyFn, calls } = makeImmediateApply(storage); - await new AutoApprovalDrainer(storage, applyFn).drain(GK); - - expect(calls).toEqual([1, 2, 3]); - for (let id of [1, 2, 3]) { - let record = getAction(storage, id); - expect(record.state).toBe("approved"); - expect(record.autoApproved).toBe(true); - expect(record.resolvedBy?.id).toBe(ENABLER.id); - } - }); - - it("stops at a manual gate without skipping ahead, then resumes once it clears", async () => { - let storage = makeStorage(); - enableRule(storage); - putAction(storage, 1); - putAction(storage, 2, { autoApprovable: false }); // manual gate - putAction(storage, 3); - - let { applyFn, calls } = makeImmediateApply(storage); - let drainer = new AutoApprovalDrainer(storage, applyFn); - await drainer.drain(GK); - - // Only the action before the gate is applied; the gate and everything behind it stay pending. - expect(calls).toEqual([1]); - expect(getAction(storage, 2).state).toBe("pending"); - expect(getAction(storage, 3).state).toBe("pending"); - - // Clear the gate (as a manual approval would) and re-drain: the rest applies, still in order. - let gate = getAction(storage, 2); - gate.state = "approved"; - storage.actions.put(gate); - await drainer.drain(GK); - - expect(calls).toEqual([1, 3]); - expect(getAction(storage, 3).state).toBe("approved"); - }); - - // Two concurrent drains for the same gatekeeper must not double-apply. The input gate is open - // across the apply await, so without the single-flight guard the second drain's pending re-check - // would see the still-"pending" record and apply it again. - it("never applies an action more than once under concurrent drains", async () => { - let storage = makeStorage(); - enableRule(storage); - putAction(storage, 1); - - let apply = makeControlledApply(storage); - let drainer = new AutoApprovalDrainer(storage, apply.applyFn); - - let first = drainer.drain(GK); // starts, calls apply(1), parks mid-apply - let second = drainer.drain(GK); // must coalesce, not start a second apply - await second; - - expect(apply.calls).toEqual([1]); - expect(apply.inFlight()).toBe(1); - - apply.releaseNext(); // resolve apply(1); record becomes approved - await first; // rerun pass re-lists: action 1 no longer pending -> no re-apply - - expect(apply.calls).toEqual([1]); - expect(getAction(storage, 1).state).toBe("approved"); - }); - - // Work that arrives while a drain is parked must still be applied -- the coalescing - // "rerun" flag must not drop the wakeup. - it("applies work submitted while a drain is parked mid-apply", async () => { - let storage = makeStorage(); - enableRule(storage); - putAction(storage, 1); - - let apply = makeControlledApply(storage); - let drainer = new AutoApprovalDrainer(storage, apply.applyFn); - - let first = drainer.drain(GK); // parks mid-apply on action 1 - - putAction(storage, 2); // new eligible action arrives mid-drain - let second = drainer.drain(GK); // coalesces -> sets the rerun flag - await second; - expect(apply.calls).toEqual([1]); - - apply.releaseNext(); // finish action 1; rerun pass should pick up action 2 - await flush(); - - expect(apply.calls).toEqual([1, 2]); - expect(apply.inFlight()).toBe(1); - - apply.releaseNext(); // finish action 2 - await first; - - expect(apply.calls).toEqual([1, 2]); - expect(getAction(storage, 1).state).toBe("approved"); - expect(getAction(storage, 2).state).toBe("approved"); - }); -}); diff --git a/packages/workshop-backend/src/action-sync.ts b/packages/workshop-backend/src/action-sync.ts new file mode 100644 index 000000000..c16aba23d --- /dev/null +++ b/packages/workshop-backend/src/action-sync.ts @@ -0,0 +1,299 @@ +// Action-sync core: reconciles this workspace's pending action records with a gatekeeper through +// one batch `applyActionsThrough(actionId, vetoes)` call per pass. A pass computes the decision +// frontier (manual approvals staged by the caller, then auto-approval rules, then deliverable +// vetoes), makes the call, and translates the result back onto the records: everything at or below +// the applied frontier becomes "approved" with the right attribution, a `stopped` action keeps its +// pending state plus a display-safe `failure`, and veto-cascade invalidations become "rejected" +// with `cascadedFrom` attribution. +// +// A per-gatekeeper single-flight guard (the DO's input gate is open across the RPC await) +// coalesces concurrent requests into the next pass, so two approvals arriving together produce one +// call at the higher frontier. The gatekeeper accessor is injected, keeping the driver +// constructible over a mock storage in tests. + +import type { Collection } from "@gadgets/typed-storage"; +import type { AiChatAuthorInfo } from "@gadgets/workshop-shared/api"; +import type { ApplyActionsThroughResult, Gatekeeper } from "@gadgets/workshop-shared/gatekeeper"; +import { createWorkshopLogger } from "./observability"; +import type { ActionRecord, AutoApproveTagRecord } from "./overseer.js"; + +const logger = createWorkshopLogger("workshop.action.sync"); + +export interface ActionSyncStorage { + actions: Collection; + autoApproveTags: Collection; +} + +/** + * The slice of the gatekeeper stub surface the driver drives, derived from the RPC contract. + * `applyActionsThrough` is optional during the migration; on a live stub the property is always a + * callable proxy and an un-migrated gatekeeper throws when it is invoked (see isMethodMissing). + */ +export type GatekeeperActionTarget = + Pick>, "applyActionsThrough" | "applyAction" | "rejectAction">; + +export type GetGatekeeperFn = (gatekeeperId: number) => GatekeeperActionTarget; + +/** + * A staged manual approval: apply every undecided action through `frontier` (a gatekeeper-local + * action ID) under `resolvedBy`'s authority. + */ +export type ManualApproval = { frontier: number, resolvedBy: AiChatAuthorInfo }; + +type 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 a snapshot first: list() is a lazy generator over storage, and we mutate the + // actions collection as we reconcile. Keyed and ordered by `record.action` (the + // gatekeeper-local ID space) -- `record.id` shares a counter with observations and hooks. + let records = [...this.storage.actions.list()].filter( + (rec): rec is ActionRecord & {type: "action"} => + rec.gatekeeperId === gatekeeperId && rec.type === "action"); + let byAction = new Map(records.map(record => [record.action, record])); + let pending = records.filter(record => record.state === "pending") + .toSorted((a, b) => a.action - b.action); + let stagedVetoes = records.filter(record => record.state === "rejected" && record.vetoPending) + .toSorted((a, b) => a.action - b.action); + + // Decide the frontier and, for every pending action it covers, the attribution to record if + // the gatekeeper applies it. Attribution is captured now, before the RPC, so a rule removed + // mid-call can't leave an applied action unattributed: this is the single pending->approved + // chokepoint, and every transition must record the resolving user and whether it was + // automatic. + let manualAscending = 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 0c1fa032f..000000000 --- a/packages/workshop-backend/src/auto-approval.ts +++ /dev/null @@ -1,97 +0,0 @@ -// Auto-approval drain core: applies eligible pending actions in id order, with a per-gatekeeper -// single-flight guard so two concurrent drains (the DO's input gate is open across the apply await) -// can't double-apply the same action. The apply is injected, keeping this constructible over a -// mock storage in tests. - -import type { Collection } from "@gadgets/typed-storage"; -import type { AiChatAuthorInfo } from "@gadgets/workshop-shared/api"; -import { createWorkshopLogger } from "./observability"; -import type { ActionRecord, AutoApproveTagRecord } from "./overseer.js"; - -const logger = createWorkshopLogger("workshop.auto.approval"); - -export interface AutoApprovalStorage { - actions: Collection; - autoApproveTags: Collection; -} - -/** - * Applies a single eligible pending action: invoke the gatekeeper, mark it approved, persist. The - * caller has already validated that the record is still pending. - */ -export type ApplyPendingActionFn = ( - record: ActionRecord & {type: "action"}, - resolvedBy: AiChatAuthorInfo, - autoApproved: boolean) => Promise; - -export class AutoApprovalDrainer { - // Per-gatekeeper single-flight state. Key present => a drain is running for that gatekeeper; the - // value is a "rerun" flag, set when another drain is requested while one is in flight, so work - // submitted during a drain isn't lost. - #draining = new Map(); - - constructor( - private storage: AutoApprovalStorage, - private applyPendingAction: ApplyPendingActionFn) {} - - async drain(gatekeeperId: number): Promise { - if (this.#draining.has(gatekeeperId)) { - this.#draining.set(gatekeeperId, true); // ask the running drain to loop again - return; - } - this.#draining.set(gatekeeperId, false); - try { - do { - this.#draining.set(gatekeeperId, false); - await this.#drainOnce(gatekeeperId); - } while (this.#draining.get(gatekeeperId)); - } finally { - this.#draining.delete(gatekeeperId); - } - } - - // Apply all currently-eligible pending actions of the gatekeeper, in ascending id order. Stops at - // the first pending action that is NOT auto-eligible (a manual gate) or that throws while applying - // -- it is never skipped ahead of. This preserves in-order application and the invariant that - // nothing is silently applied past a human gate. - // - // Eligibility requires BOTH signals: the author's `autoApprovable` verdict on the action AND a - // user-enabled rule for the action's type on this gatekeeper. - async #drainOnce(gatekeeperId: number): Promise { - // Materialize a snapshot first: list() is a lazy generator over storage, and we mutate the - // actions collection (via applyPendingAction) as we go. - let pending = [...this.storage.actions.list()].filter( - (rec): rec is ActionRecord & {type: "action"} => - rec.gatekeeperId === gatekeeperId && rec.type === "action" && rec.state === "pending"); - - for (let record of pending) { - let tag = record.description.actionKind?.tag; - let rule = tag !== undefined - ? this.storage.autoApproveTags.get(`${gatekeeperId}:${tag}`) - : undefined; - if (record.description.autoApprovable !== true || rule === undefined) { - // A manual gate. Stop rather than skipping ahead to any later auto-eligible action. - break; - } - - // Re-check immediately before applying, to guard against a concurrent drain having already - // taken this one. - let fresh = this.storage.actions.get(record.id); - if (!fresh || fresh.type !== "action" || fresh.state !== "pending") { - continue; - } - - try { - // Attribute the auto-approval to the user who enabled the rule -- it runs under their - // authority. - await this.applyPendingAction(fresh, rule.enabledBy, true); - } catch (err) { - // Leave the action pending for manual handling and stop the drain (never skip ahead). - logger.error("auto-approval failed", { - event: "auto.approval.failed", actionId: fresh.id, error: err, - }); - break; - } - } - } -} diff --git a/packages/workshop-backend/src/overseer.ts b/packages/workshop-backend/src/overseer.ts index e5fed5bb2..b1be3e72d 100644 --- a/packages/workshop-backend/src/overseer.ts +++ b/packages/workshop-backend/src/overseer.ts @@ -35,7 +35,7 @@ import { checkUsageAndBalance } from "./ai-gateway-billing/limits/usage-checker" import { completeAgentCatalogSnapshot, normalizeAgentCatalog } from "./agent-catalog"; import { refreshCachedBalance } from "./ai-gateway-billing/cloudflare/connection-service"; import { SharingManager, SharingCaller, CollaboratorRecord, ShareKeyRecord } from "./sharing"; -import { AutoApprovalDrainer } from "./auto-approval"; +import { ActionSyncDriver, ManualApproval } from "./action-sync"; import { collectSlashCommands, invokeSlashCommand } from "./slash-commands"; import { createWorkshopLogger, obsContext, traced } from "./observability"; import { wrapDoStubForTelemetry } from "./do-telemetry"; @@ -522,6 +522,25 @@ export type ActionRecord = { description: ActionDescription; resolvedBy?: AiChatAuthorInfo; // set when resolved (approved/rejected); absent while pending (or legacy) autoApproved?: boolean; // set when applied by an auto-approval rule rather than a human + + /** + * Display-safe reason the most recent application attempt stopped at this action. Only present + * while the action remains pending; cleared when it applies or is rejected. + */ + failure?: string; + + /** + * Workspace action ID of the rejected action whose veto invalidated this one. Only set when + * `state` is "rejected" and the rejection came from a gatekeeper dependency cascade. + */ + cascadedFrom?: number; + + /** + * Present while the user's rejection has not yet been delivered to the gatekeeper as a veto. + * The sync driver clears it on delivery. Absent on records rejected before this field existed, + * so legacy rejections are never re-delivered. + */ + vetoPending?: true; } | { type: "observation"; description: ObservationDescription; @@ -730,6 +749,8 @@ function actionRecordToLog(record: ActionRecord): ActionLogEntry { description: record.description, resolvedBy: record.resolvedBy, autoApproved: record.autoApproved, + cascadedFrom: record.cascadedFrom, + failure: record.failure, }; case "bindHook": return { @@ -1117,7 +1138,7 @@ class OverseerImpl implements AgentHooks { #liveChats = new Map(); #chatSubscribers: Set> = new Set(); - #autoApprovalDrainer: AutoApprovalDrainer; + #actionSync: ActionSyncDriver; #preparingChatMessages = new Map>(); @@ -1403,10 +1424,8 @@ class OverseerImpl implements AgentHooks { this.#migrateStorage(); this.defaultGadgetId = this.storage.defaultGadgetId.get(); - this.#autoApprovalDrainer = new AutoApprovalDrainer( - this.storage, - (record, resolvedBy, autoApproved) => - this.applyPendingAction(record, resolvedBy, autoApproved)); + this.#actionSync = new ActionSyncDriver( + this.storage, gatekeeperId => this.getGatekeeperFacet(gatekeeperId)); // Mirror every gadget-registry change into the owner's outputs index. Subscribing here makes // the registry the single chokepoint, so creation, acceptance, renaming, reverting and @@ -2567,35 +2586,18 @@ class OverseerImpl implements AgentHooks { }); } - // Apply a single pending action: invoke the gatekeeper, mark it approved, and persist (the put - // auto-notifies subscribeToActions). Shared by manual approval (`approveAction`) and the - // auto-approval drain (`drainAutoApprovals`). The caller is responsible for validating that the - // record is still pending before calling. - // - // `resolvedBy`/`autoApproved` are required (not defaulted) so that no apply path can omit how the - // gate was cleared: this is the single chokepoint where an action transitions to "approved", so - // requiring them here guarantees the audit log always records the resolving user and whether it - // was applied automatically. For an auto-approval, `resolvedBy` is the user who enabled the rule. - async applyPendingAction(record: ActionRecord & {type: "action"}, - resolvedBy: AiChatAuthorInfo, autoApproved: boolean): Promise { - let gatekeeper = this.getGatekeeperFacet(record.gatekeeperId); - await gatekeeper.applyAction(record.action); - record.state = "approved"; - record.appliedAt = new Date(); - record.resolvedBy = resolvedBy; - record.autoApproved = autoApproved; - this.storage.actions.put(record); + // Reconcile the gatekeeper's pending actions through one batch applyActionsThrough call: staged + // manual approvals, then auto-eligible actions (stopping at the first manual gate -- nothing is + // silently applied past one), then any deliverable staged vetoes. Resolves with the workspace + // record ids the pass decided. Delegates to the single-flight driver, which coalesces concurrent + // requests for the same gatekeeper (the DO's input gate is open across the RPC await). + syncActions(gatekeeperId: number, manualApproval?: ManualApproval): Promise { + return this.#actionSync.sync(gatekeeperId, manualApproval); } - // Apply all currently-eligible pending actions of the given gatekeeper, in ascending id order. - // Stops at the first pending action that is NOT auto-eligible (i.e. a manual gate) or that throws - // while applying -- it is never skipped ahead of. This preserves in-order application and the - // invariant that nothing is silently applied past a human gate. - // - // Delegates to the single-flight drainer, which guards against concurrent drains for the same - // gatekeeper double-applying an action (the DO's input gate is open across the apply await). - drainAutoApprovals(gatekeeperId: number): Promise { - return this.#autoApprovalDrainer.drain(gatekeeperId); + // Resolves once no sync pass is in flight for the gatekeeper (see ActionSyncDriver.settled). + settledActionSync(gatekeeperId: number): Promise { + return this.#actionSync.settled(gatekeeperId); } // Blocks other messages and agent turns for this chat until the returned object is disposed. @@ -2996,7 +2998,7 @@ class OverseerImpl implements AgentHooks { } if (willAutoApprove) { - this.ctx.waitUntil(this.drainAutoApprovals(gatekeeperId)); + this.ctx.waitUntil(this.syncActions(gatekeeperId)); } } @@ -7749,17 +7751,33 @@ class OverseerClientInterface extends RpcTarget implements Overseer { // Resolve the approver's identity before applying, so a failed profile fetch can't leave the // action applied in the world but still "pending" in storage. let profile = await this.#getClientProfile(); - await this.impl.applyPendingAction(action, profile, false); - // If this was an awaited agent action, resume only after all awaited actions in the turn are - // approved. If applyPendingAction throws, the action stays pending and the turn stays suspended. - if (action.caller.from === "agent" && action.description.awaitDecision) { - await this.#maybeResumeAfterActionDecision(action.caller.chatId); + // Approving is a decision frontier: the sync pass applies this action AND every earlier + // undecided action from the same gatekeeper (attributed to this approver), then continues + // through any auto-eligible actions the cleared gate unblocked. + let decided = await this.impl.syncActions( + action.gatekeeperId, {frontier: action.action, resolvedBy: profile}); + + let fresh = this.impl.storage.actions.get(id); + if (fresh?.type === "action" && fresh.state === "pending") { + // The gatekeeper stopped at (or before) this action; surface the display-safe reason so the + // user can resolve the problem and retry. The action stays pending. + throw new Error(fresh.failure ?? `Failed to apply action: ${id}`); } - // Clearing this manual gate may unblock later auto-eligible pending actions on the same - // gatekeeper, so cascade a drain (in-order) once this one is applied. - this.impl.ctx.waitUntil(this.impl.drainAutoApprovals(action.gatekeeperId)); + // Resume turns suspended on awaitDecision whose awaited actions this pass decided -- the batch + // may have covered earlier actions from other chats, not just the approved one. + let chatIds = new Set(); + for (let recordId of decided) { + let record = this.impl.storage.actions.get(recordId); + if (record?.type === "action" && record.caller.from === "agent" && + record.description.awaitDecision) { + chatIds.add(record.caller.chatId); + } + } + for (let chatId of chatIds) { + await this.#maybeResumeAfterActionDecision(chatId); + } } async listHooks(): Promise { @@ -7894,18 +7912,29 @@ class OverseerClientInterface extends RpcTarget implements Overseer { throw new Error(`Can't reject an observation: ${id}`); } - let gatekeeper = this.impl.getGatekeeperFacet(action.gatekeeperId); - - // Resolve the rejecter's identity before notifying the gatekeeper, so a failed profile fetch - // can't leave the action rejected with the gatekeeper but still "pending" in storage. + // Resolve the rejecter's identity first, so a failed profile fetch can't leave the action + // half-rejected. let profile = await this.#getClientProfile(); - await gatekeeper.rejectAction(action.action); + // A rejection must never interleave with an in-flight sync pass that might be applying this + // very action; wait it out, then re-check. + await this.impl.settledActionSync(action.gatekeeperId); + let fresh = this.impl.storage.actions.get(id); + if (fresh?.type !== "action" || fresh.state !== "pending") { + throw new Error(`Action is not pending: ${id}`); + } + + // The rejection is decided here and now; delivery to the gatekeeper is a staged veto. It goes + // out with the next sync pass whose frontier covers it -- immediately below, if every earlier + // action is already decided, otherwise once the actions below it are. + fresh.state = "rejected"; + fresh.appliedAt = new Date(); + fresh.resolvedBy = profile; + fresh.vetoPending = true; + delete fresh.failure; + this.impl.storage.actions.put(fresh); - action.state = "rejected"; - action.appliedAt = new Date(); - action.resolvedBy = profile; - this.impl.storage.actions.put(action); + this.impl.ctx.waitUntil(this.impl.syncActions(action.gatekeeperId)); // Deny leaves the turn ended, like denyConnectionRequest. The rejected record also prevents a // sibling approval from resuming this turn. @@ -7929,7 +7958,7 @@ class OverseerClientInterface extends RpcTarget implements Overseer { enabledBy: profile, }); // Apply the currently-visible pending action(s) with this tag right away. - this.impl.ctx.waitUntil(this.impl.drainAutoApprovals(gatekeeperId)); + this.impl.ctx.waitUntil(this.impl.syncActions(gatekeeperId)); } // Remove the auto-approval rule for `tag` on the given gatekeeper, so future matching actions diff --git a/packages/workshop-shared/src/api.ts b/packages/workshop-shared/src/api.ts index 8fb641693..ac26644fa 100644 --- a/packages/workshop-shared/src/api.ts +++ b/packages/workshop-shared/src/api.ts @@ -1467,7 +1467,7 @@ export interface CodeSubscriber { * Specifies the state of an action in the action log: * * pending: Action has not been applied yet. It is waiting for approval. * * approved: Action was approved and applied. - * * rejected: Action was rejected by the user. + * * rejected: Action was rejected by the user or invalidated by another rejected action. */ export type ActionState = "pending" | "approved" | "rejected"; @@ -1504,6 +1504,18 @@ export type ActionLogEntry = { * clicking Approve. Only ever set alongside state "approved" (there is no automatic rejection). */ autoApproved?: boolean; + + /** + * Workspace action ID whose rejection invalidated this action. Only set when `state` is + * "rejected" and the action was rejected as part of a dependency cascade. + */ + cascadedFrom?: number; + + /** + * Display-safe reason the most recent application attempt stopped at this action. Only set while + * the action remains pending. The action may be retried or rejected. + */ + failure?: string; } | { type: "observation"; description: ObservationDescription; @@ -1710,8 +1722,8 @@ export interface Overseer extends RpcTarget { listActions(): Promise; /** - * Approve an action that is currently in the "pending" state. The action will be performed on - * approval. + * Approve an action that is currently in the "pending" state. This performs the action and may + * also perform earlier pending actions from the same Gatekeeper connection. */ approveAction(id: number): Promise;