import { Effect, FiberRef } from "effect";import {  StorageError,  typedAdapter,  type DBAdapter,  type DBSchema,  type DBTransactionAdapter,  type StorageFailure,  type TypedAdapter,} from "@executor/storage-core"; import {  scopeBlobStore,  type BlobStore,} from "./blob";import {  coreSchema,  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 {  NoHandlerError,  PluginNotLoadedError,  SourceRemovalNotAllowedError,  ToolInvocationError,  ToolNotFoundError,} from "./errors";import { 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[] = []> = {  readonly scope: 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>;    readonly remove: (id: string) => Effect.Effect<void, StorageFailure>;    readonly list: () => Effect.Effect<readonly SecretRef[], StorageFailure>;    readonly providers: () => Effect.Effect<readonly string[]>;  };   readonly close: () => Effect.Effect<void, StorageFailure>;} & PluginExtensions<TPlugins>; export interface ExecutorConfig<  TPlugins extends readonly AnyPlugin[] = [],> {  readonly scope: 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,  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,  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);     const now = new Date();    yield* core.create({      model: "source",      data: {        id: input.id,        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}`,          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,      });    }  }); const deleteSourceById = (  core: TypedAdapter<CoreSchema>,  sourceId: string,): Effect.Effect<void, StorageFailure> =>  Effect.gen(function* () {    yield* core.deleteMany({      model: "tool",      where: [{ field: "source_id", value: sourceId }],    });    yield* core.deleteMany({      model: "definition",      where: [{ field: "source_id", value: sourceId }],    });    yield* core.delete({      model: "source",      where: [{ field: "id", value: sourceId }],    });  }); const writeDefinitions = (  core: TypedAdapter<CoreSchema>,  pluginId: string,  input: DefinitionsInput,): Effect.Effect<void, StorageFailure> =>  Effect.gen(function* () {    yield* core.deleteMany({      model: "definition",      where: [{ field: "source_id", value: input.sourceId }],    });    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}`,        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,); // A `DBAdapter` whose methods dispatch to the active adapter (tx handle or// root) on every call. Stable identity for consumers (plugin storage,// `typedAdapter`, etc.) — they see one adapter object, but the routing is// decided at call time via the FiberRef above.const buildAdapterRouter = (root: DBAdapter): DBAdapter => {  const pick = <A, E>(    use: (active: DBTransactionAdapter) => Effect.Effect<A, E>,  ): Effect.Effect<A, E> =>    Effect.flatMap(FiberRef.get(activeAdapterRef), (active) =>      use((active ?? (root as DBTransactionAdapter))),    );   return {    id: root.id,    create: (data) => pick((a) => a.create(data)),    createMany: (data) => pick((a) => a.createMany(data)),    findOne: (data) => pick((a) => a.findOne(data)),    findMany: (data) => pick((a) => a.findMany(data)),    count: (data) => pick((a) => a.count(data)),    update: (data) => pick((a) => a.update(data)),    updateMany: (data) => pick((a) => a.updateMany(data)),    delete: (data) => pick((a) => a.delete(data)),    deleteMany: (data) => pick((a) => a.deleteMany(data)),    // transaction() always opens a real boundary on the ROOT adapter so the    // tx uses one real connection from the pool. If we're already inside a    // parent tx (FiberRef set), skip opening a nested sql.begin — that's    // the postgres.js + Hyperdrive deadlock path — and just run the    // callback with the existing tx handle. In both cases the callback    // sees a FiberRef-substituted adapter so further nested writes thread    // through.    transaction: (callback) =>      Effect.flatMap(FiberRef.get(activeAdapterRef), (active) => {        if (active) return callback(active);        return root.transaction((trx) =>          Effect.locally(callback(trx), activeAdapterRef, trx),        );      }),  } as DBAdapter;}; // ---------------------------------------------------------------------------// createExecutor// --------------------------------------------------------------------------- interface StaticTools {  readonly source: StaticSourceDecl;  readonly tool: StaticToolDecl;  readonly pluginId: string;  readonly ctx: PluginCtx<unknown>;} interface StaticSources {  readonly source: StaticSourceDecl;  readonly pluginId: string;} interface PluginRuntime {  readonly plugin: AnyPlugin;  readonly storage: unknown;  readonly ctx: PluginCtx<unknown>;} export const createExecutor = <  const TPlugins extends readonly AnyPlugin[] = [],>(  config: ExecutorConfig<TPlugins>,): Effect.Effect<Executor<TPlugins>, Error> =>  Effect.gen(function* () {    const {      scope,      adapter: rootAdapter,      blobs,      plugins = [] as unknown as TPlugins,    } = config;     // Scope-wrap the root adapter so every read on a tenant-scoped table    // filters by the current scope stack and every write stamps the    // write target. Today the stack has one element; the adapter's    // `ScopeContext` shape already accepts an ordered list so layering    // (org → workspace → user) can land later without changing plugin    // code. Only tables whose schema declares `scope_id` are scoped.    const schema = collectSchemas(plugins);    const scopedRoot = scopeAdapter(      rootAdapter,      { read: [scope.id], write: scope.id },      schema,    );    const adapter = buildAdapterRouter(scopedRoot);    const core = typedAdapter<CoreSchema>(adapter);     // Populated once, never mutated after startup.    const staticTools = new Map<string, StaticTools>();    const staticSources = new Map<string, StaticSources>();     // Per-plugin runtime state.    const runtimes = new Map<string, PluginRuntime>();

Lines 1–500 of 1334. Load more