import { Deferred, Effect, FiberRef, Option, Schema } from "effect";import { StorageError, typedAdapter, type DBAdapter, type DBSchema, type DBTransactionAdapter, type StorageFailure, type TypedAdapter,} from "@executor/storage-core"; import { pluginBlobStore, type BlobStore,} from "./blob";import { ConnectionProviderState, ConnectionRef, ConnectionRefreshError, type ConnectionProvider, type CreateConnectionInput, type UpdateConnectionTokensInput,} from "./connections";import { coreSchema, type ConnectionRow, type CoreSchema, type DefinitionsInput, type SourceInput, type SourceRow, type ToolAnnotations, type ToolRow,} from "./core-schema";import { ElicitationDeclinedError, ElicitationResponse, FormElicitation, type ElicitationHandler, type ElicitationRequest,} from "./elicitation";import { ConnectionNotFoundError, ConnectionProviderNotRegisteredError, ConnectionReauthRequiredError, ConnectionRefreshNotSupportedError, NoHandlerError, PluginNotLoadedError, SecretOwnedByConnectionError, SourceRemovalNotAllowedError, ToolInvocationError, ToolNotFoundError,} from "./errors";import { ConnectionId, ScopeId, SecretId, ToolId } from "./ids";import type { AnyPlugin, Elicit, PluginCtx, PluginExtensions, StaticSourceDecl, StaticToolDecl, StorageDeps,} from "./plugin";import type { Scope } from "./scope";import { SecretRef, SetSecretInput, type SecretProvider,} from "./secrets";import { ToolSchema, type Source, type SourceDetectionResult, type Tool, type ToolListFilter,} from "./types";import { buildToolTypeScriptPreview } from "./schema-types";import { scopeAdapter } from "./scoped-adapter"; // ---------------------------------------------------------------------------// InvokeOptions — passed to `executor.tools.invoke(id, args, options)`.// The `onElicitation` handler is threaded into the `elicit` function// exposed on plugin ctx / InvokeToolInput. Tools that never elicit// simply don't call it.//// The "accept-all" sentinel is convenient for tests and CLI automation —// every elicitation request gets auto-accepted with an empty content// payload. For real interactive hosts, pass a real handler.// --------------------------------------------------------------------------- export interface InvokeOptions { readonly onElicitation?: ElicitationHandler | "accept-all";} const acceptAllHandler: ElicitationHandler = () => Effect.succeed(new ElicitationResponse({ action: "accept" })); const resolveElicitationHandler = ( options: InvokeOptions | undefined,): ElicitationHandler => { const handler = options?.onElicitation; if (!handler || handler === "accept-all") return acceptAllHandler; return handler;}; // ---------------------------------------------------------------------------// Executor — public surface. Every list/invoke/schema call is a direct// core-table query (for dynamic rows) unioned with the in-memory static// pool. No ToolRegistry, no SourceRegistry, no SecretStore services.// --------------------------------------------------------------------------- export type Executor<TPlugins extends readonly AnyPlugin[] = []> = { /** * Precedence-ordered scope stack this executor was configured with. * Innermost first. Consumers that need "the display scope" typically * pick `scopes.at(-1)` (outermost, e.g. the organization) or * `scopes[0]` (innermost, e.g. the current user-in-org) depending on * what they're rendering. */ readonly scopes: readonly Scope[]; readonly tools: { readonly list: ( filter?: ToolListFilter, ) => Effect.Effect<readonly Tool[], StorageFailure>; /** Fetch a tool's full schema view: JSON schemas with `$defs` * attached from the core `definition` table, plus TypeScript * preview strings rendered from them. Returns `null` for unknown * tool ids. */ readonly schema: ( toolId: string, ) => Effect.Effect<ToolSchema | null, StorageFailure>; /** Every `$defs` entry across every source, grouped by source id. * Used for bulk schema export and downstream TypeScript rendering. */ readonly definitions: () => Effect.Effect< Record<string, Record<string, unknown>>, StorageFailure >; readonly invoke: ( toolId: string, args: unknown, options?: InvokeOptions, ) => Effect.Effect< unknown, | ToolNotFoundError | PluginNotLoadedError | NoHandlerError | ToolInvocationError | ElicitationDeclinedError | StorageFailure >; }; readonly sources: { readonly list: () => Effect.Effect<readonly Source[], StorageFailure>; readonly remove: ( sourceId: string, ) => Effect.Effect<void, SourceRemovalNotAllowedError | StorageFailure>; readonly refresh: (sourceId: string) => Effect.Effect<void, StorageFailure>; /** URL autodetection — fans out to every plugin's `detect` hook * (if declared), returns every high/medium/low-confidence match. * UI picks a winner from the list. */ readonly detect: ( url: string, ) => Effect.Effect<readonly SourceDetectionResult[], StorageFailure>; /** All `$defs` registered for a single source, keyed by def name. */ readonly definitions: ( sourceId: string, ) => Effect.Effect<Record<string, unknown>, StorageFailure>; }; readonly secrets: { readonly get: ( id: string, ) => Effect.Effect<string | null, StorageFailure>; /** Fast-path existence check — hits the core `secret` routing table * only, never calls the provider. Use this for UI state ("secret * missing, prompt to add") to avoid keychain permission prompts * or 1password IPC roundtrips on a pre-flight check. */ readonly status: ( id: string, ) => Effect.Effect<"resolved" | "missing", StorageFailure>; readonly set: ( input: SetSecretInput, ) => Effect.Effect<SecretRef, StorageFailure>; /** Delete a bare (non-connection-owned) secret. Connection-owned * secrets are rejected with `SecretOwnedByConnectionError` — use * `connections.remove` instead. */ readonly remove: ( id: string, ) => Effect.Effect<void, SecretOwnedByConnectionError | StorageFailure>; readonly list: () => Effect.Effect<readonly SecretRef[], StorageFailure>; readonly providers: () => Effect.Effect<readonly string[]>; }; readonly connections: { readonly get: ( id: string, ) => Effect.Effect<ConnectionRef | null, StorageFailure>; readonly list: () => Effect.Effect<readonly ConnectionRef[], StorageFailure>; readonly create: ( input: CreateConnectionInput, ) => Effect.Effect< ConnectionRef, ConnectionProviderNotRegisteredError | StorageFailure >; readonly updateTokens: ( input: UpdateConnectionTokensInput, ) => Effect.Effect< ConnectionRef, ConnectionNotFoundError | StorageFailure >; readonly setIdentityLabel: ( id: string, label: string | null, ) => Effect.Effect<void, ConnectionNotFoundError | StorageFailure>; readonly accessToken: ( id: string, ) => Effect.Effect< string, | ConnectionNotFoundError | ConnectionProviderNotRegisteredError | ConnectionRefreshNotSupportedError | ConnectionReauthRequiredError | ConnectionRefreshError | StorageFailure >; readonly remove: (id: string) => Effect.Effect<void, StorageFailure>; readonly providers: () => Effect.Effect<readonly string[]>; }; readonly close: () => Effect.Effect<void, StorageFailure>;} & PluginExtensions<TPlugins>; export interface ExecutorConfig< TPlugins extends readonly AnyPlugin[] = [],> { /** * Precedence-ordered scope stack. Innermost first; typical shape is * `[userInOrgScope, orgScope]`. Reads on scoped tables walk the * stack (first hit wins for shadow-by-id consumers like secrets and * blobs); writes require callers to name an explicit target scope. * Must be non-empty. */ readonly scopes: readonly Scope[]; readonly adapter: DBAdapter; readonly blobs: BlobStore; readonly plugins?: TPlugins;} // ---------------------------------------------------------------------------// collectSchemas — merge coreSchema with every plugin's declared schema.// Hosts call this and pass the result to the migration runner (or to// the adapter factory for backends that auto-migrate from a schema// manifest) before constructing the executor.// --------------------------------------------------------------------------- export const collectSchemas = ( plugins: readonly AnyPlugin[],): DBSchema => { const merged: Record<string, DBSchema[string]> = { ...coreSchema }; for (const plugin of plugins) { if (!plugin.schema) continue; for (const [modelKey, model] of Object.entries(plugin.schema)) { if (merged[modelKey]) { throw new Error( `Duplicate model "${modelKey}" contributed by plugin "${plugin.id}"` + ` (reserved by core or another plugin)`, ); } merged[modelKey] = model as DBSchema[string]; } } return merged;}; // ---------------------------------------------------------------------------// Row → public projection conversions// --------------------------------------------------------------------------- const rowToSource = (row: SourceRow): Source => ({ id: row.id, scopeId: row.scope_id, kind: row.kind, name: row.name, url: row.url ?? undefined, pluginId: row.plugin_id, canRemove: Boolean(row.can_remove), canRefresh: Boolean(row.can_refresh), canEdit: Boolean(row.can_edit), runtime: false,}); const staticDeclToSource = ( decl: StaticSourceDecl, pluginId: string,): Source => ({ id: decl.id, scopeId: undefined, kind: decl.kind, name: decl.name, url: decl.url, pluginId, canRemove: decl.canRemove ?? false, canRefresh: decl.canRefresh ?? false, canEdit: decl.canEdit ?? false, runtime: true,}); const decodeJsonColumn = (value: unknown): unknown => { if (value === null || value === undefined) return undefined; if (typeof value !== "string") return value; try { return JSON.parse(value); } catch { return value; }}; const rowToTool = ( row: ToolRow, annotations?: ToolAnnotations,): Tool => ({ id: row.id, sourceId: row.source_id, pluginId: row.plugin_id, name: row.name, description: row.description, inputSchema: decodeJsonColumn(row.input_schema), outputSchema: decodeJsonColumn(row.output_schema), annotations,}); const staticDeclToTool = ( source: StaticSourceDecl, tool: StaticToolDecl, pluginId: string,): Tool => ({ id: `${source.id}.${tool.name}`, sourceId: source.id, pluginId, name: tool.name, description: tool.description, inputSchema: tool.inputSchema, outputSchema: tool.outputSchema, annotations: tool.annotations,}); // ---------------------------------------------------------------------------// Dynamic-row writers — used by ctx.core.sources.register. Static sources// never touch these functions.// --------------------------------------------------------------------------- // Upsert shape: delete any existing source + tools + definitions for// `input.id` before creating fresh rows. Keeps replayable — boot-time// sync from executor.jsonc can call register() on rows that already// exist without tripping a UNIQUE constraint.const writeSourceInput = ( core: TypedAdapter<CoreSchema>, pluginId: string, input: SourceInput,): Effect.Effect<void, StorageFailure> => Effect.gen(function* () { yield* deleteSourceById(core, input.id, input.scope); const now = new Date(); yield* core.create({ model: "source", data: { id: input.id, scope_id: input.scope, plugin_id: pluginId, kind: input.kind, name: input.name, url: input.url ?? undefined, can_remove: input.canRemove ?? true, can_refresh: input.canRefresh ?? false, can_edit: input.canEdit ?? false, created_at: now, updated_at: now, }, forceAllowId: true, }); if (input.tools.length > 0) { yield* core.createMany({ model: "tool", data: input.tools.map((tool) => ({ id: `${input.id}.${tool.name}`, scope_id: input.scope, source_id: input.id, plugin_id: pluginId, name: tool.name, description: tool.description, input_schema: tool.inputSchema ?? undefined, output_schema: tool.outputSchema ?? undefined, created_at: now, updated_at: now, })), forceAllowId: true, }); } }); // Delete a source and its tools + definitions at ONE specific scope.// The scoped adapter already narrows reads/writes to the executor's// stack via `scope_id IN (...)`, but we pin `scope_id = scopeId` here// so this helper never widens into a stack-wide wipe — a bystander// scope's rows with a colliding `source_id` must survive.const deleteSourceById = ( core: TypedAdapter<CoreSchema>, sourceId: string, scopeId: string,): Effect.Effect<void, StorageFailure> => Effect.gen(function* () { yield* core.deleteMany({ model: "tool", where: [ { field: "source_id", value: sourceId }, { field: "scope_id", value: scopeId }, ], }); yield* core.deleteMany({ model: "definition", where: [ { field: "source_id", value: sourceId }, { field: "scope_id", value: scopeId }, ], }); yield* core.delete({ model: "source", where: [ { field: "id", value: sourceId }, { field: "scope_id", value: scopeId }, ], }); }); const writeDefinitions = ( core: TypedAdapter<CoreSchema>, pluginId: string, input: DefinitionsInput,): Effect.Effect<void, StorageFailure> => Effect.gen(function* () { // Pin the delete to `input.scope` — without this, the scoped // adapter's `scope_id IN (stack)` injection would nuke definitions // at outer scopes whenever an inner-scope writer re-registers // definitions for the same source id. yield* core.deleteMany({ model: "definition", where: [ { field: "source_id", value: input.sourceId }, { field: "scope_id", value: input.scope }, ], }); const entries = Object.entries(input.definitions); if (entries.length === 0) return; const now = new Date(); yield* core.createMany({ model: "definition", data: entries.map(([name, schema]) => ({ id: `${input.sourceId}.${name}`, scope_id: input.scope, source_id: input.sourceId, plugin_id: pluginId, name, schema: schema as Record<string, unknown>, created_at: now, })), forceAllowId: true, }); }); // ---------------------------------------------------------------------------// Filtering — shared between dynamic (DB) and static (in-memory) pools// so `tools.list({ query, sourceId })` matches across both.// --------------------------------------------------------------------------- const toolMatchesFilter = (tool: Tool, filter: ToolListFilter): boolean => { if (filter.sourceId && tool.sourceId !== filter.sourceId) return false; if (filter.query) { const q = filter.query.toLowerCase(); const hay = `${tool.name} ${tool.description}`.toLowerCase(); if (!hay.includes(q)) return false; } return true;}; // ---------------------------------------------------------------------------// Active-adapter FiberRef. Nested plugin writes read this ref so that// `ctx.transaction` can swap in a tx-bound adapter handle without changing// the ctx shape — the root adapter returned by `buildAdapterRouter` below// resolves its target per call, so any Effect running inside// `Effect.locally(_, activeAdapterRef, trx)` automatically routes every// query through the same sql.begin connection. This is what makes nested// writes atomic on postgres + Hyperdrive without deadlocking a pool of 1.// ---------------------------------------------------------------------------const activeAdapterRef = FiberRef.unsafeMake<DBTransactionAdapter | null>( null,); Lines 1–500 of 2410. Load more
import { Deferred, Effect, FiberRef, Option, Schema } from "effect"; import { StorageError, typedAdapter, type DBAdapter, type DBSchema, type DBTransactionAdapter, type StorageFailure, type TypedAdapter, } from "@executor/storage-core"; import { pluginBlobStore, type BlobStore, } from "./blob"; import { ConnectionProviderState, ConnectionRef, ConnectionRefreshError, type ConnectionProvider, type CreateConnectionInput, type UpdateConnectionTokensInput, } from "./connections"; import { coreSchema, type ConnectionRow, type CoreSchema, type DefinitionsInput, type SourceInput, type SourceRow, type ToolAnnotations, type ToolRow, } from "./core-schema"; import { ElicitationDeclinedError, ElicitationResponse, FormElicitation, type ElicitationHandler, type ElicitationRequest, } from "./elicitation"; import { ConnectionNotFoundError, ConnectionProviderNotRegisteredError, ConnectionReauthRequiredError, ConnectionRefreshNotSupportedError, NoHandlerError, PluginNotLoadedError, SecretOwnedByConnectionError, SourceRemovalNotAllowedError, ToolInvocationError, ToolNotFoundError, } from "./errors"; import { ConnectionId, ScopeId, SecretId, ToolId } from "./ids"; import type { AnyPlugin, Elicit, PluginCtx, PluginExtensions, StaticSourceDecl, StaticToolDecl, StorageDeps, } from "./plugin"; import type { Scope } from "./scope"; import { SecretRef, SetSecretInput, type SecretProvider, } from "./secrets"; import { ToolSchema, type Source, type SourceDetectionResult, type Tool, type ToolListFilter, } from "./types"; import { buildToolTypeScriptPreview } from "./schema-types"; import { scopeAdapter } from "./scoped-adapter"; // --------------------------------------------------------------------------- // InvokeOptions — passed to `executor.tools.invoke(id, args, options)`. // The `onElicitation` handler is threaded into the `elicit` function // exposed on plugin ctx / InvokeToolInput. Tools that never elicit // simply don't call it. // // The "accept-all" sentinel is convenient for tests and CLI automation — // every elicitation request gets auto-accepted with an empty content // payload. For real interactive hosts, pass a real handler. // --------------------------------------------------------------------------- export interface InvokeOptions { readonly onElicitation?: ElicitationHandler | "accept-all"; } const acceptAllHandler: ElicitationHandler = () => Effect.succeed(new ElicitationResponse({ action: "accept" })); const resolveElicitationHandler = ( options: InvokeOptions | undefined, ): ElicitationHandler => { const handler = options?.onElicitation; if (!handler || handler === "accept-all") return acceptAllHandler; return handler; }; // --------------------------------------------------------------------------- // Executor — public surface. Every list/invoke/schema call is a direct // core-table query (for dynamic rows) unioned with the in-memory static // pool. No ToolRegistry, no SourceRegistry, no SecretStore services. // --------------------------------------------------------------------------- export type Executor<TPlugins extends readonly AnyPlugin[] = []> = { /** * Precedence-ordered scope stack this executor was configured with. * Innermost first. Consumers that need "the display scope" typically * pick `scopes.at(-1)` (outermost, e.g. the organization) or * `scopes[0]` (innermost, e.g. the current user-in-org) depending on * what they're rendering. */ readonly scopes: readonly Scope[]; readonly tools: { readonly list: ( filter?: ToolListFilter, ) => Effect.Effect<readonly Tool[], StorageFailure>; /** Fetch a tool's full schema view: JSON schemas with `$defs` * attached from the core `definition` table, plus TypeScript * preview strings rendered from them. Returns `null` for unknown * tool ids. */ readonly schema: ( toolId: string, ) => Effect.Effect<ToolSchema | null, StorageFailure>; /** Every `$defs` entry across every source, grouped by source id. * Used for bulk schema export and downstream TypeScript rendering. */ readonly definitions: () => Effect.Effect< Record<string, Record<string, unknown>>, StorageFailure >; readonly invoke: ( toolId: string, args: unknown, options?: InvokeOptions, ) => Effect.Effect< unknown, | ToolNotFoundError | PluginNotLoadedError | NoHandlerError | ToolInvocationError | ElicitationDeclinedError | StorageFailure >; }; readonly sources: { readonly list: () => Effect.Effect<readonly Source[], StorageFailure>; readonly remove: ( sourceId: string, ) => Effect.Effect<void, SourceRemovalNotAllowedError | StorageFailure>; readonly refresh: (sourceId: string) => Effect.Effect<void, StorageFailure>; /** URL autodetection — fans out to every plugin's `detect` hook * (if declared), returns every high/medium/low-confidence match. * UI picks a winner from the list. */ readonly detect: ( url: string, ) => Effect.Effect<readonly SourceDetectionResult[], StorageFailure>; /** All `$defs` registered for a single source, keyed by def name. */ readonly definitions: ( sourceId: string, ) => Effect.Effect<Record<string, unknown>, StorageFailure>; }; readonly secrets: { readonly get: ( id: string, ) => Effect.Effect<string | null, StorageFailure>; /** Fast-path existence check — hits the core `secret` routing table * only, never calls the provider. Use this for UI state ("secret * missing, prompt to add") to avoid keychain permission prompts * or 1password IPC roundtrips on a pre-flight check. */ readonly status: ( id: string, ) => Effect.Effect<"resolved" | "missing", StorageFailure>; readonly set: ( input: SetSecretInput, ) => Effect.Effect<SecretRef, StorageFailure>; /** Delete a bare (non-connection-owned) secret. Connection-owned * secrets are rejected with `SecretOwnedByConnectionError` — use * `connections.remove` instead. */ readonly remove: ( id: string, ) => Effect.Effect<void, SecretOwnedByConnectionError | StorageFailure>; readonly list: () => Effect.Effect<readonly SecretRef[], StorageFailure>; readonly providers: () => Effect.Effect<readonly string[]>; }; readonly connections: { readonly get: ( id: string, ) => Effect.Effect<ConnectionRef | null, StorageFailure>; readonly list: () => Effect.Effect<readonly ConnectionRef[], StorageFailure>; readonly create: ( input: CreateConnectionInput, ) => Effect.Effect< ConnectionRef, ConnectionProviderNotRegisteredError | StorageFailure >; readonly updateTokens: ( input: UpdateConnectionTokensInput, ) => Effect.Effect< ConnectionRef, ConnectionNotFoundError | StorageFailure >; readonly setIdentityLabel: ( id: string, label: string | null, ) => Effect.Effect<void, ConnectionNotFoundError | StorageFailure>; readonly accessToken: ( id: string, ) => Effect.Effect< string, | ConnectionNotFoundError | ConnectionProviderNotRegisteredError | ConnectionRefreshNotSupportedError | ConnectionReauthRequiredError | ConnectionRefreshError | StorageFailure >; readonly remove: (id: string) => Effect.Effect<void, StorageFailure>; readonly providers: () => Effect.Effect<readonly string[]>; }; readonly close: () => Effect.Effect<void, StorageFailure>; } & PluginExtensions<TPlugins>; export interface ExecutorConfig< TPlugins extends readonly AnyPlugin[] = [], > { /** * Precedence-ordered scope stack. Innermost first; typical shape is * `[userInOrgScope, orgScope]`. Reads on scoped tables walk the * stack (first hit wins for shadow-by-id consumers like secrets and * blobs); writes require callers to name an explicit target scope. * Must be non-empty. */ readonly scopes: readonly Scope[]; readonly adapter: DBAdapter; readonly blobs: BlobStore; readonly plugins?: TPlugins; } // --------------------------------------------------------------------------- // collectSchemas — merge coreSchema with every plugin's declared schema. // Hosts call this and pass the result to the migration runner (or to // the adapter factory for backends that auto-migrate from a schema // manifest) before constructing the executor. // --------------------------------------------------------------------------- export const collectSchemas = ( plugins: readonly AnyPlugin[], ): DBSchema => { const merged: Record<string, DBSchema[string]> = { ...coreSchema }; for (const plugin of plugins) { if (!plugin.schema) continue; for (const [modelKey, model] of Object.entries(plugin.schema)) { if (merged[modelKey]) { throw new Error( `Duplicate model "${modelKey}" contributed by plugin "${plugin.id}"` + ` (reserved by core or another plugin)`, ); } merged[modelKey] = model as DBSchema[string]; } } return merged; }; // --------------------------------------------------------------------------- // Row → public projection conversions // --------------------------------------------------------------------------- const rowToSource = (row: SourceRow): Source => ({ id: row.id, scopeId: row.scope_id, kind: row.kind, name: row.name, url: row.url ?? undefined, pluginId: row.plugin_id, canRemove: Boolean(row.can_remove), canRefresh: Boolean(row.can_refresh), canEdit: Boolean(row.can_edit), runtime: false, }); const staticDeclToSource = ( decl: StaticSourceDecl, pluginId: string, ): Source => ({ id: decl.id, scopeId: undefined, kind: decl.kind, name: decl.name, url: decl.url, pluginId, canRemove: decl.canRemove ?? false, canRefresh: decl.canRefresh ?? false, canEdit: decl.canEdit ?? false, runtime: true, }); const decodeJsonColumn = (value: unknown): unknown => { if (value === null || value === undefined) return undefined; if (typeof value !== "string") return value; try { return JSON.parse(value); } catch { return value; } }; const rowToTool = ( row: ToolRow, annotations?: ToolAnnotations, ): Tool => ({ id: row.id, sourceId: row.source_id, pluginId: row.plugin_id, name: row.name, description: row.description, inputSchema: decodeJsonColumn(row.input_schema), outputSchema: decodeJsonColumn(row.output_schema), annotations, }); const staticDeclToTool = ( source: StaticSourceDecl, tool: StaticToolDecl, pluginId: string, ): Tool => ({ id: `${source.id}.${tool.name}`, sourceId: source.id, pluginId, name: tool.name, description: tool.description, inputSchema: tool.inputSchema, outputSchema: tool.outputSchema, annotations: tool.annotations, }); // --------------------------------------------------------------------------- // Dynamic-row writers — used by ctx.core.sources.register. Static sources // never touch these functions. // --------------------------------------------------------------------------- // Upsert shape: delete any existing source + tools + definitions for // `input.id` before creating fresh rows. Keeps replayable — boot-time // sync from executor.jsonc can call register() on rows that already // exist without tripping a UNIQUE constraint. const writeSourceInput = ( core: TypedAdapter<CoreSchema>, pluginId: string, input: SourceInput, ): Effect.Effect<void, StorageFailure> => Effect.gen(function* () { yield* deleteSourceById(core, input.id, input.scope); const now = new Date(); yield* core.create({ model: "source", data: { id: input.id, scope_id: input.scope, plugin_id: pluginId, kind: input.kind, name: input.name, url: input.url ?? undefined, can_remove: input.canRemove ?? true, can_refresh: input.canRefresh ?? false, can_edit: input.canEdit ?? false, created_at: now, updated_at: now, }, forceAllowId: true, }); if (input.tools.length > 0) { yield* core.createMany({ model: "tool", data: input.tools.map((tool) => ({ id: `${input.id}.${tool.name}`, scope_id: input.scope, source_id: input.id, plugin_id: pluginId, name: tool.name, description: tool.description, input_schema: tool.inputSchema ?? undefined, output_schema: tool.outputSchema ?? undefined, created_at: now, updated_at: now, })), forceAllowId: true, }); } }); // Delete a source and its tools + definitions at ONE specific scope. // The scoped adapter already narrows reads/writes to the executor's // stack via `scope_id IN (...)`, but we pin `scope_id = scopeId` here // so this helper never widens into a stack-wide wipe — a bystander // scope's rows with a colliding `source_id` must survive. const deleteSourceById = ( core: TypedAdapter<CoreSchema>, sourceId: string, scopeId: string, ): Effect.Effect<void, StorageFailure> => Effect.gen(function* () { yield* core.deleteMany({ model: "tool", where: [ { field: "source_id", value: sourceId }, { field: "scope_id", value: scopeId }, ], }); yield* core.deleteMany({ model: "definition", where: [ { field: "source_id", value: sourceId }, { field: "scope_id", value: scopeId }, ], }); yield* core.delete({ model: "source", where: [ { field: "id", value: sourceId }, { field: "scope_id", value: scopeId }, ], }); }); const writeDefinitions = ( core: TypedAdapter<CoreSchema>, pluginId: string, input: DefinitionsInput, ): Effect.Effect<void, StorageFailure> => Effect.gen(function* () { // Pin the delete to `input.scope` — without this, the scoped // adapter's `scope_id IN (stack)` injection would nuke definitions // at outer scopes whenever an inner-scope writer re-registers // definitions for the same source id. yield* core.deleteMany({ model: "definition", where: [ { field: "source_id", value: input.sourceId }, { field: "scope_id", value: input.scope }, ], }); const entries = Object.entries(input.definitions); if (entries.length === 0) return; const now = new Date(); yield* core.createMany({ model: "definition", data: entries.map(([name, schema]) => ({ id: `${input.sourceId}.${name}`, scope_id: input.scope, source_id: input.sourceId, plugin_id: pluginId, name, schema: schema as Record<string, unknown>, created_at: now, })), forceAllowId: true, }); }); // --------------------------------------------------------------------------- // Filtering — shared between dynamic (DB) and static (in-memory) pools // so `tools.list({ query, sourceId })` matches across both. // --------------------------------------------------------------------------- const toolMatchesFilter = (tool: Tool, filter: ToolListFilter): boolean => { if (filter.sourceId && tool.sourceId !== filter.sourceId) return false; if (filter.query) { const q = filter.query.toLowerCase(); const hay = `${tool.name} ${tool.description}`.toLowerCase(); if (!hay.includes(q)) return false; } return true; }; // --------------------------------------------------------------------------- // Active-adapter FiberRef. Nested plugin writes read this ref so that // `ctx.transaction` can swap in a tx-bound adapter handle without changing // the ctx shape — the root adapter returned by `buildAdapterRouter` below // resolves its target per call, so any Effect running inside // `Effect.locally(_, activeAdapterRef, trx)` automatically routes every // query through the same sql.begin connection. This is what makes nested // writes atomic on postgres + Hyperdrive without deadlocking a pool of 1. // --------------------------------------------------------------------------- const activeAdapterRef = FiberRef.unsafeMake<DBTransactionAdapter | null>( null, );