diff --git a/.changeset/toolkit-list-scope.md b/.changeset/toolkit-list-scope.md new file mode 100644 index 0000000000..2af6981f88 --- /dev/null +++ b/.changeset/toolkit-list-scope.md @@ -0,0 +1,6 @@ +--- +"@executor-js/sdk": patch +"@executor-js/plugin-toolkits": patch +--- + +Toolkit sessions no longer walk the whole workspace catalog on connect, search, or describe: the toolkit's access patterns narrow the tool rows core reads. Tools reads no longer wait on re-listing catalogs that are only older than the freshness TTL; those rebuild in the background while the read answers from the persisted rows. Stale-marked and config-revised catalogs still gate the read within the grace budget. diff --git a/packages/core/sdk/src/connections.test.ts b/packages/core/sdk/src/connections.test.ts index 51c7f8d110..1fd653c093 100644 --- a/packages/core/sdk/src/connections.test.ts +++ b/packages/core/sdk/src/connections.test.ts @@ -10,6 +10,7 @@ import { Option, Predicate, Result, + Schedule, Schema, Tracer, } from "effect"; @@ -1912,6 +1913,135 @@ describe("tool catalog sync safety", () => { ), ); + // Live clock: the poll below waits on a detached rebuild fiber, not on the + // test clock. + it.live("a time-expired catalog answers from persisted rows and rebuilds in the background", () => + Effect.scoped( + Effect.gen(function* () { + const listingStarted = yield* Deferred.make(); + const releaseListing = yield* Deferred.make(); + let resolutions = 0; + const remotePlugin = definePlugin(() => ({ + id: "remote" as const, + credentialProviders: [memoryProvider()], + storage: () => ({}), + remoteToolCatalog: true, + resolveTools: () => + Effect.gen(function* () { + resolutions += 1; + if (resolutions === 1) { + return { tools: [{ name: ToolName.make("deploy"), description: "deploy" }] }; + } + yield* Deferred.succeed(listingStarted, undefined); + yield* Deferred.await(releaseListing); + return { + tools: [ + { name: ToolName.make("deploy"), description: "deploy" }, + { name: ToolName.make("list"), description: "list" }, + ], + }; + }), + invokeTool: ({ toolRow }) => Effect.succeed({ ran: toolRow.name }), + extension: (ctx) => ({ + seed: () => + ctx.core.integrations.register({ slug: INTEG, description: "Vercel", config: {} }), + }), + }))(); + // TTL 0: every catalog is time-expired on every read. + const config = { + ...makeTestConfig({ plugins: [remotePlugin] as const }), + toolsSyncTtlMs: 0, + }; + const executor = yield* createExecutor(config); + yield* executor.remote.seed(); + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + value: "secret-token", + }); + + // Let the clock move past the stamp `create` wrote, so the catalog is + // older than the zero TTL on the read below. + yield* Effect.sleep("5 millis"); + + // The upstream listing is held open. A read that waited on it would + // pay the full grace budget; this one must answer at once from the + // persisted catalog. + const startedAt = Date.now(); + const stale = yield* executor.tools.list({ integration: INTEG }); + expect(Date.now() - startedAt).toBeLessThan(1000); + expect(stale.map((tool) => String(tool.name))).toEqual(["deploy"]); + yield* Deferred.await(listingStarted); + + // Once the background rebuild lands, a later read observes it. + yield* Deferred.succeed(releaseListing, undefined); + const converged = yield* executor.tools.list({ integration: INTEG }).pipe( + Effect.map((tools) => tools.map((tool) => String(tool.name)).sort()), + Effect.repeat({ + until: (names) => names.length === 2, + schedule: Schedule.spaced("10 millis"), + }), + Effect.timeout("5 seconds"), + ); + expect(converged).toEqual(["deploy", "list"]); + }), + ), + ); + + it.effect("a stale-marked catalog still gates the read within the grace budget", () => + Effect.scoped( + Effect.gen(function* () { + let resolutions = 0; + const remotePlugin = definePlugin(() => ({ + id: "remote" as const, + credentialProviders: [memoryProvider()], + storage: () => ({}), + remoteToolCatalog: true, + resolveTools: () => + Effect.sync(() => { + resolutions += 1; + return { + tools: + resolutions === 1 + ? [{ name: ToolName.make("deploy"), description: "deploy" }] + : [ + { name: ToolName.make("deploy"), description: "deploy" }, + { name: ToolName.make("list"), description: "list" }, + ], + }; + }), + invokeTool: ({ toolRow }) => Effect.succeed({ ran: toolRow.name }), + extension: (ctx) => ({ + seed: () => + ctx.core.integrations.register({ slug: INTEG, description: "Vercel", config: {} }), + }), + }))(); + const config = makeTestConfig({ plugins: [remotePlugin] as const }); + const executor = yield* createExecutor(config); + yield* executor.remote.seed(); + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + value: "secret-token", + }); + // Stale-marked (an upstream said the catalog changed): the very next + // read reflects the rebuild. + yield* Effect.promise(() => + config.db.updateMany("connection", { + where: (b) => b.and(b("integration", "=", String(INTEG)), b("name", "=", "main")), + set: { tools_synced_at: null }, + }), + ); + const tools = yield* executor.tools.list({ integration: INTEG }); + expect(tools.map((tool) => String(tool.name)).sort()).toEqual(["deploy", "list"]); + }), + ), + ); + it.effect( "background sync preserves a nonzero remote catalog when a plugin returns authoritative empty", () => diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index cd89ab9075..1e99b4531d 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -154,6 +154,7 @@ import { import type { FirstPartyOAuthClientConfig } from "./oauth-client"; import { comparePolicyRow, + isUnboundedDynamicToolScope, isValidPattern, matchPattern, positionForNewPattern, @@ -180,6 +181,7 @@ import type { StaticIntegrationDecl, StaticToolDecl, StorageDeps, + PreparedToolPolicy, ToolPolicyProvider, ToolPolicyProviderRule, ToolInvocationCredential, @@ -3222,7 +3224,10 @@ export const createExecutor = !tool.static).map((tool) => String(tool.integration)), ); @@ -4759,7 +4764,10 @@ export const createExecutor = !tool.static) @@ -5399,10 +5407,8 @@ export const createExecutor = EffectivePolicy; + readonly resolve: PreparedToolPolicy["resolve"]; + readonly dynamicScope: PreparedToolPolicy["dynamicScope"]; }; const compareProviderPolicyRule = ( @@ -5443,9 +5449,10 @@ export const createExecutor = ({ + Effect.map((prepared) => ({ kind: "prepared" as const, - resolve, + resolve: prepared.resolve, + dynamicScope: prepared.dynamicScope, })), ) : activeToolPolicyProvider.resolve @@ -5521,116 +5528,137 @@ export const createExecutor = [row.slug, row] as const)); - // The TTL only matters when a loaded plugin actually lists a live remote - // catalog; otherwise skip it so age alone never widens the stale query. - const anyRemoteCatalog = Array.from(runtimes.values()).some( - (runtime) => runtime.plugin.remoteToolCatalog === true, - ); - const cutoff = - toolsSyncTtlMs == null || !anyRemoteCatalog ? null : Date.now() - toolsSyncTtlMs; - - // Bound the scan to potentially-stale rows: stale-marked (NULL stamp) or - // synced before the latest instant any trigger could fire at (the TTL - // cutoff / the newest config revision). Per-row trigger checks below - // re-verify against each row's own integration; in steady state this - // query returns nothing and the read pays one indexed lookup. - const latestRevision = integrations.reduce( - (max, row) => - row.config_revised_at == null - ? max - : Math.max(max ?? Number(row.config_revised_at), Number(row.config_revised_at)), - null, - ); - const staleBefore = - cutoff === null && latestRevision === null - ? null - : Math.max(cutoff ?? Number.MIN_SAFE_INTEGER, latestRevision ?? Number.MIN_SAFE_INTEGER); - - const connections = yield* core.findMany("connection", { - where: (b: AnyCb) => - staleBefore === null - ? b.isNull("tools_synced_at") - : b.or(b.isNull("tools_synced_at"), b("tools_synced_at", "<", staleBefore)), - }); - // Each rebuild is an independent upstream listing, so they run together - // rather than one after another: a host with many stale remote-catalog - // connections otherwise pays the sum of every server's latency on the - // read that trips the TTL. Only the listings overlap — `persistCatalog` - // keeps the catalog writes in a single-file queue, so this fan-out never - // opens two transactions on a one-connection database. - const rebuilds: Effect.Effect[] = []; - for (const connection of connections) { - const integrationRow = integrationBySlug.get(connection.integration); - if (!integrationRow) continue; - const runtime = runtimes.get(integrationRow.plugin_id); - // Only re-produce catalogs this executor can actually re-list — - // rebuilding under an unloaded plugin would clear a working catalog. - // (A loaded plugin without `resolveTools` still flows through: - // `produceConnectionTools` runs its clear-and-stamp cleanup path.) - if (!runtime) continue; - - const syncedAt = - connection.tools_synced_at == null ? null : Number(connection.tools_synced_at); - const revisedTime = - integrationRow.config_revised_at == null + const syncStaleConnectionTools = (mode: "converge" | "bounded") => + Effect.gen(function* () { + // The platform view can never persist a rebuilt catalog (writes are + // denied at the storage boundary), so attempting the sync would only + // fire upstream `resolveTools` calls whose results are thrown away — + // network side effects on a read-only credential. Skip it entirely: + // read-only-ness of the platform read path is a stated invariant here, + // not an accident of the best-effort catch below. + if (config.platformView === true) return; + const integrations = yield* core.findMany("integration", {}); + if (integrations.length === 0) return; + const integrationBySlug = new Map(integrations.map((row) => [row.slug, row] as const)); + // The TTL only matters when a loaded plugin actually lists a live remote + // catalog; otherwise skip it so age alone never widens the stale query. + const anyRemoteCatalog = Array.from(runtimes.values()).some( + (runtime) => runtime.plugin.remoteToolCatalog === true, + ); + const cutoff = + toolsSyncTtlMs == null || !anyRemoteCatalog ? null : Date.now() - toolsSyncTtlMs; + + // Bound the scan to potentially-stale rows: stale-marked (NULL stamp) or + // synced before the latest instant any trigger could fire at (the TTL + // cutoff / the newest config revision). Per-row trigger checks below + // re-verify against each row's own integration; in steady state this + // query returns nothing and the read pays one indexed lookup. + const latestRevision = integrations.reduce( + (max, row) => + row.config_revised_at == null + ? max + : Math.max(max ?? Number(row.config_revised_at), Number(row.config_revised_at)), + null, + ); + const staleBefore = + cutoff === null && latestRevision === null ? null - : Number(integrationRow.config_revised_at); - - const staleMarked = syncedAt === null; - const configRevised = revisedTime !== null && (syncedAt ?? 0) < revisedTime; - const expired = - cutoff !== null && - runtime.plugin.remoteToolCatalog === true && - syncedAt !== null && - syncedAt < cutoff; - if (!staleMarked && !configRevised && !expired) continue; + : Math.max( + cutoff ?? Number.MIN_SAFE_INTEGER, + latestRevision ?? Number.MIN_SAFE_INTEGER, + ); - rebuilds.push( - produceConnectionTools( - integrationRow, - { - owner: connection.owner as Owner, - integration: IntegrationSlug.make(connection.integration), - name: ConnectionName.make(connection.name), - }, - "background", - ).pipe( - // Best-effort, but never silent: the read still succeeds on the - // stale-but-working catalog and the peer rebuilds still finish, - // while the operator gets the connection that failed and why. - // Without this a connection whose upstream is permanently broken - // re-fails on every read and leaves no trace anywhere. - Effect.catch((error) => - Effect.logWarning("executor stale tool sync failed", { - integration: connection.integration, - connection: connection.name, - error: describeSyncFailure(error), - }).pipe(Effect.as([] as readonly Tool[])), - ), - Effect.withSpan("executor.tools.sync_stale", { - attributes: { - "executor.integration": connection.integration, - "executor.connection": connection.name, + const connections = yield* core.findMany("connection", { + where: (b: AnyCb) => + staleBefore === null + ? b.isNull("tools_synced_at") + : b.or(b.isNull("tools_synced_at"), b("tools_synced_at", "<", staleBefore)), + }); + // Each rebuild is an independent upstream listing, so they run together + // rather than one after another: a host with many stale remote-catalog + // connections otherwise pays the sum of every server's latency on the + // read that trips the TTL. Only the listings overlap — `persistCatalog` + // keeps the catalog writes in a single-file queue, so this fan-out never + // opens two transactions on a one-connection database. + // + // Two urgency classes. A stale-MARKED or config-revised catalog is known + // wrong (the upstream said so, or the integration's config changed), so + // the read waits for it within the grace budget. A catalog that is only + // older than the TTL is stale-but-working: in bounded mode its rebuild + // runs entirely in the background and the read answers from the + // persisted rows at once. Without that split every read after the TTL + // paid the grace budget for MCP listings it had no reason to wait on. + const urgent: Effect.Effect[] = []; + const deferred: Effect.Effect[] = []; + for (const connection of connections) { + const integrationRow = integrationBySlug.get(connection.integration); + if (!integrationRow) continue; + const runtime = runtimes.get(integrationRow.plugin_id); + // Only re-produce catalogs this executor can actually re-list — + // rebuilding under an unloaded plugin would clear a working catalog. + // (A loaded plugin without `resolveTools` still flows through: + // `produceConnectionTools` runs its clear-and-stamp cleanup path.) + if (!runtime) continue; + + const syncedAt = + connection.tools_synced_at == null ? null : Number(connection.tools_synced_at); + const revisedTime = + integrationRow.config_revised_at == null + ? null + : Number(integrationRow.config_revised_at); + + const staleMarked = syncedAt === null; + const configRevised = revisedTime !== null && (syncedAt ?? 0) < revisedTime; + const expired = + cutoff !== null && + runtime.plugin.remoteToolCatalog === true && + syncedAt !== null && + syncedAt < cutoff; + if (!staleMarked && !configRevised && !expired) continue; + + (staleMarked || configRevised || mode === "converge" ? urgent : deferred).push( + produceConnectionTools( + integrationRow, + { + owner: connection.owner as Owner, + integration: IntegrationSlug.make(connection.integration), + name: ConnectionName.make(connection.name), }, - }), - ), - ); - } - yield* Effect.all(rebuilds, { - concurrency: STALE_TOOLS_SYNC_CONCURRENCY, + "background", + ).pipe( + // Best-effort, but never silent: the read still succeeds on the + // stale-but-working catalog and the peer rebuilds still finish, + // while the operator gets the connection that failed and why. + // Without this a connection whose upstream is permanently broken + // re-fails on every read and leaves no trace anywhere. + Effect.catch((error) => + Effect.logWarning("executor stale tool sync failed", { + integration: connection.integration, + connection: connection.name, + error: describeSyncFailure(error), + }).pipe(Effect.as([] as readonly Tool[])), + ), + Effect.withSpan("executor.tools.sync_stale", { + attributes: { + "executor.integration": connection.integration, + "executor.connection": connection.name, + }, + }), + ), + ); + } + if (deferred.length > 0) { + const background = yield* Effect.forkDetach( + Effect.all(deferred, { concurrency: STALE_TOOLS_SYNC_CONCURRENCY }), + ); + config.waitUntil?.( + new Promise((resolve) => background.addObserver(() => resolve(undefined))), + ); + } + yield* Effect.all(urgent, { + concurrency: STALE_TOOLS_SYNC_CONCURRENCY, + }); }); - }); // How long a tools read waits for the stale sync before answering from // the persisted rows (`ExecutorConfig.toolsSyncGraceMs`; `null` blocks @@ -5648,51 +5676,97 @@ export const createExecutor = - Effect.gen(function* () { - const fiber = yield* Effect.forkDetach( - syncStaleConnectionTools.pipe( - Effect.catch((error) => - Effect.logWarning("executor stale tool sync scan failed", { - error: describeSyncFailure(error), - }), + const startStaleSync = Effect.gen(function* () { + const fiber = yield* Effect.forkDetach( + syncStaleConnectionTools("bounded").pipe( + Effect.catch((error) => + Effect.logWarning("executor stale tool sync scan failed", { + error: describeSyncFailure(error), + }), + ), + ), + ); + // On hosts that cancel request-scoped I/O once the response settles + // (Cloudflare Workers), hand the host the rebuilds' completion so the + // catalog still converges after the read stops waiting. + config.waitUntil?.( + new Promise((resolve) => fiber.addObserver(() => resolve(undefined))), + ); + return fiber; + }); + + // Restrict a tool-row read to the prefixes an allowlist policy source can + // reach. `null` = no restriction; `false` = nothing reachable, skip the + // read. Any unbounded prefix (a bare `*`, or wildcards in every position) + // makes the whole scope unrestricted. + const dynamicScopeCondition = ( + ruleSet: ActivePolicyRuleSet, + ): ((b: AnyCb) => Condition | boolean) | null | false => { + if (ruleSet.kind !== "prepared" || ruleSet.dynamicScope === undefined) return null; + const scopes = ruleSet.dynamicScope; + if (scopes.length === 0) return false; + if (scopes.some(isUnboundedDynamicToolScope)) return null; + return (b: AnyCb) => + b.or( + ...scopes.map((scope) => + b.and( + scope.integration === null ? true : b("integration", "=", scope.integration), + scope.owner === null ? true : b("owner", "=", scope.owner), + scope.connection === null ? true : b("connection", "=", scope.connection), ), ), ); - // On hosts that cancel request-scoped I/O once the response settles - // (Cloudflare Workers), hand the host the rebuilds' completion so the - // catalog still converges after the read stops waiting. - config.waitUntil?.( - new Promise((resolve) => fiber.addObserver(() => resolve(undefined))), - ); - yield* Fiber.await(fiber).pipe(Effect.timeoutOption(graceMs), Effect.asVoid); - }); + }; - const toolsList = (filter?: ToolListFilter): Effect.Effect => + // `awaitStaleSync: false` still starts the bounded background sync (so + // catalogs converge for sessions that only ever list connections) but + // answers without waiting on it. Visibility-only readers (which + // connections and integrations exist under the active policy) use it: a + // stale catalog does not change which connection a tool belongs to, so + // gating those reads on upstream MCP listings only added latency to every + // session start. In strict (`null` grace) mode the wait is unconditional. + const readTools = ( + filter: ToolListFilter | undefined, + options: { readonly awaitStaleSync: boolean }, + ): Effect.Effect => Effect.gen(function* () { + let syncFiber: Fiber.Fiber | null = null; if (toolsSyncGraceMs === null) { - yield* syncStaleConnectionTools; + yield* syncStaleConnectionTools("converge"); } else { - yield* awaitStaleSyncWithinGrace(toolsSyncGraceMs); + syncFiber = yield* startStaleSync; } + // Fetch the policy snapshot while the sync runs: the scope it carries + // decides which rows to read at all. + const policyRules = yield* listActivePolicyRuleSet(); + if (syncFiber && options.awaitStaleSync) { + yield* Fiber.await(syncFiber).pipe( + Effect.timeoutOption(toolsSyncGraceMs ?? 0), + Effect.asVoid, + ); + } + const scopeCondition = dynamicScopeCondition(policyRules); // Projected: the list surface is metadata (address, description, // annotations) — loading every tool's input/output schema JSON made // an unbounded list scale with schema bytes, not tool count. - const rows = yield* core.findMany("tool", { - where: (b: AnyCb) => - b.and( - filter?.integration === undefined - ? true - : b("integration", "=", String(filter.integration)), - filter?.owner === undefined ? true : b("owner", "=", filter.owner), - filter?.connection === undefined - ? true - : b("connection", "=", String(filter.connection)), - ), - select: TOOL_INVOCATION_COLUMNS, - }); + const rows = + scopeCondition === false + ? [] + : yield* core.findMany("tool", { + where: (b: AnyCb) => + b.and( + filter?.integration === undefined + ? true + : b("integration", "=", String(filter.integration)), + filter?.owner === undefined ? true : b("owner", "=", filter.owner), + filter?.connection === undefined + ? true + : b("connection", "=", String(filter.connection)), + scopeCondition === null ? true : scopeCondition(b), + ), + select: TOOL_INVOCATION_COLUMNS, + }); const includeBlocked = filter?.includeBlocked ?? false; - const policyRules = yield* listActivePolicyRuleSet(); // Only tools whose integration is still in the catalog. A tool row // whose integration was removed is an orphan (a removal that could // not reach this subject's rows): listing it invites an invoke that @@ -5729,6 +5803,9 @@ export const createExecutor = => + readTools(filter, { awaitStaleSync: true }); + const toolSchema = ( address: ToolAddress, ): Effect.Effect => diff --git a/packages/core/sdk/src/index.ts b/packages/core/sdk/src/index.ts index 56d7d5b777..627d3de1de 100644 --- a/packages/core/sdk/src/index.ts +++ b/packages/core/sdk/src/index.ts @@ -192,7 +192,10 @@ export { export { matchPattern, isValidPattern, + dynamicToolScopeForPattern, + isUnboundedDynamicToolScope, effectivePolicyFromSorted, + type DynamicToolScope, ToolPolicyActionSchema, type ToolPolicy, type CreateToolPolicyInput, @@ -388,6 +391,7 @@ export { type AnyPlugin, type StorageDeps, type OwnerBinding, + type PreparedToolPolicy, type ToolPolicyProvider, type ToolPolicyProviderRule, type IntegrationRecord, diff --git a/packages/core/sdk/src/plugin.ts b/packages/core/sdk/src/plugin.ts index 2ace32f891..e079f5ab8a 100644 --- a/packages/core/sdk/src/plugin.ts +++ b/packages/core/sdk/src/plugin.ts @@ -55,6 +55,7 @@ import type { CredentialProvider, ProviderEntry } from "./provider"; import type { PluginStorageConfig, PluginStorageFacade } from "./plugin-storage"; import type { CreateToolPolicyInput, + DynamicToolScope, EffectivePolicy, RemoveToolPolicyInput, ToolPolicy, @@ -131,13 +132,25 @@ export interface ToolPolicyProvider { * requests), so caching on it would serve stale policy state. Each operation * gets a fresh snapshot. */ - readonly prepare?: () => Effect.Effect< - (input: { - readonly toolId: string; - readonly defaultRequiresApproval?: boolean; - }) => EffectivePolicy, - StorageFailure - >; + readonly prepare?: () => Effect.Effect; +} + +/** What `ToolPolicyProvider.prepare` hands core for one operation. */ +export interface PreparedToolPolicy { + /** Pure resolver over the snapshot `prepare` fetched. */ + readonly resolve: (input: { + readonly toolId: string; + readonly defaultRequiresApproval?: boolean; + }) => EffectivePolicy; + /** + * The dynamic-tool prefixes this policy source can ever approve. When set, + * core restricts the tool rows it loads on a list to these prefixes instead + * of reading the whole catalog and blocking most of it in memory — the read + * then scales with the allowlist, not the workspace. An empty array means no + * dynamic tool is reachable. Omit when the source is not an allowlist (any + * row may be approved) so core keeps the unrestricted read. + */ + readonly dynamicScope?: readonly DynamicToolScope[]; } // --------------------------------------------------------------------------- diff --git a/packages/core/sdk/src/policies.test.ts b/packages/core/sdk/src/policies.test.ts index 18c5d29a72..98af2a65e7 100644 --- a/packages/core/sdk/src/policies.test.ts +++ b/packages/core/sdk/src/policies.test.ts @@ -16,6 +16,7 @@ import { ElicitationResponse, type ElicitationHandler } from "./elicitation"; import { createExecutor } from "./executor"; import type { FumaDb } from "./fuma-runtime"; import { + dynamicToolScopeForPattern, effectivePolicyFromSorted, isValidPattern, matchPattern, @@ -108,6 +109,56 @@ describe("isValidPattern", () => { }); }); +describe("dynamicToolScopeForPattern", () => { + it("reads the connection prefix out of subtree patterns", () => { + expect(dynamicToolScopeForPattern("github.org.main.*")).toEqual({ + integration: "github", + owner: "org", + connection: "main", + }); + expect(dynamicToolScopeForPattern("github.org.*")).toEqual({ + integration: "github", + owner: "org", + connection: null, + }); + expect(dynamicToolScopeForPattern("github.*")).toEqual({ + integration: "github", + owner: null, + connection: null, + }); + }); + + it("treats a mid-segment wildcard as any value for that position", () => { + expect(dynamicToolScopeForPattern("github.*.*.repos.list")).toEqual({ + integration: "github", + owner: null, + connection: null, + }); + expect(dynamicToolScopeForPattern("github.user.*.repos.*")).toEqual({ + integration: "github", + owner: "user", + connection: null, + }); + }); + + it("is unbounded for the universal pattern", () => { + expect(dynamicToolScopeForPattern("*")).toEqual({ + integration: null, + owner: null, + connection: null, + }); + }); + + it("yields no scope for patterns that can only reach static tools", () => { + // Exact ids shorter than a dynamic address. + expect(dynamicToolScopeForPattern("github")).toBeNull(); + expect(dynamicToolScopeForPattern("github.org.main")).toBeNull(); + // A literal owner that is neither org nor user is a static namespace. + expect(dynamicToolScopeForPattern("executor.coreTools.*")).toBeNull(); + expect(dynamicToolScopeForPattern("executor.coreTools.connections.list")).toBeNull(); + }); +}); + describe("resolveToolPolicy", () => { // v2: policy rows carry `owner` (org|user) instead of a scope id. const ROW = ( @@ -714,6 +765,100 @@ describe("active tool-policy provider", () => { ); }); +describe("prepared tool policy provider with a dynamic scope", () => { + const scopedProviderPlugin = ( + dynamicScope: readonly { + integration: string | null; + owner: string | null; + connection: string | null; + }[], + ) => + definePlugin(() => ({ + id: "scoped-policy-provider" as const, + storage: () => ({}), + toolPolicyProvider: () => ({ + list: () => Effect.succeed([]), + // Approves everything it is asked about: only the scope decides what + // core reads, so anything missing from the list was never loaded. + prepare: () => + Effect.succeed({ + resolve: () => ({ action: "approve" as const, source: "user" as const, pattern: "*" }), + dynamicScope, + }), + }), + }))(); + + const setupScoped = ( + dynamicScope: readonly { + integration: string | null; + owner: string | null; + connection: string | null; + }[], + ) => + makeTestExecutor({ + plugins: [policyTestPlugin(), scopedProviderPlugin(dynamicScope)] as const, + }).pipe( + Effect.tap((executor) => + Effect.gen(function* () { + yield* executor.ptest.seed(); + for (const integration of [VERCEL, GITHUB]) { + yield* executor.connections.create({ + owner: "org", + name: CONN, + integration, + template: TEMPLATE, + value: "v", + }); + } + }), + ), + ); + + const dynamicAddresses = (tools: readonly { address: unknown; static?: boolean }[]) => + tools + .filter((tool) => !tool.static) + .map((tool) => String(tool.address)) + .sort(); + + it.effect("restricts the list to the scoped connection", () => + Effect.gen(function* () { + const executor = yield* setupScoped([ + { integration: String(VERCEL), owner: "org", connection: String(CONN) }, + ]); + const tools = yield* executor.tools.list(); + expect(dynamicAddresses(tools)).toEqual([ + String(addr(VERCEL, "delete")), + String(addr(VERCEL, "deploy")), + ]); + const connections = yield* executor.connections.list(); + expect(connections.map((connection) => String(connection.integration))).toEqual([ + String(VERCEL), + ]); + }), + ); + + it.effect("a wildcard position widens the scope to every value", () => + Effect.gen(function* () { + const executor = yield* setupScoped([{ integration: null, owner: "org", connection: null }]); + const tools = yield* executor.tools.list(); + expect(dynamicAddresses(tools)).toEqual([ + String(addr(GITHUB, "list")), + String(addr(VERCEL, "delete")), + String(addr(VERCEL, "deploy")), + ]); + }), + ); + + it.effect("an empty scope reads no dynamic rows", () => + Effect.gen(function* () { + const executor = yield* setupScoped([]); + const tools = yield* executor.tools.list(); + expect(dynamicAddresses(tools)).toEqual([]); + expect(yield* executor.connections.list()).toEqual([]); + }), + ); +}); + describe("approve / require_approval interaction with annotations", () => { it.effect("approve skips the elicitation prompt even when plugin requires approval", () => Effect.gen(function* () { diff --git a/packages/core/sdk/src/policies.ts b/packages/core/sdk/src/policies.ts index 8620d9c6d3..b9e1f1ecad 100644 --- a/packages/core/sdk/src/policies.ts +++ b/packages/core/sdk/src/policies.ts @@ -120,6 +120,49 @@ export const isValidPattern = (pattern: string): boolean => { return true; }; +// --------------------------------------------------------------------------- +// Dynamic-tool scope — the (integration, owner, connection) prefix a pattern +// can reach. Lets a policy source that is an allowlist (a toolkit) narrow the +// tool rows core loads to the connections the allowlist names, instead of +// walking the whole catalog and blocking almost all of it in memory. +// --------------------------------------------------------------------------- + +/** One reachable prefix of a dynamic tool id `integration.owner.connection.tool`. + * `null` in a position means any value. All three `null` = unbounded. */ +export interface DynamicToolScope { + readonly integration: string | null; + readonly owner: string | null; + readonly connection: string | null; +} + +export const isUnboundedDynamicToolScope = (scope: DynamicToolScope): boolean => + scope.integration === null && scope.owner === null && scope.connection === null; + +/** + * The prefix of dynamic tool ids a pattern can match, or `null` when it can + * match none. A dynamic tool id has at least four segments, so an exact + * pattern shorter than that reaches only static tools. A trailing `*` covers + * every deeper segment; a mid-pattern `*` covers exactly that segment. + */ +export const dynamicToolScopeForPattern = (pattern: string): DynamicToolScope | null => { + if (pattern === "*") return { integration: null, owner: null, connection: null }; + const segments = pattern.split("."); + const subtree = segments.at(-1) === "*"; + if (!subtree && segments.length < 4) return null; + const at = (index: number): string | null => { + const segment = segments[index]; + if (segment === undefined) return null; + // Only the trailing `*` reaches past its own position; a mid `*` is one + // segment, which is also "any value" for that position. + return segment === "*" ? null : segment; + }; + const owner = at(1); + // A dynamic tool id's owner segment is always `org` or `user`; any other + // literal there names a static namespace (`executor.coreTools.*`). + if (owner !== null && owner !== "org" && owner !== "user") return null; + return { integration: at(0), owner, connection: at(2) }; +}; + // --------------------------------------------------------------------------- // Resolution — each owner contributes its first matching rule by local // position; the most restrictive matched action across owners wins. Caller diff --git a/packages/plugins/mcp/src/sdk/catalog-sync.test.ts b/packages/plugins/mcp/src/sdk/catalog-sync.test.ts index f3328876b3..34ec9f24f8 100644 --- a/packages/plugins/mcp/src/sdk/catalog-sync.test.ts +++ b/packages/plugins/mcp/src/sdk/catalog-sync.test.ts @@ -14,7 +14,7 @@ // --------------------------------------------------------------------------- import { describe, expect, it } from "@effect/vitest"; -import { Deferred, Effect, Fiber, Option, Ref, Schema } from "effect"; +import { Deferred, Effect, Fiber, Option, Ref, Schedule, Schema } from "effect"; import { HttpServerResponse } from "effect/unstable/http"; import { @@ -133,11 +133,13 @@ describe("MCP tool-catalog sync (end-to-end)", () => { }), ); - it.effect("expired catalogs re-list on read once older than the freshness TTL", () => + // Live clock: the rebuild is real I/O against the test server, so the poll + // must advance on wall time rather than the test clock. + it.live("expired catalogs re-list in the background once older than the freshness TTL", () => Effect.gen(function* () { const mutable = makeMutableCatalogMcpServer(); const server = yield* serveMcpServer(mutable.factory); - // Everything is instantly stale — every tools read re-lists. + // Everything is instantly stale — every tools read starts a re-list. const executor = yield* makeCatalogTestExecutor(server.url, { toolsSyncTtlMs: 0 }); expect(toolNames(yield* executor.tools.list())).toContain(mutable.initialToolName); @@ -145,7 +147,17 @@ describe("MCP tool-catalog sync (end-to-end)", () => { // Server-side change with no notification and no executor signal at all. mutable.renameTool(); - const refreshed = toolNames(yield* executor.tools.list()); + // A time-expired catalog is stale-but-working: the read answers from the + // persisted rows without waiting on the upstream listing, and a later + // read observes the rebuilt catalog. + const refreshed = yield* executor.tools.list().pipe( + Effect.map(toolNames), + Effect.repeat({ + until: (names) => names.includes(mutable.renamedToolName), + schedule: Schedule.spaced("20 millis"), + }), + Effect.timeout("5 seconds"), + ); expect(refreshed).toContain(mutable.renamedToolName); expect(refreshed).not.toContain(mutable.initialToolName); }), diff --git a/packages/plugins/toolkits/src/server.test.ts b/packages/plugins/toolkits/src/server.test.ts index bab67eb0e9..3db165f6e6 100644 --- a/packages/plugins/toolkits/src/server.test.ts +++ b/packages/plugins/toolkits/src/server.test.ts @@ -132,6 +132,50 @@ describe("toolkitsPlugin", () => { }), ); + it.effect("prepares a dynamic scope from the toolkit's access patterns", () => + Effect.gen(function* () { + const executor = yield* makeTestExecutor({ + plugins: [toolkitsPlugin()] as const, + }); + + const orgKit = yield* executor.toolkits.create({ owner: "org", name: "Org Kit" }); + for (const pattern of [ + "github.org.main.*", + "slack.*", + "linear.*.*.issues.list", + "github.user.alice.*", + "executor.coreTools.*", + ]) { + yield* executor.toolkits.createConnection(orgKit.id, { pattern }); + } + const prepared = yield* executor.toolkits.preparePolicyResolverForSlug(orgKit.slug); + // An org toolkit never reaches personal rows: unowned prefixes pin to + // org, user-only prefixes drop, and static-only patterns contribute none. + const byIntegration = ( + a: { integration: string | null }, + b: { integration: string | null }, + ) => String(a.integration).localeCompare(String(b.integration)); + expect([...(prepared.dynamicScope ?? [])].sort(byIntegration)).toEqual([ + { integration: "github", owner: "org", connection: "main" }, + { integration: "linear", owner: "org", connection: null }, + { integration: "slack", owner: "org", connection: null }, + ]); + expect(prepared.resolve({ toolId: "github.org.main.repos.list" }).action).toBe("approve"); + expect(prepared.resolve({ toolId: "github.user.alice.repos.list" }).action).toBe("block"); + + const personalKit = yield* executor.toolkits.create({ owner: "user", name: "Me Kit" }); + yield* executor.toolkits.createConnection(personalKit.id, { pattern: "github.user.alice.*" }); + const personal = yield* executor.toolkits.preparePolicyResolverForSlug(personalKit.slug); + expect(personal.dynamicScope).toEqual([ + { integration: "github", owner: "user", connection: "alice" }, + ]); + + const missing = yield* executor.toolkits.preparePolicyResolverForSlug("no-such-kit"); + expect(missing.dynamicScope).toEqual([]); + expect(missing.resolve({ toolId: "github.org.main.repos.list" }).action).toBe("block"); + }), + ); + it.effect("treats a persisted connection-root approve as an access policy", () => Effect.gen(function* () { const executor = yield* makeTestExecutor({ diff --git a/packages/plugins/toolkits/src/server.ts b/packages/plugins/toolkits/src/server.ts index 7dacc3e456..eef15de797 100644 --- a/packages/plugins/toolkits/src/server.ts +++ b/packages/plugins/toolkits/src/server.ts @@ -2,16 +2,19 @@ import { Context, definePlugin, definePluginStorageCollection, + dynamicToolScopeForPattern, Effect, HttpApiBuilder, isValidPattern, matchPattern, Schema, + type DynamicToolScope, type EffectivePolicy, type Owner, type PluginCtx, type PluginStorageFacade, type PluginStorageCollectionFacade, + type PreparedToolPolicy, type StorageFailure, type ToolPolicyAction, type ToolPolicyProvider, @@ -150,22 +153,42 @@ const isLegacyConnectionPolicy = (policy: ToolkitPolicyRecord): boolean => { return parts.at(-1) === "*" && (parts.length === 3 || parts.length === 4); }; -const resolveToolkitPolicy = ( - toolId: string, +// The toolkit's rules, digested once so resolving a tool is a scan over +// already-sorted patterns. A tools list resolves every candidate row against +// the same snapshot, so the legacy split and the sort must not be redone per +// tool. +interface ToolkitRuleSnapshot { + /** Patterns granting access: connection records plus legacy approve rows. */ + readonly accessPatterns: readonly string[]; + /** Non-legacy policies in precedence order. */ + readonly orderedPolicies: readonly ToolkitPolicyRecord[]; +} + +const digestToolkitRules = ( connections: readonly ToolkitConnectionRecord[], policies: readonly ToolkitPolicyRecord[], +): ToolkitRuleSnapshot => { + const legacyPolicyIds = legacyConnectionPolicyIds(policies, connections); + return { + accessPatterns: [ + ...connections.map((connection) => connection.pattern), + ...policies.filter((policy) => legacyPolicyIds.has(policy.id)).map((p) => p.pattern), + ], + orderedPolicies: policies + .filter((policy) => !legacyPolicyIds.has(policy.id)) + .sort(comparePositioned), + }; +}; + +const resolveToolkitPolicy = ( + toolId: string, + rules: ToolkitRuleSnapshot, defaultRequiresApproval?: boolean, ): EffectivePolicy => { - const legacyPolicyIds = legacyConnectionPolicyIds(policies, connections); - const connected = - connections.some((connection) => matchPattern(connection.pattern, toolId)) || - policies.some( - (policy) => legacyPolicyIds.has(policy.id) && matchPattern(policy.pattern, toolId), - ); + const connected = rules.accessPatterns.some((pattern) => matchPattern(pattern, toolId)); if (!connected) return blockedPolicy(); - for (const policy of [...policies].sort(comparePositioned)) { - if (legacyPolicyIds.has(policy.id)) continue; + for (const policy of rules.orderedPolicies) { if (!matchPattern(policy.pattern, toolId)) continue; return { action: policy.action, @@ -177,6 +200,28 @@ const resolveToolkitPolicy = ( return pluginDefaultPolicy(defaultRequiresApproval); }; +// The dynamic-tool prefixes the access patterns can reach. An org toolkit +// never grants personal tools, so its prefixes are pinned to org rows and +// user-only prefixes drop out; the per-tool check still enforces the same +// rule for anything the prefix cannot express. +const toolkitDynamicScope = ( + rules: ToolkitRuleSnapshot, + isOrg: boolean, +): readonly DynamicToolScope[] => { + const scopes: DynamicToolScope[] = []; + for (const pattern of rules.accessPatterns) { + const scope = dynamicToolScopeForPattern(pattern); + if (!scope) continue; + if (!isOrg) { + scopes.push(scope); + continue; + } + if (scope.owner === "user") continue; + scopes.push(scope.owner === null ? { ...scope, owner: "org" } : scope); + } + return scopes; +}; + const legacyConnectionPolicyIds = ( policies: readonly ToolkitPolicyRecord[], connections: readonly ToolkitConnectionRecord[], @@ -507,38 +552,36 @@ const makeToolkitsExtension = (ctx: PluginCtx) => { if (toolkit.owner === "org" && isPersonalDynamicToolId(toolId)) return blockedPolicy(); const policies = yield* listPoliciesForRecord(toolkit.data.id); const connections = yield* listConnectionsForRecord(toolkit.data.id); - return resolveToolkitPolicy(toolId, connections, policies, defaultRequiresApproval); + return resolveToolkitPolicy( + toolId, + digestToolkitRules(connections, policies), + defaultRequiresApproval, + ); }); // Batched form of `resolvePolicyForSlug`: fetch the toolkit, its policies, and // its connections ONCE, then hand back a pure resolver core can run for every - // tool in a single tools/list or tools/call. `resolvePolicyForSlug` re-fetches - // policies + connections on every tool, which is the per-tool N+1 that scales - // with the whole catalog on the list surface. This is byte-for-byte the same - // resolution, just hoisted out of the loop. + // tool in a single tools/list or tools/call, plus the prefixes those rules + // can reach so core reads only the toolkit's rows instead of the whole + // catalog. `resolvePolicyForSlug` re-fetches policies + connections on every + // tool, which is the per-tool N+1 that scales with the whole catalog on the + // list surface. This is the same resolution, hoisted out of the loop. const preparePolicyResolverForSlug = ( slug: string, - ): Effect.Effect< - (input: { - readonly toolId: string; - readonly defaultRequiresApproval?: boolean; - }) => EffectivePolicy, - StorageFailure - > => + ): Effect.Effect => Effect.gen(function* () { const toolkit = yield* getBySlugEntry(slug); - if (!toolkit) return () => blockedPolicy(); + if (!toolkit) return { resolve: () => blockedPolicy(), dynamicScope: [] }; const isOrg = toolkit.owner === "org"; const policies = yield* listPoliciesForRecord(toolkit.data.id); const connections = yield* listConnectionsForRecord(toolkit.data.id); - return (input: { readonly toolId: string; readonly defaultRequiresApproval?: boolean }) => { - if (isOrg && isPersonalDynamicToolId(input.toolId)) return blockedPolicy(); - return resolveToolkitPolicy( - input.toolId, - connections, - policies, - input.defaultRequiresApproval, - ); + const rules = digestToolkitRules(connections, policies); + return { + resolve: (input) => { + if (isOrg && isPersonalDynamicToolId(input.toolId)) return blockedPolicy(); + return resolveToolkitPolicy(input.toolId, rules, input.defaultRequiresApproval); + }, + dynamicScope: toolkitDynamicScope(rules, isOrg), }; });