From b1f3407ec222e3e08a0d6a4fbafbd798819d3f80 Mon Sep 17 00:00:00 2001 From: Dan Carter Date: Wed, 19 Aug 2026 09:09:08 -0400 Subject: [PATCH 1/6] Revalidate MCP actions before dispatch --- .../mcp-shared/__tests__/action-store.test.ts | 145 ++++++-- .../__tests__/client-pagination.test.ts | 18 + .../mcp-shared/__tests__/connection.test.ts | 29 +- packages/mcp-shared/__tests__/facet.test.ts | 349 +++++++++++++++++- packages/mcp-shared/__tests__/session.test.ts | 58 ++- packages/mcp-shared/__tests__/tools.test.ts | 16 + packages/mcp-shared/src/account.ts | 26 +- packages/mcp-shared/src/action-store.ts | 112 +++++- packages/mcp-shared/src/client.ts | 24 +- packages/mcp-shared/src/connection.ts | 42 ++- packages/mcp-shared/src/facet.ts | 135 ++++++- packages/mcp-shared/src/session.ts | 52 ++- packages/mcp-shared/src/tools.ts | 6 + .../__tests__/auto-approval.test.ts | 174 ++++++++- .../workshop-backend/src/auto-approval.ts | 48 ++- packages/workshop-backend/src/overseer.ts | 38 +- packages/workshop-shared/src/api.ts | 5 +- packages/workshop-shared/src/gatekeeper.ts | 58 +++ 18 files changed, 1215 insertions(+), 120 deletions(-) diff --git a/packages/mcp-shared/__tests__/action-store.test.ts b/packages/mcp-shared/__tests__/action-store.test.ts index 2f2afd8e7..f0797a4b6 100644 --- a/packages/mcp-shared/__tests__/action-store.test.ts +++ b/packages/mcp-shared/__tests__/action-store.test.ts @@ -1,8 +1,15 @@ import { describe, expect, it } from "vitest"; import { DatabaseSync, type SQLInputValue } from "node:sqlite"; -import { ActionStore } from "../src/action-store.js"; -import { McpProtocolError, McpSessionExpiredError } from "../src/client.js"; +import { ActionInvalidatedError, ActionStore } from "../src/action-store.js"; +import { + ACTION_INVALIDATED_ERROR_CODE, + getActionInvalidationReason, +} from "@gadgets/workshop-shared/gatekeeper"; +import { + McpProtocolError, + McpSessionExpiredError, +} from "../src/client.js"; type TestSql = ConstructorParameters[0]; @@ -12,7 +19,7 @@ function fakeSql(): TestSql { exec(query: string, ...bindings: SQLInputValue[]) { const rows = bindings.length > 0 ? db.prepare(query).all(...bindings) - : /^\s*(?:SELECT|INSERT.*RETURNING)/is.test(query) + : /^\s*(?:SELECT|INSERT.*RETURNING)/is.test(query) ? db.prepare(query).all() : (db.exec(query), []); return { @@ -28,11 +35,81 @@ function fakeSql(): TestSql { const log = { debug() {}, info() {}, warn() {}, error() {}, with() { return log; } }; const ok = async () => ({ content: [{ type: "text" as const, text: "done" }] }); +const snapshot = { policyFingerprint: "action:manual", connectionGeneration: 1 }; +const stage = (store: ActionStore, args: Record) => + store.stage("send", args, snapshot); describe("ActionStore", () => { + it("persists the approved policy and account generation", () => { + const store = new ActionStore(fakeSql()); + const staged = store.stage("send", {}, { + policyFingerprint: "vetted:w01", + connectionGeneration: 7, + }); + + expect(store.get(staged.id)).toMatchObject({ + policyFingerprint: "vetted:w01", + connectionGeneration: 7, + }); + }); + + it("closes an action invalidated before dispatch without claiming it may have landed", async () => { + const store = new ActionStore(fakeSql()); + const staged = stage(store, {}); + + await expect(store.apply(staged.id, async () => { + throw new ActionInvalidatedError("Policy changed. Stage the call again."); + }, log)).rejects.toMatchObject({ + errorCode: ACTION_INVALIDATED_ERROR_CODE, + message: `${ACTION_INVALIDATED_ERROR_CODE}: Policy changed. Stage the call again.`, + }); + await expect(store.apply(staged.id, async () => { + throw new Error("must not dispatch again"); + }, log)).rejects.toMatchObject({ + errorCode: ACTION_INVALIDATED_ERROR_CODE, + message: `${ACTION_INVALIDATED_ERROR_CODE}: Policy changed. Stage the call again.`, + }); + + expect(store.get(staged.id)).toMatchObject({ + state: "failed", + retryable: false, + error: "Policy changed. Stage the call again.", + }); + }); + + it("retains the invalidation discriminator in the serialized error message", async () => { + const store = new ActionStore(fakeSql()); + const staged = stage(store, {}); + + const error = await store.apply(staged.id, async () => { + throw new ActionInvalidatedError("Policy changed."); + }, log).catch((caught: unknown) => caught); + + expect(error).toBeInstanceOf(Error); + expect((error as Error).message).toContain(ACTION_INVALIDATED_ERROR_CODE); + expect(getActionInvalidationReason(new Error((error as Error).message))) + .toBe("Policy changed."); + }); + + it("keeps validation failures retryable when tools/call was never reached", async () => { + const store = new ActionStore(fakeSql()); + const staged = stage(store, {}); + + await expect(store.apply(staged.id, async () => { + throw new McpProtocolError("validation failed"); + }, log)).rejects.toThrow(/validation failed/); + + expect(store.get(staged.id)).toMatchObject({ + state: "failed", + retryable: true, + error: "validation failed", + dispatched: false, + }); + }); + it("keeps a decided action available until the Gadget collects it", async () => { const store = new ActionStore(fakeSql()); - const staged = store.stage("send", { to: "a@b.c" }); + const staged = stage(store, { to: "a@b.c" }); await store.apply(staged.id, fn => fn({ callTool: ok } as never), log); expect(store.get(staged.id)?.state).toBe("applied"); expect(store.get(staged.id)?.result?.text).toBe("done"); @@ -41,9 +118,9 @@ describe("ActionStore", () => { it("retains only recent terminal actions without pruning pending ones", async () => { const sql = fakeSql(); const store = new ActionStore(sql); - const pending = store.stage("send", { pending: true }); + const pending = stage(store, { pending: true }); for (let i = 0; i < 250; i++) { - const staged = store.stage("send", { i }); + const staged = stage(store, { i }); await store.apply(staged.id, fn => fn({ callTool: ok } as never), log); } const { count } = sql.exec<{ count: number }>("SELECT count(*) AS count FROM mcp_actions").one(); @@ -59,8 +136,8 @@ describe("ActionStore", () => { // Pending records cannot be pruned, so without a cap they are an unbounded write primitive: a // Gadget that queues calls and never collects them grows the Durable Object forever. const store = new ActionStore(fakeSql()); - for (let i = 0; i < 50; i++) store.stage("send", { i }); - expect(() => store.stage("send", { overflow: true })).toThrow(/awaiting approval/); + for (let i = 0; i < 50; i++) stage(store, { i }); + expect(() => stage(store, { overflow: true })).toThrow(/awaiting approval/); }); it("rejects arguments that cannot be stored safely", () => { @@ -68,27 +145,27 @@ describe("ActionStore", () => { const circular: Record = {}; circular.self = circular; - expect(() => store.stage("send", circular)).toThrow(/JSON-compatible/); - expect(() => store.stage("send", { body: "x".repeat(70_000) })).toThrow(/too large/); + expect(() => stage(store, circular)).toThrow(/JSON-compatible/); + expect(() => stage(store, { body: "x".repeat(70_000) })).toThrow(/too large/); }); it("frees a pending slot however the call was decided", async () => { const store = new ActionStore(fakeSql()); - for (let i = 0; i < 48; i++) store.stage("send", { i }); - const applied = store.stage("send", { applied: true }); - const rejected = store.stage("send", { rejected: true }); - expect(() => store.stage("send", {})).toThrow(); + for (let i = 0; i < 48; i++) stage(store, { i }); + const applied = stage(store, { applied: true }); + const rejected = stage(store, { rejected: true }); + expect(() => stage(store, {})).toThrow(); await store.apply(applied.id, fn => fn({ callTool: ok } as never), log); store.reject(rejected.id); - expect(() => store.stage("send", {})).not.toThrow(); - expect(() => store.stage("send", {})).not.toThrow(); + expect(() => stage(store, {})).not.toThrow(); + expect(() => stage(store, {})).not.toThrow(); }); it("records a declined call as retryable, distinct from a rejection", async () => { // The server answered, so the tool did not run and another attempt cannot duplicate anything. const store = new ActionStore(fakeSql()); - const staged = store.stage("send", {}); + const staged = stage(store, {}); const declined = async () => { throw new McpProtocolError("MCP server rejected: unknown tool", -32601, "declined"); }; @@ -110,7 +187,7 @@ describe("ActionStore", () => { // server acted, so the request may already have been carried out. Retrying is how one approval // becomes two writes, and MCP has no inverse operation to undo the second. const store = new ActionStore(fakeSql()); - const staged = store.stage("send", {}); + const staged = stage(store, {}); let calls = 0; const dropped = async () => { calls++; @@ -133,7 +210,7 @@ describe("ActionStore", () => { it("treats an unrecognised failure as possibly performed", async () => { // Fails safe. A throw site that has not said what it means is not evidence the tool never ran. const store = new ActionStore(fakeSql()); - const staged = store.stage("send", {}); + const staged = stage(store, {}); const boom = async () => { throw new Error("upstream exploded"); }; await expect(store.apply(staged.id, fn => fn({ callTool: boom } as never), log)) .rejects.toThrow(/may or may not have taken effect/); @@ -151,7 +228,7 @@ describe("ActionStore", () => { }; for (let attempt = 0; attempt < 250; attempt++) { - const staged = store.stage("send", { attempt }); + const staged = stage(store, { attempt }); await expect(store.apply(staged.id, fn => fn({ callTool: declined } as never), log)) .rejects.toThrow(); } @@ -164,7 +241,7 @@ describe("ActionStore", () => { // Pruning must not reach the record that just failed: the Gadget learns the outcome by asking // for it afterwards. const store = new ActionStore(fakeSql()); - const staged = store.stage("send", {}); + const staged = stage(store, {}); const declined = async () => { throw new McpProtocolError("MCP server rejected: unknown tool", -32601, "declined"); }; @@ -178,7 +255,7 @@ describe("ActionStore", () => { // MCP says a session-bearing 404 means no dispatch, but a fronting proxy can produce the same // response after the upstream accepted the write. The two cases are indistinguishable here. const store = new ActionStore(fakeSql()); - const staged = store.stage("send", {}); + const staged = stage(store, {}); let calls = 0; const expired = async () => { calls++; @@ -199,7 +276,7 @@ describe("ActionStore", () => { // tool. MCP describes no inverse operation, so the duplicate write cannot be undone -- the // claim has to be taken before the first await, not after the call returns. const store = new ActionStore(fakeSql()); - const staged = store.stage("send", { to: "a@b.c" }); + const staged = stage(store, { to: "a@b.c" }); let calls = 0; const slow = async () => { calls++; @@ -221,7 +298,7 @@ describe("ActionStore", () => { it("does not reject an action after application has started", async () => { const store = new ActionStore(fakeSql()); - const staged = store.stage("send", {}); + const staged = stage(store, {}); let release!: () => void; const blocked = new Promise(resolve => { release = resolve; }); const applying = store.apply(staged.id, async fn => { @@ -243,7 +320,7 @@ describe("ActionStore", () => { // write permanently. The action is closed as failed and a person decides what to do. const sql = fakeSql(); const store = new ActionStore(sql); - const staged = store.stage("send", {}); + const staged = stage(store, {}); sql.exec( "UPDATE mcp_actions SET state = 'applying', claimed_at = ? WHERE id = ?", Date.now() - 5 * 60 * 1000, @@ -265,19 +342,19 @@ describe("ActionStore", () => { // Object never makes one. Counting those rows retired the binding one interruption at a time. const sql = fakeSql(); const store = new ActionStore(sql); - for (let n = 0; n < 50; n++) store.stage("send", { n }); + for (let n = 0; n < 50; n++) stage(store, { n }); sql.exec( "UPDATE mcp_actions SET state = 'applying', claimed_at = ?", Date.now() - 5 * 60 * 1000, ); - expect(new ActionStore(sql).stage("send", { n: 50 }).state).toBe("pending"); + expect(stage(new ActionStore(sql), { n: 50 }).state).toBe("pending"); }); it("does not retire a live slow call when another action is staged", async () => { const sql = fakeSql(); const store = new ActionStore(sql); - const staged = store.stage("send", {}); + const staged = stage(store, {}); let release!: () => void; const applying = store.apply(staged.id, async fn => { await new Promise(resolve => { release = resolve; }); @@ -289,7 +366,7 @@ describe("ActionStore", () => { staged.id, ); - store.stage("send", { another: true }); + stage(store, { another: true }); expect(store.get(staged.id)?.state).toBe("applying"); release(); await applying; @@ -299,14 +376,14 @@ describe("ActionStore", () => { const sql = fakeSql(); let store = new ActionStore(sql); for (let batch = 0; batch < 3; batch++) { - for (let n = 0; n < 50; n++) store.stage("send", { batch, n }); + for (let n = 0; n < 50; n++) stage(store, { batch, n }); sql.exec( "UPDATE mcp_actions SET state = 'applying', claimed_at = ? WHERE state = 'pending'", Date.now() - 5 * 60 * 1000, ); store = new ActionStore(sql); } - store.stage("send", { final: true }); + stage(store, { final: true }); const { applying } = sql.exec<{ applying: number }>( "SELECT count(*) AS applying FROM mcp_actions WHERE state = 'applying'").one(); @@ -322,7 +399,7 @@ describe("ActionStore", () => { // and the next attempt re-ran a tool call that had already taken effect. Losing the result is // recoverable; repeating the write is not. const store = new ActionStore(fakeSql()); - const staged = store.stage("send", {}); + const staged = stage(store, {}); let calls = 0; const poison = async () => { calls++; @@ -342,7 +419,7 @@ describe("ActionStore", () => { it("bounds retained results by UTF-8 bytes", async () => { const store = new ActionStore(fakeSql()); - const staged = store.stage("send", {}); + const staged = stage(store, {}); const big = async () => ({ content: [{ type: "text" as const, text: "😀".repeat(40_000) }] }); await store.apply(staged.id, fn => fn({ callTool: big } as never), log); expect(store.get(staged.id)?.state).toBe("applied"); @@ -351,7 +428,7 @@ describe("ActionStore", () => { it("is idempotent, because the Workshop may retry a call whose result it never saw", async () => { const store = new ActionStore(fakeSql()); - const staged = store.stage("send", {}); + const staged = stage(store, {}); let calls = 0; const counting = async () => { calls++; return { content: [] }; }; await store.apply(staged.id, fn => fn({ callTool: counting } as never), log); diff --git a/packages/mcp-shared/__tests__/client-pagination.test.ts b/packages/mcp-shared/__tests__/client-pagination.test.ts index 1d82bfe1f..c3df576e9 100644 --- a/packages/mcp-shared/__tests__/client-pagination.test.ts +++ b/packages/mcp-shared/__tests__/client-pagination.test.ts @@ -183,6 +183,24 @@ describe("McpClient.listTools", () => { expect(calls()).toBe(2); }); + it("does not carry an exact lookup deadline into a later tool call", async () => { + let calls = 0; + vi.stubGlobal("fetch", async (_input: unknown, init?: RequestInit) => { + calls++; + const request = JSON.parse(String(init?.body)); + return new Response(JSON.stringify({ + jsonrpc: "2.0", + id: request.id, + result: { content: [] }, + }), { headers: { "Content-Type": "application/json" } }); + }); + const client = new McpClient("https://mcp.example.com/mcp", async () => null); + + await expect(client.findTool("send", 0)).rejects.toThrow(/timed out|timeout/i); + await expect(client.callTool("send", {})).resolves.toEqual({ content: [] }); + expect(calls).toBe(1); + }); + it("stops paging when a bounded search has enough matches", async () => { const calls = stubPages([{ tools: Array.from({ length: 25 }, (_, i) => ({ name: `jira_tool_${i}` })), diff --git a/packages/mcp-shared/__tests__/connection.test.ts b/packages/mcp-shared/__tests__/connection.test.ts index de7d029f3..4378b3c36 100644 --- a/packages/mcp-shared/__tests__/connection.test.ts +++ b/packages/mcp-shared/__tests__/connection.test.ts @@ -5,7 +5,12 @@ import { McpCallNotDispatchedError, McpClient, } from "../src/client.js"; -import { withClient, type ConnectionAccount } from "../src/connection.js"; +import { + createMcpConnectionChangedRpcError, + McpConnectionChangedError, + withClient, + type ConnectionAccount, +} from "../src/connection.js"; afterEach(() => vi.unstubAllGlobals()); @@ -62,6 +67,24 @@ it("classifies credential lookup failure as not dispatched", async () => { expect(callMayHaveTakenEffect(error)).toBe(false); }); +it("preserves a connection change detected during credential lookup", async () => { + const account: ConnectionAccount = { + async getConnection() { + throw createMcpConnectionChangedRpcError("The account was repointed."); + }, + async assertConnectionCurrent() {}, + async setMcpSessionId() { return true; }, + async noteCredentialsExpired() {}, + }; + + const error = await withClient({}, account, "https://mcp.example.com", + client => client.callTool("send", {}), { retryOnExpiry: false }).catch(err => err); + + expect(error).toBeInstanceOf(McpConnectionChangedError); + expect(error.message).toBe("The account was repointed."); + expect(callMayHaveTakenEffect(error)).toBe(false); +}); + it("classifies initialization failure as not dispatched", async () => { vi.stubGlobal("fetch", async () => new Response(null, { status: 500 })); const account: ConnectionAccount = { @@ -102,7 +125,7 @@ it("rechecks account generation immediately before the tool request", async () = return { authorization: "token", sessionId: "session", generation: 1 }; }, async assertConnectionCurrent() { - if (!current) throw new Error("connection replaced"); + if (!current) throw createMcpConnectionChangedRpcError("connection replaced"); }, async setMcpSessionId() { return true; }, async noteCredentialsExpired() {}, @@ -113,7 +136,7 @@ it("rechecks account generation immediately before the tool request", async () = return client.callTool("send", {}); }, { retryOnExpiry: false }).catch(err => err); - expect(error).toBeInstanceOf(McpCallNotDispatchedError); + expect(error).toBeInstanceOf(McpConnectionChangedError); expect(requests).toBe(0); }); diff --git a/packages/mcp-shared/__tests__/facet.test.ts b/packages/mcp-shared/__tests__/facet.test.ts index 38112314e..929497367 100644 --- a/packages/mcp-shared/__tests__/facet.test.ts +++ b/packages/mcp-shared/__tests__/facet.test.ts @@ -1,13 +1,24 @@ import { expect, it, vi } from "vitest"; +import { DatabaseSync, type SQLInputValue } from "node:sqlite"; import { McpFacetBase } from "../src/facet.js"; import { McpSessionBase } from "../src/session.js"; -import { classifyTool, type ServerTrust } from "../src/tools.js"; +import { classifyTool, toolPolicyFingerprint, type ServerTrust } from "../src/tools.js"; import type { McpClient, McpTool } from "../src/client.js"; import type { ToolScope } from "../src/scope.js"; import type { ScopedCatalog } from "../src/catalog.js"; -import type { ConnectionAccount } from "../src/connection.js"; -import type { ResourceDescription } from "@gadgets/workshop-shared/gatekeeper"; +import { + McpConnectionChangedError, + type ConnectionAccount, + type WithClientOptions, +} from "../src/connection.js"; +import { + ACTION_INVALIDATED_ERROR_CODE, + ACTION_RESTAGE_REQUIRED_ERROR_CODE, + getActionInvalidationReason, + getActionRestageRequiredReason, + type ResourceDescription, +} from "@gadgets/workshop-shared/gatekeeper"; const log = { debug() {}, info() {}, error() {}, @@ -16,6 +27,23 @@ const log = { with() { return this; }, }; +function fakeSql(): SqlStorage { + const db = new DatabaseSync(":memory:"); + return { + exec(query: string, ...bindings: SQLInputValue[]) { + const rows = bindings.length > 0 + ? db.prepare(query).all(...bindings) + : /^\s*(?:SELECT|INSERT.*RETURNING)/is.test(query) + ? db.prepare(query).all() + : (db.exec(query), []); + return { + toArray: () => rows as T[], + one: () => rows[0] as T, + }; + }, + } as unknown as SqlStorage; +} + class TestSession extends McpSessionBase {} class TestFacet extends McpFacetBase = []; + discoveryDeadlines: Array = []; + connectionFailure: Error | undefined; + toolCallFailure: Error | undefined; + beforeToolLookup: (() => Promise) | undefined; + beforeToolCall: (() => Promise) | undefined; beforeCatalogRead: (() => Promise) | undefined; protected get log() { return log; } @@ -49,21 +85,60 @@ class TestFacet extends McpFacetBase( - fn: (client: McpClient) => Promise, + fn: (client: McpClient, connectionGeneration: number) => Promise, + options?: WithClientOptions, ): Promise { + if (this.connectionFailure) throw this.connectionFailure; this.remoteCalls++; + this.callOptions.push(options); const client = { - findTool: async (name: string) => this.remoteTools.find(tool => tool.name === name), + findTool: async (name: string, deadline?: number) => { + this.discoveryDeadlines.push(deadline); + await this.beforeToolLookup?.(); + return this.remoteTools.find(tool => tool.name === name); + }, listTools: async ( _maxTools: number, include: (tool: McpTool) => boolean, - ) => ({ tools: this.remoteTools.filter(include), truncated: false }), + deadline?: number, + ) => { + this.discoveryDeadlines.push(deadline); + await this.beforeToolLookup?.(); + return { tools: this.remoteTools.filter(include), truncated: false }; + }, listMatchingToolSummaries: async ( maxTools: number, include: (tool: McpTool) => boolean, ) => this.remoteTools.filter(include).slice(0, maxTools), + callTool: async () => { + if (this.toolCallFailure) throw this.toolCallFailure; + this.toolCalls++; + await this.beforeToolCall?.(); + return { content: [] }; + }, } as unknown as McpClient; - return fn(client); + return fn(client, this.connectionGeneration); + } + setScope(scope: ToolScope) { this.ctx.props.scope = scope; } + removeActionSnapshot(action: number) { + this.ctx.storage.sql.exec( + "UPDATE mcp_actions SET policy_fingerprint = NULL, connection_generation = NULL WHERE id = ?", + action, + ); + } + markActionOutcomeUnknown(action: number) { + this.ctx.storage.sql.exec( + `UPDATE mcp_actions SET state = 'failed', retryable = 0, dispatched = NULL, + error = 'This call may or may not have taken effect.' WHERE id = ?`, + action, + ); + } + markActionLegacyRetryableFailure(action: number) { + this.ctx.storage.sql.exec( + `UPDATE mcp_actions SET state = 'failed', retryable = NULL, dispatched = NULL, + error = 'Temporary failure.' WHERE id = ?`, + action, + ); } runDiscoveryTest(operation: () => Promise): Promise { return this.runDiscovery(operation); @@ -71,9 +146,9 @@ class TestFacet extends McpFacetBase { + const subject = facet(); + subject.catalogResult = Promise.resolve({ tools: [], isPortal: false, truncated: true }); + subject.remoteTools = [{ name: "search_issues", annotations: { readOnlyHint: true } }]; + + await expect(subject.findTool("search_issues")).resolves.toMatchObject({ mode: "read" }); + subject.remoteTools = [{ name: "search_issues", annotations: { readOnlyHint: false } }]; + await expect(subject.resolveToolForCall("search_issues")) + .resolves.toMatchObject({ entry: { mode: "action" } }); + expect(subject.remoteCalls).toBe(2); +}); + +it("refreshes portal identity before dispatching a newly portal-native tool", async () => { + const subject = facet(); + subject.remoteTools = [ + { name: "portal_list_servers" }, + { name: "portal_toggle_servers" }, + ]; + + await expect(subject.resolveToolForCall("portal_toggle_servers")).resolves.toBeUndefined(); +}); + it("bounds concurrent discovery work across distinct requests", async () => { const subject = facet(); let active = 0; @@ -189,3 +286,235 @@ it("bounds concurrent catalog reads", async () => { release(); await Promise.all(searches); }); + +it("bounds approval-time tool revalidation", async () => { + const subject = facet(); + const tool: McpTool = { name: "send", annotations: { readOnlyHint: false } }; + subject.catalogResult = Promise.resolve({ + isPortal: false, + truncated: false, + tools: [classifyTool(tool, "byo")], + }); + subject.remoteTools = [tool]; + let active = 0; + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + subject.beforeToolLookup = async () => { + active++; + await gate; + active--; + }; + const snapshot = { + policyFingerprint: toolPolicyFingerprint(tool, "byo"), + connectionGeneration: 1, + }; + const actions = Array.from( + { length: 12 }, (_, i) => subject.stageAction("send", { i }, snapshot).id); + + const applying = actions.map(action => subject.applyAction(action)); + + await vi.waitFor(() => expect(active).toBeGreaterThan(0)); + expect(active).toBe(4); + release(); + await Promise.all(applying); +}); + +it("bounds revalidation without imposing its deadline on action dispatch", async () => { + const subject = facet(); + const tool: McpTool = { name: "send" }; + subject.remoteTools = [tool]; + const action = subject.stageAction("send", {}, { + policyFingerprint: toolPolicyFingerprint(tool, "byo"), + connectionGeneration: 1, + }); + + await subject.applyAction(action.id); + + expect(subject.discoveryDeadlines).toHaveLength(1); + expect(subject.discoveryDeadlines[0]).toEqual(expect.any(Number)); + expect(subject.callOptions).toEqual([{ retryOnExpiry: false }]); +}); + +it("releases discovery capacity before a validated action is dispatched", async () => { + const subject = facet(); + const tool: McpTool = { name: "send" }; + subject.remoteTools = [tool]; + let activeDispatches = 0; + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + subject.beforeToolCall = async () => { + activeDispatches++; + await gate; + activeDispatches--; + }; + const snapshot = { + policyFingerprint: toolPolicyFingerprint(tool, "byo"), + connectionGeneration: 1, + }; + const actions = Array.from({ length: 4 }, () => subject.stageAction("send", {}, snapshot).id); + const applying = actions.map(action => subject.applyAction(action)); + await vi.waitFor(() => expect(activeDispatches).toBe(4)); + + let discoveryStarted = false; + const discovery = subject.runDiscoveryTest(async () => { discoveryStarted = true; }); + try { + await vi.waitFor(() => expect(discoveryStarted).toBe(true)); + } finally { + release(); + await Promise.all([...applying, discovery]); + } +}); + +it("requires a legacy action to be restaged without misreporting a connection change", async () => { + const subject = facet(); + const action = subject.stageAction("send", {}, { + policyFingerprint: toolPolicyFingerprint({ name: "send" }, "byo"), + connectionGeneration: 1, + }); + subject.removeActionSnapshot(action.id); + + const error = await subject.applyAction(action.id).catch((caught: unknown) => caught); + + expect(error).toMatchObject({ errorCode: ACTION_RESTAGE_REQUIRED_ERROR_CODE }); + expect(getActionRestageRequiredReason(new Error((error as Error).message))) + .toMatch(/predates current approval policy checks/i); + expect(subject.lookupAction(action.id)).toMatchObject({ + state: "failed", + retryable: false, + dispatched: false, + error: expect.stringMatching(/predates current approval policy checks/i), + }); + await expect(subject.applyAction(action.id)) + .rejects.toMatchObject({ errorCode: ACTION_RESTAGE_REQUIRED_ERROR_CODE }); + const session = await subject.startSession(queue as never); + await expect(session.getActionResult(action.id)).resolves.toMatchObject({ + status: "failed", + message: expect.stringMatching(/predates current approval policy checks/i), + }); + expect(subject.toolCalls).toBe(0); +}); + +it("preserves an outcome-unknown legacy action instead of declaring it safe to restage", async () => { + const subject = facet(); + const action = subject.stageAction("send", {}, { + policyFingerprint: toolPolicyFingerprint({ name: "send" }, "byo"), + connectionGeneration: 1, + }); + subject.removeActionSnapshot(action.id); + subject.markActionOutcomeUnknown(action.id); + + await expect(subject.applyAction(action.id)).rejects.toThrow(/may or may not have taken effect/i); + expect(subject.lookupAction(action.id)).toMatchObject({ + state: "failed", + retryable: false, + error: "This call may or may not have taken effect.", + }); +}); + +it("restages a legacy failed action whose absent retryable flag means retryable", async () => { + const subject = facet(); + const action = subject.stageAction("send", {}, { + policyFingerprint: toolPolicyFingerprint({ name: "send" }, "byo"), + connectionGeneration: 1, + }); + subject.removeActionSnapshot(action.id); + subject.markActionLegacyRetryableFailure(action.id); + + await expect(subject.applyAction(action.id)) + .rejects.toMatchObject({ errorCode: ACTION_RESTAGE_REQUIRED_ERROR_CODE }); + expect(subject.lookupAction(action.id)).toMatchObject({ + state: "failed", + retryable: false, + dispatched: false, + error: expect.stringMatching(/predates current approval policy checks/i), + }); +}); + +it("does not dispatch after the connection generation changes", async () => { + const subject = facet(); + const tool: McpTool = { name: "send" }; + subject.remoteTools = [tool]; + const action = subject.stageAction("send", {}, { + policyFingerprint: toolPolicyFingerprint(tool, "byo"), + connectionGeneration: 1, + }); + subject.connectionGeneration = 2; + + await expect(subject.applyAction(action.id)).rejects.toThrow(/connection changed/i); + expect(subject.toolCalls).toBe(0); +}); + +it("invalidates an action when the connection changes before a client opens", async () => { + const subject = facet(); + const tool: McpTool = { name: "send" }; + const action = subject.stageAction("send", {}, { + policyFingerprint: toolPolicyFingerprint(tool, "byo"), + connectionGeneration: 1, + }); + subject.connectionFailure = new McpConnectionChangedError("The account was repointed."); + + const error = await subject.applyAction(action.id).catch((caught: unknown) => caught); + + expect(error).toMatchObject({ errorCode: ACTION_INVALIDATED_ERROR_CODE }); + expect(getActionInvalidationReason(new Error((error as Error).message))) + .toMatch(/connection changed/i); + expect(subject.lookupAction(action.id)).toMatchObject({ + state: "failed", + retryable: false, + dispatched: false, + }); + expect(subject.toolCalls).toBe(0); +}); + +it("replays an assert-time connection invalidation as not dispatched", async () => { + const subject = facet(); + const tool: McpTool = { name: "send" }; + subject.remoteTools = [tool]; + const action = subject.stageAction("send", {}, { + policyFingerprint: toolPolicyFingerprint(tool, "byo"), + connectionGeneration: 1, + }); + subject.toolCallFailure = new McpConnectionChangedError("The account was repointed."); + + await expect(subject.applyAction(action.id)) + .rejects.toMatchObject({ errorCode: ACTION_INVALIDATED_ERROR_CODE }); + expect(subject.lookupAction(action.id)).toMatchObject({ + state: "failed", + retryable: false, + dispatched: false, + }); + await expect(subject.applyAction(action.id)) + .rejects.toMatchObject({ errorCode: ACTION_INVALIDATED_ERROR_CODE }); + expect(subject.toolCalls).toBe(0); +}); + +it("does not dispatch after effective tool policy changes", async () => { + const subject = facet(); + const tool: McpTool = { name: "send" }; + subject.remoteTools = [{ name: "send", annotations: { readOnlyHint: true } }]; + const action = subject.stageAction("send", {}, { + policyFingerprint: toolPolicyFingerprint(tool, "byo"), + connectionGeneration: 1, + }); + + await expect(subject.applyAction(action.id)).rejects.toThrow(/policy changed/i); + expect(subject.toolCalls).toBe(0); +}); + +it("does not dispatch a removed or newly out-of-scope tool", async () => { + const tool: McpTool = { name: "send" }; + for (const configure of [ + (subject: TestFacet) => { subject.remoteTools = []; }, + (subject: TestFacet) => { subject.remoteTools = [tool]; subject.setScope({ tools: ["other"] }); }, + ]) { + const subject = facet(); + const action = subject.stageAction("send", {}, { + policyFingerprint: toolPolicyFingerprint(tool, "byo"), + connectionGeneration: 1, + }); + configure(subject); + + await expect(subject.applyAction(action.id)).rejects.toThrow(/no longer allowed|policy changed/i); + expect(subject.toolCalls).toBe(0); + } +}); diff --git a/packages/mcp-shared/__tests__/session.test.ts b/packages/mcp-shared/__tests__/session.test.ts index 081e83ff4..712fdb47d 100644 --- a/packages/mcp-shared/__tests__/session.test.ts +++ b/packages/mcp-shared/__tests__/session.test.ts @@ -4,6 +4,12 @@ import { McpSessionBase, type McpSessionHost, type StoredAction } from "../src/s import { MAX_TOOL_NAME_CHARS } from "../src/client.js"; import { classifyTool } from "../src/tools.js"; +const resolved = (entry: ReturnType) => ({ + entry, + policyFingerprint: "byo:w--", + connectionGeneration: 1, +}); + it("reports an execution failure distinctly from a rejected approval", async () => { const failed: StoredAction = { id: 1, @@ -41,7 +47,7 @@ it("tells an agent to return a pending action so its approval can appear in chat serverName: "Jira", endpoint: "https://mcp.example.com", scope: { serverId: "jira" }, - findTool: async () => entry, + resolveToolForCall: async () => resolved(entry), stageAction: () => staged, discardStagedAction() {}, actionKindFor: () => ({ tag: "jira:create", label: "Create issue" }), @@ -95,10 +101,10 @@ it("calls a tool resolved beyond the initial generated catalog", async () => { endpoint: "https://mcp.example.com", scope: { serverId: "jira" }, tools: async () => [], - findTool: async () => expanded, - call: async (fn: (client: never) => Promise) => fn({ + resolveToolForCall: async () => resolved(expanded), + call: async (fn: (client: never, connectionGeneration: number) => Promise) => fn({ callTool: async () => ({ content: [{ type: "text", text: "PROJ-1" }] }), - } as never), + } as never, 1), } as unknown as McpSessionHost; const queue = { authorizeObservation() {} }; const session = new McpSessionBase(host, queue as never); @@ -109,6 +115,49 @@ it("calls a tool resolved beyond the initial generated catalog", async () => { }); }); +it("resolves current tool policy before deciding whether a call needs approval", async () => { + let resolvedPolicy = false; + const host = { + serverName: "Jira", + endpoint: "https://mcp.example.com", + scope: { serverId: "jira" }, + resolveToolForCall: async () => { + resolvedPolicy = true; + return resolved(classifyTool( + { name: "jira_search_issues", annotations: { readOnlyHint: true } }, "byo")); + }, + call: async (fn: (client: never, connectionGeneration: number) => Promise) => fn({ + callTool: async () => ({ content: [] }), + } as never, 1), + } as unknown as McpSessionHost; + const session = new McpSessionBase(host, { authorizeObservation() {} } as never); + + await session.callTool("jira_search_issues"); + + expect(resolvedPolicy).toBe(true); +}); + +it("does not dispatch a read after its connection generation changes", async () => { + let called = false; + const entry = classifyTool({ + name: "jira_search_issues", + annotations: { readOnlyHint: true }, + }, "byo"); + const host = { + serverName: "Jira", + endpoint: "https://mcp.example.com", + scope: { serverId: "jira" }, + resolveToolForCall: async () => resolved(entry), + call: async (fn: (client: never, connectionGeneration: number) => Promise) => + fn({ callTool: async () => { called = true; return { content: [] }; } } as never, + 2), + } as unknown as McpSessionHost; + const session = new McpSessionBase(host, { authorizeObservation() {} } as never); + + await expect(session.callTool("jira_search_issues")).rejects.toThrow(/connection changed/i); + expect(called).toBe(false); +}); + it("identifies the tool in a describe observation", async () => { const observations: { description: string }[] = []; const host = { @@ -136,6 +185,7 @@ it("names the grant, not the server, when a scoped binding lacks the tool", asyn endpoint: "https://mcp.example.com", scope: { serverId: "jira" }, findTool: async () => undefined, + resolveToolForCall: async () => undefined, } as unknown as McpSessionHost; const session = new McpSessionBase(host, { authorizeObservation() {} } as never); diff --git a/packages/mcp-shared/__tests__/tools.test.ts b/packages/mcp-shared/__tests__/tools.test.ts index 275ebd45a..625a9f47d 100644 --- a/packages/mcp-shared/__tests__/tools.test.ts +++ b/packages/mcp-shared/__tests__/tools.test.ts @@ -4,6 +4,7 @@ import { catalogRevision, classifyTool, describeCall, + toolPolicyFingerprint, toolInfo, } from "../src/tools.js"; import type { McpTool } from "../src/client.js"; @@ -69,6 +70,21 @@ describe("classifyTool", () => { }); }); +describe("toolPolicyFingerprint", () => { + it("tracks effective dispatch policy rather than inert raw claims", () => { + expect(toolPolicyFingerprint(tool(), "byo")).toBe(toolPolicyFingerprint(tool({ + destructiveHint: false, + idempotentHint: true, + }), "byo")); + expect(toolPolicyFingerprint(tool(), "vetted")) + .not.toBe(toolPolicyFingerprint(tool({ readOnlyHint: true }), "vetted")); + expect(toolPolicyFingerprint(tool(), "vetted")).not.toBe(toolPolicyFingerprint(tool({ + destructiveHint: false, + idempotentHint: true, + }), "vetted")); + }); +}); + describe("toolInfo", () => { it("records where the read classification came from", () => { // Delegated trust is fine; invisible trust is not. Consumers can always see which it was. diff --git a/packages/mcp-shared/src/account.ts b/packages/mcp-shared/src/account.ts index 02c07b9c8..0e64a2256 100644 --- a/packages/mcp-shared/src/account.ts +++ b/packages/mcp-shared/src/account.ts @@ -24,7 +24,12 @@ import { } from "@modelcontextprotocol/client"; import { McpAuthRequiredError, McpClient, type McpServerInfo } from "./client.js"; -import { clientName, type ConnectionEnv, type McpConnection } from "./connection.js"; +import { + clientName, + createMcpConnectionChangedRpcError, + type ConnectionEnv, + type McpConnection, +} from "./connection.js"; import { ACCESS_TOKEN_SAFETY_MS, CONNECT_TIMEOUT_MS, @@ -705,15 +710,25 @@ export abstract class McpAccountBase // is about to contact; reject before reading or refreshing credentials if the account has since // moved. Otherwise a bearer token minted for the new portal would be sent to the old endpoint. if (!sameEndpoint(endpoint, server.endpoint)) { - throw new Error( + throw createMcpConnectionChangedRpcError( `This binding is for ${hostOf(endpoint)}, but the account is now connected to ` + `${hostOf(server.endpoint)}. Replace the binding before using it again.`); } - const authorization = await this.#getAuthorization(server, generation); + let authorization: string | null; + try { + authorization = await this.#getAuthorization(server, generation); + } catch (error) { + if (!this.isCurrentConnection(server, generation)) { + throw createMcpConnectionChangedRpcError( + "This MCP connection changed while credentials were being prepared. Try again."); + } + throw error; + } // `#getAuthorization` may await a token refresh. A reconnect can interleave there, so recheck // before returning the credential to a caller that still intends to contact the old endpoint. if (!this.isCurrentConnection(server, generation)) { - throw new Error("This MCP connection changed while credentials were being prepared. Try again."); + throw createMcpConnectionChangedRpcError( + "This MCP connection changed while credentials were being prepared. Try again."); } return { authorization, @@ -727,7 +742,8 @@ export abstract class McpAccountBase const server = this.server(); if (!server || !sameEndpoint(endpoint, server.endpoint) || generation !== this.connectionGeneration()) { - throw new Error("This MCP connection changed before the request was sent. Try again."); + throw createMcpConnectionChangedRpcError( + "This MCP connection changed before the request was sent. Try again."); } } diff --git a/packages/mcp-shared/src/action-store.ts b/packages/mcp-shared/src/action-store.ts index bc78811e8..1c0e2410e 100644 --- a/packages/mcp-shared/src/action-store.ts +++ b/packages/mcp-shared/src/action-store.ts @@ -1,6 +1,12 @@ // Durable lifecycle for approval-gated MCP tool calls. The owning facet supplies its isolated SQLite // database; claims are persisted before external I/O so an interrupted write is never replayed. +import { + ACTION_INVALIDATED_ERROR_CODE, + createActionInvalidatedError, + createActionRestageRequiredError, + getActionRestageRequiredReason, +} from "@gadgets/workshop-shared/gatekeeper"; import { callMayHaveTakenEffect, type McpClient, type McpToolCallResult } from "./client.js"; import type { McpLog } from "./log.js"; import type { StoredAction } from "./session.js"; @@ -18,6 +24,9 @@ type ActionRow = { args_json: string; state: StoredAction["state"]; submitted_at: number; + policy_fingerprint: string | null; + connection_generation: number | null; + dispatched: number | null; claimed_at: number | null; retryable: number | null; result_json: string | null; @@ -31,6 +40,9 @@ function fromRow(row: ActionRow): StoredAction { args: JSON.parse(row.args_json) as Record, state: row.state, submittedAt: row.submitted_at, + policyFingerprint: row.policy_fingerprint ?? undefined, + connectionGeneration: row.connection_generation ?? undefined, + dispatched: row.dispatched === null ? undefined : row.dispatched === 1, claimedAt: row.claimed_at ?? undefined, retryable: row.retryable === null ? undefined : row.retryable === 1, result: row.result_json @@ -45,6 +57,20 @@ export const APPLY_OUTCOME_UNKNOWN_MESSAGE = "This call was interrupted after it had been sent, so it may or may not have taken effect. " + "Check the server before trying it again."; +/** The approved tool or account policy changed before dispatch, so the action must be restaged. */ +export class ActionInvalidatedError extends Error { + /** Stable code retained when this error crosses the gatekeeper RPC boundary. */ + readonly errorCode = ACTION_INVALIDATED_ERROR_CODE; + /** Human-readable reason stored separately from the serialized discriminator. */ + readonly reason: string; + + constructor(reason: string) { + super(createActionInvalidatedError(reason).message); + this.name = "ActionInvalidatedError"; + this.reason = reason; + } +} + /** Stores queued MCP actions in one facet-local SQLite table. */ export class ActionStore { #sql: SqlStorage; @@ -57,18 +83,40 @@ export class ActionStore { args_json TEXT NOT NULL CHECK (json_valid(args_json) AND json_type(args_json) = 'object'), state TEXT NOT NULL CHECK (state IN ('pending', 'applying', 'applied', 'rejected', 'failed')), submitted_at INTEGER NOT NULL, + policy_fingerprint TEXT, + connection_generation INTEGER, + dispatched INTEGER CHECK (dispatched IS NULL OR dispatched IN (0, 1)), claimed_at INTEGER, retryable INTEGER CHECK (retryable IS NULL OR retryable IN (0, 1)), result_json TEXT CHECK (result_json IS NULL OR json_valid(result_json)), error TEXT ) STRICT`); + const columns = new Set( + sql.exec<{ name: string }>( + "SELECT name FROM pragma_table_info('mcp_actions')", + ).toArray().map(column => column.name), + ); + if (!columns.has("policy_fingerprint")) { + sql.exec("ALTER TABLE mcp_actions ADD COLUMN policy_fingerprint TEXT"); + } + if (!columns.has("connection_generation")) { + sql.exec("ALTER TABLE mcp_actions ADD COLUMN connection_generation INTEGER"); + } + if (!columns.has("dispatched")) { + sql.exec("ALTER TABLE mcp_actions ADD COLUMN dispatched INTEGER"); + } // A fresh store means a fresh Durable Object activation. Any persisted claim belonged to an // interrupted prior activation and must never be replayed because the write may have landed. sql.exec( `UPDATE mcp_actions SET state = 'failed', retryable = 0, error = ? - WHERE state = 'applying'`, + WHERE state = 'applying' AND dispatched IS NOT 0`, APPLY_OUTCOME_UNKNOWN_MESSAGE, ); + sql.exec( + `UPDATE mcp_actions SET state = 'failed', retryable = 1, + error = 'This call was interrupted before it was sent. Try applying it again.' + WHERE state = 'applying' AND dispatched = 0`, + ); this.#prune(); } @@ -80,18 +128,24 @@ export class ActionStore { #save(action: StoredAction): void { this.#sql.exec( - `UPDATE mcp_actions SET state = ?, claimed_at = ?, retryable = ?, result_json = ?, error = ? + `UPDATE mcp_actions SET state = ?, claimed_at = ?, retryable = ?, result_json = ?, error = ?, + dispatched = ? WHERE id = ?`, action.state, action.claimedAt ?? null, action.retryable === undefined ? null : Number(action.retryable), action.result === undefined ? null : JSON.stringify(action.result), action.error ?? null, + action.dispatched === undefined ? null : Number(action.dispatched), action.id, ); } - stage(toolName: string, args: Record): StoredAction { + stage( + toolName: string, + args: Record, + snapshot: { policyFingerprint: string; connectionGeneration: number }, + ): StoredAction { let argsJson: string; let storedArgs: Record; try { @@ -116,17 +170,32 @@ export class ActionStore { const submittedAt = Date.now(); const { id } = this.#sql.exec<{ id: number }>( - `INSERT INTO mcp_actions (tool_name, args_json, state, submitted_at) - VALUES (?, ?, 'pending', ?) RETURNING id`, - toolName, argsJson, submittedAt, + `INSERT INTO mcp_actions ( + tool_name, args_json, state, submitted_at, policy_fingerprint, connection_generation + ) VALUES (?, ?, 'pending', ?, ?, ?) RETURNING id`, + toolName, argsJson, submittedAt, snapshot.policyFingerprint, snapshot.connectionGeneration, ).one(); - return { id, toolName, args: storedArgs, state: "pending", submittedAt }; + return { id, toolName, args: storedArgs, state: "pending", submittedAt, ...snapshot }; } discard(id: number): void { this.#sql.exec("DELETE FROM mcp_actions WHERE id = ? AND state = 'pending'", id); } + /** Closes an old queued action that cannot be checked against an approval-policy snapshot. */ + markRestageRequired(id: number, reason: string): void { + const stored = this.get(id); + if (!stored) throw new Error(`MCP action ${id} is unknown.`); + stored.state = "failed"; + stored.claimedAt = undefined; + stored.retryable = false; + stored.dispatched = false; + stored.result = undefined; + stored.error = createActionRestageRequiredError(reason).message; + this.#save(stored); + this.#prune(); + } + async apply( id: number, call: (fn: (client: McpClient) => Promise) => Promise, @@ -137,6 +206,12 @@ export class ActionStore { if (stored.state === "applied") return; if (stored.state === "rejected") throw new Error(`MCP action ${id} was already rejected.`); if (stored.state === "failed" && stored.retryable === false) { + if (stored.dispatched === false) { + const restageReason = getActionRestageRequiredReason(new Error(stored.error ?? "")); + if (restageReason !== undefined) throw createActionRestageRequiredError(restageReason); + throw new ActionInvalidatedError( + stored.error ?? "This MCP action became invalid before dispatch."); + } throw new Error(stored.error ?? `MCP action ${id} cannot be retried.`); } if (stored.state === "applying") { @@ -145,15 +220,34 @@ export class ActionStore { stored.state = "applying"; stored.claimedAt = Date.now(); + stored.dispatched = false; stored.error = undefined; stored.result = undefined; this.#save(stored); let result: McpToolCallResult; + let dispatched = false; try { - result = await call(client => client.callTool(stored.toolName, stored.args)); + result = await call(client => { + dispatched = true; + stored.dispatched = true; + this.#save(stored); + return client.callTool(stored.toolName, stored.args); + }); } catch (err) { - const mayHaveLanded = callMayHaveTakenEffect(err); + if (err instanceof ActionInvalidatedError) { + stored.state = "failed"; + stored.retryable = false; + stored.dispatched = false; + stored.error = err.reason; + this.#save(stored); + this.#prune(); + log.warn("tool call invalidated before dispatch", { + event: "action.apply.invalidated", actionId: id, toolName: stored.toolName, error: err, + }); + throw err; + } + const mayHaveLanded = dispatched && callMayHaveTakenEffect(err); stored.state = "failed"; stored.retryable = !mayHaveLanded; stored.error = mayHaveLanded diff --git a/packages/mcp-shared/src/client.ts b/packages/mcp-shared/src/client.ts index 3d5dcfe79..a78928e91 100644 --- a/packages/mcp-shared/src/client.ts +++ b/packages/mcp-shared/src/client.ts @@ -416,7 +416,7 @@ export class McpClient { return headers; } - async #post(body: unknown): Promise { + async #post(body: unknown, deadline?: number): Promise { let headers: Headers; try { const method = typeof body === "object" && body !== null && "method" in body @@ -433,7 +433,7 @@ export class McpClient { method: "POST", headers, body: JSON.stringify(body), - }, this.#fetchOptions); + }, deadline === undefined ? this.#fetchOptions : { ...this.#fetchOptions, deadline }); } catch (err) { if (err instanceof FetchNotStartedError) { throw new McpCallNotDispatchedError(err.message, err); @@ -466,12 +466,13 @@ export class McpClient { async #callMeasured( method: string, params?: unknown, + deadline?: number, ): Promise<{ result: T; responseBytes: number }> { // A transport session is persisted on the account and can be used by several short-lived client // instances concurrently. Prefixing IDs per instance prevents two active requests from both // being JSON-RPC id 1 and confusing the server's SSE response routing. const id = `${this.#requestPrefix}:${++this.#requestId}`; - const response = await this.#post({ jsonrpc: "2.0", id, method, params }); + const response = await this.#post({ jsonrpc: "2.0", id, method, params }, deadline); if (!response.ok) { await response.body?.cancel().catch(() => undefined); @@ -565,9 +566,13 @@ export class McpClient { * * `include`, when present, is applied before count and byte budgets so an aggregator's unrelated * tools cannot crowd the requested server or exact grant names out of the bounded result. + * `deadline` bounds this listing only; later requests on the same client keep their original + * fetch options. */ - async listTools(maxTools: number, include?: McpToolFilter): Promise { - return this.#list(maxTools, include, clampToolDefinition); + async listTools( + maxTools: number, include?: McpToolFilter, deadline?: number, + ): Promise { + return this.#list(maxTools, include, clampToolDefinition, false, false, deadline); } /** @@ -583,11 +588,11 @@ export class McpClient { return this.#list(maxTools, include, indexTool, true); } - /** Finds one exact tool without reading pages after the match. */ - async findTool(name: string): Promise { + /** Finds one exact tool without reading pages after the match, optionally before one deadline. */ + async findTool(name: string, deadline?: number): Promise { if (!isValidToolName(name)) return undefined; return (await this.#list( - 1, tool => tool.name === name, clampToolDefinition, true, true)).tools[0]; + 1, tool => tool.name === name, clampToolDefinition, true, true, deadline)).tools[0]; } /** Collects at most `maxTools` bounded matching summaries without scanning later pages. */ @@ -603,6 +608,7 @@ export class McpClient { project: (tool: McpWireTool) => T, stopWhenFull = false, failOnScanLimit = false, + deadline?: number, ): Promise<{ tools: T[]; truncated: boolean }> { const tools: T[] = []; let budget = MAX_CATALOG_BYTES; @@ -619,7 +625,7 @@ export class McpClient { for (let page = 0; page < MAX_TOOL_PAGES; page++) { const measured = await this.#callMeasured<{ tools?: McpWireTool[]; nextCursor?: string }>( - "tools/list", cursor === undefined ? {} : { cursor }); + "tools/list", cursor === undefined ? {} : { cursor }, deadline); const body = measured.result; scannedBytes += measured.responseBytes; if (scannedBytes > MAX_SCANNED_TOOL_BYTES) return scanLimit(); diff --git a/packages/mcp-shared/src/connection.ts b/packages/mcp-shared/src/connection.ts index da3c59fd8..7f79a6991 100644 --- a/packages/mcp-shared/src/connection.ts +++ b/packages/mcp-shared/src/connection.ts @@ -83,12 +83,34 @@ function notDispatched(err: unknown): McpCallNotDispatchedError { ); } +const CONNECTION_CHANGED_ERROR_PREFIX = "MCP_CONNECTION_CHANGED: "; + +/** A connector-side connection change known to have preceded tool dispatch. */ +export class McpConnectionChangedError extends McpCallNotDispatchedError { + constructor(reason: string) { + super(reason); + this.name = "McpConnectionChangedError"; + } +} + +/** Creates a connection-change error whose discriminator survives account Durable Object RPC. */ +export function createMcpConnectionChangedRpcError(reason: string): Error { + return new Error(`${CONNECTION_CHANGED_ERROR_PREFIX}${reason}`); +} + +function connectionChangedReason(error: unknown): string | undefined { + const message = error instanceof Error ? error.message : undefined; + return message?.startsWith(CONNECTION_CHANGED_ERROR_PREFIX) + ? message.slice(CONNECTION_CHANGED_ERROR_PREFIX.length) + : undefined; +} + /** Runs `fn` against an initialized client for `endpoint`, using the account's credentials. */ export async function withClient( env: ConnectionEnv, account: ConnectionAccount, endpoint: string, - fn: (client: McpClient) => Promise, + fn: (client: McpClient, connectionGeneration: number) => Promise, options: WithClientOptions = {}, ): Promise { // Read once for the whole operation. The account refreshes a token a minute before expiry, so one @@ -97,13 +119,21 @@ export async function withClient( try { connection = await account.getConnection(endpoint); } catch (err) { + const reason = connectionChangedReason(err); + if (reason !== undefined) throw new McpConnectionChangedError(reason); throw notDispatched(err); } const { authorization, sessionId, generation } = connection; const client = new McpClient( endpoint, async method => { if (method === "tools/call") { - await account.assertConnectionCurrent(endpoint, generation); + try { + await account.assertConnectionCurrent(endpoint, generation); + } catch (err) { + const reason = connectionChangedReason(err); + if (reason !== undefined) throw new McpConnectionChangedError(reason); + throw err; + } } return authorization; }, sessionId, { @@ -157,8 +187,12 @@ export async function withClient( } } try { - return await fn(client); + return await fn(client, generation); } catch (err) { + if (err instanceof McpCallNotDispatchedError && + err.cause instanceof McpConnectionChangedError) { + throw err.cause; + } if (!(err instanceof McpSessionExpiredError)) throw err; if (options.retryOnExpiry !== false) { client.sessionId = null; @@ -168,7 +202,7 @@ export async function withClient( if (initializeError instanceof McpAuthRequiredError) throw initializeError; throw notDispatched(initializeError); } - return await fn(client); + return await fn(client, generation); } // This call is not retried, since it may already have taken effect. The session is gone all // the same, so the cached id is dead: left in place it would fail every later call for a diff --git a/packages/mcp-shared/src/facet.ts b/packages/mcp-shared/src/facet.ts index 009594dbe..762c39d96 100644 --- a/packages/mcp-shared/src/facet.ts +++ b/packages/mcp-shared/src/facet.ts @@ -2,23 +2,29 @@ // identity, props, labels, trust source, and account lookup. import { DurableObject, type RpcStub } from "cloudflare:workers"; -import type { - ActionKind, - ApprovalQueue, - Gatekeeper, - GatekeeperUserVerifier, - ResourceDescription, +import { + createActionRestageRequiredError, + type ActionKind, + type ApprovalQueue, + type Gatekeeper, + type GatekeeperUserVerifier, + type ResourceDescription, } from "@gadgets/workshop-shared/gatekeeper"; -import { ActionStore, REVERT_UNSUPPORTED_MESSAGE } from "./action-store.js"; +import { + ActionInvalidatedError, + ActionStore, + REVERT_UNSUPPORTED_MESSAGE, +} from "./action-store.js"; import { CATALOG_TTL_MS, HydratedTools, scopedCatalog, type ScopedCatalog, } from "./catalog.js"; -import type { McpClient } from "./client.js"; +import type { McpClient, McpTool } from "./client.js"; import { + McpConnectionChangedError, withClient, type ConnectionAccount, type ConnectionEnv, @@ -26,6 +32,11 @@ import { } from "./connection.js"; import type { McpLog } from "./log.js"; import { DEFAULT_REQUEST_TIMEOUT_MS } from "./fetch.js"; +import { + isPortalNativeTool, + looksLikePortal, + PORTAL_LIST_SERVERS_TOOL, +} from "./portal.js"; import { formatToolScope, scopeAllows, type ToolScope } from "./scope.js"; import { matchesToolQuery, toolQueryTerms, MAX_SEARCH_RESULTS } from "./tool-search.js"; import { McpSessionBase, type McpSessionHost, type StoredAction } from "./session.js"; @@ -34,6 +45,7 @@ import { observerRefusalMessage } from "./sharing-policy.js"; import { actionKindFor, classifyTool, + toolPolicyFingerprint, type ClassifiedTool, type ServerTrust, } from "./tools.js"; @@ -64,6 +76,11 @@ export abstract class McpFacetBase< #hydrated = new HydratedTools(); #activeDiscoveries = 0; #waitingDiscoveries: Array<() => void> = []; + #resolvingCalls = new Map>(); #actions(): ActionStore { return this.#actionStore ??= new ActionStore(this.ctx.storage.sql); @@ -215,6 +232,53 @@ export abstract class McpFacetBase< }); } + async #freshTool( + client: McpClient, + name: string, + catalog: ScopedCatalog, + deadline?: number, + ): Promise { + if (!catalog.isPortal && isPortalNativeTool(name)) { + const probe = await client.listTools( + 2, + tool => tool.name === name || tool.name === PORTAL_LIST_SERVERS_TOOL, + deadline, + ); + if (looksLikePortal(probe.tools, { truncated: probe.truncated, cap: 2 })) return undefined; + return probe.tools.find(tool => tool.name === name); + } + return client.findTool(name, deadline); + } + + /** Resolves current dispatch policy and the account generation it belongs to. */ + resolveToolForCall(name: string): Promise<{ + entry: ClassifiedTool; + policyFingerprint: string; + connectionGeneration: number; + } | undefined> { + const existing = this.#resolvingCalls.get(name); + if (existing) return existing; + const resolving = this.runDiscovery(async deadline => { + if (!scopeAllows(this.scope, name, this.scope.serverId !== undefined)) return undefined; + const catalog = await this.catalog(deadline); + if (!scopeAllows(this.scope, name, catalog.isPortal)) return undefined; + return this.call(async (client, connectionGeneration) => { + const tool = await this.#freshTool(client, name, catalog); + if (!tool) return undefined; + const entry = classifyTool(tool, this.trust); + return { + entry, + policyFingerprint: toolPolicyFingerprint(tool, this.trust), + connectionGeneration, + }; + }, { deadline }); + }).finally(() => { + if (this.#resolvingCalls.get(name) === resolving) this.#resolvingCalls.delete(name); + }); + this.#resolvingCalls.set(name, resolving); + return resolving; + } + /** Returns action kinds that this facet's current catalog permits auto-approving. */ async getAutoApprovableActions(): Promise { return (await this.tools()) @@ -244,8 +308,12 @@ export abstract class McpFacetBase< async removeObserver(_id: string): Promise {} /** Stages an MCP action for approval. */ - stageAction(toolName: string, args: Record): StoredAction { - return this.#actions().stage(toolName, args); + stageAction( + toolName: string, + args: Record, + snapshot: { policyFingerprint: string; connectionGeneration: number }, + ): StoredAction { + return this.#actions().stage(toolName, args, snapshot); } /** Discards an action whose approval submission failed. */ @@ -260,8 +328,49 @@ export abstract class McpFacetBase< /** Applies an approved action without retrying an outcome-unknown write. */ async applyAction(action: number): Promise { - await this.#actions().apply( - action, fn => this.call(fn, { retryOnExpiry: false }), this.log); + const stored = this.#actions().get(action); + if (!stored) throw new Error(`MCP action ${action} is unknown.`); + const snapshotMissing = stored.connectionGeneration === undefined || + stored.policyFingerprint === undefined; + const safeToRestage = stored.state === "pending" || + (stored.state === "failed" && stored.retryable !== false); + if (snapshotMissing && safeToRestage) { + const reason = + "This MCP action predates current approval policy checks. Stage the call again."; + this.#actions().markRestageRequired(action, reason); + throw createActionRestageRequiredError(reason); + } + return this.#actions().apply( + action, + fn => this.call(async (client, connectionGeneration) => { + const dispatch = await this.runDiscovery(async deadline => { + const catalog = await this.catalog(deadline); + if (stored.connectionGeneration === undefined || stored.policyFingerprint === undefined + || connectionGeneration !== stored.connectionGeneration) { + throw new ActionInvalidatedError( + "This MCP connection changed after approval was requested. Stage the call again."); + } + if (!scopeAllows(this.scope, stored.toolName, catalog.isPortal)) { + throw new ActionInvalidatedError( + "This MCP tool is no longer allowed by the binding. Stage the call again."); + } + const tool = await this.#freshTool(client, stored.toolName, catalog, deadline); + if (!tool || toolPolicyFingerprint(tool, this.trust) !== stored.policyFingerprint) { + throw new ActionInvalidatedError( + "This MCP tool's approval policy changed. Review and stage the call again."); + } + return () => fn(client); + }); + return dispatch(); + }, { retryOnExpiry: false }).catch(error => { + if (error instanceof McpConnectionChangedError) { + throw new ActionInvalidatedError( + "This MCP connection changed after approval was requested. Stage the call again."); + } + throw error; + }), + this.log, + ); } /** Rejects a pending action. */ @@ -276,7 +385,7 @@ export abstract class McpFacetBase< /** Runs a call against this facet's endpoint and account. */ call( - fn: (client: McpClient) => Promise, + fn: (client: McpClient, connectionGeneration: number) => Promise, options?: WithClientOptions, ): Promise { return withClient(this.env, this.account(), this.endpoint, fn, options); diff --git a/packages/mcp-shared/src/session.ts b/packages/mcp-shared/src/session.ts index d40fb99ab..3ffd49644 100644 --- a/packages/mcp-shared/src/session.ts +++ b/packages/mcp-shared/src/session.ts @@ -5,11 +5,16 @@ // supplies. The base never touches the Durable Object, the account, or the endpoint's credentials. import { RpcTarget, type RpcStub } from "cloudflare:workers"; -import type { ActionDescription, ActionKind, ApprovalQueue } - from "@gadgets/workshop-shared/gatekeeper"; +import { + getActionRestageRequiredReason, + type ActionDescription, + type ActionKind, + type ApprovalQueue, +} from "@gadgets/workshop-shared/gatekeeper"; import { MAX_TOOL_NAME_CHARS, + McpCallNotDispatchedError, type McpClient, } from "./client.js"; import type { WithClientOptions } from "./connection.js"; @@ -46,6 +51,12 @@ export type StoredAction = { */ state: "pending" | "applying" | "applied" | "rejected" | "failed"; submittedAt: number; + /** Policy fingerprint the user approved, absent on actions staged by older code. */ + policyFingerprint?: string; + /** Account connection generation the user approved, absent on actions staged by older code. */ + connectionGeneration?: number; + /** Set immediately before `tools/call`; false proves a failure preceded dispatch. */ + dispatched?: boolean; /** When the in-flight apply was claimed, for recovering a claim whose Durable Object died mid-call. */ claimedAt?: number; /** @@ -57,7 +68,7 @@ export type StoredAction = { retryable?: boolean; /** Populated once applied; delivered to the Gadget as an observation. */ result?: Extract; - /** Terminal failure reason retained for later collection. */ + /** Terminal failure or invalidation reason retained for later collection. */ error?: string; }; @@ -77,16 +88,27 @@ export interface McpSessionHost { searchTools(query: string): Promise; /** Finds one granted tool definition by exact wire name. */ findTool(name: string): Promise; - /** Runs `fn` against an initialized client for this binding's endpoint. */ + /** Resolves current dispatch policy and the account generation it belongs to. */ + resolveToolForCall(name: string): Promise<{ + entry: ClassifiedTool; + policyFingerprint: string; + connectionGeneration: number; + } | undefined>; + + /** Runs `fn` against an initialized client and captured generation for this binding's endpoint. */ call( - fn: (client: McpClient) => Promise, + fn: (client: McpClient, connectionGeneration: number) => Promise, options?: WithClientOptions, ): Promise; /** The approval-kind tag for one tool, namespaced so pre-approvals cannot cross servers. */ actionKindFor(toolName: string): ActionKind; - stageAction(toolName: string, args: Record): StoredAction; + stageAction( + toolName: string, + args: Record, + snapshot: { policyFingerprint: string; connectionGeneration: number }, + ): StoredAction; discardStagedAction(id: number): void; lookupAction(id: number): StoredAction | undefined; } @@ -190,8 +212,9 @@ export class McpSessionBase extends RpcTarget { } const host = this.#host; - const entry = await host.findTool(name); - if (!entry) throw new Error(this.#noSuchToolMessage(name)); + const resolved = await host.resolveToolForCall(name); + if (!resolved) throw new Error(this.#noSuchToolMessage(name)); + const { entry } = resolved; const described = describeCall({ serverName: host.serverName, @@ -203,13 +226,19 @@ export class McpSessionBase extends RpcTarget { }); if (entry.mode === "read") { - const result = await host.call(client => client.callTool(name, toolArgs)); + const result = await host.call((client, connectionGeneration) => { + if (connectionGeneration !== resolved.connectionGeneration) { + throw new McpCallNotDispatchedError( + "This MCP connection changed while the tool was being resolved. Try again."); + } + return client.callTool(name, toolArgs); + }); // Authorize before the data is handed back, per the gatekeeper contract. await this.#queue.authorizeObservation(described); return toCallResult(result); } - const staged = host.stageAction(name, toolArgs); + const staged = host.stageAction(name, toolArgs, resolved); const description: ActionDescription = { ...described, // MCP describes no inverse operation for a tool call. @@ -262,7 +291,8 @@ export class McpSessionBase extends RpcTarget { case "failed": return { status: "failed", - message: stored.error + message: (stored.error && getActionRestageRequiredReason(new Error(stored.error))) + ?? stored.error ?? `Calling "${stored.toolName}" on ${host.serverName} failed.`, }; case "applied": { diff --git a/packages/mcp-shared/src/tools.ts b/packages/mcp-shared/src/tools.ts index b89dfbc27..7058d6239 100644 --- a/packages/mcp-shared/src/tools.ts +++ b/packages/mcp-shared/src/tools.ts @@ -134,6 +134,12 @@ function policyClaims(tool: McpTool): string { ].join(""); } +/** Stable snapshot of one tool's effective read/action and auto-approval policy. */ +export function toolPolicyFingerprint(tool: McpTool, trust: ServerTrust): string { + const policy = classifyTool(tool, trust); + return `${policy.mode}:${policy.autoApprovable ? "auto" : "manual"}`; +} + /** * Stable fingerprint of a tool catalog, for detecting that an endpoint changed under us. * diff --git a/packages/workshop-backend/__tests__/auto-approval.test.ts b/packages/workshop-backend/__tests__/auto-approval.test.ts index 0b4ce7389..019e569d6 100644 --- a/packages/workshop-backend/__tests__/auto-approval.test.ts +++ b/packages/workshop-backend/__tests__/auto-approval.test.ts @@ -1,8 +1,18 @@ import { describe, it, expect } from "vitest"; import { createTypedStorage, collection } from "@gadgets/typed-storage"; -import { AutoApprovalDrainer, AutoApprovalStorage, ApplyPendingActionFn } from "../src/auto-approval.js"; +import { + AutoApprovalDrainer, + AutoApprovalStorage, + ApplyPendingActionFn, + clearAutoApprovalRules, + handleActionApplyFailure, +} from "../src/auto-approval.js"; import type { ActionRecord, AutoApproveTagRecord } from "../src/overseer.js"; import type { AiChatAuthorInfo } from "@gadgets/workshop-shared/api"; +import { + createActionInvalidatedError, + createActionRestageRequiredError, +} from "@gadgets/workshop-shared/gatekeeper"; import { makeMockStorage } from "./mock-storage.js"; function makeStorage(): AutoApprovalStorage { @@ -66,6 +76,7 @@ function makeImmediateApply(storage: AutoApprovalStorage) { fresh.autoApproved = autoApproved; storage.actions.put(fresh); } + return "approved"; }; return { applyFn, calls }; } @@ -78,7 +89,7 @@ function makeControlledApply(storage: AutoApprovalStorage) { let gates: Array<() => void> = []; let applyFn: ApplyPendingActionFn = (record, resolvedBy, autoApproved) => { calls.push(record.id); - return new Promise((resolve) => { + return new Promise<"approved">((resolve) => { gates.push(() => { let fresh = storage.actions.get(record.id); if (fresh && fresh.type === "action") { @@ -88,7 +99,7 @@ function makeControlledApply(storage: AutoApprovalStorage) { fresh.autoApproved = autoApproved; storage.actions.put(fresh); } - resolve(); + resolve("approved"); }); }); }; @@ -156,6 +167,93 @@ describe("AutoApprovalDrainer.drain", () => { expect(getAction(storage, 3).state).toBe("approved"); }); + it("allows auto-approval again after the user explicitly re-enables a cleared rule", async () => { + let storage = makeStorage(); + enableRule(storage); + putAction(storage, 1); + putAction(storage, 2); + const calls: number[] = []; + const apply: ApplyPendingActionFn = async record => { + calls.push(record.id); + if (record.id === 1) { + record.state = "rejected"; + record.invalidationReason = "Policy changed."; + clearAutoApprovalRules(storage, GK); + storage.actions.put(record); + return "stopped"; + } + record.state = "approved"; + storage.actions.put(record); + return "approved"; + }; + + await new AutoApprovalDrainer(storage, apply).drain(GK); + + expect(calls).toEqual([1]); + expect(getAction(storage, 2).state).toBe("pending"); + + expect(storage.autoApproveTags.get(`${GK}:edit`)).toBeUndefined(); + enableRule(storage); + await new AutoApprovalDrainer(storage, apply).drain(GK); + + expect(calls).toEqual([1, 2]); + expect(getAction(storage, 2).state).toBe("approved"); + }); + + it("does not restart after invalidation when a concurrent drain requested a rerun", async () => { + let storage = makeStorage(); + enableRule(storage); + putAction(storage, 1); + putAction(storage, 2); + const calls: number[] = []; + let release!: () => void; + const apply: ApplyPendingActionFn = record => { + calls.push(record.id); + return new Promise(resolve => { + release = () => { + record.state = "rejected"; + record.invalidationReason = "Policy changed."; + storage.actions.put(record); + resolve("stopped"); + }; + }); + }; + const drainer = new AutoApprovalDrainer(storage, apply); + + const first = drainer.drain(GK); + await flush(); + await drainer.drain(GK); + release(); + await first; + + expect(calls).toEqual([1]); + expect(getAction(storage, 2).state).toBe("pending"); + }); + + it("stops when a later candidate is invalidated after the drain snapshot", async () => { + let storage = makeStorage(); + enableRule(storage); + putAction(storage, 1); + putAction(storage, 2); + putAction(storage, 3); + let apply = makeControlledApply(storage); + let draining = new AutoApprovalDrainer(storage, apply.applyFn).drain(GK); + await flush(); + + let invalidated = getAction(storage, 2); + invalidated.state = "rejected"; + invalidated.invalidationReason = "Policy changed."; + storage.actions.put(invalidated); + apply.releaseNext(); + await flush(); + const callsAfterInvalidation = [...apply.calls]; + if (apply.inFlight() > 0) apply.releaseNext(); + await draining; + + expect(callsAfterInvalidation).toEqual([1]); + expect(getAction(storage, 3).state).toBe("pending"); + }); + // 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. @@ -212,3 +310,73 @@ describe("AutoApprovalDrainer.drain", () => { expect(getAction(storage, 2).state).toBe("approved"); }); }); + +describe("clearAutoApprovalRules", () => { + it("removes every rule for the invalidated gatekeeper and leaves other connections alone", () => { + const storage = makeStorage(); + enableRule(storage, "edit", GK); + enableRule(storage, "delete", GK); + enableRule(storage, "edit", 2); + + clearAutoApprovalRules(storage, GK); + + expect(storage.autoApproveTags.get(`${GK}:edit`)).toBeUndefined(); + expect(storage.autoApproveTags.get(`${GK}:delete`)).toBeUndefined(); + expect(storage.autoApproveTags.get("2:edit")).toBeDefined(); + }); + + it("materializes the lazy rule list before deleting from its collection", () => { + let iterating = false; + const deleted: string[] = []; + const rules: AutoApproveTagRecord[] = [ + { gatekeeperId: GK, actionKind: { tag: "edit", label: "Edits" }, enabledBy: ENABLER }, + { gatekeeperId: GK, actionKind: { tag: "delete", label: "Deletes" }, enabledBy: ENABLER }, + ]; + const autoApproveTags = { + *list() { + iterating = true; + try { + yield* rules; + } finally { + iterating = false; + } + }, + delete(key: string) { + if (iterating) throw new Error("collection iterator invalidated"); + deleted.push(key); + }, + } as unknown as AutoApprovalStorage["autoApproveTags"]; + + expect(() => clearAutoApprovalRules({ autoApproveTags }, GK)).not.toThrow(); + expect(deleted).toEqual([`${GK}:edit`, `${GK}:delete`]); + }); +}); + +describe("handleActionApplyFailure", () => { + it("preserves rules when an action predates approval snapshots", () => { + const storage = makeStorage(); + enableRule(storage, "edit"); + enableRule(storage, "delete"); + + expect(handleActionApplyFailure( + storage, + GK, + createActionRestageRequiredError("Stage the call again."), + )).toBe("Stage the call again."); + expect([...storage.autoApproveTags.list()].map(rule => rule.actionKind.tag).toSorted()) + .toEqual(["delete", "edit"]); + }); + + it("clears rules when the approval context changed", () => { + const storage = makeStorage(); + enableRule(storage, "edit"); + enableRule(storage, "delete"); + + expect(handleActionApplyFailure( + storage, + GK, + createActionInvalidatedError("The connection changed."), + )).toBe("The connection changed."); + expect([...storage.autoApproveTags.list()]).toEqual([]); + }); +}); diff --git a/packages/workshop-backend/src/auto-approval.ts b/packages/workshop-backend/src/auto-approval.ts index 0c1fa032f..215a47820 100644 --- a/packages/workshop-backend/src/auto-approval.ts +++ b/packages/workshop-backend/src/auto-approval.ts @@ -5,6 +5,10 @@ import type { Collection } from "@gadgets/typed-storage"; import type { AiChatAuthorInfo } from "@gadgets/workshop-shared/api"; +import { + getActionInvalidationReason, + getActionRestageRequiredReason, +} from "@gadgets/workshop-shared/gatekeeper"; import { createWorkshopLogger } from "./observability"; import type { ActionRecord, AutoApproveTagRecord } from "./overseer.js"; @@ -22,7 +26,34 @@ export interface AutoApprovalStorage { export type ApplyPendingActionFn = ( record: ActionRecord & {type: "action"}, resolvedBy: AiChatAuthorInfo, - autoApproved: boolean) => Promise; + autoApproved: boolean) => Promise<"approved" | "stopped">; + +/** Removes every auto-approval rule for a gatekeeper after its approval context becomes invalid. */ +export function clearAutoApprovalRules( + storage: Pick, gatekeeperId: number): void { + const keys = [...storage.autoApproveTags.list()] + .filter(rule => rule.gatekeeperId === gatekeeperId) + .map(rule => `${gatekeeperId}:${rule.actionKind.tag}`); + for (const key of keys) { + storage.autoApproveTags.delete(key); + } +} + +/** + * Recognizes a pre-dispatch action failure and returns its user-facing reason. A genuine context + * invalidation clears every rule for the connection; a deploy-migration restage preserves them. + */ +export function handleActionApplyFailure( + storage: Pick, + gatekeeperId: number, + error: unknown): string | undefined { + const invalidationReason = getActionInvalidationReason(error); + if (invalidationReason !== undefined) { + clearAutoApprovalRules(storage, gatekeeperId); + return invalidationReason; + } + return getActionRestageRequiredReason(error); +} export class AutoApprovalDrainer { // Per-gatekeeper single-flight state. Key present => a drain is running for that gatekeeper; the @@ -43,7 +74,7 @@ export class AutoApprovalDrainer { try { do { this.#draining.set(gatekeeperId, false); - await this.#drainOnce(gatekeeperId); + if (await this.#drainOnce(gatekeeperId) === "stopped") return; } while (this.#draining.get(gatekeeperId)); } finally { this.#draining.delete(gatekeeperId); @@ -51,13 +82,12 @@ export class AutoApprovalDrainer { } // 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. + // the first pending action that is NOT auto-eligible (a manual gate), cannot be dispatched, or + // throws while applying -- none is skipped ahead of. // // 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 { + async #drainOnce(gatekeeperId: number): Promise<"complete" | "stopped"> { // 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( @@ -77,6 +107,8 @@ export class AutoApprovalDrainer { // 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?.type === "action" && fresh.state === "rejected" && + fresh.invalidationReason !== undefined) return "stopped"; if (!fresh || fresh.type !== "action" || fresh.state !== "pending") { continue; } @@ -84,7 +116,8 @@ export class AutoApprovalDrainer { try { // Attribute the auto-approval to the user who enabled the rule -- it runs under their // authority. - await this.applyPendingAction(fresh, rule.enabledBy, true); + const outcome = await this.applyPendingAction(fresh, rule.enabledBy, true); + if (outcome === "stopped") return "stopped"; } catch (err) { // Leave the action pending for manual handling and stop the drain (never skip ahead). logger.error("auto-approval failed", { @@ -93,5 +126,6 @@ export class AutoApprovalDrainer { break; } } + return "complete"; } } diff --git a/packages/workshop-backend/src/overseer.ts b/packages/workshop-backend/src/overseer.ts index fa05cf4a6..0e468b9c8 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 { AutoApprovalDrainer, handleActionApplyFailure } from "./auto-approval"; import { collectSlashCommands, invokeSlashCommand } from "./slash-commands"; import { createWorkshopLogger, obsContext, traced } from "./observability"; import { wrapDoStubForTelemetry } from "./do-telemetry"; @@ -528,8 +528,12 @@ export type ActionRecord = { appliedAt?: Date; action: number; // action key assigned by the gatekeeper, passed back on apply/reject/revert 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 + /** Set when resolved (approved/rejected); absent while pending or on legacy records. */ + resolvedBy?: AiChatAuthorInfo; + /** True when applied by an auto-approval rule rather than a human. */ + autoApproved?: boolean; + /** Why the action became invalid before dispatch. */ + invalidationReason?: string; } | { type: "observation"; description: ObservationDescription; @@ -738,6 +742,7 @@ function actionRecordToLog(record: ActionRecord): ActionLogEntry { description: record.description, resolvedBy: record.resolvedBy, autoApproved: record.autoApproved, + invalidationReason: record.invalidationReason, }; case "bindHook": return { @@ -2657,14 +2662,32 @@ class OverseerImpl implements AgentHooks { // 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 { + resolvedBy: AiChatAuthorInfo, autoApproved: boolean) + : Promise<"approved" | "stopped"> { let gatekeeper = this.getGatekeeperFacet(record.gatekeeperId); - await gatekeeper.applyAction(record.action); + let failureReason: string | undefined; + try { + await gatekeeper.applyAction(record.action); + } catch (error) { + failureReason = handleActionApplyFailure(this.storage, record.gatekeeperId, error); + if (failureReason === undefined) throw error; + } + if (failureReason !== undefined) { + // Keep the established wire state so an older browser renders this as denied rather than + // claiming the undispatched action was approved. The reason distinguishes invalidation. + record.state = "rejected"; + record.appliedAt = new Date(); + record.resolvedBy = resolvedBy; + record.invalidationReason = failureReason; + this.storage.actions.put(record); + return "stopped"; + } record.state = "approved"; record.appliedAt = new Date(); record.resolvedBy = resolvedBy; record.autoApproved = autoApproved; this.storage.actions.put(record); + return "approved"; } // Apply all currently-eligible pending actions of the given gatekeeper, in ascending id order. @@ -7828,7 +7851,8 @@ 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); + const outcome = await this.impl.applyPendingAction(action, profile, false); + if (outcome === "stopped") return; // 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. @@ -7944,7 +7968,7 @@ class OverseerClientInterface extends RpcTarget implements Overseer { // Only resume when every awaited action in the turn has been decided and all were approved. if (awaited.length === 0) return; // No awaited action in current turn. if (awaited.some(r => r.state === "pending")) return; // Still waiting on a decision. - if (awaited.some(r => r.state === "rejected")) return; // Denial leaves the turn ended. + if (awaited.some(r => r.state === "rejected")) return; // Persist one note for replay; raw action cards are not surfaced to the LLM. Concurrent // approvals could both pass the gate above and append duplicate notes (the DO input gate is diff --git a/packages/workshop-shared/src/api.ts b/packages/workshop-shared/src/api.ts index d858dd2d5..01d3a6bbe 100644 --- a/packages/workshop-shared/src/api.ts +++ b/packages/workshop-shared/src/api.ts @@ -1470,7 +1470,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 became invalid before dispatch. */ export type ActionState = "pending" | "approved" | "rejected"; @@ -1507,6 +1507,9 @@ export type ActionLogEntry = { * clicking Approve. Only ever set alongside state "approved" (there is no automatic rejection). */ autoApproved?: boolean; + + /** Why an approved action became invalid before dispatch. Present only when state is rejected. */ + invalidationReason?: string; } | { type: "observation"; description: ObservationDescription; diff --git a/packages/workshop-shared/src/gatekeeper.ts b/packages/workshop-shared/src/gatekeeper.ts index 48bcdd729..84a480c8d 100644 --- a/packages/workshop-shared/src/gatekeeper.ts +++ b/packages/workshop-shared/src/gatekeeper.ts @@ -18,6 +18,59 @@ import type { WorkerEntrypoint, DurableObject, RpcTarget, RpcStub } from "cloudflare:workers"; +/** Stable RPC error code thrown when an approved action became invalid before dispatch. */ +export const ACTION_INVALIDATED_ERROR_CODE = "GATEKEEPER_ACTION_INVALIDATED"; + +const ACTION_INVALIDATED_ERROR_PREFIX = `${ACTION_INVALIDATED_ERROR_CODE}: `; + +function codedActionErrorReason( + error: unknown, code: string, prefix: string): string | undefined { + const candidate = typeof error === "object" && error !== null && "errorCode" in error + ? error.errorCode + : undefined; + const message = typeof error === "object" && error !== null && "message" in error + && typeof error.message === "string" + ? error.message + : undefined; + return message?.startsWith(prefix) ? message.slice(prefix.length) + : candidate === code ? message : undefined; +} + +/** Creates an invalidation error whose discriminator survives Workers RPC error serialization. */ +export function createActionInvalidatedError(reason: string): Error & { + errorCode: typeof ACTION_INVALIDATED_ERROR_CODE; +} { + return Object.assign(new Error(`${ACTION_INVALIDATED_ERROR_PREFIX}${reason}`), { + errorCode: ACTION_INVALIDATED_ERROR_CODE, + } as const); +} + +/** Reads an invalidation reason from a local coded error or its message-only RPC representation. */ +export function getActionInvalidationReason(error: unknown): string | undefined { + return codedActionErrorReason( + error, ACTION_INVALIDATED_ERROR_CODE, ACTION_INVALIDATED_ERROR_PREFIX); +} + +/** Stable RPC error code thrown when an old queued action lacks an approval-policy snapshot. */ +export const ACTION_RESTAGE_REQUIRED_ERROR_CODE = "GATEKEEPER_ACTION_RESTAGE_REQUIRED"; + +const ACTION_RESTAGE_REQUIRED_ERROR_PREFIX = `${ACTION_RESTAGE_REQUIRED_ERROR_CODE}: `; + +/** Creates a restage-required error whose discriminator survives Workers RPC serialization. */ +export function createActionRestageRequiredError(reason: string): Error & { + errorCode: typeof ACTION_RESTAGE_REQUIRED_ERROR_CODE; +} { + return Object.assign(new Error(`${ACTION_RESTAGE_REQUIRED_ERROR_PREFIX}${reason}`), { + errorCode: ACTION_RESTAGE_REQUIRED_ERROR_CODE, + } as const); +} + +/** Reads a restage-required reason from a local coded error or its message-only RPC form. */ +export function getActionRestageRequiredReason(error: unknown): string | undefined { + return codedActionErrorReason( + error, ACTION_RESTAGE_REQUIRED_ERROR_CODE, ACTION_RESTAGE_REQUIRED_ERROR_PREFIX); +} + /** * A pagination cursor. * @@ -811,6 +864,11 @@ export interface Gatekeeper extends DurableObject { * If this throws an exception, the user will be informed that the action failed and given the * opportunity to retry or discard. * + * If policy or authority changed after approval but before dispatch, throw an error created by + * `createActionInvalidatedError()`. New Workshop versions record that distinct terminal outcome; + * old versions leave the action pending rather than recording a write that was never dispatched + * as approved. + * * 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. From 644dc5b4d55a910fce4797e5762ec1035197e8ee Mon Sep 17 00:00:00 2001 From: Dan Carter Date: Wed, 19 Aug 2026 09:09:08 -0400 Subject: [PATCH 2/6] Render invalidated MCP actions --- .../workshop-frontend/src/Activity.test.tsx | 134 ++++++++++++++++++ packages/workshop-frontend/src/Activity.tsx | 20 ++- .../workshop-frontend/src/ChatInterface.tsx | 42 ++---- .../workshop-frontend/src/useResolveAction.ts | 8 +- 4 files changed, 163 insertions(+), 41 deletions(-) create mode 100644 packages/workshop-frontend/src/Activity.test.tsx diff --git a/packages/workshop-frontend/src/Activity.test.tsx b/packages/workshop-frontend/src/Activity.test.tsx new file mode 100644 index 000000000..d4ed03088 --- /dev/null +++ b/packages/workshop-frontend/src/Activity.test.tsx @@ -0,0 +1,134 @@ +// @vitest-environment jsdom +/* eslint-disable react/react-in-jsx-scope */ + +import { act } from 'react' +import { createRoot, type Root } from 'react-dom/client' +import { afterEach, describe, expect, it, vi } from 'vitest' +import type { ActionLogEntry, Overseer } from '@gadgets/workshop-shared/api' +import type { RpcStub } from 'capnweb' + +import Activity from './Activity' + +(globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true + +const testState = vi.hoisted(() => ({ + actionsById: new Map(), + refresh: vi.fn<() => Promise>(), +})) + +vi.mock('@cloudflare/kumo', () => ({ + Switch: () => null, + useKumoToastManager: () => ({ add: vi.fn<(toast: unknown) => void>() }), +})) +vi.mock('./useActions', () => ({ + useActions: () => ({ actionsById: testState.actionsById, isReady: true }), +})) +vi.mock('./useResolveAction', () => ({ + useResolveAction: () => vi.fn<() => Promise>(), +})) +vi.mock('./useAlwaysApproveTag', () => ({ + useAlwaysApproveTag: () => ({ + alwaysApproveTag: vi.fn<() => Promise>(), + isTagAutoApproved: () => false, + }), +})) +vi.mock('./useAutoApproval', () => ({ + autoApprovalKey: () => '', + useAutoApproval: () => ({ + entries: [], isLoading: false, pending: new Set(), + refresh: testState.refresh, + setEnabled: vi.fn<() => Promise>(), + }), +})) +vi.mock('./AuthContext', () => ({ useAuthenticatedApi: () => ({ authenticatedApi: {} }) })) +vi.mock('./useAvatar', () => ({ useAvatar: () => undefined })) +vi.mock('./useVendorBranding', () => ({ useVendorBranding: () => new Map() })) +vi.mock('./components/GatekeeperIcon', () => ({ GatekeeperIcon: () => null })) +vi.mock('./components/HookToggle', () => ({ HookToggle: () => null })) +vi.mock('./components/ResolveButton', () => ({ + AlwaysApproveButton: () => null, + ResolveButton: () => null, +})) +vi.mock('./components/WorkshopControls', () => ({ WorkshopButton: () => null })) +vi.mock('./components/AutoApproveConfirmDialog', () => ({ default: () => null })) + +describe('Activity history', () => { + let root: Root | undefined + let container: HTMLDivElement | undefined + + afterEach(() => { + act(() => root?.unmount()) + container?.remove() + testState.actionsById = new Map() + testState.refresh.mockClear() + }) + + it('shows the reason an action was invalidated', async () => { + testState.actionsById = new Map([[1, { + id: 1, + gatekeeperId: 1, + resourceTitle: 'Issue tracker', + createdAt: new Date(), + appliedAt: new Date(), + state: 'rejected', + type: 'action', + description: { + title: 'Create issue', + description: 'Creates an issue.', + implementsRevert: false, + }, + invalidationReason: 'The connection changed. Stage the call again.', + }]]) + container = document.createElement('div') + document.body.append(container) + root = createRoot(container) + await act(async () => root!.render( + } + view="history" + onViewChange={() => {}} + />, + )) + + const row = [...container.querySelectorAll('button')] + .find(button => button.textContent?.includes('Create issue')) + await act(async () => row!.click()) + + expect(container.textContent).toContain('The connection changed. Stage the call again.') + }) + + it('refreshes auto-approval rules for out-of-order invalidations', async () => { + container = document.createElement('div') + document.body.append(container) + root = createRoot(container) + const render = () => root!.render( + } view="auto" onViewChange={() => {}} />, + ) + await act(async () => render()) + testState.refresh.mockClear() + + const firstInvalidation: ActionLogEntry = { + id: 10, + gatekeeperId: 1, + resourceTitle: 'Issue tracker', + createdAt: new Date(), + state: 'rejected', + type: 'action', + description: { title: 'Create issue', description: '', implementsRevert: false }, + invalidationReason: 'Policy changed.', + } + testState.actionsById = new Map([[firstInvalidation.id, firstInvalidation]]) + await act(async () => render()) + + expect(testState.refresh).toHaveBeenCalledOnce() + + const secondInvalidation = { ...firstInvalidation, id: 5 } + testState.actionsById = new Map([ + [firstInvalidation.id, firstInvalidation], + [secondInvalidation.id, secondInvalidation], + ]) + await act(async () => render()) + + expect(testState.refresh).toHaveBeenCalledTimes(2) + }) +}) diff --git a/packages/workshop-frontend/src/Activity.tsx b/packages/workshop-frontend/src/Activity.tsx index f4ab21e53..5614780cd 100644 --- a/packages/workshop-frontend/src/Activity.tsx +++ b/packages/workshop-frontend/src/Activity.tsx @@ -96,6 +96,9 @@ function activityStatus( if (record.state === 'pending') { return { label: 'Waiting', dotClass: 'bg-kumo-brand', textClass: 'text-kumo-strong' } } + if (record.type === 'action' && record.invalidationReason) { + return { label: 'Invalidated', dotClass: 'bg-kumo-danger', textClass: 'text-kumo-danger' } + } if (record.state === 'rejected') { return { label: 'Denied', dotClass: 'bg-kumo-danger', textClass: 'text-kumo-danger' } } @@ -130,7 +133,7 @@ export default function Activity({ } | null>(null) const toasts = useKumoToastManager() - const { pendingActions, historyGroups, historyTotal, historyShown } = useMemo(() => { + const { pendingActions, historyGroups, historyTotal, historyShown, invalidationCount } = useMemo(() => { const records = [...actionsById.values()] const pending = records .filter(record => record.state === 'pending') @@ -152,6 +155,9 @@ export default function Activity({ historyGroups: groups, historyTotal: resolved.length, historyShown: filtered.length, + invalidationCount: records.filter( + record => record.type === 'action' && record.invalidationReason, + ).length, } }, [actionsById, historyFilter]) @@ -324,7 +330,10 @@ export default function Activity({ )} ) : ( - + )} {confirmAutoApprove && ( @@ -351,7 +360,7 @@ function AutoApprovalPanel({ reloadTrigger, }: { overseer: RpcStub - reloadTrigger?: number + reloadTrigger?: string | number }) { const { entries, isLoading, loadError, pending, refresh, setEnabled } = useAutoApproval(overseer) const { authenticatedApi } = useAuthenticatedApi() @@ -612,6 +621,11 @@ function HistoryRow({ {record.description.description}

)} + {record.type === 'action' && record.invalidationReason && ( +

+ {record.invalidationReason} +

+ )}
{formatFullDate(at)} {record.resourceTitle} diff --git a/packages/workshop-frontend/src/ChatInterface.tsx b/packages/workshop-frontend/src/ChatInterface.tsx index 23988f3e2..7c07c0304 100644 --- a/packages/workshop-frontend/src/ChatInterface.tsx +++ b/packages/workshop-frontend/src/ChatInterface.tsx @@ -5830,32 +5830,6 @@ function ChatInterface({ return changed; }; - const applyOptimisticActionState = (actionId: number, state: "approved" | "rejected"): boolean => { - let changed = false; - const locations = cacheRef.current.actionMessages.get(actionId); - if (!locations) return false; - - for (const [key, location] of locations) { - const messages = cacheRef.current.messages.get(location.chatId); - const msg = messages?.[location.sequence]; - if (msg?.type !== "action" || msg.actionId !== actionId || !msg.actionLog) { - locations.delete(key); - continue; - } - - const nextMessages = [...messages!]; - nextMessages[location.sequence] = { - ...msg, - actionLog: { ...msg.actionLog, state, appliedAt: new Date() }, - }; - cacheRef.current.messages.set(location.chatId, nextMessages); - changed = true; - } - - if (locations.size === 0) cacheRef.current.actionMessages.delete(actionId); - return changed; - }; - const applyOptimisticHookEnabled = (actionId: number, enabled: boolean): boolean => { let changed = false; const locations = cacheRef.current.actionMessages.get(actionId); @@ -5907,9 +5881,7 @@ function ChatInterface({ const { alwaysApproveTag, isTagAutoApproved } = useAlwaysApproveTag(overseer, setProcessingActions, onAutoApproveChange); - const resolveAction = useResolveAction(overseer, setProcessingActions, (actionId, state) => { - if (applyOptimisticActionState(actionId, state)) forceUpdate(); - }); + const resolveAction = useResolveAction(overseer, setProcessingActions); // Handle enabling/disabling a bound hook from the chat thread. const handleToggleHook = async (actionId: number, hookId: number, enabled: boolean) => { @@ -6462,7 +6434,8 @@ function ChatInterface({ const isPending = state === "pending"; const isApproved = state === "approved"; - const isRejected = state === "rejected"; + const isInvalidated = log.invalidationReason !== undefined; + const isRejected = state === "rejected" && !isInvalidated; // A blocking (awaitDecision) pending action suspends the agent turn and blocks the composer, so // present it as a prominent callout with its details expanded by default. const isBlocking = isPending && log.description.awaitDecision === true; @@ -6475,8 +6448,10 @@ function ChatInterface({ ? "Approved" : isRejected ? "Denied" - : null; - const stateLabelCls = isRejected + : isInvalidated + ? "Invalidated" + : null; + const stateLabelCls = isRejected || isInvalidated ? "text-kumo-danger" : "text-kumo-inactive"; // Auto-approval target: offer "Always approve this type" only when enabling a rule would @@ -6627,6 +6602,9 @@ function ChatInterface({
+ {log.invalidationReason && ( +

{log.invalidationReason}

+ )} {resourceMeta}
)} diff --git a/packages/workshop-frontend/src/useResolveAction.ts b/packages/workshop-frontend/src/useResolveAction.ts index 11607851b..22b1d8031 100644 --- a/packages/workshop-frontend/src/useResolveAction.ts +++ b/packages/workshop-frontend/src/useResolveAction.ts @@ -1,25 +1,21 @@ -import { useCallback, useRef, type Dispatch, type SetStateAction } from 'react' +import { useCallback, type Dispatch, type SetStateAction } from 'react' import { useKumoToastManager } from '@cloudflare/kumo' import type { RpcStub } from 'capnweb' -import type { ActionState, Overseer } from '@gadgets/workshop-shared/api' +import type { Overseer } from '@gadgets/workshop-shared/api' type ActionDecision = 'approve' | 'deny' export function useResolveAction( overseer: RpcStub, setProcessing: Dispatch>>, - onResolved?: (actionId: number, state: Extract) => void, ) { const toasts = useKumoToastManager() - const onResolvedRef = useRef(onResolved) - onResolvedRef.current = onResolved return useCallback(async (actionId: number, decision: ActionDecision) => { setProcessing(previous => new Set(previous).add(actionId)) try { if (decision === 'approve') await overseer.approveAction(actionId) else await overseer.rejectAction(actionId) - onResolvedRef.current?.(actionId, decision === 'approve' ? 'approved' : 'rejected') } catch (error) { console.error(`Failed to ${decision} action:`, error) toasts.add({ title: `Failed to ${decision} action`, variant: 'error' }) From 82c3d9845d0f42b4dea9bba927a04f8234c9da7d Mon Sep 17 00:00:00 2001 From: Dan Carter Date: Wed, 19 Aug 2026 10:19:01 -0400 Subject: [PATCH 3/6] Refresh auto-approval affordances after invalidation --- packages/workshop-frontend/src/Activity.tsx | 26 ++++++- .../workshop-frontend/src/ChatInterface.tsx | 22 +++++- .../src/useAlwaysApproveTag.test.tsx | 67 +++++++++++++++++++ .../src/useAlwaysApproveTag.ts | 25 ++++++- 4 files changed, 135 insertions(+), 5 deletions(-) create mode 100644 packages/workshop-frontend/src/useAlwaysApproveTag.test.tsx diff --git a/packages/workshop-frontend/src/Activity.tsx b/packages/workshop-frontend/src/Activity.tsx index 5614780cd..80ab12d87 100644 --- a/packages/workshop-frontend/src/Activity.tsx +++ b/packages/workshop-frontend/src/Activity.tsx @@ -133,8 +133,24 @@ export default function Activity({ } | null>(null) const toasts = useKumoToastManager() - const { pendingActions, historyGroups, historyTotal, historyShown, invalidationCount } = useMemo(() => { + const { + pendingActions, + historyGroups, + historyTotal, + historyShown, + invalidationCount, + autoApprovalInvalidations, + } = useMemo(() => { const records = [...actionsById.values()] + const invalidations = new Map() + for (const record of records) { + if (record.type === 'action' && record.invalidationReason && record.gatekeeperId !== undefined) { + invalidations.set( + record.gatekeeperId, + Math.max(record.id, invalidations.get(record.gatekeeperId) ?? -1), + ) + } + } const pending = records .filter(record => record.state === 'pending') .toSorted((a, b) => timeValue(a.createdAt) - timeValue(b.createdAt) || a.id - b.id) @@ -158,6 +174,7 @@ export default function Activity({ invalidationCount: records.filter( record => record.type === 'action' && record.invalidationReason, ).length, + autoApprovalInvalidations: invalidations, } }, [actionsById, historyFilter]) @@ -181,7 +198,12 @@ export default function Activity({ } const { alwaysApproveTag, isTagAutoApproved } = - useAlwaysApproveTag(overseer, setProcessingActions, onAutoApproveChange) + useAlwaysApproveTag( + overseer, + setProcessingActions, + onAutoApproveChange, + autoApprovalInvalidations, + ) const toggleExpanded = (id: number) => { setExpandedActionId(previous => (previous === id ? null : id)) diff --git a/packages/workshop-frontend/src/ChatInterface.tsx b/packages/workshop-frontend/src/ChatInterface.tsx index 7c07c0304..9938ccc4a 100644 --- a/packages/workshop-frontend/src/ChatInterface.tsx +++ b/packages/workshop-frontend/src/ChatInterface.tsx @@ -5875,11 +5875,31 @@ function ChatInterface({ actionKind: ActionKind; actionLabel: string } | null >(null); + const autoApprovalInvalidations = useMemo(() => { + const invalidations = new Map(); + for (const message of currentMessages ?? []) { + const log = message.type === "action" ? message.actionLog : undefined; + if (log?.type !== "action" || !log.invalidationReason || log.gatekeeperId === undefined) { + continue; + } + invalidations.set( + log.gatekeeperId, + Math.max(message.actionId, invalidations.get(log.gatekeeperId) ?? -1), + ); + } + return invalidations; + }, [currentMessages]); + // Enable auto-approval of an action tag on its connection (gated by the confirm dialog). The // server applies the now-eligible pending action(s) via its drain, and the action state flips to // "approved" through the actions subscription -- so we don't optimistically mutate it here. const { alwaysApproveTag, isTagAutoApproved } = - useAlwaysApproveTag(overseer, setProcessingActions, onAutoApproveChange); + useAlwaysApproveTag( + overseer, + setProcessingActions, + onAutoApproveChange, + autoApprovalInvalidations, + ); const resolveAction = useResolveAction(overseer, setProcessingActions); diff --git a/packages/workshop-frontend/src/useAlwaysApproveTag.test.tsx b/packages/workshop-frontend/src/useAlwaysApproveTag.test.tsx new file mode 100644 index 000000000..3de60321f --- /dev/null +++ b/packages/workshop-frontend/src/useAlwaysApproveTag.test.tsx @@ -0,0 +1,67 @@ +// @vitest-environment jsdom +/* eslint-disable react/react-in-jsx-scope */ + +import { act, type Dispatch, type SetStateAction } from 'react' +import { createRoot, type Root } from 'react-dom/client' +import { afterEach, describe, expect, it, vi } from 'vitest' +import type { RpcStub } from 'capnweb' +import type { Overseer } from '@gadgets/workshop-shared/api' +import type { ActionKind } from '@gadgets/workshop-shared/gatekeeper' +import { useAlwaysApproveTag } from './useAlwaysApproveTag' + +(globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true + +vi.mock('@cloudflare/kumo', () => ({ + useKumoToastManager: () => ({ add: vi.fn<() => void>() }), +})) + +const ACTION_KIND: ActionKind = { tag: 'send', label: 'Send' } + +describe('useAlwaysApproveTag', () => { + let root: Root | undefined + let container: HTMLDivElement | undefined + + afterEach(() => { + act(() => root?.unmount()) + container?.remove() + vi.restoreAllMocks() + }) + + it('forgets locally enabled tags only when a newer invalidation arrives', async () => { + const overseer = { + setAutoApprovedActionKind: + vi.fn<(_gatekeeperId: number, _actionKind: ActionKind) => Promise>(async () => {}), + } as unknown as RpcStub + const setProcessingActions = + vi.fn<(_value: SetStateAction>) => void>() as unknown as + Dispatch>> + let state: ReturnType | undefined + + function Probe({ invalidations }: { invalidations: ReadonlyMap }) { + state = useAlwaysApproveTag( + overseer, + setProcessingActions, + undefined, + invalidations, + ) + return null + } + + container = document.createElement('div') + document.body.append(container) + root = createRoot(container) + await act(async () => root!.render()) + await act(async () => { await state!.alwaysApproveTag(1, 7, ACTION_KIND) }) + expect(state!.isTagAutoApproved(7, ACTION_KIND.tag)).toBe(true) + + await act(async () => root!.render()) + expect(state!.isTagAutoApproved(7, ACTION_KIND.tag)).toBe(false) + + await act(async () => { await state!.alwaysApproveTag(2, 7, ACTION_KIND) }) + await act(async () => root!.render()) + expect(state!.isTagAutoApproved(7, ACTION_KIND.tag)).toBe(true) + + await act(async () => root!.render()) + expect(state!.isTagAutoApproved(7, ACTION_KIND.tag)).toBe(false) + }) +}) diff --git a/packages/workshop-frontend/src/useAlwaysApproveTag.ts b/packages/workshop-frontend/src/useAlwaysApproveTag.ts index f415be233..d838c22ae 100644 --- a/packages/workshop-frontend/src/useAlwaysApproveTag.ts +++ b/packages/workshop-frontend/src/useAlwaysApproveTag.ts @@ -1,4 +1,4 @@ -import { useCallback, useState, type Dispatch, type SetStateAction } from 'react' +import { useCallback, useEffect, useRef, useState, type Dispatch, type SetStateAction } from 'react' import { useKumoToastManager } from '@cloudflare/kumo' import { RpcStub } from 'capnweb' import { Overseer } from '@gadgets/workshop-shared/api' @@ -14,9 +14,30 @@ export function useAlwaysApproveTag( setProcessingActions: Dispatch>>, // Invoked after a rule is successfully enabled, so other views (e.g. the Connections rule list) // can refresh without waiting to be re-opened. - onEnabled?: () => void) { + onEnabled?: () => void, + latestInvalidations?: ReadonlyMap) { const toasts = useKumoToastManager() const [enabledTags, setEnabledTags] = useState>(new Set()) + const seenInvalidations = useRef(new Map()) + + useEffect(() => { + if (!latestInvalidations) return + const changedGatekeepers = new Set() + for (const [gatekeeperId, actionId] of latestInvalidations) { + if (actionId > (seenInvalidations.current.get(gatekeeperId) ?? -1)) { + changedGatekeepers.add(gatekeeperId) + seenInvalidations.current.set(gatekeeperId, actionId) + } + } + if (changedGatekeepers.size === 0) return + setEnabledTags(previous => { + const next = new Set([...previous].filter(key => { + const separator = key.indexOf(':') + return !changedGatekeepers.has(Number(key.slice(0, separator))) + })) + return next.size === previous.size ? previous : next + }) + }, [latestInvalidations]) // Enable auto-approval for the action's class. Returns true on success, false on failure (the // error is surfaced via a toast) so the caller can decide whether to dismiss a confirm dialog. From e6a6ebe62a15e53a28e235467cbcb5ded9237564 Mon Sep 17 00:00:00 2001 From: Dan Carter Date: Wed, 19 Aug 2026 10:30:27 -0400 Subject: [PATCH 4/6] Track invalidations across workspace chats --- .../workshop-frontend/src/ChatInterface.tsx | 29 +++++++++---------- 1 file changed, 13 insertions(+), 16 deletions(-) diff --git a/packages/workshop-frontend/src/ChatInterface.tsx b/packages/workshop-frontend/src/ChatInterface.tsx index 9938ccc4a..44048e235 100644 --- a/packages/workshop-frontend/src/ChatInterface.tsx +++ b/packages/workshop-frontend/src/ChatInterface.tsx @@ -5462,9 +5462,21 @@ function ChatInterface({ }; }, [overseer]); - // Patch cached chat messages on action upserts. + const [autoApprovalInvalidations, setAutoApprovalInvalidations] = + useState>(new Map()); + + // Patch cached chat messages and reconcile auto-approval state on workspace-wide action upserts. useActionEntries(overseer, (record) => { if (applyActionLogUpdateToCachedMessages(record)) scheduleUpdate(); + if (record.type === "action" && record.invalidationReason && record.gatekeeperId !== undefined) { + const gatekeeperId = record.gatekeeperId; + setAutoApprovalInvalidations(previous => { + if (record.id <= (previous.get(gatekeeperId) ?? -1)) return previous; + const next = new Map(previous); + next.set(gatekeeperId, record.id); + return next; + }); + } }); // Reset per-chat UI state when selectedChatId changes @@ -5875,21 +5887,6 @@ function ChatInterface({ actionKind: ActionKind; actionLabel: string } | null >(null); - const autoApprovalInvalidations = useMemo(() => { - const invalidations = new Map(); - for (const message of currentMessages ?? []) { - const log = message.type === "action" ? message.actionLog : undefined; - if (log?.type !== "action" || !log.invalidationReason || log.gatekeeperId === undefined) { - continue; - } - invalidations.set( - log.gatekeeperId, - Math.max(message.actionId, invalidations.get(log.gatekeeperId) ?? -1), - ); - } - return invalidations; - }, [currentMessages]); - // Enable auto-approval of an action tag on its connection (gated by the confirm dialog). The // server applies the now-eligible pending action(s) via its drain, and the action state flips to // "approved" through the actions subscription -- so we don't optimistically mutate it here. From 44682aed36058dc6662baeb0209f4b593ed32b52 Mon Sep 17 00:00:00 2001 From: Dan Carter Date: Wed, 19 Aug 2026 12:56:32 -0400 Subject: [PATCH 5/6] Simplify MCP action invalidation state --- .../mcp-shared/__tests__/action-store.test.ts | 121 +++++++++++++----- packages/mcp-shared/__tests__/facet.test.ts | 48 ++++--- packages/mcp-shared/src/action-store.ts | 111 ++++++++-------- packages/mcp-shared/src/facet.ts | 18 +-- packages/mcp-shared/src/session.ts | 7 +- .../__tests__/auto-approval.test.ts | 7 +- .../workshop-backend/src/auto-approval.ts | 10 +- packages/workshop-shared/src/gatekeeper.ts | 83 +++++------- 8 files changed, 217 insertions(+), 188 deletions(-) diff --git a/packages/mcp-shared/__tests__/action-store.test.ts b/packages/mcp-shared/__tests__/action-store.test.ts index f0797a4b6..d1e240c0b 100644 --- a/packages/mcp-shared/__tests__/action-store.test.ts +++ b/packages/mcp-shared/__tests__/action-store.test.ts @@ -1,10 +1,10 @@ import { describe, expect, it } from "vitest"; import { DatabaseSync, type SQLInputValue } from "node:sqlite"; -import { ActionInvalidatedError, ActionStore } from "../src/action-store.js"; +import { ActionStore } from "../src/action-store.js"; import { - ACTION_INVALIDATED_ERROR_CODE, - getActionInvalidationReason, + createActionDispatchStoppedError, + getActionDispatchStopped, } from "@gadgets/workshop-shared/gatekeeper"; import { McpProtocolError, @@ -57,38 +57,26 @@ describe("ActionStore", () => { const store = new ActionStore(fakeSql()); const staged = stage(store, {}); - await expect(store.apply(staged.id, async () => { - throw new ActionInvalidatedError("Policy changed. Stage the call again."); - }, log)).rejects.toMatchObject({ - errorCode: ACTION_INVALIDATED_ERROR_CODE, - message: `${ACTION_INVALIDATED_ERROR_CODE}: Policy changed. Stage the call again.`, - }); - await expect(store.apply(staged.id, async () => { + const first = await store.apply(staged.id, async () => { + throw createActionDispatchStoppedError( + "invalidated", "Policy changed. Stage the call again."); + }, log).catch((caught: unknown) => caught); + const replay = await store.apply(staged.id, async () => { throw new Error("must not dispatch again"); - }, log)).rejects.toMatchObject({ - errorCode: ACTION_INVALIDATED_ERROR_CODE, - message: `${ACTION_INVALIDATED_ERROR_CODE}: Policy changed. Stage the call again.`, + }, log).catch((caught: unknown) => caught); + + expect(getActionDispatchStopped(first)).toEqual({ + kind: "invalidated", + reason: "Policy changed. Stage the call again.", }); + expect(getActionDispatchStopped(replay)).toEqual(getActionDispatchStopped(first)); - expect(store.get(staged.id)).toMatchObject({ + const stored = store.get(staged.id); + expect(stored).toMatchObject({ state: "failed", retryable: false, - error: "Policy changed. Stage the call again.", }); - }); - - it("retains the invalidation discriminator in the serialized error message", async () => { - const store = new ActionStore(fakeSql()); - const staged = stage(store, {}); - - const error = await store.apply(staged.id, async () => { - throw new ActionInvalidatedError("Policy changed."); - }, log).catch((caught: unknown) => caught); - - expect(error).toBeInstanceOf(Error); - expect((error as Error).message).toContain(ACTION_INVALIDATED_ERROR_CODE); - expect(getActionInvalidationReason(new Error((error as Error).message))) - .toBe("Policy changed."); + expect(getActionDispatchStopped(stored?.error)).toEqual(getActionDispatchStopped(first)); }); it("keeps validation failures retryable when tools/call was never reached", async () => { @@ -103,7 +91,6 @@ describe("ActionStore", () => { state: "failed", retryable: true, error: "validation failed", - dispatched: false, }); }); @@ -337,6 +324,80 @@ describe("ActionStore", () => { expect(recovered.get(staged.id)?.state).toBe("failed"); }); + it("keeps a claim retryable when interruption preceded the dispatch boundary", async () => { + const sql = fakeSql(); + const store = new ActionStore(sql); + const staged = stage(store, {}); + sql.exec( + "UPDATE mcp_actions SET state = 'applying', retryable = 1 WHERE id = ?", + staged.id, + ); + const recovered = new ActionStore(sql); + + await recovered.apply(staged.id, fn => fn({ callTool: ok } as never), log); + + expect(recovered.get(staged.id)?.state).toBe("applied"); + }); + + it("conservatively closes interrupted claims when migrating the main-era schema", () => { + const sql = fakeSql(); + sql.exec(`CREATE TABLE mcp_actions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + tool_name TEXT NOT NULL, + args_json TEXT NOT NULL, + state TEXT NOT NULL, + submitted_at INTEGER NOT NULL, + claimed_at INTEGER, + retryable INTEGER, + result_json TEXT, + error TEXT + ) STRICT`); + sql.exec( + `INSERT INTO mcp_actions ( + tool_name, args_json, state, submitted_at, retryable + ) VALUES ('send', '{}', 'applying', 1, 1)`, + ); + + const migrated = new ActionStore(sql); + + expect(migrated.get(1)).toMatchObject({ state: "failed", retryable: false }); + }); + + it("uses the prior dispatch marker once when migrating interrupted claims", async () => { + const sql = fakeSql(); + sql.exec(`CREATE TABLE mcp_actions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + tool_name TEXT NOT NULL, + args_json TEXT NOT NULL, + state TEXT NOT NULL, + submitted_at INTEGER NOT NULL, + policy_fingerprint TEXT, + connection_generation INTEGER, + dispatched INTEGER, + claimed_at INTEGER, + retryable INTEGER, + result_json TEXT, + error TEXT + ) STRICT`); + sql.exec( + `INSERT INTO mcp_actions ( + id, tool_name, args_json, state, submitted_at, policy_fingerprint, + connection_generation, dispatched, retryable + ) VALUES + (1, 'send', '{}', 'applying', 1, 'action:manual', 1, 1, 1), + (2, 'send', '{}', 'applying', 1, 'action:manual', 1, 0, 1)`, + ); + + const migrated = new ActionStore(sql); + + expect(migrated.get(1)).toMatchObject({ state: "failed", retryable: false }); + expect(migrated.get(2)).toMatchObject({ state: "failed", retryable: true }); + await expect(migrated.apply(1, fn => fn({ callTool: ok } as never), log)) + .rejects.toThrow(/may or may not have taken effect/); + await migrated.apply(2, fn => fn({ callTool: ok } as never), log); + expect(migrated.get(2)?.state).toBe("applied"); + }); + it("does not let claims nobody will settle consume the queue permanently", async () => { // Only the next `apply()` for that same id settles an expired claim, and an evicted Durable // Object never makes one. Counting those rows retired the binding one interruption at a time. diff --git a/packages/mcp-shared/__tests__/facet.test.ts b/packages/mcp-shared/__tests__/facet.test.ts index 929497367..ee293de75 100644 --- a/packages/mcp-shared/__tests__/facet.test.ts +++ b/packages/mcp-shared/__tests__/facet.test.ts @@ -13,10 +13,7 @@ import { type WithClientOptions, } from "../src/connection.js"; import { - ACTION_INVALIDATED_ERROR_CODE, - ACTION_RESTAGE_REQUIRED_ERROR_CODE, - getActionInvalidationReason, - getActionRestageRequiredReason, + getActionDispatchStopped, type ResourceDescription, } from "@gadgets/workshop-shared/gatekeeper"; @@ -128,15 +125,15 @@ class TestFacet extends McpFacetBase caught); - expect(error).toMatchObject({ errorCode: ACTION_RESTAGE_REQUIRED_ERROR_CODE }); - expect(getActionRestageRequiredReason(new Error((error as Error).message))) - .toMatch(/predates current approval policy checks/i); - expect(subject.lookupAction(action.id)).toMatchObject({ + expect(getActionDispatchStopped(error)).toMatchObject({ + kind: "restage", + reason: expect.stringMatching(/predates current approval policy checks/i), + }); + const stored = subject.lookupAction(action.id); + expect(stored).toMatchObject({ state: "failed", retryable: false, - dispatched: false, error: expect.stringMatching(/predates current approval policy checks/i), }); - await expect(subject.applyAction(action.id)) - .rejects.toMatchObject({ errorCode: ACTION_RESTAGE_REQUIRED_ERROR_CODE }); + const replay = await subject.applyAction(action.id).catch((caught: unknown) => caught); + expect(getActionDispatchStopped(replay)).toEqual(getActionDispatchStopped(stored?.error)); const session = await subject.startSession(queue as never); await expect(session.getActionResult(action.id)).resolves.toMatchObject({ status: "failed", @@ -420,12 +418,11 @@ it("restages a legacy failed action whose absent retryable flag means retryable" subject.removeActionSnapshot(action.id); subject.markActionLegacyRetryableFailure(action.id); - await expect(subject.applyAction(action.id)) - .rejects.toMatchObject({ errorCode: ACTION_RESTAGE_REQUIRED_ERROR_CODE }); + const error = await subject.applyAction(action.id).catch((caught: unknown) => caught); + expect(getActionDispatchStopped(error)?.kind).toBe("restage"); expect(subject.lookupAction(action.id)).toMatchObject({ state: "failed", retryable: false, - dispatched: false, error: expect.stringMatching(/predates current approval policy checks/i), }); }); @@ -455,13 +452,13 @@ it("invalidates an action when the connection changes before a client opens", as const error = await subject.applyAction(action.id).catch((caught: unknown) => caught); - expect(error).toMatchObject({ errorCode: ACTION_INVALIDATED_ERROR_CODE }); - expect(getActionInvalidationReason(new Error((error as Error).message))) - .toMatch(/connection changed/i); + expect(getActionDispatchStopped(error)).toMatchObject({ + kind: "invalidated", + reason: expect.stringMatching(/connection changed/i), + }); expect(subject.lookupAction(action.id)).toMatchObject({ state: "failed", retryable: false, - dispatched: false, }); expect(subject.toolCalls).toBe(0); }); @@ -476,15 +473,14 @@ it("replays an assert-time connection invalidation as not dispatched", async () }); subject.toolCallFailure = new McpConnectionChangedError("The account was repointed."); - await expect(subject.applyAction(action.id)) - .rejects.toMatchObject({ errorCode: ACTION_INVALIDATED_ERROR_CODE }); + const first = await subject.applyAction(action.id).catch((caught: unknown) => caught); + expect(getActionDispatchStopped(first)?.kind).toBe("invalidated"); expect(subject.lookupAction(action.id)).toMatchObject({ state: "failed", retryable: false, - dispatched: false, }); - await expect(subject.applyAction(action.id)) - .rejects.toMatchObject({ errorCode: ACTION_INVALIDATED_ERROR_CODE }); + const replay = await subject.applyAction(action.id).catch((caught: unknown) => caught); + expect(getActionDispatchStopped(replay)).toEqual(getActionDispatchStopped(first)); expect(subject.toolCalls).toBe(0); }); diff --git a/packages/mcp-shared/src/action-store.ts b/packages/mcp-shared/src/action-store.ts index 1c0e2410e..a1102867b 100644 --- a/packages/mcp-shared/src/action-store.ts +++ b/packages/mcp-shared/src/action-store.ts @@ -2,10 +2,8 @@ // database; claims are persisted before external I/O so an interrupted write is never replayed. import { - ACTION_INVALIDATED_ERROR_CODE, - createActionInvalidatedError, - createActionRestageRequiredError, - getActionRestageRequiredReason, + createActionDispatchStoppedError, + getActionDispatchStopped, } from "@gadgets/workshop-shared/gatekeeper"; import { callMayHaveTakenEffect, type McpClient, type McpToolCallResult } from "./client.js"; import type { McpLog } from "./log.js"; @@ -26,7 +24,6 @@ type ActionRow = { submitted_at: number; policy_fingerprint: string | null; connection_generation: number | null; - dispatched: number | null; claimed_at: number | null; retryable: number | null; result_json: string | null; @@ -42,7 +39,6 @@ function fromRow(row: ActionRow): StoredAction { submittedAt: row.submitted_at, policyFingerprint: row.policy_fingerprint ?? undefined, connectionGeneration: row.connection_generation ?? undefined, - dispatched: row.dispatched === null ? undefined : row.dispatched === 1, claimedAt: row.claimed_at ?? undefined, retryable: row.retryable === null ? undefined : row.retryable === 1, result: row.result_json @@ -57,19 +53,8 @@ export const APPLY_OUTCOME_UNKNOWN_MESSAGE = "This call was interrupted after it had been sent, so it may or may not have taken effect. " + "Check the server before trying it again."; -/** The approved tool or account policy changed before dispatch, so the action must be restaged. */ -export class ActionInvalidatedError extends Error { - /** Stable code retained when this error crosses the gatekeeper RPC boundary. */ - readonly errorCode = ACTION_INVALIDATED_ERROR_CODE; - /** Human-readable reason stored separately from the serialized discriminator. */ - readonly reason: string; - - constructor(reason: string) { - super(createActionInvalidatedError(reason).message); - this.name = "ActionInvalidatedError"; - this.reason = reason; - } -} +const APPLY_NOT_DISPATCHED_MESSAGE = + "This call was interrupted before it was sent. Try applying it again."; /** Stores queued MCP actions in one facet-local SQLite table. */ export class ActionStore { @@ -85,7 +70,6 @@ export class ActionStore { submitted_at INTEGER NOT NULL, policy_fingerprint TEXT, connection_generation INTEGER, - dispatched INTEGER CHECK (dispatched IS NULL OR dispatched IN (0, 1)), claimed_at INTEGER, retryable INTEGER CHECK (retryable IS NULL OR retryable IN (0, 1)), result_json TEXT CHECK (result_json IS NULL OR json_valid(result_json)), @@ -102,21 +86,45 @@ export class ActionStore { if (!columns.has("connection_generation")) { sql.exec("ALTER TABLE mcp_actions ADD COLUMN connection_generation INTEGER"); } - if (!columns.has("dispatched")) { - sql.exec("ALTER TABLE mcp_actions ADD COLUMN dispatched INTEGER"); - } + sql.exec(`CREATE TABLE IF NOT EXISTS mcp_action_store_meta ( + version INTEGER NOT NULL + ) STRICT`); + const version = sql.exec<{ version: number }>( + "SELECT version FROM mcp_action_store_meta LIMIT 1").toArray()[0]?.version ?? 0; // A fresh store means a fresh Durable Object activation. Any persisted claim belonged to an // interrupted prior activation and must never be replayed because the write may have landed. - sql.exec( - `UPDATE mcp_actions SET state = 'failed', retryable = 0, error = ? - WHERE state = 'applying' AND dispatched IS NOT 0`, - APPLY_OUTCOME_UNKNOWN_MESSAGE, - ); - sql.exec( - `UPDATE mcp_actions SET state = 'failed', retryable = 1, - error = 'This call was interrupted before it was sent. Try applying it again.' - WHERE state = 'applying' AND dispatched = 0`, - ); + if (version === 0) { + if (columns.has("dispatched")) { + sql.exec( + `UPDATE mcp_actions SET state = 'failed', retryable = 0, error = ? + WHERE state = 'applying' AND dispatched IS NOT 0`, + APPLY_OUTCOME_UNKNOWN_MESSAGE, + ); + sql.exec( + `UPDATE mcp_actions SET state = 'failed', retryable = 1, error = ? + WHERE state = 'applying' AND dispatched = 0`, + APPLY_NOT_DISPATCHED_MESSAGE, + ); + } else { + sql.exec( + `UPDATE mcp_actions SET state = 'failed', retryable = 0, error = ? + WHERE state = 'applying'`, + APPLY_OUTCOME_UNKNOWN_MESSAGE, + ); + } + sql.exec("INSERT INTO mcp_action_store_meta (version) VALUES (1)"); + } else { + sql.exec( + `UPDATE mcp_actions SET state = 'failed', retryable = 0, error = ? + WHERE state = 'applying' AND retryable IS NOT 1`, + APPLY_OUTCOME_UNKNOWN_MESSAGE, + ); + sql.exec( + `UPDATE mcp_actions SET state = 'failed', retryable = 1, error = ? + WHERE state = 'applying' AND retryable = 1`, + APPLY_NOT_DISPATCHED_MESSAGE, + ); + } this.#prune(); } @@ -128,15 +136,13 @@ export class ActionStore { #save(action: StoredAction): void { this.#sql.exec( - `UPDATE mcp_actions SET state = ?, claimed_at = ?, retryable = ?, result_json = ?, error = ?, - dispatched = ? + `UPDATE mcp_actions SET state = ?, claimed_at = ?, retryable = ?, result_json = ?, error = ? WHERE id = ?`, action.state, action.claimedAt ?? null, action.retryable === undefined ? null : Number(action.retryable), action.result === undefined ? null : JSON.stringify(action.result), action.error ?? null, - action.dispatched === undefined ? null : Number(action.dispatched), action.id, ); } @@ -189,9 +195,8 @@ export class ActionStore { stored.state = "failed"; stored.claimedAt = undefined; stored.retryable = false; - stored.dispatched = false; stored.result = undefined; - stored.error = createActionRestageRequiredError(reason).message; + stored.error = createActionDispatchStoppedError("restage", reason).message; this.#save(stored); this.#prune(); } @@ -206,12 +211,8 @@ export class ActionStore { if (stored.state === "applied") return; if (stored.state === "rejected") throw new Error(`MCP action ${id} was already rejected.`); if (stored.state === "failed" && stored.retryable === false) { - if (stored.dispatched === false) { - const restageReason = getActionRestageRequiredReason(new Error(stored.error ?? "")); - if (restageReason !== undefined) throw createActionRestageRequiredError(restageReason); - throw new ActionInvalidatedError( - stored.error ?? "This MCP action became invalid before dispatch."); - } + const stopped = getActionDispatchStopped(stored.error ?? ""); + if (stopped) throw createActionDispatchStoppedError(stopped.kind, stopped.reason); throw new Error(stored.error ?? `MCP action ${id} cannot be retried.`); } if (stored.state === "applying") { @@ -220,34 +221,34 @@ export class ActionStore { stored.state = "applying"; stored.claimedAt = Date.now(); - stored.dispatched = false; + stored.retryable = true; stored.error = undefined; stored.result = undefined; this.#save(stored); let result: McpToolCallResult; - let dispatched = false; + let crossedDispatchBoundary = false; try { result = await call(client => { - dispatched = true; - stored.dispatched = true; + crossedDispatchBoundary = true; + stored.retryable = false; this.#save(stored); return client.callTool(stored.toolName, stored.args); }); } catch (err) { - if (err instanceof ActionInvalidatedError) { + const stopped = getActionDispatchStopped(err); + if (stopped) { stored.state = "failed"; stored.retryable = false; - stored.dispatched = false; - stored.error = err.reason; + stored.error = createActionDispatchStoppedError(stopped.kind, stopped.reason).message; this.#save(stored); this.#prune(); - log.warn("tool call invalidated before dispatch", { - event: "action.apply.invalidated", actionId: id, toolName: stored.toolName, error: err, + log.warn("tool call stopped before dispatch", { + event: "action.apply.stopped", actionId: id, toolName: stored.toolName, error: err, }); - throw err; + throw createActionDispatchStoppedError(stopped.kind, stopped.reason); } - const mayHaveLanded = dispatched && callMayHaveTakenEffect(err); + const mayHaveLanded = crossedDispatchBoundary && callMayHaveTakenEffect(err); stored.state = "failed"; stored.retryable = !mayHaveLanded; stored.error = mayHaveLanded diff --git a/packages/mcp-shared/src/facet.ts b/packages/mcp-shared/src/facet.ts index 762c39d96..b0118d776 100644 --- a/packages/mcp-shared/src/facet.ts +++ b/packages/mcp-shared/src/facet.ts @@ -3,7 +3,7 @@ import { DurableObject, type RpcStub } from "cloudflare:workers"; import { - createActionRestageRequiredError, + createActionDispatchStoppedError, type ActionKind, type ApprovalQueue, type Gatekeeper, @@ -11,11 +11,7 @@ import { type ResourceDescription, } from "@gadgets/workshop-shared/gatekeeper"; -import { - ActionInvalidatedError, - ActionStore, - REVERT_UNSUPPORTED_MESSAGE, -} from "./action-store.js"; +import { ActionStore, REVERT_UNSUPPORTED_MESSAGE } from "./action-store.js"; import { CATALOG_TTL_MS, HydratedTools, @@ -338,7 +334,7 @@ export abstract class McpFacetBase< const reason = "This MCP action predates current approval policy checks. Stage the call again."; this.#actions().markRestageRequired(action, reason); - throw createActionRestageRequiredError(reason); + throw createActionDispatchStoppedError("restage", reason); } return this.#actions().apply( action, @@ -347,16 +343,16 @@ export abstract class McpFacetBase< const catalog = await this.catalog(deadline); if (stored.connectionGeneration === undefined || stored.policyFingerprint === undefined || connectionGeneration !== stored.connectionGeneration) { - throw new ActionInvalidatedError( + throw createActionDispatchStoppedError("invalidated", "This MCP connection changed after approval was requested. Stage the call again."); } if (!scopeAllows(this.scope, stored.toolName, catalog.isPortal)) { - throw new ActionInvalidatedError( + throw createActionDispatchStoppedError("invalidated", "This MCP tool is no longer allowed by the binding. Stage the call again."); } const tool = await this.#freshTool(client, stored.toolName, catalog, deadline); if (!tool || toolPolicyFingerprint(tool, this.trust) !== stored.policyFingerprint) { - throw new ActionInvalidatedError( + throw createActionDispatchStoppedError("invalidated", "This MCP tool's approval policy changed. Review and stage the call again."); } return () => fn(client); @@ -364,7 +360,7 @@ export abstract class McpFacetBase< return dispatch(); }, { retryOnExpiry: false }).catch(error => { if (error instanceof McpConnectionChangedError) { - throw new ActionInvalidatedError( + throw createActionDispatchStoppedError("invalidated", "This MCP connection changed after approval was requested. Stage the call again."); } throw error; diff --git a/packages/mcp-shared/src/session.ts b/packages/mcp-shared/src/session.ts index 3ffd49644..de4ca5c6a 100644 --- a/packages/mcp-shared/src/session.ts +++ b/packages/mcp-shared/src/session.ts @@ -6,7 +6,7 @@ import { RpcTarget, type RpcStub } from "cloudflare:workers"; import { - getActionRestageRequiredReason, + getActionDispatchStopped, type ActionDescription, type ActionKind, type ApprovalQueue, @@ -55,8 +55,6 @@ export type StoredAction = { policyFingerprint?: string; /** Account connection generation the user approved, absent on actions staged by older code. */ connectionGeneration?: number; - /** Set immediately before `tools/call`; false proves a failure preceded dispatch. */ - dispatched?: boolean; /** When the in-flight apply was claimed, for recovering a claim whose Durable Object died mid-call. */ claimedAt?: number; /** @@ -291,8 +289,7 @@ export class McpSessionBase extends RpcTarget { case "failed": return { status: "failed", - message: (stored.error && getActionRestageRequiredReason(new Error(stored.error))) - ?? stored.error + message: (stored.error && getActionDispatchStopped(stored.error)?.reason) ?? stored.error ?? `Calling "${stored.toolName}" on ${host.serverName} failed.`, }; case "applied": { diff --git a/packages/workshop-backend/__tests__/auto-approval.test.ts b/packages/workshop-backend/__tests__/auto-approval.test.ts index 019e569d6..2867fe9ee 100644 --- a/packages/workshop-backend/__tests__/auto-approval.test.ts +++ b/packages/workshop-backend/__tests__/auto-approval.test.ts @@ -10,8 +10,7 @@ import { import type { ActionRecord, AutoApproveTagRecord } from "../src/overseer.js"; import type { AiChatAuthorInfo } from "@gadgets/workshop-shared/api"; import { - createActionInvalidatedError, - createActionRestageRequiredError, + createActionDispatchStoppedError, } from "@gadgets/workshop-shared/gatekeeper"; import { makeMockStorage } from "./mock-storage.js"; @@ -361,7 +360,7 @@ describe("handleActionApplyFailure", () => { expect(handleActionApplyFailure( storage, GK, - createActionRestageRequiredError("Stage the call again."), + createActionDispatchStoppedError("restage", "Stage the call again."), )).toBe("Stage the call again."); expect([...storage.autoApproveTags.list()].map(rule => rule.actionKind.tag).toSorted()) .toEqual(["delete", "edit"]); @@ -375,7 +374,7 @@ describe("handleActionApplyFailure", () => { expect(handleActionApplyFailure( storage, GK, - createActionInvalidatedError("The connection changed."), + createActionDispatchStoppedError("invalidated", "The connection changed."), )).toBe("The connection changed."); expect([...storage.autoApproveTags.list()]).toEqual([]); }); diff --git a/packages/workshop-backend/src/auto-approval.ts b/packages/workshop-backend/src/auto-approval.ts index 215a47820..9ac33911f 100644 --- a/packages/workshop-backend/src/auto-approval.ts +++ b/packages/workshop-backend/src/auto-approval.ts @@ -6,8 +6,7 @@ import type { Collection } from "@gadgets/typed-storage"; import type { AiChatAuthorInfo } from "@gadgets/workshop-shared/api"; import { - getActionInvalidationReason, - getActionRestageRequiredReason, + getActionDispatchStopped, } from "@gadgets/workshop-shared/gatekeeper"; import { createWorkshopLogger } from "./observability"; import type { ActionRecord, AutoApproveTagRecord } from "./overseer.js"; @@ -47,12 +46,11 @@ export function handleActionApplyFailure( storage: Pick, gatekeeperId: number, error: unknown): string | undefined { - const invalidationReason = getActionInvalidationReason(error); - if (invalidationReason !== undefined) { + const stopped = getActionDispatchStopped(error); + if (stopped?.kind === "invalidated") { clearAutoApprovalRules(storage, gatekeeperId); - return invalidationReason; } - return getActionRestageRequiredReason(error); + return stopped?.reason; } export class AutoApprovalDrainer { diff --git a/packages/workshop-shared/src/gatekeeper.ts b/packages/workshop-shared/src/gatekeeper.ts index 84a480c8d..469aef4b4 100644 --- a/packages/workshop-shared/src/gatekeeper.ts +++ b/packages/workshop-shared/src/gatekeeper.ts @@ -18,57 +18,36 @@ import type { WorkerEntrypoint, DurableObject, RpcTarget, RpcStub } from "cloudflare:workers"; -/** Stable RPC error code thrown when an approved action became invalid before dispatch. */ -export const ACTION_INVALIDATED_ERROR_CODE = "GATEKEEPER_ACTION_INVALIDATED"; - -const ACTION_INVALIDATED_ERROR_PREFIX = `${ACTION_INVALIDATED_ERROR_CODE}: `; - -function codedActionErrorReason( - error: unknown, code: string, prefix: string): string | undefined { - const candidate = typeof error === "object" && error !== null && "errorCode" in error - ? error.errorCode - : undefined; - const message = typeof error === "object" && error !== null && "message" in error - && typeof error.message === "string" - ? error.message - : undefined; - return message?.startsWith(prefix) ? message.slice(prefix.length) - : candidate === code ? message : undefined; -} - -/** Creates an invalidation error whose discriminator survives Workers RPC error serialization. */ -export function createActionInvalidatedError(reason: string): Error & { - errorCode: typeof ACTION_INVALIDATED_ERROR_CODE; -} { - return Object.assign(new Error(`${ACTION_INVALIDATED_ERROR_PREFIX}${reason}`), { - errorCode: ACTION_INVALIDATED_ERROR_CODE, - } as const); -} - -/** Reads an invalidation reason from a local coded error or its message-only RPC representation. */ -export function getActionInvalidationReason(error: unknown): string | undefined { - return codedActionErrorReason( - error, ACTION_INVALIDATED_ERROR_CODE, ACTION_INVALIDATED_ERROR_PREFIX); -} - -/** Stable RPC error code thrown when an old queued action lacks an approval-policy snapshot. */ -export const ACTION_RESTAGE_REQUIRED_ERROR_CODE = "GATEKEEPER_ACTION_RESTAGE_REQUIRED"; +/** Why an approved action stopped before dispatch. */ +export type ActionDispatchStopped = { + /** Whether current authority changed or an older action lacks the snapshot needed to prove it. */ + kind: "invalidated" | "restage"; + /** Human-readable explanation shown to the user. */ + reason: string; +}; -const ACTION_RESTAGE_REQUIRED_ERROR_PREFIX = `${ACTION_RESTAGE_REQUIRED_ERROR_CODE}: `; +const ACTION_DISPATCH_STOPPED_PREFIX = "GATEKEEPER_ACTION_DISPATCH_STOPPED:"; -/** Creates a restage-required error whose discriminator survives Workers RPC serialization. */ -export function createActionRestageRequiredError(reason: string): Error & { - errorCode: typeof ACTION_RESTAGE_REQUIRED_ERROR_CODE; -} { - return Object.assign(new Error(`${ACTION_RESTAGE_REQUIRED_ERROR_PREFIX}${reason}`), { - errorCode: ACTION_RESTAGE_REQUIRED_ERROR_CODE, - } as const); +/** Creates a dispatch-stopped error whose kind survives Workers RPC error serialization. */ +export function createActionDispatchStoppedError( + kind: ActionDispatchStopped["kind"], reason: string): Error { + return new Error(`${ACTION_DISPATCH_STOPPED_PREFIX}${kind}: ${reason}`); } -/** Reads a restage-required reason from a local coded error or its message-only RPC form. */ -export function getActionRestageRequiredReason(error: unknown): string | undefined { - return codedActionErrorReason( - error, ACTION_RESTAGE_REQUIRED_ERROR_CODE, ACTION_RESTAGE_REQUIRED_ERROR_PREFIX); +/** Reads a dispatch-stopped result from a local error, RPC error, or persisted message. */ +export function getActionDispatchStopped(error: unknown): ActionDispatchStopped | undefined { + const message = typeof error === "string" ? error + : typeof error === "object" && error !== null && "message" in error + && typeof error.message === "string" + ? error.message + : undefined; + if (!message?.startsWith(ACTION_DISPATCH_STOPPED_PREFIX)) return undefined; + const encoded = message.slice(ACTION_DISPATCH_STOPPED_PREFIX.length); + const separator = encoded.indexOf(": "); + if (separator < 0) return undefined; + const kind = encoded.slice(0, separator); + if (kind !== "invalidated" && kind !== "restage") return undefined; + return { kind, reason: encoded.slice(separator + 2) }; } /** @@ -864,10 +843,12 @@ export interface Gatekeeper extends DurableObject { * If this throws an exception, the user will be informed that the action failed and given the * opportunity to retry or discard. * - * If policy or authority changed after approval but before dispatch, throw an error created by - * `createActionInvalidatedError()`. New Workshop versions record that distinct terminal outcome; - * old versions leave the action pending rather than recording a write that was never dispatched - * as approved. + * If policy or authority changed after approval but before dispatch, throw an `"invalidated"` + * error created by `createActionDispatchStoppedError()`. New Workshop versions record that + * distinct terminal outcome; old versions leave the action pending rather than recording a write + * that was never dispatched as approved. `"restage"` is reserved for an older queued action that + * lacks the snapshot needed for current validation. Both stop the current drain, but only + * `"invalidated"` clears auto-approval rules whose authority may have changed. * * 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 From 728b9858b02b18513937a7db48f710e3bf72dd8d Mon Sep 17 00:00:00 2001 From: Dan Carter Date: Wed, 19 Aug 2026 12:56:43 -0400 Subject: [PATCH 6/6] Centralize auto-approval invalidation updates --- packages/workshop-frontend/src/Activity.tsx | 18 +-------- .../workshop-frontend/src/ChatInterface.tsx | 21 +--------- .../src/useAlwaysApproveTag.test.tsx | 39 ++++++++++++------- .../src/useAlwaysApproveTag.ts | 31 +++++++-------- 4 files changed, 43 insertions(+), 66 deletions(-) diff --git a/packages/workshop-frontend/src/Activity.tsx b/packages/workshop-frontend/src/Activity.tsx index 80ab12d87..c3f9d6445 100644 --- a/packages/workshop-frontend/src/Activity.tsx +++ b/packages/workshop-frontend/src/Activity.tsx @@ -139,18 +139,8 @@ export default function Activity({ historyTotal, historyShown, invalidationCount, - autoApprovalInvalidations, } = useMemo(() => { const records = [...actionsById.values()] - const invalidations = new Map() - for (const record of records) { - if (record.type === 'action' && record.invalidationReason && record.gatekeeperId !== undefined) { - invalidations.set( - record.gatekeeperId, - Math.max(record.id, invalidations.get(record.gatekeeperId) ?? -1), - ) - } - } const pending = records .filter(record => record.state === 'pending') .toSorted((a, b) => timeValue(a.createdAt) - timeValue(b.createdAt) || a.id - b.id) @@ -174,7 +164,6 @@ export default function Activity({ invalidationCount: records.filter( record => record.type === 'action' && record.invalidationReason, ).length, - autoApprovalInvalidations: invalidations, } }, [actionsById, historyFilter]) @@ -198,12 +187,7 @@ export default function Activity({ } const { alwaysApproveTag, isTagAutoApproved } = - useAlwaysApproveTag( - overseer, - setProcessingActions, - onAutoApproveChange, - autoApprovalInvalidations, - ) + useAlwaysApproveTag(overseer, setProcessingActions, onAutoApproveChange) const toggleExpanded = (id: number) => { setExpandedActionId(previous => (previous === id ? null : id)) diff --git a/packages/workshop-frontend/src/ChatInterface.tsx b/packages/workshop-frontend/src/ChatInterface.tsx index 44048e235..7c07c0304 100644 --- a/packages/workshop-frontend/src/ChatInterface.tsx +++ b/packages/workshop-frontend/src/ChatInterface.tsx @@ -5462,21 +5462,9 @@ function ChatInterface({ }; }, [overseer]); - const [autoApprovalInvalidations, setAutoApprovalInvalidations] = - useState>(new Map()); - - // Patch cached chat messages and reconcile auto-approval state on workspace-wide action upserts. + // Patch cached chat messages on action upserts. useActionEntries(overseer, (record) => { if (applyActionLogUpdateToCachedMessages(record)) scheduleUpdate(); - if (record.type === "action" && record.invalidationReason && record.gatekeeperId !== undefined) { - const gatekeeperId = record.gatekeeperId; - setAutoApprovalInvalidations(previous => { - if (record.id <= (previous.get(gatekeeperId) ?? -1)) return previous; - const next = new Map(previous); - next.set(gatekeeperId, record.id); - return next; - }); - } }); // Reset per-chat UI state when selectedChatId changes @@ -5891,12 +5879,7 @@ function ChatInterface({ // server applies the now-eligible pending action(s) via its drain, and the action state flips to // "approved" through the actions subscription -- so we don't optimistically mutate it here. const { alwaysApproveTag, isTagAutoApproved } = - useAlwaysApproveTag( - overseer, - setProcessingActions, - onAutoApproveChange, - autoApprovalInvalidations, - ); + useAlwaysApproveTag(overseer, setProcessingActions, onAutoApproveChange); const resolveAction = useResolveAction(overseer, setProcessingActions); diff --git a/packages/workshop-frontend/src/useAlwaysApproveTag.test.tsx b/packages/workshop-frontend/src/useAlwaysApproveTag.test.tsx index 3de60321f..acabc8986 100644 --- a/packages/workshop-frontend/src/useAlwaysApproveTag.test.tsx +++ b/packages/workshop-frontend/src/useAlwaysApproveTag.test.tsx @@ -5,7 +5,7 @@ import { act, type Dispatch, type SetStateAction } from 'react' import { createRoot, type Root } from 'react-dom/client' import { afterEach, describe, expect, it, vi } from 'vitest' import type { RpcStub } from 'capnweb' -import type { Overseer } from '@gadgets/workshop-shared/api' +import type { ActionLogEntry, Overseer } from '@gadgets/workshop-shared/api' import type { ActionKind } from '@gadgets/workshop-shared/gatekeeper' import { useAlwaysApproveTag } from './useAlwaysApproveTag' @@ -15,6 +15,16 @@ vi.mock('@cloudflare/kumo', () => ({ useKumoToastManager: () => ({ add: vi.fn<() => void>() }), })) +const actionEntries = vi.hoisted(() => ({ + listener: undefined as ((record: ActionLogEntry) => void) | undefined, +})) + +vi.mock('./useActions', () => ({ + useActionEntries: (_overseer: unknown, listener: (record: ActionLogEntry) => void) => { + actionEntries.listener = listener + }, +})) + const ACTION_KIND: ActionKind = { tag: 'send', label: 'Send' } describe('useAlwaysApproveTag', () => { @@ -27,7 +37,7 @@ describe('useAlwaysApproveTag', () => { vi.restoreAllMocks() }) - it('forgets locally enabled tags only when a newer invalidation arrives', async () => { + it('forgets locally enabled tags for each newly invalidated action, regardless of action order', async () => { const overseer = { setAutoApprovedActionKind: vi.fn<(_gatekeeperId: number, _actionKind: ActionKind) => Promise>(async () => {}), @@ -37,31 +47,34 @@ describe('useAlwaysApproveTag', () => { Dispatch>> let state: ReturnType | undefined - function Probe({ invalidations }: { invalidations: ReadonlyMap }) { - state = useAlwaysApproveTag( - overseer, - setProcessingActions, - undefined, - invalidations, - ) + function Probe() { + state = useAlwaysApproveTag(overseer, setProcessingActions) return null } + const invalidate = (id: number) => actionEntries.listener!({ + id, + type: 'action', + gatekeeperId: 7, + invalidationReason: 'Connection changed.', + } as ActionLogEntry) + container = document.createElement('div') document.body.append(container) root = createRoot(container) - await act(async () => root!.render()) + await act(async () => root!.render()) + act(() => invalidate(10)) await act(async () => { await state!.alwaysApproveTag(1, 7, ACTION_KIND) }) expect(state!.isTagAutoApproved(7, ACTION_KIND.tag)).toBe(true) - await act(async () => root!.render()) + act(() => invalidate(5)) expect(state!.isTagAutoApproved(7, ACTION_KIND.tag)).toBe(false) await act(async () => { await state!.alwaysApproveTag(2, 7, ACTION_KIND) }) - await act(async () => root!.render()) + act(() => invalidate(5)) expect(state!.isTagAutoApproved(7, ACTION_KIND.tag)).toBe(true) - await act(async () => root!.render()) + act(() => invalidate(4)) expect(state!.isTagAutoApproved(7, ACTION_KIND.tag)).toBe(false) }) }) diff --git a/packages/workshop-frontend/src/useAlwaysApproveTag.ts b/packages/workshop-frontend/src/useAlwaysApproveTag.ts index d838c22ae..a807b76ce 100644 --- a/packages/workshop-frontend/src/useAlwaysApproveTag.ts +++ b/packages/workshop-frontend/src/useAlwaysApproveTag.ts @@ -3,6 +3,7 @@ import { useKumoToastManager } from '@cloudflare/kumo' import { RpcStub } from 'capnweb' import { Overseer } from '@gadgets/workshop-shared/api' import { ActionKind } from '@gadgets/workshop-shared/gatekeeper' +import { useActionEntries } from './useActions' /** * Enables an auto-approval rule for the action's (gatekeeperId, actionKind.tag), and tracks @@ -14,30 +15,26 @@ export function useAlwaysApproveTag( setProcessingActions: Dispatch>>, // Invoked after a rule is successfully enabled, so other views (e.g. the Connections rule list) // can refresh without waiting to be re-opened. - onEnabled?: () => void, - latestInvalidations?: ReadonlyMap) { + onEnabled?: () => void) { const toasts = useKumoToastManager() const [enabledTags, setEnabledTags] = useState>(new Set()) - const seenInvalidations = useRef(new Map()) + const seenInvalidations = useRef(new Set()) useEffect(() => { - if (!latestInvalidations) return - const changedGatekeepers = new Set() - for (const [gatekeeperId, actionId] of latestInvalidations) { - if (actionId > (seenInvalidations.current.get(gatekeeperId) ?? -1)) { - changedGatekeepers.add(gatekeeperId) - seenInvalidations.current.set(gatekeeperId, actionId) - } - } - if (changedGatekeepers.size === 0) return + seenInvalidations.current.clear() + setEnabledTags(new Set()) + }, [overseer]) + + useActionEntries(overseer, record => { + if (record.type !== 'action' || !record.invalidationReason || record.gatekeeperId === undefined || + seenInvalidations.current.has(record.id)) return + seenInvalidations.current.add(record.id) + const prefix = `${record.gatekeeperId}:` setEnabledTags(previous => { - const next = new Set([...previous].filter(key => { - const separator = key.indexOf(':') - return !changedGatekeepers.has(Number(key.slice(0, separator))) - })) + const next = new Set([...previous].filter(key => !key.startsWith(prefix))) return next.size === previous.size ? previous : next }) - }, [latestInvalidations]) + }) // Enable auto-approval for the action's class. Returns true on success, false on failure (the // error is surfaced via a toast) so the caller can decide whether to dismiss a confirm dialog.