From 4eaf0ed480facc1ce5c913456a7828f181ad47ad Mon Sep 17 00:00:00 2001 From: MarioCadenas Date: Fri, 21 Aug 2026 17:12:40 +0200 Subject: [PATCH 1/8] refactor(appkit): extract pure agents-plugin helpers into sibling modules Step 1 of splitting the ~2.5k-line agents plugin. Moves module-scope pure functions/constants out of agents.ts verbatim (behavior-preserving): - approval.ts requiresApproval - prompt.ts composePromptForAgent - builtin-tools.ts LOAD_SKILL_TOOL_DEF, READ_SKILL_FILE_TOOL_DEF - adapter-extensions.ts buildAdapterExtensions, supervisorToolDescription, warnOnCapabilityMismatch agents.ts: 2512 -> 2331 lines. typecheck + 394 agent tests green. Signed-off-by: MarioCadenas --- .../src/plugins/agents/adapter-extensions.ts | 97 +++++++++ packages/appkit/src/plugins/agents/agents.ts | 203 +----------------- .../appkit/src/plugins/agents/approval.ts | 30 +++ .../src/plugins/agents/builtin-tools.ts | 40 ++++ packages/appkit/src/plugins/agents/prompt.ts | 43 ++++ 5 files changed, 219 insertions(+), 194 deletions(-) create mode 100644 packages/appkit/src/plugins/agents/adapter-extensions.ts create mode 100644 packages/appkit/src/plugins/agents/approval.ts create mode 100644 packages/appkit/src/plugins/agents/builtin-tools.ts create mode 100644 packages/appkit/src/plugins/agents/prompt.ts diff --git a/packages/appkit/src/plugins/agents/adapter-extensions.ts b/packages/appkit/src/plugins/agents/adapter-extensions.ts new file mode 100644 index 000000000..467fa07a4 --- /dev/null +++ b/packages/appkit/src/plugins/agents/adapter-extensions.ts @@ -0,0 +1,97 @@ +import type { AgentAdapter } from "shared"; + +import { + SUPERVISOR_EXTENSION_KEY, + type SupervisorTool, +} from "../../agents/supervisor-api"; +import type { ResolvedToolEntry } from "../../core/agent/types"; +import { createLogger } from "../../logging/logger"; + +const logger = createLogger("agents"); + +/** + * Pulls the LLM-readable description off any {@link SupervisorTool} kind. + * Used to populate the synthetic placeholder `def.description` on + * hosted-supervisor tool-index entries. + */ +export function supervisorToolDescription(spec: SupervisorTool): string { + switch (spec.type) { + case "genie_space": + return spec.genie_space.description; + case "uc_function": + return spec.uc_function.description; + case "knowledge_assistant": + return spec.knowledge_assistant.description; + case "app": + return spec.app.description; + case "uc_connection": + return spec.uc_connection.description; + } +} + +/** + * Builds the `AgentInput.extensions` payload from a tool index, aggregating + * the hosted-supervisor specs under {@link SUPERVISOR_EXTENSION_KEY}. Returns + * `undefined` when there are no adapter-side hosted tools so the field stays + * absent on the wire — adapters that don't read extensions never see it. + */ +export function buildAdapterExtensions( + toolIndex: Map, +): Readonly> | undefined { + const supervisorSpecs: SupervisorTool[] = []; + for (const entry of toolIndex.values()) { + if (entry.source === "hosted-supervisor") { + supervisorSpecs.push(entry.spec); + } + } + if (supervisorSpecs.length === 0) return undefined; + return { + [SUPERVISOR_EXTENSION_KEY]: { hostedTools: supervisorSpecs }, + }; +} + +/** + * Compares the adapter's declared capabilities against the tool index and + * logs a warning when the agent's tool declarations would be silently + * dropped at runtime. Warn-not-throw: misconfiguration is loud enough to + * notice without taking the whole app down. + */ +export function warnOnCapabilityMismatch( + agentName: string, + adapter: AgentAdapter, + toolIndex: Map, +): void { + const accepted = new Set(adapter.acceptsExtensions ?? []); + + const hostedSupervisorKeys: string[] = []; + const inputToolKeys: string[] = []; + for (const [key, entry] of toolIndex) { + if (entry.source === "hosted-supervisor") { + hostedSupervisorKeys.push(key); + } else { + inputToolKeys.push(key); + } + } + + if ( + hostedSupervisorKeys.length > 0 && + !accepted.has(SUPERVISOR_EXTENSION_KEY) + ) { + logger.warn( + `Agent '${agentName}' declares hosted-supervisor tools (${hostedSupervisorKeys.join(", ")}) ` + + "but its model adapter does not accept the 'databricks.supervisor' extension. " + + "These tools will not reach the model. Pair them with `DatabricksAdapter.fromSupervisorApi(...)`, or remove them.", + ); + } + + // `consumesInputTools` defaults to true. Only warn when an adapter + // explicitly opts out (`false`) and an input tool would be silently + // ignored. + if (adapter.consumesInputTools === false && inputToolKeys.length > 0) { + logger.warn( + `Agent '${agentName}' declares function tools / sub-agents / MCP tools (${inputToolKeys.join(", ")}) ` + + "but its model adapter does not consume input.tools (Supervisor API owns its own tool loop). " + + "These tools will not be exposed to the model. See docs/plugins/agents.md.", + ); + } +} diff --git a/packages/appkit/src/plugins/agents/agents.ts b/packages/appkit/src/plugins/agents/agents.ts index 39acdaf2b..37072c894 100644 --- a/packages/appkit/src/plugins/agents/agents.ts +++ b/packages/appkit/src/plugins/agents/agents.ts @@ -14,15 +14,10 @@ import type { ResponseOutputMessage, ResponseStreamEvent, Thread, - ToolAnnotations, ToolProvider, } from "shared"; -import { - isSupervisorTool, - SUPERVISOR_EXTENSION_KEY, - type SupervisorTool, -} from "../../agents/supervisor-api"; +import { isSupervisorTool } from "../../agents/supervisor-api"; import { FilesConnector } from "../../connectors/files"; import { AppKitMcpClient, buildMcpHostPolicy } from "../../connectors/mcp"; import { getWorkspaceClient } from "../../context"; @@ -41,15 +36,10 @@ import { type ResolvedSkillCatalog, readSkillResource, renderLoadedSkill, - renderSkillCatalog, resolveSkill, resolveSkillCatalog, type SkillDefinition, } from "../../core/agent/skills"; -import { - buildBaseSystemPrompt, - composeSystemPrompt, -} from "../../core/agent/system-prompt"; import { resolveToolkitFromProvider } from "../../core/agent/toolkit-resolver"; import { functionToolToDefinition, @@ -61,10 +51,8 @@ import type { AgentDefinition, AgentsPluginConfig, AgentTools, - BaseSystemPromptOption, Plugins, PluginToolkitProvider, - PromptContext, RegisteredAgent, ResolvedToolEntry, } from "../../core/agent/types"; @@ -73,6 +61,13 @@ import { createLogger } from "../../logging/logger"; import { Plugin, toPlugin } from "../../plugin"; import { defineManifest } from "../../registry"; import type { WorkspaceClient } from "../../workspace-client"; +import { + buildAdapterExtensions, + supervisorToolDescription, + warnOnCapabilityMismatch, +} from "./adapter-extensions"; +import { requiresApproval } from "./approval"; +import { LOAD_SKILL_TOOL_DEF, READ_SKILL_FILE_TOOL_DEF } from "./builtin-tools"; import { agentStreamDefaults } from "./defaults"; import { EventChannel } from "./event-channel"; import { AgentEventTranslator } from "./event-translator"; @@ -84,6 +79,7 @@ import { traceAgent, traceTool, } from "./mlflow"; +import { composePromptForAgent } from "./prompt"; import { approvalRequestSchema, cancelRequestSchema, @@ -107,33 +103,6 @@ interface AgentSource { origin: "file" | "code"; } -/** - * Decide whether a tool call must traverse the approval gate. Honours both - * the modern `effect` field (mutating values: write / update / destructive) - * and the legacy `destructive: true` boolean. The contract is documented on - * `ToolAnnotations.effect` in shared/agent.ts. - * - * Without this, a tool authored only with `effect: "destructive"` (the - * preferred API) bypassed the gate entirely. - */ -function requiresApproval(annotations: ToolAnnotations | undefined): boolean { - if (!annotations) return false; - if (annotations.destructive === true) return true; - switch (annotations.effect) { - case "write": - case "update": - case "destructive": - return true; - case "read": - case undefined: - return false; - default: { - const _exhaustive: never = annotations.effect; - return false; - } - } -} - /** * Per-stream state shared between the top-level `executeTool` and any * `runSubAgent` calls below it. Carrying the budget counter, abort signal, @@ -2338,160 +2307,6 @@ function normalizeAutoInherit(value: AgentsPluginConfig["autoInheritTools"]): { return { file: value.file ?? false, code: value.code ?? false }; } -/** Built-in tool the model calls to load a skill's full instructions on demand. */ -const LOAD_SKILL_TOOL_DEF: AgentToolDefinition = { - name: "load_skill", - description: - "Load the full instructions for one of the available skills by name. Call this before acting on a task that matches a skill's description. Returns the skill's instructions plus a list of any bundled files you can read with read_skill_file.", - parameters: { - type: "object", - properties: { - skill: { - type: "string", - description: - "The exact skill name to load, as shown in the available-skills list.", - }, - }, - required: ["skill"], - }, - annotations: { effect: "read" }, -}; - -/** Built-in tool for reading a resource file that a loaded skill references. */ -const READ_SKILL_FILE_TOOL_DEF: AgentToolDefinition = { - name: "read_skill_file", - description: - "Read a bundled resource file that a loaded skill references (e.g. a reference doc). Only files listed by load_skill for that skill are readable.", - parameters: { - type: "object", - properties: { - skill: { type: "string", description: "The skill that owns the file." }, - path: { - type: "string", - description: - "Relative path of the file within the skill, as listed by load_skill.", - }, - }, - required: ["skill", "path"], - }, - annotations: { effect: "read" }, -}; - -function composePromptForAgent( - registered: RegisteredAgent, - pluginLevel: BaseSystemPromptOption | undefined, - ctx: PromptContext, -): string { - const perAgent = registered.baseSystemPrompt; - const resolved = perAgent !== undefined ? perAgent : pluginLevel; - - let base = ""; - if (resolved === false) { - base = ""; - } else if (typeof resolved === "string") { - base = resolved; - } else if (typeof resolved === "function") { - base = resolved(ctx); - } else { - base = buildBaseSystemPrompt(ctx); - } - - const composed = composeSystemPrompt(base, registered.instructions); - - // Append the always-on skill catalog (name + description only). Done here, - // after composeSystemPrompt, so it survives a custom/`false` base prompt. - const catalog = registered.skills?.catalog; - if (!catalog || catalog.length === 0) return composed; - return `${composed}\n\n${renderSkillCatalog(catalog)}`; -} - -/** - * Pulls the LLM-readable description off any {@link SupervisorTool} kind. - * Used to populate the synthetic placeholder `def.description` on - * hosted-supervisor tool-index entries. - */ -function supervisorToolDescription(spec: SupervisorTool): string { - switch (spec.type) { - case "genie_space": - return spec.genie_space.description; - case "uc_function": - return spec.uc_function.description; - case "knowledge_assistant": - return spec.knowledge_assistant.description; - case "app": - return spec.app.description; - case "uc_connection": - return spec.uc_connection.description; - } -} - -/** - * Builds the `AgentInput.extensions` payload from a tool index, aggregating - * the hosted-supervisor specs under {@link SUPERVISOR_EXTENSION_KEY}. Returns - * `undefined` when there are no adapter-side hosted tools so the field stays - * absent on the wire — adapters that don't read extensions never see it. - */ -function buildAdapterExtensions( - toolIndex: Map, -): Readonly> | undefined { - const supervisorSpecs: SupervisorTool[] = []; - for (const entry of toolIndex.values()) { - if (entry.source === "hosted-supervisor") { - supervisorSpecs.push(entry.spec); - } - } - if (supervisorSpecs.length === 0) return undefined; - return { - [SUPERVISOR_EXTENSION_KEY]: { hostedTools: supervisorSpecs }, - }; -} - -/** - * Compares the adapter's declared capabilities against the tool index and - * logs a warning when the agent's tool declarations would be silently - * dropped at runtime. Warn-not-throw: misconfiguration is loud enough to - * notice without taking the whole app down. - */ -function warnOnCapabilityMismatch( - agentName: string, - adapter: AgentAdapter, - toolIndex: Map, -): void { - const accepted = new Set(adapter.acceptsExtensions ?? []); - - const hostedSupervisorKeys: string[] = []; - const inputToolKeys: string[] = []; - for (const [key, entry] of toolIndex) { - if (entry.source === "hosted-supervisor") { - hostedSupervisorKeys.push(key); - } else { - inputToolKeys.push(key); - } - } - - if ( - hostedSupervisorKeys.length > 0 && - !accepted.has(SUPERVISOR_EXTENSION_KEY) - ) { - logger.warn( - `Agent '${agentName}' declares hosted-supervisor tools (${hostedSupervisorKeys.join(", ")}) ` + - "but its model adapter does not accept the 'databricks.supervisor' extension. " + - "These tools will not reach the model. Pair them with `DatabricksAdapter.fromSupervisorApi(...)`, or remove them.", - ); - } - - // `consumesInputTools` defaults to true. Only warn when an adapter - // explicitly opts out (`false`) and an input tool would be silently - // ignored. - if (adapter.consumesInputTools === false && inputToolKeys.length > 0) { - logger.warn( - `Agent '${agentName}' declares function tools / sub-agents / MCP tools (${inputToolKeys.join(", ")}) ` + - "but its model adapter does not consume input.tools (Supervisor API owns its own tool loop). " + - "These tools will not be exposed to the model. See docs/plugins/agents.md.", - ); - } -} - /** * Plugin factory for the agents plugin. Discovers agents from * `server/agents//agent.{ts,md}` by default (markdown still in diff --git a/packages/appkit/src/plugins/agents/approval.ts b/packages/appkit/src/plugins/agents/approval.ts new file mode 100644 index 000000000..f3831139f --- /dev/null +++ b/packages/appkit/src/plugins/agents/approval.ts @@ -0,0 +1,30 @@ +import type { ToolAnnotations } from "shared"; + +/** + * Decide whether a tool call must traverse the approval gate. Honours both + * the modern `effect` field (mutating values: write / update / destructive) + * and the legacy `destructive: true` boolean. The contract is documented on + * `ToolAnnotations.effect` in shared/agent.ts. + * + * Without this, a tool authored only with `effect: "destructive"` (the + * preferred API) bypassed the gate entirely. + */ +export function requiresApproval( + annotations: ToolAnnotations | undefined, +): boolean { + if (!annotations) return false; + if (annotations.destructive === true) return true; + switch (annotations.effect) { + case "write": + case "update": + case "destructive": + return true; + case "read": + case undefined: + return false; + default: { + const _exhaustive: never = annotations.effect; + return false; + } + } +} diff --git a/packages/appkit/src/plugins/agents/builtin-tools.ts b/packages/appkit/src/plugins/agents/builtin-tools.ts new file mode 100644 index 000000000..6689be49f --- /dev/null +++ b/packages/appkit/src/plugins/agents/builtin-tools.ts @@ -0,0 +1,40 @@ +import type { AgentToolDefinition } from "shared"; + +/** Built-in tool the model calls to load a skill's full instructions on demand. */ +export const LOAD_SKILL_TOOL_DEF: AgentToolDefinition = { + name: "load_skill", + description: + "Load the full instructions for one of the available skills by name. Call this before acting on a task that matches a skill's description. Returns the skill's instructions plus a list of any bundled files you can read with read_skill_file.", + parameters: { + type: "object", + properties: { + skill: { + type: "string", + description: + "The exact skill name to load, as shown in the available-skills list.", + }, + }, + required: ["skill"], + }, + annotations: { effect: "read" }, +}; + +/** Built-in tool for reading a resource file that a loaded skill references. */ +export const READ_SKILL_FILE_TOOL_DEF: AgentToolDefinition = { + name: "read_skill_file", + description: + "Read a bundled resource file that a loaded skill references (e.g. a reference doc). Only files listed by load_skill for that skill are readable.", + parameters: { + type: "object", + properties: { + skill: { type: "string", description: "The skill that owns the file." }, + path: { + type: "string", + description: + "Relative path of the file within the skill, as listed by load_skill.", + }, + }, + required: ["skill", "path"], + }, + annotations: { effect: "read" }, +}; diff --git a/packages/appkit/src/plugins/agents/prompt.ts b/packages/appkit/src/plugins/agents/prompt.ts new file mode 100644 index 000000000..90cc34ec1 --- /dev/null +++ b/packages/appkit/src/plugins/agents/prompt.ts @@ -0,0 +1,43 @@ +import { renderSkillCatalog } from "../../core/agent/skills"; +import { + buildBaseSystemPrompt, + composeSystemPrompt, +} from "../../core/agent/system-prompt"; +import type { + BaseSystemPromptOption, + PromptContext, + RegisteredAgent, +} from "../../core/agent/types"; + +/** + * Composes an agent's full system prompt: resolves the base prompt (per-agent + * override, else plugin-level, else the built-in default), combines it with the + * agent's instructions, then appends the always-on skill catalog. + */ +export function composePromptForAgent( + registered: RegisteredAgent, + pluginLevel: BaseSystemPromptOption | undefined, + ctx: PromptContext, +): string { + const perAgent = registered.baseSystemPrompt; + const resolved = perAgent !== undefined ? perAgent : pluginLevel; + + let base = ""; + if (resolved === false) { + base = ""; + } else if (typeof resolved === "string") { + base = resolved; + } else if (typeof resolved === "function") { + base = resolved(ctx); + } else { + base = buildBaseSystemPrompt(ctx); + } + + const composed = composeSystemPrompt(base, registered.instructions); + + // Append the always-on skill catalog (name + description only). Done here, + // after composeSystemPrompt, so it survives a custom/`false` base prompt. + const catalog = registered.skills?.catalog; + if (!catalog || catalog.length === 0) return composed; + return `${composed}\n\n${renderSkillCatalog(catalog)}`; +} From b65a8457252d0e32f6ff4ec45cdc04b9141ac392 Mon Sep 17 00:00:00 2001 From: MarioCadenas Date: Fri, 21 Aug 2026 17:20:35 +0200 Subject: [PATCH 2/8] refactor(appkit): extract agents skill loading/dispatch into skill-loader Step 2 of splitting the agents plugin. Moves skill discovery, per-agent catalog resolution, and the load_skill/read_skill_file dispatch into skill-loader.ts as free functions; the class keeps thin delegators (call sites unchanged) and skillWorkspaceClient() as the OBO credential seam. agents.ts: 2331 -> 2152 lines. typecheck + 394 agent tests green. Signed-off-by: MarioCadenas --- packages/appkit/src/plugins/agents/agents.ts | 223 ++-------------- .../appkit/src/plugins/agents/skill-loader.ts | 244 ++++++++++++++++++ 2 files changed, 268 insertions(+), 199 deletions(-) create mode 100644 packages/appkit/src/plugins/agents/skill-loader.ts diff --git a/packages/appkit/src/plugins/agents/agents.ts b/packages/appkit/src/plugins/agents/agents.ts index 37072c894..404c32eb2 100644 --- a/packages/appkit/src/plugins/agents/agents.ts +++ b/packages/appkit/src/plugins/agents/agents.ts @@ -18,7 +18,6 @@ import type { } from "shared"; import { isSupervisorTool } from "../../agents/supervisor-api"; -import { FilesConnector } from "../../connectors/files"; import { AppKitMcpClient, buildMcpHostPolicy } from "../../connectors/mcp"; import { getWorkspaceClient } from "../../context"; import { consumeAdapterStream } from "../../core/agent/consume-adapter-stream"; @@ -30,15 +29,9 @@ import { } from "../../core/agent/load-code-agents"; import { normalizeToolResult } from "../../core/agent/normalize-result"; import { createPluginsProxy } from "../../core/agent/plugins-map"; -import { - loadSkillsFromDir, - parseSkill, - type ResolvedSkillCatalog, - readSkillResource, - renderLoadedSkill, - resolveSkill, - resolveSkillCatalog, - type SkillDefinition, +import type { + ResolvedSkillCatalog, + SkillDefinition, } from "../../core/agent/skills"; import { resolveToolkitFromProvider } from "../../core/agent/toolkit-resolver"; import { @@ -86,6 +79,13 @@ import { chatRequestSchema, invocationsRequestSchema, } from "./schemas"; +import { + dispatchSkillTool, + loadGlobalSkills, + loadVolumeSkills, + renderForcedSkill, + resolveAgentSkills, +} from "./skill-loader"; import { InMemoryThreadStore } from "./thread-store"; import { ToolApprovalGate } from "./tool-approval-gate"; @@ -640,17 +640,8 @@ export class AgentsPlugin extends Plugin implements ToolProvider { } /** Loads the shared global skill pool from `/skills/`. */ - private async loadGlobalSkills(): Promise { - const dir = this.resolvedAgentsDir(); - if (!dir) return []; - return loadSkillsFromDir(path.join(dir, "skills"), "bundle-global"); - } - - /** Configured catalog-skills volume path, or null when none is set. */ - private resolveSkillsVolume(): string | null { - const configured = - this.config.skillsVolume ?? process.env.DATABRICKS_VOLUME_AGENT_SKILLS; - return configured && configured.trim() !== "" ? configured.trim() : null; + private loadGlobalSkills(): Promise { + return loadGlobalSkills(this.resolvedAgentsDir()); } /** @@ -663,136 +654,25 @@ export class AgentsPlugin extends Plugin implements ToolProvider { return getWorkspaceClient(); } - /** - * Discovers catalog skills from the configured UC Volume, read as the - * service principal at boot (and on `reload()`). Each `//` - * folder with a `SKILL.md` becomes a `source: "volume"` skill. Best-effort: - * a missing volume, unavailable workspace client, or a malformed individual - * skill is logged and skipped rather than failing the whole registry build. - */ - private async loadVolumeSkills(): Promise { - const volume = this.resolveSkillsVolume(); - if (!volume) return []; - - if ((this.config.skillCredentialMode ?? "sp") === "obo") { - logger.warn( - "skillCredentialMode 'obo' is not wired yet; reading catalog skills as the service principal.", - ); - } - - let client: WorkspaceClient; - try { - client = this.skillWorkspaceClient(); - } catch (err) { - logger.warn( - "Skipping catalog skills at '%s': no workspace client available (%s).", - volume, - err instanceof Error ? err.message : String(err), - ); - return []; - } - - const connector = new FilesConnector({ defaultVolume: volume }); - let entries: Awaited>; - try { - entries = await connector.list(client, volume); - } catch (err) { - logger.warn( - "Failed to list catalog skills volume '%s': %s", - volume, - err instanceof Error ? err.message : String(err), - ); - return []; - } - - // Read every skill concurrently — each is a network round-trip to the - // volume, so serial reads would add up. A per-skill failure skips only - // that skill; Promise.all preserves the (sorted) entry order. - const loaded = await Promise.all( - entries.map(async (entry): Promise => { - if (!entry.is_directory || !entry.name || !entry.path) return null; - const skillDir = entry.path; - const skillFile = `${skillDir}/SKILL.md`; - try { - const raw = await connector.read(client, skillFile); - const parsed = parseSkill(raw, skillFile); - const files = await this.listVolumeSkillFiles( - connector, - client, - skillDir, - ); - return { - name: parsed.name, - description: parsed.description, - body: parsed.body, - source: "volume", - dir: skillDir, - files, - allowedTools: parsed.allowedTools, - }; - } catch (err) { - logger.warn( - "Skipping catalog skill '%s': %s", - skillDir, - err instanceof Error ? err.message : String(err), - ); - return null; - } - }), - ); - return loaded.filter((s): s is SkillDefinition => s !== null); + /** Discovers catalog skills from the configured UC Volume (best-effort). */ + private loadVolumeSkills(): Promise { + return loadVolumeSkills(this.config, () => this.skillWorkspaceClient()); } - /** Lists a volume skill's resource files (one level, excluding SKILL.md). */ - private async listVolumeSkillFiles( - connector: FilesConnector, - client: WorkspaceClient, - skillDir: string, - ): Promise { - try { - const entries = await connector.list(client, skillDir); - return entries - .filter((e) => !e.is_directory && e.name && e.name !== "SKILL.md") - .map((e) => e.name as string) - .sort(); - } catch { - return []; - } - } - - /** - * Resolves the per-agent skill catalog: loads this agent's private skills - * (`//skills/`, file-origin only), then applies visibility - * (opt-in via `def.skills` or `autoInheritSkills`) and collision rules - * against the shared global pool. Returns `undefined` when nothing is - * visible so the prompt catalog and dispatch can cheaply skip skills. - */ private async resolveAgentSkills( name: string, def: AgentDefinition, src: AgentSource, ): Promise { - const dir = this.resolvedAgentsDir(); - const perAgentSkills = - src.origin === "file" && dir - ? await loadSkillsFromDir( - path.join(dir, name, "skills"), - "bundle-agent", - ) - : []; - const inherit = normalizeAutoInherit(this.config.autoInheritSkills); - const autoInherit = src.origin === "file" ? inherit.file : inherit.code; - - const catalog = resolveSkillCatalog({ - agentName: name, - agentSkillNames: def.skills, - perAgentSkills, + return resolveAgentSkills({ + name, + def, + agentsDir: this.resolvedAgentsDir(), + isFileOrigin: src.origin === "file", + autoInherit: src.origin === "file" ? inherit.file : inherit.code, globalSkills: this.globalSkills, - autoInherit, }); - - return catalog.byAddress.size > 0 ? catalog : undefined; } private async resolveAdapter( @@ -1952,55 +1832,11 @@ export class AgentsPlugin extends Plugin implements ToolProvider { return normalizeToolResult(toolResult); } - /** - * Executes the built-in `load_skill` / `read_skill_file` tools against the - * agent's resolved skill catalog. `load_skill` returns a skill's body plus a - * manifest of bundled files; `read_skill_file` returns the contents of one - * of those files (bundle skills only in v1 — catalog-volume resource reads - * arrive with the volume source). - */ - private async dispatchSkillTool( + private dispatchSkillTool( entry: Extract, args: unknown, ): Promise { - const obj = - typeof args === "object" && args !== null - ? (args as Record) - : {}; - const skillName = typeof obj.skill === "string" ? obj.skill.trim() : ""; - if (!skillName) { - throw new Error( - `'${entry.builtin}' requires a 'skill' argument naming the skill to use.`, - ); - } - - const skill = resolveSkill(entry.catalog, skillName); - - if (entry.builtin === "load_skill") { - return renderLoadedSkill(skill); - } - - // read_skill_file - const filePath = typeof obj.path === "string" ? obj.path.trim() : ""; - if (!filePath) { - throw new Error("'read_skill_file' requires a 'path' argument."); - } - if (!skill.files.includes(filePath)) { - throw new Error( - `Skill '${skill.name}' has no bundled file '${filePath}'. Available: ${ - skill.files.join(", ") || "" - }.`, - ); - } - - if (skill.source === "volume") { - const connector = new FilesConnector({ defaultVolume: skill.dir }); - return connector.read( - this.skillWorkspaceClient(), - `${skill.dir}/${filePath}`, - ); - } - return readSkillResource(skill.dir, filePath); + return dispatchSkillTool(entry, args, () => this.skillWorkspaceClient()); } /** @@ -2013,18 +1849,7 @@ export class AgentsPlugin extends Plugin implements ToolProvider { registered: RegisteredAgent, name: string, ): string | null { - if (!registered.skills) return null; - try { - const skill = resolveSkill(registered.skills, name); - return `The user explicitly requested the "${skill.name}" skill for this turn. Its instructions:\n\n${renderLoadedSkill(skill)}`; - } catch (err) { - logger.warn( - "Ignoring forced skill '%s': %s", - name, - err instanceof Error ? err.message : String(err), - ); - return null; - } + return renderForcedSkill(registered, name); } /** diff --git a/packages/appkit/src/plugins/agents/skill-loader.ts b/packages/appkit/src/plugins/agents/skill-loader.ts new file mode 100644 index 000000000..2c2474495 --- /dev/null +++ b/packages/appkit/src/plugins/agents/skill-loader.ts @@ -0,0 +1,244 @@ +import path from "node:path"; + +import { FilesConnector } from "../../connectors/files"; +import { + loadSkillsFromDir, + readSkillResource, + renderLoadedSkill, + type ResolvedSkillCatalog, + resolveSkill, + resolveSkillCatalog, + type SkillDefinition, +} from "../../core/agent/skills"; +import { parseSkill } from "../../core/agent/skills"; +import type { + AgentDefinition, + AgentsPluginConfig, + RegisteredAgent, + ResolvedToolEntry, +} from "../../core/agent/types"; +import { createLogger } from "../../logging/logger"; +import type { WorkspaceClient } from "../../workspace-client"; + +const logger = createLogger("agents"); + +/** Loads the shared global skill pool from `/skills/`. */ +export async function loadGlobalSkills( + agentsDir: string, +): Promise { + if (!agentsDir) return []; + return loadSkillsFromDir(path.join(agentsDir, "skills"), "bundle-global"); +} + +/** Configured catalog-skills volume path, or null when none is set. */ +function resolveSkillsVolume(config: AgentsPluginConfig): string | null { + const configured = + config.skillsVolume ?? process.env.DATABRICKS_VOLUME_AGENT_SKILLS; + return configured && configured.trim() !== "" ? configured.trim() : null; +} + +/** + * Discovers catalog skills from the configured UC Volume, read as the + * service principal at boot (and on `reload()`). Each `//` + * folder with a `SKILL.md` becomes a `source: "volume"` skill. Best-effort: + * a missing volume, unavailable workspace client, or a malformed individual + * skill is logged and skipped rather than failing the whole registry build. + * + * `getClient` is a thunk so the workspace-client resolution (the single + * switch point for a future OBO mode) stays with the caller and is resolved + * lazily inside the try/catch here. + */ +export async function loadVolumeSkills( + config: AgentsPluginConfig, + getClient: () => WorkspaceClient, +): Promise { + const volume = resolveSkillsVolume(config); + if (!volume) return []; + + if ((config.skillCredentialMode ?? "sp") === "obo") { + logger.warn( + "skillCredentialMode 'obo' is not wired yet; reading catalog skills as the service principal.", + ); + } + + let client: WorkspaceClient; + try { + client = getClient(); + } catch (err) { + logger.warn( + "Skipping catalog skills at '%s': no workspace client available (%s).", + volume, + err instanceof Error ? err.message : String(err), + ); + return []; + } + + const connector = new FilesConnector({ defaultVolume: volume }); + let entries: Awaited>; + try { + entries = await connector.list(client, volume); + } catch (err) { + logger.warn( + "Failed to list catalog skills volume '%s': %s", + volume, + err instanceof Error ? err.message : String(err), + ); + return []; + } + + // Read every skill concurrently — each is a network round-trip to the + // volume, so serial reads would add up. A per-skill failure skips only + // that skill; Promise.all preserves the (sorted) entry order. + const loaded = await Promise.all( + entries.map(async (entry): Promise => { + if (!entry.is_directory || !entry.name || !entry.path) return null; + const skillDir = entry.path; + const skillFile = `${skillDir}/SKILL.md`; + try { + const raw = await connector.read(client, skillFile); + const parsed = parseSkill(raw, skillFile); + const files = await listVolumeSkillFiles(connector, client, skillDir); + return { + name: parsed.name, + description: parsed.description, + body: parsed.body, + source: "volume", + dir: skillDir, + files, + allowedTools: parsed.allowedTools, + }; + } catch (err) { + logger.warn( + "Skipping catalog skill '%s': %s", + skillDir, + err instanceof Error ? err.message : String(err), + ); + return null; + } + }), + ); + return loaded.filter((s): s is SkillDefinition => s !== null); +} + +/** Lists a volume skill's resource files (one level, excluding SKILL.md). */ +async function listVolumeSkillFiles( + connector: FilesConnector, + client: WorkspaceClient, + skillDir: string, +): Promise { + try { + const entries = await connector.list(client, skillDir); + return entries + .filter((e) => !e.is_directory && e.name && e.name !== "SKILL.md") + .map((e) => e.name as string) + .sort(); + } catch { + return []; + } +} + +/** + * Resolves the per-agent skill catalog: loads this agent's private skills + * (`//skills/`, file-origin only), then applies visibility + * (opt-in via `def.skills` or the resolved `autoInherit`) and collision rules + * against the shared global pool. Returns `undefined` when nothing is visible + * so the prompt catalog and dispatch can cheaply skip skills. + */ +export async function resolveAgentSkills(opts: { + name: string; + def: AgentDefinition; + agentsDir: string; + isFileOrigin: boolean; + autoInherit: boolean; + globalSkills: SkillDefinition[]; +}): Promise { + const perAgentSkills = + opts.isFileOrigin && opts.agentsDir + ? await loadSkillsFromDir( + path.join(opts.agentsDir, opts.name, "skills"), + "bundle-agent", + ) + : []; + + const catalog = resolveSkillCatalog({ + agentName: opts.name, + agentSkillNames: opts.def.skills, + perAgentSkills, + globalSkills: opts.globalSkills, + autoInherit: opts.autoInherit, + }); + + return catalog.byAddress.size > 0 ? catalog : undefined; +} + +/** + * Executes the built-in `load_skill` / `read_skill_file` tools against the + * agent's resolved skill catalog. `load_skill` returns a skill's body plus a + * manifest of bundled files; `read_skill_file` returns the contents of one + * of those files. + */ +export async function dispatchSkillTool( + entry: Extract, + args: unknown, + getClient: () => WorkspaceClient, +): Promise { + const obj = + typeof args === "object" && args !== null + ? (args as Record) + : {}; + const skillName = typeof obj.skill === "string" ? obj.skill.trim() : ""; + if (!skillName) { + throw new Error( + `'${entry.builtin}' requires a 'skill' argument naming the skill to use.`, + ); + } + + const skill = resolveSkill(entry.catalog, skillName); + + if (entry.builtin === "load_skill") { + return renderLoadedSkill(skill); + } + + // read_skill_file + const filePath = typeof obj.path === "string" ? obj.path.trim() : ""; + if (!filePath) { + throw new Error("'read_skill_file' requires a 'path' argument."); + } + if (!skill.files.includes(filePath)) { + throw new Error( + `Skill '${skill.name}' has no bundled file '${filePath}'. Available: ${ + skill.files.join(", ") || "" + }.`, + ); + } + + if (skill.source === "volume") { + const connector = new FilesConnector({ defaultVolume: skill.dir }); + return connector.read(getClient(), `${skill.dir}/${filePath}`); + } + return readSkillResource(skill.dir, filePath); +} + +/** + * Renders the prompt addendum for a force-loaded skill (`/skill-name`). + * Returns null when the agent has no catalog or the name doesn't resolve — + * an unusable request shouldn't fail the whole turn, so it's logged and the + * model proceeds with the catalog + `load_skill` still available. + */ +export function renderForcedSkill( + registered: RegisteredAgent, + name: string, +): string | null { + if (!registered.skills) return null; + try { + const skill = resolveSkill(registered.skills, name); + return `The user explicitly requested the "${skill.name}" skill for this turn. Its instructions:\n\n${renderLoadedSkill(skill)}`; + } catch (err) { + logger.warn( + "Ignoring forced skill '%s': %s", + name, + err instanceof Error ? err.message : String(err), + ); + return null; + } +} From 8eb01853dcaaaab50f095b28829ac239a37cb544 Mon Sep 17 00:00:00 2001 From: MarioCadenas Date: Fri, 21 Aug 2026 17:34:47 +0200 Subject: [PATCH 3/8] refactor(appkit): extract agents registry assembly into registry.ts Step 3 of splitting the agents plugin. Moves the decoupled boot-time assembly helpers into registry.ts: loadCodeAgents, hasCodeAgentSources (now internal), resolveDefaultAgent, and the AgentSource type. buildAgentRegistry stays as the orchestrator that wires them. Also merges a duplicate import in skill-loader.ts. agents.ts: 2152 -> 2088 lines. typecheck + 394 agent tests green. Signed-off-by: MarioCadenas --- packages/appkit/src/plugins/agents/agents.ts | 98 +++-------------- .../appkit/src/plugins/agents/registry.ts | 101 ++++++++++++++++++ .../appkit/src/plugins/agents/skill-loader.ts | 2 +- 3 files changed, 119 insertions(+), 82 deletions(-) create mode 100644 packages/appkit/src/plugins/agents/registry.ts diff --git a/packages/appkit/src/plugins/agents/agents.ts b/packages/appkit/src/plugins/agents/agents.ts index 404c32eb2..fb2ddcc4c 100644 --- a/packages/appkit/src/plugins/agents/agents.ts +++ b/packages/appkit/src/plugins/agents/agents.ts @@ -1,5 +1,5 @@ import { randomUUID } from "node:crypto"; -import { existsSync, readdirSync } from "node:fs"; +import { existsSync } from "node:fs"; import path from "node:path"; import type express from "express"; @@ -22,11 +22,7 @@ import { AppKitMcpClient, buildMcpHostPolicy } from "../../connectors/mcp"; import { getWorkspaceClient } from "../../context"; import { consumeAdapterStream } from "../../core/agent/consume-adapter-stream"; import { loadAgentsFromDir } from "../../core/agent/load-agents"; -import { - CODE_AGENTS_SOURCE_DIR, - loadCodeAgentsFromDir, - resolveCodeAgentsDir, -} from "../../core/agent/load-code-agents"; +import { CODE_AGENTS_SOURCE_DIR } from "../../core/agent/load-code-agents"; import { normalizeToolResult } from "../../core/agent/normalize-result"; import { createPluginsProxy } from "../../core/agent/plugins-map"; import type { @@ -73,6 +69,11 @@ import { traceTool, } from "./mlflow"; import { composePromptForAgent } from "./prompt"; +import { + type AgentSource, + loadCodeAgents, + resolveDefaultAgent, +} from "./registry"; import { approvalRequestSchema, cancelRequestSchema, @@ -94,15 +95,6 @@ const logger = createLogger("agents"); /** Deprecated markdown location, read as a fallback with a one-time warning. */ const LEGACY_MARKDOWN_DIR = "config/agents"; -/** - * Context flag recorded on the in-memory AgentDefinition to indicate whether - * it came from markdown (file) or from user code. Drives the asymmetric - * `autoInheritTools` default. - */ -interface AgentSource { - origin: "file" | "code"; -} - /** * Per-stream state shared between the top-level `executeTool` and any * `runSubAgent` calls below it. Carrying the budget counter, abort signal, @@ -440,25 +432,12 @@ export class AgentsPlugin extends Plugin implements ToolProvider { merged: Record, fileDefault: string | null, ): string | null { - if (this.config.defaultAgent) { - if (!agents.has(this.config.defaultAgent)) { - throw new Error( - `defaultAgent '${this.config.defaultAgent}' is not registered. Available: ${Array.from(agents.keys()).join(", ")}`, - ); - } - return this.config.defaultAgent; - } - - const codeDefault = Object.keys(merged) - .filter( - (id) => merged[id].src.origin === "code" && merged[id].def.default, - ) - .sort()[0]; - if (codeDefault) return codeDefault; - - if (fileDefault && agents.has(fileDefault)) return fileDefault; - - return agents.keys().next().value ?? null; + return resolveDefaultAgent( + agents, + merged, + fileDefault, + this.config.defaultAgent, + ); } /** @@ -494,54 +473,11 @@ export class AgentsPlugin extends Plugin implements ToolProvider { return path.resolve(process.cwd(), CODE_AGENTS_SOURCE_DIR); } - /** - * Discovers code agents (see {@link resolveCodeAgentsDir} and - * {@link loadCodeAgentsFromDir}). Warns if sources exist but nothing was - * discovered — usually the build didn't emit the compiled agents — unless - * the deprecated `agents({ agents })` map is carrying them instead. - */ - private async loadCodeAgents(): Promise> { - const resolved = resolveCodeAgentsDir({ - cwd: process.cwd(), - exists: existsSync, - }); - - const discovered = await loadCodeAgentsFromDir(resolved.dir, { - extensions: resolved.extensions, + private loadCodeAgents(): Promise> { + return loadCodeAgents({ + agentsDir: this.resolvedAgentsDir(), + hasDeprecatedMap: Object.keys(this.config.agents ?? {}).length > 0, }); - - const usingDeprecatedMap = Object.keys(this.config.agents ?? {}).length > 0; - const sourceDir = this.resolvedAgentsDir(); - if ( - Object.keys(discovered).length === 0 && - !usingDeprecatedMap && - this.hasCodeAgentSources(sourceDir) - ) { - logger.warn( - "Found code-agent sources in %s but discovered no code agents (scanned %s). " + - "In a production build, ensure `/*/agent.ts` is included as tsdown entries so the compiled agents are emitted.", - sourceDir, - resolved.dir, - ); - } - - return discovered; - } - - /** True when `dir` holds at least one `/agent.ts` folder. */ - private hasCodeAgentSources(dir: string): boolean { - try { - return readdirSync(dir, { withFileTypes: true }).some((e) => { - if (!e.isDirectory() && !e.isSymbolicLink()) return false; - try { - return readdirSync(path.join(dir, e.name)).includes("agent.ts"); - } catch { - return false; - } - }); - } catch { - return false; - } } private async loadFileDefinitions( diff --git a/packages/appkit/src/plugins/agents/registry.ts b/packages/appkit/src/plugins/agents/registry.ts new file mode 100644 index 000000000..0f10df958 --- /dev/null +++ b/packages/appkit/src/plugins/agents/registry.ts @@ -0,0 +1,101 @@ +import { existsSync, readdirSync } from "node:fs"; +import path from "node:path"; + +import { + loadCodeAgentsFromDir, + resolveCodeAgentsDir, +} from "../../core/agent/load-code-agents"; +import type { AgentDefinition, RegisteredAgent } from "../../core/agent/types"; +import { createLogger } from "../../logging/logger"; + +const logger = createLogger("agents"); + +/** + * Context flag recorded on the in-memory AgentDefinition to indicate whether + * it came from markdown (file) or from user code. Drives the asymmetric + * `autoInheritTools` default. + */ +export interface AgentSource { + origin: "file" | "code"; +} + +/** True when `dir` holds at least one `/agent.ts` folder. */ +function hasCodeAgentSources(dir: string): boolean { + try { + return readdirSync(dir, { withFileTypes: true }).some((e) => { + if (!e.isDirectory() && !e.isSymbolicLink()) return false; + try { + return readdirSync(path.join(dir, e.name)).includes("agent.ts"); + } catch { + return false; + } + }); + } catch { + return false; + } +} + +/** + * Discovers code agents (see {@link resolveCodeAgentsDir} and + * {@link loadCodeAgentsFromDir}). Warns if sources exist but nothing was + * discovered — usually the build didn't emit the compiled agents — unless + * the deprecated `agents({ agents })` map is carrying them instead. + */ +export async function loadCodeAgents(opts: { + agentsDir: string; + hasDeprecatedMap: boolean; +}): Promise> { + const resolved = resolveCodeAgentsDir({ + cwd: process.cwd(), + exists: existsSync, + }); + + const discovered = await loadCodeAgentsFromDir(resolved.dir, { + extensions: resolved.extensions, + }); + + if ( + Object.keys(discovered).length === 0 && + !opts.hasDeprecatedMap && + hasCodeAgentSources(opts.agentsDir) + ) { + logger.warn( + "Found code-agent sources in %s but discovered no code agents (scanned %s). " + + "In a production build, ensure `/*/agent.ts` is included as tsdown entries so the compiled agents are emitted.", + opts.agentsDir, + resolved.dir, + ); + } + + return discovered; +} + +/** + * Resolves the default agent. Precedence: explicit `configDefault` > + * a code/discovered agent flagged `default: true` (stable id order) > + * markdown `default: true` > first registered (insertion order). + */ +export function resolveDefaultAgent( + agents: Map, + merged: Record, + fileDefault: string | null, + configDefault: string | undefined, +): string | null { + if (configDefault) { + if (!agents.has(configDefault)) { + throw new Error( + `defaultAgent '${configDefault}' is not registered. Available: ${Array.from(agents.keys()).join(", ")}`, + ); + } + return configDefault; + } + + const codeDefault = Object.keys(merged) + .filter((id) => merged[id].src.origin === "code" && merged[id].def.default) + .sort()[0]; + if (codeDefault) return codeDefault; + + if (fileDefault && agents.has(fileDefault)) return fileDefault; + + return agents.keys().next().value ?? null; +} diff --git a/packages/appkit/src/plugins/agents/skill-loader.ts b/packages/appkit/src/plugins/agents/skill-loader.ts index 2c2474495..4d9c493e3 100644 --- a/packages/appkit/src/plugins/agents/skill-loader.ts +++ b/packages/appkit/src/plugins/agents/skill-loader.ts @@ -3,6 +3,7 @@ import path from "node:path"; import { FilesConnector } from "../../connectors/files"; import { loadSkillsFromDir, + parseSkill, readSkillResource, renderLoadedSkill, type ResolvedSkillCatalog, @@ -10,7 +11,6 @@ import { resolveSkillCatalog, type SkillDefinition, } from "../../core/agent/skills"; -import { parseSkill } from "../../core/agent/skills"; import type { AgentDefinition, AgentsPluginConfig, From 31a4c30796f86d4c12501f350e5ae0a788fad8d3 Mon Sep 17 00:00:00 2001 From: MarioCadenas Date: Fri, 21 Aug 2026 17:57:42 +0200 Subject: [PATCH 4/8] refactor(appkit): extract agents tool-dispatch engine into tool-dispatch.ts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Step 4 (final) of splitting the agents plugin. Moves dispatchToolCall + runSubAgent — the tool-call budget, approval gate, and sub-agent recursion — into tool-dispatch.ts as free functions over RunState + a ToolDispatchDeps object. The plugin builds deps via toolDispatchDeps(); the two executeTool closures call the free function. RunState moves with them. Tests updated to invoke the free functions (deps built from the plugin's own builder). agents.ts: 2088 -> 1839 lines (2512 -> 1839 across all four steps). typecheck + 394 agent tests green. Signed-off-by: MarioCadenas --- packages/appkit/src/plugins/agents/agents.ts | 290 ++--------------- .../agents/tests/dispatch-tool-call.test.ts | 47 ++- .../plugins/agents/tests/dos-limits.test.ts | 25 +- .../src/plugins/agents/tool-dispatch.ts | 303 ++++++++++++++++++ 4 files changed, 373 insertions(+), 292 deletions(-) create mode 100644 packages/appkit/src/plugins/agents/tool-dispatch.ts diff --git a/packages/appkit/src/plugins/agents/agents.ts b/packages/appkit/src/plugins/agents/agents.ts index fb2ddcc4c..a16105e26 100644 --- a/packages/appkit/src/plugins/agents/agents.ts +++ b/packages/appkit/src/plugins/agents/agents.ts @@ -6,7 +6,6 @@ import type express from "express"; import pc from "picocolors"; import type { AgentAdapter, - AgentRunContext, AgentToolDefinition, IAppRouter, Message, @@ -23,7 +22,6 @@ import { getWorkspaceClient } from "../../context"; import { consumeAdapterStream } from "../../core/agent/consume-adapter-stream"; import { loadAgentsFromDir } from "../../core/agent/load-agents"; import { CODE_AGENTS_SOURCE_DIR } from "../../core/agent/load-code-agents"; -import { normalizeToolResult } from "../../core/agent/normalize-result"; import { createPluginsProxy } from "../../core/agent/plugins-map"; import type { ResolvedSkillCatalog, @@ -66,7 +64,6 @@ import { initAgentTracing, linkTraceToRun, traceAgent, - traceTool, } from "./mlflow"; import { composePromptForAgent } from "./prompt"; import { @@ -89,40 +86,17 @@ import { } from "./skill-loader"; import { InMemoryThreadStore } from "./thread-store"; import { ToolApprovalGate } from "./tool-approval-gate"; +import { + dispatchToolCall, + type RunState, + type ToolDispatchDeps, +} from "./tool-dispatch"; const logger = createLogger("agents"); /** Deprecated markdown location, read as a fallback with a one-time warning. */ const LEGACY_MARKDOWN_DIR = "config/agents"; -/** - * Per-stream state shared between the top-level `executeTool` and any - * `runSubAgent` calls below it. Carrying the budget counter, abort signal, - * approval policy, and event-channel through one object is what lets the - * sub-agent path enforce the same limits and approval gate as the parent. - * - * Without this shared state the sub-agent path silently bypassed both the - * tool-call budget and the destructive-tool approval gate. - */ -interface RunState { - req: express.Request; - userId: string; - requestId: string; - abortController: AbortController; - signal: AbortSignal; - approvalPolicy: { requireForDestructive: boolean; timeoutMs: number }; - limits: { - maxConcurrentStreamsPerUser: number; - maxToolCalls: number; - maxSubAgentDepth: number; - toolCallTimeoutMs: number; - }; - translator: AgentEventTranslator; - outboundEvents: EventChannel; - /** Boxed mutable counter shared across parent + all sub-agent dispatches. */ - toolCallsUsed: { count: number }; -} - export class AgentsPlugin extends Plugin implements ToolProvider { static manifest = defineManifest<"agents">(manifest); static phase: PluginPhase = "deferred"; @@ -1294,8 +1268,9 @@ export class AgentsPlugin extends Plugin implements ToolProvider { toolCallsUsed: { count: 0 }, }; + const deps = this.toolDispatchDeps(); const executeTool = (name: string, args: unknown): Promise => - this.dispatchToolCall(runState, registered.toolIndex, name, args, 0); + dispatchToolCall(deps, runState, registered.toolIndex, name, args, 0); // Drive the adapter and the approval-event side-channel concurrently. // Outbound events from both sources flow through `outboundEvents`; the @@ -1515,8 +1490,9 @@ export class AgentsPlugin extends Plugin implements ToolProvider { toolCallsUsed: { count: 0 }, }; + const deps = this.toolDispatchDeps(); const executeTool = (name: string, args: unknown): Promise => - this.dispatchToolCall(runState, registered.toolIndex, name, args, 0); + dispatchToolCall(deps, runState, registered.toolIndex, name, args, 0); let fullContent = ""; try { @@ -1638,136 +1614,6 @@ export class AgentsPlugin extends Plugin implements ToolProvider { }); } - /** - * Dispatch a single tool call from either the top-level adapter or a - * sub-agent. Centralising this in one method is what makes the budget - * counter, approval gate, and abort signal observe sub-agent activity: - * `runSubAgent` reuses the same `runState` and so increments the same - * counter and emits approval events through the same channel. - * - * `depth` is the current sub-agent recursion depth (0 at the top level). - * It is forwarded to `runSubAgent` when the dispatched entry is itself a - * sub-agent, so depth limits remain enforced. - */ - private async dispatchToolCall( - runState: RunState, - toolIndex: Map, - name: string, - args: unknown, - depth: number, - ): Promise { - if (runState.toolCallsUsed.count >= runState.limits.maxToolCalls) { - runState.abortController.abort( - new Error( - `Tool-call budget exhausted (limit ${runState.limits.maxToolCalls}).`, - ), - ); - throw new Error( - `Tool-call budget exhausted (limit ${runState.limits.maxToolCalls}). Raise agents({ limits: { maxToolCalls } }) or review the agent's tool-selection logic.`, - ); - } - runState.toolCallsUsed.count++; - - const entry = toolIndex.get(name); - if (!entry) throw new Error(`Unknown tool: ${name}`); - - if ( - runState.approvalPolicy.requireForDestructive && - requiresApproval(entry.def.annotations) - ) { - const approvalId = randomUUID(); - for (const ev of runState.translator.translate({ - type: "approval_pending", - approvalId, - streamId: runState.requestId, - toolName: name, - args, - annotations: entry.def.annotations, - })) { - runState.outboundEvents.push(ev); - } - const decision = await this.approvalGate.wait({ - approvalId, - streamId: runState.requestId, - userId: runState.userId, - timeoutMs: runState.approvalPolicy.timeoutMs, - }); - if (decision === "deny") { - return `Tool execution denied by user approval gate (tool: ${name}).`; - } - } - - // Traced from here so the span covers execution only, not the approval - // wait above (which is human latency). - const toolResult = await traceTool(name, args, async () => { - let result: unknown; - if (entry.source === "toolkit") { - if (!this.context) { - throw new Error( - "Plugin tool execution requires PluginContext; this should never happen through createApp", - ); - } - result = await this.context.executeTool( - runState.req, - entry.pluginName, - entry.localName, - args, - runState.signal, - runState.limits.toolCallTimeoutMs, - ); - } else if (entry.source === "function") { - // Function tools declare their parameters as a JSON-object schema, - // so adapters always serialize `args` as an object. A non-object - // value here means the upstream model emitted malformed tool-call - // JSON; surface a clear error rather than silently passing through - // a wrong-shape value the tool will then choke on. - if (typeof args !== "object" || args === null || Array.isArray(args)) { - throw new Error( - `Function tool '${name}' received non-object arguments (got ${args === null ? "null" : Array.isArray(args) ? "array" : typeof args}); expected a JSON object.`, - ); - } - result = await entry.functionTool.execute( - args as Record, - ); - } else if (entry.source === "mcp") { - if (!this.mcpClient) throw new Error("MCP client not connected"); - const oboToken = runState.req.headers["x-forwarded-access-token"]; - const mcpAuth = - typeof oboToken === "string" - ? { Authorization: `Bearer ${oboToken}` } - : undefined; - result = await this.mcpClient.callTool( - entry.mcpToolName, - args, - mcpAuth, - ); - } else if (entry.source === "subagent") { - const childAgent = this.agents.get(entry.agentName); - if (!childAgent) - throw new Error(`Sub-agent not found: ${entry.agentName}`); - result = await this.runSubAgent(runState, childAgent, args, depth + 1); - } else if (entry.source === "hosted-supervisor") { - // Defense-in-depth: should never fire. Hosted-supervisor entries are - // routed via `AgentInput.extensions` and the SA endpoint executes - // them server-side; their `def` is filtered out of the adapter's - // `tools` array, so the model never sees a callable schema for them. - // If we reach here, the agent is paired with a non-SA adapter that - // somehow surfaced the placeholder def to the model — surface a - // clear error rather than crash later in `normalizeToolResult`. - throw new Error( - `Tool '${name}' is a hosted-supervisor tool and cannot be invoked from the Node process. ` + - "It is executed server-side by the Databricks AI Gateway and is only reachable when the agent's model is a Supervisor API adapter.", - ); - } else if (entry.source === "skill") { - result = await this.dispatchSkillTool(entry, args); - } - - return result; - }); - - return normalizeToolResult(toolResult); - } - private dispatchSkillTool( entry: Extract, args: unknown, @@ -1788,115 +1634,17 @@ export class AgentsPlugin extends Plugin implements ToolProvider { return renderForcedSkill(registered, name); } - /** - * Runs a sub-agent in response to an `agent-` tool call. Returns the - * concatenated text output to hand back to the parent adapter as the tool - * result. - * - * `depth` starts at 1 for a top-level sub-agent invocation (i.e. the - * outer `_streamAgent` calls `runSubAgent(..., 1)`) and increments on - * each nested `runSubAgent` call. Depths exceeding - * `limits.maxSubAgentDepth` are rejected before any adapter work. - * - * Sub-agent tool calls run through `dispatchToolCall` with the same - * `runState` as the parent — the budget counter and approval gate are - * therefore enforced for every nested call, not only at the top level. - */ - private async runSubAgent( - runState: RunState, - child: RegisteredAgent, - args: unknown, - depth: number, - ): Promise { - if (depth > runState.limits.maxSubAgentDepth) { - throw new Error( - `Sub-agent depth exceeded (limit ${runState.limits.maxSubAgentDepth}). ` + - `Raise agents({ limits: { maxSubAgentDepth } }) or break the delegation cycle.`, - ); - } - - const input = - typeof args === "object" && - args !== null && - typeof (args as { input?: unknown }).input === "string" - ? (args as { input: string }).input - : JSON.stringify(args); - // Same filter as the top-level path: hosted-supervisor `def` is a - // placeholder, not a callable function — exclude from the adapter's - // `tools` array. The specs are routed via `extensions` instead. - const childTools = Array.from(child.toolIndex.values()) - .filter((e) => e.source !== "hosted-supervisor") - .map((e) => e.def); - - const childExecute = (name: string, childArgs: unknown): Promise => - this.dispatchToolCall(runState, child.toolIndex, name, childArgs, depth); - - const runContext: AgentRunContext = { - executeTool: childExecute, - signal: runState.signal, + /** Collaborators the tool-dispatch path needs, bound to this instance. */ + private toolDispatchDeps(): ToolDispatchDeps { + return { + approvalGate: this.approvalGate, + context: this.context, + getMcpClient: () => this.mcpClient, + agents: this.agents, + dispatchSkillTool: (entry, args) => this.dispatchSkillTool(entry, args), + pluginName: this.name, + baseSystemPrompt: this.config.baseSystemPrompt, }; - - const pluginNames = this.context - ? this.context - .getPluginNames() - .filter((n) => n !== this.name && n !== "server") - : []; - const systemPrompt = composePromptForAgent( - child, - this.config.baseSystemPrompt, - { - agentName: child.name, - pluginNames, - toolNames: childTools.map((t) => t.name), - }, - ); - - const messages: Message[] = [ - { - id: "system", - role: "system", - content: systemPrompt, - createdAt: new Date(), - }, - { - id: randomUUID(), - role: "user", - content: input, - createdAt: new Date(), - }, - ]; - - return consumeAdapterStream( - child.adapter.run( - { - messages, - tools: childTools, - threadId: randomUUID(), - signal: runState.signal, - extensions: buildAdapterExtensions(child.toolIndex), - }, - runContext, - ), - { - signal: runState.signal, - // Forward every sub-agent event into the parent's outbound SSE - // stream so the client sees nested tool_call / tool_result events - // (UI-action tools like apply_filter / highlight_period rely on - // this) and the sub-agent's streaming text as it's generated. - // - // `metadata` is the one exception: sub-agents have their own - // threadId, and forwarding it would overwrite the parent's - // thread state on the client and break multi-turn continuity. - // Approval-pending events emitted by `dispatchToolCall` already - // reach `outboundEvents` directly, so they are not routed here. - onEvent: (event) => { - if (event.type === "metadata") return; - for (const translated of runState.translator.translate(event)) { - runState.outboundEvents.push(translated); - } - }, - }, - ); } private async _handleCancel(req: express.Request, res: express.Response) { diff --git a/packages/appkit/src/plugins/agents/tests/dispatch-tool-call.test.ts b/packages/appkit/src/plugins/agents/tests/dispatch-tool-call.test.ts index 0cf596ef3..7459b7137 100644 --- a/packages/appkit/src/plugins/agents/tests/dispatch-tool-call.test.ts +++ b/packages/appkit/src/plugins/agents/tests/dispatch-tool-call.test.ts @@ -8,8 +8,15 @@ import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; import { CacheManager } from "../../../cache"; import { resolveSkillCatalog } from "../../../core/agent/skills/resolve-catalog"; import type { SkillDefinition } from "../../../core/agent/skills/types"; +import type { ResolvedToolEntry } from "../../../core/agent/types"; import { createTestPluginContext } from "../../../testing"; import { AgentsPlugin } from "../agents"; +import { + dispatchToolCall, + type RunState, + runSubAgent, + type ToolDispatchDeps, +} from "../tool-dispatch"; /** * Verifies that `dispatchToolCall` is the single source of truth for the @@ -83,6 +90,13 @@ function makeRunState(plugin: AgentsPlugin) { return { runState, pushed, plugin }; } +/** The plugin's own dispatch-deps builder, reused so tests dispatch exactly as the plugin does. */ +function depsOf(plugin: AgentsPlugin): ToolDispatchDeps { + return ( + plugin as unknown as { toolDispatchDeps: () => ToolDispatchDeps } + ).toolDispatchDeps(); +} + function callDispatch( plugin: AgentsPlugin, args: { @@ -93,19 +107,10 @@ function callDispatch( depth?: number; }, ): Promise { - return ( - plugin as unknown as { - dispatchToolCall: ( - runState: unknown, - toolIndex: Map, - name: string, - args: unknown, - depth: number, - ) => Promise; - } - ).dispatchToolCall( - args.runState, - args.toolIndex, + return dispatchToolCall( + depsOf(plugin), + args.runState as RunState, + args.toolIndex as unknown as Map, args.name, args.args, args.depth ?? 0, @@ -427,7 +432,13 @@ describe("runSubAgent — sub-agent event forwarding", () => { } as any; await expect( - (plugin as any).runSubAgent(runState, child, { input: "go" }, 3), + runSubAgent( + depsOf(plugin), + runState as unknown as RunState, + child, + { input: "go" }, + 3, + ), ).rejects.toThrow(/Sub-agent depth exceeded \(limit 2\)/); expect(childRun).not.toHaveBeenCalled(); }); @@ -455,7 +466,13 @@ describe("runSubAgent — sub-agent event forwarding", () => { toolIndex: new Map(), } as any; - await (plugin as any).runSubAgent(runState, child, { input: "go" }, 1); + await runSubAgent( + depsOf(plugin), + runState as unknown as RunState, + child, + { input: "go" }, + 1, + ); const types = pushed.map((e) => (e as { type: string }).type); expect(types).not.toContain("metadata"); diff --git a/packages/appkit/src/plugins/agents/tests/dos-limits.test.ts b/packages/appkit/src/plugins/agents/tests/dos-limits.test.ts index dac3fba56..9104979d3 100644 --- a/packages/appkit/src/plugins/agents/tests/dos-limits.test.ts +++ b/packages/appkit/src/plugins/agents/tests/dos-limits.test.ts @@ -4,6 +4,11 @@ import { beforeEach, describe, expect, test, vi } from "vitest"; import { CacheManager } from "../../../cache"; import { AgentsPlugin } from "../agents"; import { chatRequestSchema, invocationsRequestSchema } from "../schemas"; +import { + type RunState, + runSubAgent, + type ToolDispatchDeps, +} from "../tool-dispatch"; /** * Exercises the four DoS caps landed for MVP: @@ -290,6 +295,12 @@ describe("resolvedLimits — default values", () => { }); }); +function depsOf(plugin: AgentsPlugin): ToolDispatchDeps { + return ( + plugin as unknown as { toolDispatchDeps: () => ToolDispatchDeps } + ).toolDispatchDeps(); +} + describe("runSubAgent — depth guard", () => { /** * Builds a minimal `RunState` matching the shape carried by `_streamAgent` @@ -336,9 +347,10 @@ describe("runSubAgent — depth guard", () => { }); const runState = makeRunState(plugin, { maxSubAgentDepth: 2 }); await expect( - (plugin as any).runSubAgent( - runState, - { name: "child", toolIndex: new Map() }, + runSubAgent( + depsOf(plugin), + runState as unknown as RunState, + { name: "child", toolIndex: new Map() } as any, {}, 3, // exceeds limit 2 ), @@ -364,9 +376,10 @@ describe("runSubAgent — depth guard", () => { }; const runState = makeRunState(plugin, { maxSubAgentDepth: 3 }); - const result = await (plugin as any).runSubAgent( - runState, - child, + const result = await runSubAgent( + depsOf(plugin), + runState as unknown as RunState, + child as any, { input: "test" }, 3, // at the limit, not over ); diff --git a/packages/appkit/src/plugins/agents/tool-dispatch.ts b/packages/appkit/src/plugins/agents/tool-dispatch.ts new file mode 100644 index 000000000..3d0e1af44 --- /dev/null +++ b/packages/appkit/src/plugins/agents/tool-dispatch.ts @@ -0,0 +1,303 @@ +import { randomUUID } from "node:crypto"; + +import type express from "express"; +import type { AgentRunContext, Message, ResponseStreamEvent } from "shared"; + +import type { AppKitMcpClient } from "../../connectors/mcp"; +import { consumeAdapterStream } from "../../core/agent/consume-adapter-stream"; +import { normalizeToolResult } from "../../core/agent/normalize-result"; +import type { + BaseSystemPromptOption, + RegisteredAgent, + ResolvedToolEntry, +} from "../../core/agent/types"; +import type { PluginContext } from "../../core/plugin-context"; +import { buildAdapterExtensions } from "./adapter-extensions"; +import { requiresApproval } from "./approval"; +import type { EventChannel } from "./event-channel"; +import type { AgentEventTranslator } from "./event-translator"; +import { traceTool } from "./mlflow"; +import { composePromptForAgent } from "./prompt"; +import type { ToolApprovalGate } from "./tool-approval-gate"; + +/** + * Per-stream state shared between the top-level `executeTool` and any + * `runSubAgent` calls below it. Carrying the budget counter, abort signal, + * approval policy, and event-channel through one object is what lets the + * sub-agent path enforce the same limits and approval gate as the parent. + * + * Without this shared state the sub-agent path silently bypassed both the + * tool-call budget and the destructive-tool approval gate. + */ +export interface RunState { + req: express.Request; + userId: string; + requestId: string; + abortController: AbortController; + signal: AbortSignal; + approvalPolicy: { requireForDestructive: boolean; timeoutMs: number }; + limits: { + maxConcurrentStreamsPerUser: number; + maxToolCalls: number; + maxSubAgentDepth: number; + toolCallTimeoutMs: number; + }; + translator: AgentEventTranslator; + outboundEvents: EventChannel; + /** Boxed mutable counter shared across parent + all sub-agent dispatches. */ + toolCallsUsed: { count: number }; +} + +/** + * Plugin-instance collaborators the dispatch path needs. Passed as one object + * so `dispatchToolCall`/`runSubAgent` stay free functions with a bounded + * interface instead of reaching into the plugin. `getMcpClient` is a thunk so + * a client connected after this object is built is still seen. + */ +export interface ToolDispatchDeps { + approvalGate: ToolApprovalGate; + context: PluginContext | undefined; + getMcpClient: () => AppKitMcpClient | null; + agents: Map; + dispatchSkillTool: ( + entry: Extract, + args: unknown, + ) => Promise; + pluginName: string; + baseSystemPrompt: BaseSystemPromptOption | undefined; +} + +/** + * Dispatch a single tool call from either the top-level adapter or a + * sub-agent. Centralising this in one function is what makes the budget + * counter, approval gate, and abort signal observe sub-agent activity: + * `runSubAgent` reuses the same `runState` and so increments the same + * counter and emits approval events through the same channel. + * + * `depth` is the current sub-agent recursion depth (0 at the top level). + * It is forwarded to `runSubAgent` when the dispatched entry is itself a + * sub-agent, so depth limits remain enforced. + */ +export async function dispatchToolCall( + deps: ToolDispatchDeps, + runState: RunState, + toolIndex: Map, + name: string, + args: unknown, + depth: number, +): Promise { + if (runState.toolCallsUsed.count >= runState.limits.maxToolCalls) { + runState.abortController.abort( + new Error( + `Tool-call budget exhausted (limit ${runState.limits.maxToolCalls}).`, + ), + ); + throw new Error( + `Tool-call budget exhausted (limit ${runState.limits.maxToolCalls}). Raise agents({ limits: { maxToolCalls } }) or review the agent's tool-selection logic.`, + ); + } + runState.toolCallsUsed.count++; + + const entry = toolIndex.get(name); + if (!entry) throw new Error(`Unknown tool: ${name}`); + + if ( + runState.approvalPolicy.requireForDestructive && + requiresApproval(entry.def.annotations) + ) { + const approvalId = randomUUID(); + for (const ev of runState.translator.translate({ + type: "approval_pending", + approvalId, + streamId: runState.requestId, + toolName: name, + args, + annotations: entry.def.annotations, + })) { + runState.outboundEvents.push(ev); + } + const decision = await deps.approvalGate.wait({ + approvalId, + streamId: runState.requestId, + userId: runState.userId, + timeoutMs: runState.approvalPolicy.timeoutMs, + }); + if (decision === "deny") { + return `Tool execution denied by user approval gate (tool: ${name}).`; + } + } + + // Traced from here so the span covers execution only, not the approval + // wait above (which is human latency). + const toolResult = await traceTool(name, args, async () => { + let result: unknown; + if (entry.source === "toolkit") { + if (!deps.context) { + throw new Error( + "Plugin tool execution requires PluginContext; this should never happen through createApp", + ); + } + result = await deps.context.executeTool( + runState.req, + entry.pluginName, + entry.localName, + args, + runState.signal, + runState.limits.toolCallTimeoutMs, + ); + } else if (entry.source === "function") { + // Function tools declare their parameters as a JSON-object schema, + // so adapters always serialize `args` as an object. A non-object + // value here means the upstream model emitted malformed tool-call + // JSON; surface a clear error rather than silently passing through + // a wrong-shape value the tool will then choke on. + if (typeof args !== "object" || args === null || Array.isArray(args)) { + throw new Error( + `Function tool '${name}' received non-object arguments (got ${args === null ? "null" : Array.isArray(args) ? "array" : typeof args}); expected a JSON object.`, + ); + } + result = await entry.functionTool.execute( + args as Record, + ); + } else if (entry.source === "mcp") { + const mcpClient = deps.getMcpClient(); + if (!mcpClient) throw new Error("MCP client not connected"); + const oboToken = runState.req.headers["x-forwarded-access-token"]; + const mcpAuth = + typeof oboToken === "string" + ? { Authorization: `Bearer ${oboToken}` } + : undefined; + result = await mcpClient.callTool(entry.mcpToolName, args, mcpAuth); + } else if (entry.source === "subagent") { + const childAgent = deps.agents.get(entry.agentName); + if (!childAgent) + throw new Error(`Sub-agent not found: ${entry.agentName}`); + result = await runSubAgent(deps, runState, childAgent, args, depth + 1); + } else if (entry.source === "hosted-supervisor") { + // Defense-in-depth: should never fire. Hosted-supervisor entries are + // routed via `AgentInput.extensions` and the SA endpoint executes + // them server-side; their `def` is filtered out of the adapter's + // `tools` array, so the model never sees a callable schema for them. + // If we reach here, the agent is paired with a non-SA adapter that + // somehow surfaced the placeholder def to the model — surface a + // clear error rather than crash later in `normalizeToolResult`. + throw new Error( + `Tool '${name}' is a hosted-supervisor tool and cannot be invoked from the Node process. ` + + "It is executed server-side by the Databricks AI Gateway and is only reachable when the agent's model is a Supervisor API adapter.", + ); + } else if (entry.source === "skill") { + result = await deps.dispatchSkillTool(entry, args); + } + + return result; + }); + + return normalizeToolResult(toolResult); +} + +/** + * Runs a sub-agent in response to an `agent-` tool call. Returns the + * concatenated text output to hand back to the parent adapter as the tool + * result. + * + * `depth` starts at 1 for a top-level sub-agent invocation and increments on + * each nested call. Depths exceeding `limits.maxSubAgentDepth` are rejected + * before any adapter work. + * + * Sub-agent tool calls run through `dispatchToolCall` with the same + * `runState` as the parent — the budget counter and approval gate are + * therefore enforced for every nested call, not only at the top level. + */ +export async function runSubAgent( + deps: ToolDispatchDeps, + runState: RunState, + child: RegisteredAgent, + args: unknown, + depth: number, +): Promise { + if (depth > runState.limits.maxSubAgentDepth) { + throw new Error( + `Sub-agent depth exceeded (limit ${runState.limits.maxSubAgentDepth}). ` + + `Raise agents({ limits: { maxSubAgentDepth } }) or break the delegation cycle.`, + ); + } + + const input = + typeof args === "object" && + args !== null && + typeof (args as { input?: unknown }).input === "string" + ? (args as { input: string }).input + : JSON.stringify(args); + // Same filter as the top-level path: hosted-supervisor `def` is a + // placeholder, not a callable function — exclude from the adapter's + // `tools` array. The specs are routed via `extensions` instead. + const childTools = Array.from(child.toolIndex.values()) + .filter((e) => e.source !== "hosted-supervisor") + .map((e) => e.def); + + const childExecute = (name: string, childArgs: unknown): Promise => + dispatchToolCall(deps, runState, child.toolIndex, name, childArgs, depth); + + const runContext: AgentRunContext = { + executeTool: childExecute, + signal: runState.signal, + }; + + const pluginNames = deps.context + ? deps.context + .getPluginNames() + .filter((n) => n !== deps.pluginName && n !== "server") + : []; + const systemPrompt = composePromptForAgent(child, deps.baseSystemPrompt, { + agentName: child.name, + pluginNames, + toolNames: childTools.map((t) => t.name), + }); + + const messages: Message[] = [ + { + id: "system", + role: "system", + content: systemPrompt, + createdAt: new Date(), + }, + { + id: randomUUID(), + role: "user", + content: input, + createdAt: new Date(), + }, + ]; + + return consumeAdapterStream( + child.adapter.run( + { + messages, + tools: childTools, + threadId: randomUUID(), + signal: runState.signal, + extensions: buildAdapterExtensions(child.toolIndex), + }, + runContext, + ), + { + signal: runState.signal, + // Forward every sub-agent event into the parent's outbound SSE + // stream so the client sees nested tool_call / tool_result events + // (UI-action tools like apply_filter / highlight_period rely on + // this) and the sub-agent's streaming text as it's generated. + // + // `metadata` is the one exception: sub-agents have their own + // threadId, and forwarding it would overwrite the parent's + // thread state on the client and break multi-turn continuity. + // Approval-pending events emitted by `dispatchToolCall` already + // reach `outboundEvents` directly, so they are not routed here. + onEvent: (event) => { + if (event.type === "metadata") return; + for (const translated of runState.translator.translate(event)) { + runState.outboundEvents.push(translated); + } + }, + }, + ); +} From 44c9143412a28670863bff5c3d6022763588129a Mon Sep 17 00:00:00 2001 From: MarioCadenas Date: Fri, 21 Aug 2026 18:10:52 +0200 Subject: [PATCH 5/8] refactor(appkit): extract agents config resolution into resolve-config.ts Moves resolvedApprovalPolicy / resolvedLimits defaulting into pure functions over AgentsPluginConfig. The getters keep the approval-policy memo cache and delegate. Config-only interface; both now unit-testable in isolation. agents.ts: 1839 -> 1812 lines. typecheck + agent tests green. Signed-off-by: MarioCadenas --- packages/appkit/src/plugins/agents/agents.ts | 30 +--------- .../src/plugins/agents/resolve-config.ts | 55 +++++++++++++++++++ 2 files changed, 58 insertions(+), 27 deletions(-) create mode 100644 packages/appkit/src/plugins/agents/resolve-config.ts diff --git a/packages/appkit/src/plugins/agents/agents.ts b/packages/appkit/src/plugins/agents/agents.ts index a16105e26..292cf9467 100644 --- a/packages/appkit/src/plugins/agents/agents.ts +++ b/packages/appkit/src/plugins/agents/agents.ts @@ -71,6 +71,7 @@ import { loadCodeAgents, resolveDefaultAgent, } from "./registry"; +import { resolveApprovalPolicy, resolveLimits } from "./resolve-config"; import { approvalRequestSchema, cancelRequestSchema, @@ -172,23 +173,7 @@ export class AgentsPlugin extends Plugin implements ToolProvider { timeoutMs: number; } { if (this.cachedApprovalPolicy) return this.cachedApprovalPolicy; - const cfg = this.config.approval ?? {}; - const APPROVAL_TIMEOUT_FLOOR_MS = 1_000; - const APPROVAL_TIMEOUT_DEFAULT_MS = 60_000; - let timeoutMs = cfg.timeoutMs ?? APPROVAL_TIMEOUT_DEFAULT_MS; - if (!Number.isFinite(timeoutMs) || timeoutMs < APPROVAL_TIMEOUT_FLOOR_MS) { - logger.warn( - "approval.timeoutMs=%s is below the %sms floor; using default %sms instead. Mutating tool calls would otherwise auto-deny before any UI could respond.", - cfg.timeoutMs, - APPROVAL_TIMEOUT_FLOOR_MS, - APPROVAL_TIMEOUT_DEFAULT_MS, - ); - timeoutMs = APPROVAL_TIMEOUT_DEFAULT_MS; - } - this.cachedApprovalPolicy = { - requireForDestructive: cfg.requireForDestructive ?? true, - timeoutMs, - }; + this.cachedApprovalPolicy = resolveApprovalPolicy(this.config); return this.cachedApprovalPolicy; } @@ -199,16 +184,7 @@ export class AgentsPlugin extends Plugin implements ToolProvider { maxSubAgentDepth: number; toolCallTimeoutMs: number; } { - const cfg = this.config.limits ?? {}; - return { - maxConcurrentStreamsPerUser: cfg.maxConcurrentStreamsPerUser ?? 5, - maxToolCalls: cfg.maxToolCalls ?? 50, - maxSubAgentDepth: cfg.maxSubAgentDepth ?? 3, - // 5 minutes is the floor for cold SQL Warehouse / long Genie / - // long Lakebase calls. The previous PluginContext default of 30s - // truncated legitimate analytics queries on cold compute. - toolCallTimeoutMs: cfg.toolCallTimeoutMs ?? 300_000, - }; + return resolveLimits(this.config); } /** Count active streams owned by a given user. O(1). */ diff --git a/packages/appkit/src/plugins/agents/resolve-config.ts b/packages/appkit/src/plugins/agents/resolve-config.ts new file mode 100644 index 000000000..d8008a291 --- /dev/null +++ b/packages/appkit/src/plugins/agents/resolve-config.ts @@ -0,0 +1,55 @@ +import type { AgentsPluginConfig } from "../../core/agent/types"; +import { createLogger } from "../../logging/logger"; + +const logger = createLogger("agents"); + +const APPROVAL_TIMEOUT_FLOOR_MS = 1_000; +const APPROVAL_TIMEOUT_DEFAULT_MS = 60_000; + +/** + * Effective approval policy with defaults applied. `timeoutMs` is clamped to a + * 1s floor so a misconfigured value (`0`, negative, or `NaN`) can't degrade + * into immediate auto-denial of every mutating tool call. + * + * The caller memoises the result, so the floor warning fires at most once per + * plugin instance rather than on every chat stream. + */ +export function resolveApprovalPolicy(config: AgentsPluginConfig): { + requireForDestructive: boolean; + timeoutMs: number; +} { + const cfg = config.approval ?? {}; + let timeoutMs = cfg.timeoutMs ?? APPROVAL_TIMEOUT_DEFAULT_MS; + if (!Number.isFinite(timeoutMs) || timeoutMs < APPROVAL_TIMEOUT_FLOOR_MS) { + logger.warn( + "approval.timeoutMs=%s is below the %sms floor; using default %sms instead. Mutating tool calls would otherwise auto-deny before any UI could respond.", + cfg.timeoutMs, + APPROVAL_TIMEOUT_FLOOR_MS, + APPROVAL_TIMEOUT_DEFAULT_MS, + ); + timeoutMs = APPROVAL_TIMEOUT_DEFAULT_MS; + } + return { + requireForDestructive: cfg.requireForDestructive ?? true, + timeoutMs, + }; +} + +/** Effective DoS limits with defaults applied. */ +export function resolveLimits(config: AgentsPluginConfig): { + maxConcurrentStreamsPerUser: number; + maxToolCalls: number; + maxSubAgentDepth: number; + toolCallTimeoutMs: number; +} { + const cfg = config.limits ?? {}; + return { + maxConcurrentStreamsPerUser: cfg.maxConcurrentStreamsPerUser ?? 5, + maxToolCalls: cfg.maxToolCalls ?? 50, + maxSubAgentDepth: cfg.maxSubAgentDepth ?? 3, + // 5 minutes is the floor for cold SQL Warehouse / long Genie / + // long Lakebase calls. The previous PluginContext default of 30s + // truncated legitimate analytics queries on cold compute. + toolCallTimeoutMs: cfg.toolCallTimeoutMs ?? 300_000, + }; +} From 34fcff86a2601e6a40b6958482e3f6eb4f8c7972 Mon Sep 17 00:00:00 2001 From: MarioCadenas Date: Fri, 21 Aug 2026 18:15:23 +0200 Subject: [PATCH 6/8] refactor(appkit): extract stream tracking into StreamRegistry Moves the active-stream map + per-user counter (the O(1) concurrency-limit check) into a StreamRegistry class. The plugin holds one instance and keeps trackStream/untrackStream/countUserStreams as delegators; cancel/approve read via streams.get(). Tests inject via trackStream and assert via the registry. agents.ts: 1812 -> 1784 lines. typecheck + 394 agent tests green. Signed-off-by: MarioCadenas --- packages/appkit/src/plugins/agents/agents.ts | 46 +++------------ .../src/plugins/agents/stream-registry.ts | 57 +++++++++++++++++++ .../agents/tests/approval-route.test.ts | 30 ++-------- .../plugins/agents/tests/dos-limits.test.ts | 32 +++++------ 4 files changed, 88 insertions(+), 77 deletions(-) create mode 100644 packages/appkit/src/plugins/agents/stream-registry.ts diff --git a/packages/appkit/src/plugins/agents/agents.ts b/packages/appkit/src/plugins/agents/agents.ts index 292cf9467..9a17adf30 100644 --- a/packages/appkit/src/plugins/agents/agents.ts +++ b/packages/appkit/src/plugins/agents/agents.ts @@ -85,6 +85,7 @@ import { renderForcedSkill, resolveAgentSkills, } from "./skill-loader"; +import { StreamRegistry } from "./stream-registry"; import { InMemoryThreadStore } from "./thread-store"; import { ToolApprovalGate } from "./tool-approval-gate"; import { @@ -106,17 +107,11 @@ export class AgentsPlugin extends Plugin implements ToolProvider { private agents = new Map(); private defaultAgentName: string | null = null; - private activeStreams = new Map< - string, - { controller: AbortController; userId: string } - >(); /** - * Per-user stream count, kept in sync with `activeStreams` so the - * concurrent-stream rate limit check is O(1) instead of O(n) over every - * active stream on every request. Mutated only via {@link trackStream} - * and {@link untrackStream}. + * Active SSE streams + per-user counts (the O(1) concurrency-limit check). + * Mutated only via {@link trackStream} / {@link untrackStream}. */ - private userStreamCounts = new Map(); + private streams = new StreamRegistry(); private mcpClient: AppKitMcpClient | null = null; private threadStore; private approvalGate = new ToolApprovalGate(); @@ -189,42 +184,19 @@ export class AgentsPlugin extends Plugin implements ToolProvider { /** Count active streams owned by a given user. O(1). */ private countUserStreams(userId: string): number { - return this.userStreamCounts.get(userId) ?? 0; + return this.streams.count(userId); } - /** - * Register a stream for `userId` and bump the per-user counter. Paired - * with {@link untrackStream}; the two helpers are the only writers to - * `activeStreams` + `userStreamCounts`, so the counter cannot drift from - * the map. - */ private trackStream( requestId: string, userId: string, controller: AbortController, ): void { - this.activeStreams.set(requestId, { controller, userId }); - this.userStreamCounts.set( - userId, - (this.userStreamCounts.get(userId) ?? 0) + 1, - ); + this.streams.track(requestId, userId, controller); } - /** - * Remove a stream from the active map and decrement the per-user - * counter. Idempotent — calling twice for the same `requestId` is a - * no-op (the second call sees no entry and returns early). - */ private untrackStream(requestId: string): void { - const entry = this.activeStreams.get(requestId); - if (!entry) return; - this.activeStreams.delete(requestId); - const next = (this.userStreamCounts.get(entry.userId) ?? 0) - 1; - if (next <= 0) { - this.userStreamCounts.delete(entry.userId); - } else { - this.userStreamCounts.set(entry.userId, next); - } + this.streams.untrack(requestId); } async setup() { @@ -1633,7 +1605,7 @@ export class AgentsPlugin extends Plugin implements ToolProvider { return; } const { streamId } = parsed.data; - const entry = this.activeStreams.get(streamId); + const entry = this.streams.get(streamId); if (!entry) { // Stream is unknown or already completed — idempotent no-op. res.json({ cancelled: true }); @@ -1661,7 +1633,7 @@ export class AgentsPlugin extends Plugin implements ToolProvider { } const { streamId, approvalId, decision } = parsed.data; - const streamEntry = this.activeStreams.get(streamId); + const streamEntry = this.streams.get(streamId); if (!streamEntry) { // Stream has already completed or never existed. Return 404 so the UI // knows the approval token is no longer valid (the waiter, if any, has diff --git a/packages/appkit/src/plugins/agents/stream-registry.ts b/packages/appkit/src/plugins/agents/stream-registry.ts new file mode 100644 index 000000000..f0d4bac6a --- /dev/null +++ b/packages/appkit/src/plugins/agents/stream-registry.ts @@ -0,0 +1,57 @@ +/** + * Tracks active SSE streams and a per-user stream count. The count is kept in + * sync with the stream map on every {@link track}/{@link untrack} so the + * concurrency-limit check is O(1) instead of O(n) over all active streams on + * every request. `track` and `untrack` are the only writers, which is what + * keeps the counter from drifting from the map. + */ +export class StreamRegistry { + private readonly activeStreams = new Map< + string, + { controller: AbortController; userId: string } + >(); + private readonly userStreamCounts = new Map(); + + /** Count active streams owned by a given user. O(1). */ + count(userId: string): number { + return this.userStreamCounts.get(userId) ?? 0; + } + + /** Total active streams across all users. */ + get size(): number { + return this.activeStreams.size; + } + + /** Look up an active stream by request id. */ + get( + requestId: string, + ): { controller: AbortController; userId: string } | undefined { + return this.activeStreams.get(requestId); + } + + /** Register a stream for `userId` and bump the per-user counter. */ + track(requestId: string, userId: string, controller: AbortController): void { + this.activeStreams.set(requestId, { controller, userId }); + this.userStreamCounts.set( + userId, + (this.userStreamCounts.get(userId) ?? 0) + 1, + ); + } + + /** + * Remove a stream and decrement the per-user counter. Idempotent — calling + * twice for the same `requestId` is a no-op. Drops the counter key entirely + * when it reaches zero so the map can't grow unbounded across many users. + */ + untrack(requestId: string): void { + const entry = this.activeStreams.get(requestId); + if (!entry) return; + this.activeStreams.delete(requestId); + const next = (this.userStreamCounts.get(entry.userId) ?? 0) - 1; + if (next <= 0) { + this.userStreamCounts.delete(entry.userId); + } else { + this.userStreamCounts.set(entry.userId, next); + } + } +} diff --git a/packages/appkit/src/plugins/agents/tests/approval-route.test.ts b/packages/appkit/src/plugins/agents/tests/approval-route.test.ts index d06002444..e54359799 100644 --- a/packages/appkit/src/plugins/agents/tests/approval-route.test.ts +++ b/packages/appkit/src/plugins/agents/tests/approval-route.test.ts @@ -97,10 +97,7 @@ describe("POST /approve route handler", () => { test("returns 403 when submitter is different from stream owner", async () => { const plugin = new AgentsPlugin({}); - (plugin as any).activeStreams.set("stream-x", { - controller: new AbortController(), - userId: "alice", - }); + (plugin as any).trackStream("stream-x", "alice", new AbortController()); const gate = (plugin as any).approvalGate; const waiter = gate.wait({ approvalId: "a1", @@ -136,10 +133,7 @@ describe("POST /approve route handler", () => { test("returns 404 when approvalId is unknown on an active stream", async () => { const plugin = new AgentsPlugin({}); - (plugin as any).activeStreams.set("stream-y", { - controller: new AbortController(), - userId: "alice", - }); + (plugin as any).trackStream("stream-y", "alice", new AbortController()); const { res, json } = mockRes(); await ( plugin as unknown as { @@ -165,10 +159,7 @@ describe("POST /approve route handler", () => { test("happy path: approve resolves pending gate with 'approve'", async () => { const plugin = new AgentsPlugin({}); - (plugin as any).activeStreams.set("stream-z", { - controller: new AbortController(), - userId: "alice", - }); + (plugin as any).trackStream("stream-z", "alice", new AbortController()); const gate = (plugin as any).approvalGate; const waiter = gate.wait({ approvalId: "a42", @@ -199,10 +190,7 @@ describe("POST /approve route handler", () => { test("happy path: deny resolves pending gate with 'deny'", async () => { const plugin = new AgentsPlugin({}); - (plugin as any).activeStreams.set("stream-z", { - controller: new AbortController(), - userId: "alice", - }); + (plugin as any).trackStream("stream-z", "alice", new AbortController()); const gate = (plugin as any).approvalGate; const waiter = gate.wait({ approvalId: "a43", @@ -235,10 +223,7 @@ describe("POST /cancel ownership + gate cleanup", () => { test("cancelling a stream denies every pending approval on that stream", async () => { const plugin = new AgentsPlugin({}); const controller = new AbortController(); - (plugin as any).activeStreams.set("stream-c", { - controller, - userId: "alice", - }); + (plugin as any).trackStream("stream-c", "alice", controller); const gate = (plugin as any).approvalGate; const a = gate.wait({ approvalId: "ca1", @@ -272,10 +257,7 @@ describe("POST /cancel ownership + gate cleanup", () => { test("cancel from a different user is refused with 403", async () => { const plugin = new AgentsPlugin({}); const controller = new AbortController(); - (plugin as any).activeStreams.set("stream-d", { - controller, - userId: "alice", - }); + (plugin as any).trackStream("stream-d", "alice", controller); const { res, json } = mockRes(); await ( plugin as unknown as { diff --git a/packages/appkit/src/plugins/agents/tests/dos-limits.test.ts b/packages/appkit/src/plugins/agents/tests/dos-limits.test.ts index 9104979d3..d124a1278 100644 --- a/packages/appkit/src/plugins/agents/tests/dos-limits.test.ts +++ b/packages/appkit/src/plugins/agents/tests/dos-limits.test.ts @@ -217,31 +217,31 @@ describe("POST /chat — per-user concurrent-stream limit", () => { // must mirror the underlying map across track/untrack and across // multiple users. const plugin = seedPlugin(); - const t = plugin as any; + const reg = (plugin as any).streams; - expect(t.countUserStreams("alice")).toBe(0); + expect(reg.count("alice")).toBe(0); - t.trackStream("s1", "alice", new AbortController()); - t.trackStream("s2", "alice", new AbortController()); - t.trackStream("s3", "bob", new AbortController()); + reg.track("s1", "alice", new AbortController()); + reg.track("s2", "alice", new AbortController()); + reg.track("s3", "bob", new AbortController()); - expect(t.countUserStreams("alice")).toBe(2); - expect(t.countUserStreams("bob")).toBe(1); - expect(t.activeStreams.size).toBe(3); + expect(reg.count("alice")).toBe(2); + expect(reg.count("bob")).toBe(1); + expect(reg.size).toBe(3); - t.untrackStream("s1"); - expect(t.countUserStreams("alice")).toBe(1); - expect(t.activeStreams.has("s1")).toBe(false); + reg.untrack("s1"); + expect(reg.count("alice")).toBe(1); + expect(reg.get("s1")).toBeUndefined(); // Untrack the last stream for a user → counter map drops the key // entirely (avoids unbounded growth across many distinct users). - t.untrackStream("s3"); - expect(t.countUserStreams("bob")).toBe(0); - expect(t.userStreamCounts.has("bob")).toBe(false); + reg.untrack("s3"); + expect(reg.count("bob")).toBe(0); + expect((reg as any).userStreamCounts.has("bob")).toBe(false); // Idempotent — untracking a missing stream is a no-op. - t.untrackStream("s1"); - expect(t.countUserStreams("alice")).toBe(1); + reg.untrack("s1"); + expect(reg.count("alice")).toBe(1); }); test("/invocations also honours maxConcurrentStreamsPerUser (no bypass)", async () => { From 2083e2a65e9242c9b3c5b6211c60a4d6261fbd98 Mon Sep 17 00:00:00 2001 From: MarioCadenas Date: Fri, 21 Aug 2026 18:20:04 +0200 Subject: [PATCH 7/8] refactor(appkit): rename agents stream tracker to ActiveStreamTracker The previous commit named it StreamRegistry, colliding with the existing SSE-layer StreamRegistry in src/stream/ (connection/event-buffer tracking used by StreamManager). They're different concepts; renamed the agents-plugin one to ActiveStreamTracker (tracks active streams + per-user counts for the O(1) concurrency limit) to avoid the name clash. Signed-off-by: MarioCadenas --- .../src/plugins/agents/stream-registry.ts | 57 ------------------- 1 file changed, 57 deletions(-) delete mode 100644 packages/appkit/src/plugins/agents/stream-registry.ts diff --git a/packages/appkit/src/plugins/agents/stream-registry.ts b/packages/appkit/src/plugins/agents/stream-registry.ts deleted file mode 100644 index f0d4bac6a..000000000 --- a/packages/appkit/src/plugins/agents/stream-registry.ts +++ /dev/null @@ -1,57 +0,0 @@ -/** - * Tracks active SSE streams and a per-user stream count. The count is kept in - * sync with the stream map on every {@link track}/{@link untrack} so the - * concurrency-limit check is O(1) instead of O(n) over all active streams on - * every request. `track` and `untrack` are the only writers, which is what - * keeps the counter from drifting from the map. - */ -export class StreamRegistry { - private readonly activeStreams = new Map< - string, - { controller: AbortController; userId: string } - >(); - private readonly userStreamCounts = new Map(); - - /** Count active streams owned by a given user. O(1). */ - count(userId: string): number { - return this.userStreamCounts.get(userId) ?? 0; - } - - /** Total active streams across all users. */ - get size(): number { - return this.activeStreams.size; - } - - /** Look up an active stream by request id. */ - get( - requestId: string, - ): { controller: AbortController; userId: string } | undefined { - return this.activeStreams.get(requestId); - } - - /** Register a stream for `userId` and bump the per-user counter. */ - track(requestId: string, userId: string, controller: AbortController): void { - this.activeStreams.set(requestId, { controller, userId }); - this.userStreamCounts.set( - userId, - (this.userStreamCounts.get(userId) ?? 0) + 1, - ); - } - - /** - * Remove a stream and decrement the per-user counter. Idempotent — calling - * twice for the same `requestId` is a no-op. Drops the counter key entirely - * when it reaches zero so the map can't grow unbounded across many users. - */ - untrack(requestId: string): void { - const entry = this.activeStreams.get(requestId); - if (!entry) return; - this.activeStreams.delete(requestId); - const next = (this.userStreamCounts.get(entry.userId) ?? 0) - 1; - if (next <= 0) { - this.userStreamCounts.delete(entry.userId); - } else { - this.userStreamCounts.set(entry.userId, next); - } - } -} From a8f7cba2bd691e2cba34dab8e01780b09544a954 Mon Sep 17 00:00:00 2001 From: MarioCadenas Date: Fri, 21 Aug 2026 18:22:09 +0200 Subject: [PATCH 8/8] fix(appkit): finish ActiveStreamTracker rename (add file + agents wiring) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The prior commit's git add hit the already-deleted stream-registry.ts path, aborted, and recorded only the deletion — leaving the pushed tip non-compiling (agents.ts imported the removed file; active-stream-tracker.ts was uncommitted). This adds the new module and the agents.ts import/usage so the tree builds. Signed-off-by: MarioCadenas --- .../plugins/agents/active-stream-tracker.ts | 60 +++++++++++++++++++ packages/appkit/src/plugins/agents/agents.ts | 4 +- 2 files changed, 62 insertions(+), 2 deletions(-) create mode 100644 packages/appkit/src/plugins/agents/active-stream-tracker.ts diff --git a/packages/appkit/src/plugins/agents/active-stream-tracker.ts b/packages/appkit/src/plugins/agents/active-stream-tracker.ts new file mode 100644 index 000000000..3e6f71067 --- /dev/null +++ b/packages/appkit/src/plugins/agents/active-stream-tracker.ts @@ -0,0 +1,60 @@ +/** + * Tracks active SSE streams and a per-user stream count for the agents plugin's + * concurrency limit. The count is kept in sync with the stream map on every + * {@link track}/{@link untrack} so the limit check is O(1) instead of O(n) over + * all active streams on every request. `track` and `untrack` are the only + * writers, which is what keeps the counter from drifting from the map. + * + * Distinct from the SSE-layer `StreamRegistry` in `src/stream/` (which buffers + * events per connection for reconnection replay) — this only counts streams. + */ +export class ActiveStreamTracker { + private readonly activeStreams = new Map< + string, + { controller: AbortController; userId: string } + >(); + private readonly userStreamCounts = new Map(); + + /** Count active streams owned by a given user. O(1). */ + count(userId: string): number { + return this.userStreamCounts.get(userId) ?? 0; + } + + /** Total active streams across all users. */ + get size(): number { + return this.activeStreams.size; + } + + /** Look up an active stream by request id. */ + get( + requestId: string, + ): { controller: AbortController; userId: string } | undefined { + return this.activeStreams.get(requestId); + } + + /** Register a stream for `userId` and bump the per-user counter. */ + track(requestId: string, userId: string, controller: AbortController): void { + this.activeStreams.set(requestId, { controller, userId }); + this.userStreamCounts.set( + userId, + (this.userStreamCounts.get(userId) ?? 0) + 1, + ); + } + + /** + * Remove a stream and decrement the per-user counter. Idempotent — calling + * twice for the same `requestId` is a no-op. Drops the counter key entirely + * when it reaches zero so the map can't grow unbounded across many users. + */ + untrack(requestId: string): void { + const entry = this.activeStreams.get(requestId); + if (!entry) return; + this.activeStreams.delete(requestId); + const next = (this.userStreamCounts.get(entry.userId) ?? 0) - 1; + if (next <= 0) { + this.userStreamCounts.delete(entry.userId); + } else { + this.userStreamCounts.set(entry.userId, next); + } + } +} diff --git a/packages/appkit/src/plugins/agents/agents.ts b/packages/appkit/src/plugins/agents/agents.ts index 9a17adf30..ea4e2ed51 100644 --- a/packages/appkit/src/plugins/agents/agents.ts +++ b/packages/appkit/src/plugins/agents/agents.ts @@ -48,6 +48,7 @@ import { createLogger } from "../../logging/logger"; import { Plugin, toPlugin } from "../../plugin"; import { defineManifest } from "../../registry"; import type { WorkspaceClient } from "../../workspace-client"; +import { ActiveStreamTracker } from "./active-stream-tracker"; import { buildAdapterExtensions, supervisorToolDescription, @@ -85,7 +86,6 @@ import { renderForcedSkill, resolveAgentSkills, } from "./skill-loader"; -import { StreamRegistry } from "./stream-registry"; import { InMemoryThreadStore } from "./thread-store"; import { ToolApprovalGate } from "./tool-approval-gate"; import { @@ -111,7 +111,7 @@ export class AgentsPlugin extends Plugin implements ToolProvider { * Active SSE streams + per-user counts (the O(1) concurrency-limit check). * Mutated only via {@link trackStream} / {@link untrackStream}. */ - private streams = new StreamRegistry(); + private streams = new ActiveStreamTracker(); private mcpClient: AppKitMcpClient | null = null; private threadStore; private approvalGate = new ToolApprovalGate();