diff --git a/packages/core/sdk/src/executor-plugin-runtime.ts b/packages/core/sdk/src/executor-plugin-runtime.ts new file mode 100644 index 000000000..d0d969959 --- /dev/null +++ b/packages/core/sdk/src/executor-plugin-runtime.ts @@ -0,0 +1,323 @@ +import { Effect, type Layer } from "effect"; +import { FetchHttpClient, type HttpClient } from "effect/unstable/http"; + +import { type BlobStore, pluginBlobStore } from "./blob"; +import type { ConnectionProvider } from "./connections"; +import type { CredentialBindingsFacade } from "./credential-bindings"; +import type { DefinitionsInput, SourceInput } from "./core-schema"; +import { + EXECUTOR_SOURCE, + EXECUTOR_SOURCE_ID, + byScopedId, + deleteSourceById, + makeCoreDb, + pluginStorageFailure, + writeDefinitions, + writeSourceInput, +} from "./executor-helpers"; +import { StorageError, makeFumaClient, type FumaDb, type StorageFailure } from "./fuma-runtime"; +import type { OAuthService } from "./oauth"; +import type { AnyPlugin, PluginCtx, StaticSourceDecl, StaticToolDecl, StorageDeps } from "./plugin"; +import type { Scope } from "./scope"; +import type { SecretProvider } from "./secrets"; +import type { RemoveSourceInput } from "./types"; + +export interface ExecutorStaticTool { + readonly source: StaticSourceDecl; + readonly tool: StaticToolDecl; + readonly pluginId: string; + readonly ctx: PluginCtx; +} + +export interface ExecutorStaticSource { + readonly source: StaticSourceDecl; + readonly pluginId: string; +} + +export interface ExecutorPluginRuntime { + readonly plugin: AnyPlugin; + readonly storage: unknown; + readonly ctx: PluginCtx; +} + +export const registerExecutorPlugins = (deps: { + readonly plugins: readonly AnyPlugin[]; + readonly scopes: readonly Scope[]; + readonly rootDb: FumaDb; + readonly blobs: BlobStore; + readonly scopeIds: readonly string[]; + readonly core: ReturnType; + readonly staticTools: Map; + readonly staticSources: Map; + readonly runtimes: Map; + readonly secretProviders: Map; + readonly connectionProviders: Map; + readonly extensions: Record; + readonly transaction: (effect: Effect.Effect) => Effect.Effect; + readonly assertScopeInStack: ( + label: string, + scopeId: string, + ) => Effect.Effect; + readonly httpClientLayer?: Layer.Layer; + readonly secrets: PluginCtx["secrets"]; + readonly connections: PluginCtx["connections"]; + readonly credentialBindings: CredentialBindingsFacade; + readonly oauth: OAuthService; +}): Effect.Effect => + Effect.gen(function* () { + const { + plugins, + scopes, + rootDb, + blobs, + scopeIds, + core, + staticTools, + staticSources, + runtimes, + secretProviders, + connectionProviders, + extensions, + transaction, + assertScopeInStack, + httpClientLayer, + secrets, + connections, + credentialBindings, + oauth, + } = deps; + + const secretsGet = secrets.get; + const secretsGetAtScope = secrets.getAtScope; + const secretsListForCtx = secrets.list; + const secretsSet = secrets.set; + const secretsRemove = secrets.remove; + const connectionsGet = connections.get; + const connectionsGetAtScope = connections.getAtScope; + const connectionsListForCtx = connections.list; + const connectionsCreate = connections.create; + const connectionsUpdateTokens = connections.updateTokens; + const connectionsSetIdentityLabel = connections.setIdentityLabel; + const connectionsAccessToken = connections.accessToken; + const connectionsAccessTokenAtScope = connections.accessTokenAtScope; + const connectionsRemove = connections.remove; + + // ------------------------------------------------------------------ + // Plugin wiring — build ctx, run extension, populate static pools, + // register secret providers. No adapter reads here. + // ------------------------------------------------------------------ + for (const plugin of plugins) { + if (runtimes.has(plugin.id)) { + return yield* new StorageError({ + message: `Duplicate plugin id: ${plugin.id}`, + cause: undefined, + }); + } + + const pluginFuma = makeFumaClient( + rootDb, + plugin.schema ? { tables: new Set(Object.keys(plugin.schema)) } : { tables: new Set() }, + ); + const storageDeps: StorageDeps = { + scopes, + fuma: pluginFuma, + // Blob keys are namespaced by `/` so two tenants + // sharing a backing BlobStore can't collide or leak on the + // same `(plugin, key)` pair. The store's `get`/`has` walk the + // scope stack (innermost first); `put`/`delete` require the + // plugin to name a target scope explicitly. + blobs: pluginBlobStore(blobs, scopeIds, plugin.id), + }; + const storage = plugin.storage(storageDeps); + + const ctx: PluginCtx = { + scopes, + storage, + httpClientLayer: httpClientLayer ?? FetchHttpClient.layer, + core: { + sources: { + register: (input: SourceInput) => + Effect.gen(function* () { + // Guard: reject a dynamic source whose id collides with + // a static source id, or any of whose would-be tool ids + // collide with a static tool id. Tool ids are + // `${source_id}.${tool.name}` — static and dynamic + // share the same string space. Fails as `StorageError` + // so the HTTP edge surfaces it as `InternalError(traceId)`. + if (staticSources.has(input.id)) { + return yield* new StorageError({ + message: `Source id "${input.id}" collides with a static source`, + cause: undefined, + }); + } + for (const tool of input.tools) { + const fqid = `${input.id}.${tool.name}`; + if (staticTools.has(fqid)) { + return yield* new StorageError({ + message: `Tool id "${fqid}" collides with a static tool`, + cause: undefined, + }); + } + } + yield* transaction(writeSourceInput(core, plugin.id, input)); + }), + unregister: (input: RemoveSourceInput) => + // `unregister` is scoped to a caller-named source row. The + // plugin already knows which source owner it is updating, + // so the core path must not infer an innermost target. + transaction( + Effect.gen(function* () { + yield* assertScopeInStack("source unregister targetScope", input.targetScope); + const row = yield* core.findFirst("source", { + where: byScopedId(input.targetScope, input.id), + }); + if (!row) return; + yield* deleteSourceById(core, input.id, input.targetScope); + }), + ), + update: (input) => + core + .updateMany("source", { + where: byScopedId(input.scope, input.id), + set: { + ...(input.name !== undefined ? { name: input.name } : {}), + ...(input.url !== undefined ? { url: input.url ?? null } : {}), + updated_at: new Date(), + }, + }) + .pipe(Effect.asVoid), + }, + definitions: { + register: (input: DefinitionsInput) => + transaction(writeDefinitions(core, plugin.id, input)), + }, + }, + secrets: { + get: (id) => secretsGet(id), + getAtScope: (id, scope) => secretsGetAtScope(id, scope), + list: () => secretsListForCtx(), + set: (input) => secretsSet(input), + remove: (input) => secretsRemove(input), + }, + connections: { + get: (id) => connectionsGet(id), + getAtScope: (id, scope) => connectionsGetAtScope(id, scope), + list: () => connectionsListForCtx(), + create: (input) => connectionsCreate(input), + updateTokens: (input) => connectionsUpdateTokens(input), + setIdentityLabel: (id, label) => connectionsSetIdentityLabel(id, label), + accessToken: (id) => connectionsAccessToken(id), + accessTokenAtScope: (id, scope) => connectionsAccessTokenAtScope(id, scope), + remove: (input) => connectionsRemove(input), + }, + credentialBindings, + oauth: oauth, + transaction: (effect: Effect.Effect) => transaction(effect), + }; + + // Build extension FIRST so it's available as `self` when resolving + // staticSources. Field ordering in the plugin spec matters — TS + // infers TExtension from `extension`'s return type, then NoInfer + // locks `self` to that inferred type on `staticSources`. + const extension: object = plugin.extension ? plugin.extension(ctx) : {}; + if (plugin.extension) { + extensions[plugin.id] = extension; + } + + // Resolve static declarations to the in-memory pools. NO DB WRITES. + // Plugin-owned executor tools are intentionally mounted under the + // single `executor` namespace so source inventory is about configured + // integrations, not plugin management surfaces: + // openapi.addSource -> executor.openapi.addSource + const decls = plugin.staticSources ? plugin.staticSources(extension) : []; + for (const source of decls) { + const mountUnderExecutor = source.kind === "executor" && source.id === plugin.id; + const mountedSource = mountUnderExecutor ? EXECUTOR_SOURCE : source; + + if (mountUnderExecutor) { + if (!staticSources.has(EXECUTOR_SOURCE_ID)) { + staticSources.set(EXECUTOR_SOURCE_ID, { + source: EXECUTOR_SOURCE, + pluginId: EXECUTOR_SOURCE_ID, + }); + } + } else { + if (staticSources.has(source.id)) { + return yield* new StorageError({ + message: `Duplicate static source id: ${source.id} (plugin ${plugin.id})`, + cause: undefined, + }); + } + staticSources.set(source.id, { source, pluginId: plugin.id }); + } + + for (const tool of source.tools) { + const mountedTool = mountUnderExecutor + ? { + ...tool, + name: `${plugin.id}.${tool.name}`, + } + : tool; + const fqid = `${mountedSource.id}.${mountedTool.name}`; + if (staticTools.has(fqid)) { + return yield* new StorageError({ + message: `Duplicate static tool id: ${fqid} (plugin ${plugin.id})`, + cause: undefined, + }); + } + staticTools.set(fqid, { + source: mountedSource, + tool: mountedTool, + pluginId: plugin.id, + ctx, + }); + } + } + + runtimes.set(plugin.id, { plugin, storage, ctx }); + + if (plugin.secretProviders) { + const raw = + typeof plugin.secretProviders === "function" + ? plugin.secretProviders(ctx) + : plugin.secretProviders; + const providers = Effect.isEffect(raw) + ? yield* raw.pipe( + Effect.mapError((cause) => pluginStorageFailure(plugin.id, "secretProviders", cause)), + ) + : raw; + for (const provider of providers) { + if (secretProviders.has(provider.key)) { + return yield* new StorageError({ + message: `Duplicate secret provider key: ${provider.key} (from plugin ${plugin.id})`, + cause: undefined, + }); + } + secretProviders.set(provider.key, provider); + } + } + + if (plugin.connectionProviders) { + const raw = + typeof plugin.connectionProviders === "function" + ? plugin.connectionProviders(ctx) + : plugin.connectionProviders; + const providers = Effect.isEffect(raw) + ? yield* raw.pipe( + Effect.mapError((cause) => + pluginStorageFailure(plugin.id, "connectionProviders", cause), + ), + ) + : raw; + for (const provider of providers) { + if (connectionProviders.has(provider.key)) { + return yield* new StorageError({ + message: `Duplicate connection provider key: ${provider.key} (from plugin ${plugin.id})`, + cause: undefined, + }); + } + connectionProviders.set(provider.key, provider); + } + } + } + }); diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index ca49460b5..ab978152d 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -1,5 +1,5 @@ import { Duration, Effect, Layer } from "effect"; -import { FetchHttpClient, type HttpClient } from "effect/unstable/http"; +import type { HttpClient } from "effect/unstable/http"; import { withQueryContext } from "fumadb/query"; import type { OAuthEndpointUrlPolicy } from "./oauth-helpers"; import { @@ -10,7 +10,7 @@ import { type StorageFailure, } from "./fuma-runtime"; -import { makeFumaBlobStore, pluginBlobStore } from "./blob"; +import { makeFumaBlobStore } from "./blob"; import { ConnectionRef, ConnectionRefreshError, @@ -20,13 +20,7 @@ import { type UpdateConnectionTokensInput, } from "./connections"; import { type CredentialBindingsFacade } from "./credential-bindings"; -import { - type ConnectionRow, - type DefinitionsInput, - type SecretRow, - type SourceInput, - type SourceRow, -} from "./core-schema"; +import { type ConnectionRow, type SecretRow, type SourceRow } from "./core-schema"; import { ElicitationDeclinedError, ElicitationResponse, @@ -56,14 +50,7 @@ import { type ToolPolicy, type UpdateToolPolicyInput, } from "./policies"; -import type { - AnyPlugin, - PluginCtx, - PluginExtensions, - StaticSourceDecl, - StaticToolDecl, - StorageDeps, -} from "./plugin"; +import type { AnyPlugin, PluginExtensions } from "./plugin"; import type { Scope } from "./scope"; import { RemoveSecretInput, SecretRef, SetSecretInput, type SecretProvider } from "./secrets"; import { Usage } from "./usages"; @@ -79,22 +66,23 @@ import { import type { ExecutorScopePolicyContext } from "./scope-policy"; import { makeCredentialBindings } from "./executor-credential-bindings"; import { makeConnectionsFacade } from "./executor-connections"; +import { + registerExecutorPlugins, + type ExecutorPluginRuntime, + type ExecutorStaticSource, + type ExecutorStaticTool, +} from "./executor-plugin-runtime"; import { makeSecretsFacade } from "./executor-secrets"; import { makeExecutorSurface } from "./executor-surface"; import { - EXECUTOR_SOURCE, - EXECUTOR_SOURCE_ID, byScopedId, collectTables, createDefaultMemoryDb, - deleteSourceById, makeCoreDb, pluginStorageFailure, storageFailureFromUnknown, validateExecutorDbTables, validateExecutorScopePolicyTables, - writeDefinitions, - writeSourceInput, } from "./executor-helpers"; // --------------------------------------------------------------------------- @@ -352,24 +340,6 @@ export { collectTables }; // createExecutor // --------------------------------------------------------------------------- -interface StaticTools { - readonly source: StaticSourceDecl; - readonly tool: StaticToolDecl; - readonly pluginId: string; - readonly ctx: PluginCtx; -} - -interface StaticSources { - readonly source: StaticSourceDecl; - readonly pluginId: string; -} - -interface PluginRuntime { - readonly plugin: AnyPlugin; - readonly storage: unknown; - readonly ctx: PluginCtx; -} - export const createExecutor = ( config: ExecutorConfig, ): Effect.Effect, StorageFailure> => @@ -416,11 +386,11 @@ export const createExecutor = (effect: Effect.Effect) => fuma.transaction(effect); // Populated once, never mutated after startup. - const staticTools = new Map(); - const staticSources = new Map(); + const staticTools = new Map(); + const staticSources = new Map(); // Per-plugin runtime state. - const runtimes = new Map(); + const runtimes = new Map(); // Secret providers keyed by `provider.key`. const secretProviders = new Map(); // Connection providers keyed by `provider.key` — drive the refresh @@ -606,224 +576,43 @@ export const createExecutor = /` so two tenants - // sharing a backing BlobStore can't collide or leak on the - // same `(plugin, key)` pair. The store's `get`/`has` walk the - // scope stack (innermost first); `put`/`delete` require the - // plugin to name a target scope explicitly. - blobs: pluginBlobStore(blobs, scopeIds, plugin.id), - }; - const storage = plugin.storage(storageDeps); - - const ctx: PluginCtx = { - scopes, - storage, - httpClientLayer: config.httpClientLayer ?? FetchHttpClient.layer, - core: { - sources: { - register: (input: SourceInput) => - Effect.gen(function* () { - // Guard: reject a dynamic source whose id collides with - // a static source id, or any of whose would-be tool ids - // collide with a static tool id. Tool ids are - // `${source_id}.${tool.name}` — static and dynamic - // share the same string space. Fails as `StorageError` - // so the HTTP edge surfaces it as `InternalError(traceId)`. - if (staticSources.has(input.id)) { - return yield* new StorageError({ - message: `Source id "${input.id}" collides with a static source`, - cause: undefined, - }); - } - for (const tool of input.tools) { - const fqid = `${input.id}.${tool.name}`; - if (staticTools.has(fqid)) { - return yield* new StorageError({ - message: `Tool id "${fqid}" collides with a static tool`, - cause: undefined, - }); - } - } - yield* transaction(writeSourceInput(core, plugin.id, input)); - }), - unregister: (input: RemoveSourceInput) => - // `unregister` is scoped to a caller-named source row. The - // plugin already knows which source owner it is updating, - // so the core path must not infer an innermost target. - transaction( - Effect.gen(function* () { - yield* assertScopeInStack("source unregister targetScope", input.targetScope); - const row = yield* core.findFirst("source", { - where: byScopedId(input.targetScope, input.id), - }); - if (!row) return; - yield* deleteSourceById(core, input.id, input.targetScope); - }), - ), - update: (input) => - core - .updateMany("source", { - where: byScopedId(input.scope, input.id), - set: { - ...(input.name !== undefined ? { name: input.name } : {}), - ...(input.url !== undefined ? { url: input.url ?? null } : {}), - updated_at: new Date(), - }, - }) - .pipe(Effect.asVoid), - }, - definitions: { - register: (input: DefinitionsInput) => - transaction(writeDefinitions(core, plugin.id, input)), - }, - }, - secrets: { - get: (id) => secretsGet(id), - getAtScope: (id, scope) => secretsGetAtScope(id, scope), - list: () => secretsListForCtx(), - set: (input) => secretsSet(input), - remove: (input) => secretsRemove(input), - }, - connections: { - get: (id) => connectionsGet(id), - getAtScope: (id, scope) => connectionsGetAtScope(id, scope), - list: () => connectionsListForCtx(), - create: (input) => connectionsCreate(input), - updateTokens: (input) => connectionsUpdateTokens(input), - setIdentityLabel: (id, label) => connectionsSetIdentityLabel(id, label), - accessToken: (id) => connectionsAccessToken(id), - accessTokenAtScope: (id, scope) => connectionsAccessTokenAtScope(id, scope), - remove: (input) => connectionsRemove(input), - }, - credentialBindings, - oauth: oauthBundle.service, - transaction: (effect: Effect.Effect) => transaction(effect), - }; - - // Build extension FIRST so it's available as `self` when resolving - // staticSources. Field ordering in the plugin spec matters — TS - // infers TExtension from `extension`'s return type, then NoInfer - // locks `self` to that inferred type on `staticSources`. - const extension: object = plugin.extension ? plugin.extension(ctx) : {}; - if (plugin.extension) { - extensions[plugin.id] = extension; - } - - // Resolve static declarations to the in-memory pools. NO DB WRITES. - // Plugin-owned executor tools are intentionally mounted under the - // single `executor` namespace so source inventory is about configured - // integrations, not plugin management surfaces: - // openapi.addSource -> executor.openapi.addSource - const decls = plugin.staticSources ? plugin.staticSources(extension) : []; - for (const source of decls) { - const mountUnderExecutor = source.kind === "executor" && source.id === plugin.id; - const mountedSource = mountUnderExecutor ? EXECUTOR_SOURCE : source; - - if (mountUnderExecutor) { - if (!staticSources.has(EXECUTOR_SOURCE_ID)) { - staticSources.set(EXECUTOR_SOURCE_ID, { - source: EXECUTOR_SOURCE, - pluginId: EXECUTOR_SOURCE_ID, - }); - } - } else { - if (staticSources.has(source.id)) { - return yield* new StorageError({ - message: `Duplicate static source id: ${source.id} (plugin ${plugin.id})`, - cause: undefined, - }); - } - staticSources.set(source.id, { source, pluginId: plugin.id }); - } - - for (const tool of source.tools) { - const mountedTool = mountUnderExecutor - ? { - ...tool, - name: `${plugin.id}.${tool.name}`, - } - : tool; - const fqid = `${mountedSource.id}.${mountedTool.name}`; - if (staticTools.has(fqid)) { - return yield* new StorageError({ - message: `Duplicate static tool id: ${fqid} (plugin ${plugin.id})`, - cause: undefined, - }); - } - staticTools.set(fqid, { - source: mountedSource, - tool: mountedTool, - pluginId: plugin.id, - ctx, - }); - } - } - - runtimes.set(plugin.id, { plugin, storage, ctx }); - - if (plugin.secretProviders) { - const raw = - typeof plugin.secretProviders === "function" - ? plugin.secretProviders(ctx) - : plugin.secretProviders; - const providers = Effect.isEffect(raw) - ? yield* raw.pipe( - Effect.mapError((cause) => pluginStorageFailure(plugin.id, "secretProviders", cause)), - ) - : raw; - for (const provider of providers) { - if (secretProviders.has(provider.key)) { - return yield* new StorageError({ - message: `Duplicate secret provider key: ${provider.key} (from plugin ${plugin.id})`, - cause: undefined, - }); - } - secretProviders.set(provider.key, provider); - } - } - - if (plugin.connectionProviders) { - const raw = - typeof plugin.connectionProviders === "function" - ? plugin.connectionProviders(ctx) - : plugin.connectionProviders; - const providers = Effect.isEffect(raw) - ? yield* raw.pipe( - Effect.mapError((cause) => - pluginStorageFailure(plugin.id, "connectionProviders", cause), - ), - ) - : raw; - for (const provider of providers) { - if (connectionProviders.has(provider.key)) { - return yield* new StorageError({ - message: `Duplicate connection provider key: ${provider.key} (from plugin ${plugin.id})`, - cause: undefined, - }); - } - connectionProviders.set(provider.key, provider); - } - } - } + yield* registerExecutorPlugins({ + plugins, + scopes, + rootDb, + blobs, + scopeIds, + core, + staticTools, + staticSources, + runtimes, + secretProviders, + connectionProviders, + extensions, + transaction, + assertScopeInStack, + httpClientLayer: config.httpClientLayer, + secrets: { + get: secretsGet, + getAtScope: secretsGetAtScope, + list: secretsListForCtx, + set: secretsSet, + remove: secretsRemove, + }, + connections: { + get: connectionsGet, + getAtScope: connectionsGetAtScope, + list: connectionsListForCtx, + create: connectionsCreate, + updateTokens: connectionsUpdateTokens, + setIdentityLabel: connectionsSetIdentityLabel, + accessToken: connectionsAccessToken, + accessTokenAtScope: connectionsAccessTokenAtScope, + remove: connectionsRemove, + }, + credentialBindings, + oauth: oauthBundle.service, + }); // ------------------------------------------------------------------ // Executor surface