diff --git a/.changeset/quiet-rockets-serve.md b/.changeset/quiet-rockets-serve.md new file mode 100644 index 00000000..9593cbf9 --- /dev/null +++ b/.changeset/quiet-rockets-serve.md @@ -0,0 +1,7 @@ +--- +"caplets": minor +--- + +# Hot reload serve config + +Add default hot reload for `caplets serve`, including live config and Caplet file reconciliation without restarting the MCP process. diff --git a/README.md b/README.md index 1e72d1f8..ab1417ab 100644 --- a/README.md +++ b/README.md @@ -138,8 +138,10 @@ caplets config paths caplets config paths --json ``` -Caplets validates this file at startup. Config changes take effect after restarting the -Caplets MCP server. +Caplets validates this file at startup and hot reloads config changes while `caplets serve` +is running. Invalid edits are ignored until fixed, so the MCP server keeps serving the last +known-good config instead of dropping every tool because of a transient JSON or validation +error. The optional `$schema` field points editors at the generated JSON Schema in [`schemas/caplets-config.schema.json`](schemas/caplets-config.schema.json). CI verifies that @@ -458,6 +460,12 @@ Configure your MCP client to run Caplets as a stdio server: If your client starts the configured command directly, `caplets` without arguments also starts the MCP server. `serve` is explicit and recommended for clarity. +`caplets serve` watches the effective user config, project config, user Caplet files, and +trusted project Caplet files. Adding, editing, disabling, or removing a Caplet updates the +top-level MCP tool list without restarting Caplets. When an MCP-backed Caplet changes or is +removed, Caplets closes only that affected downstream connection; unrelated Caplets and +their downstream connections keep running. + ## How Agents Use It Caplets initially exposes one MCP tool per enabled Caplet. If the config has `filesystem`, @@ -536,9 +544,10 @@ pnpm schema:check pnpm verify ``` -`pnpm dev` rebuilds on source changes and restarts the local stdio MCP server from +`pnpm dev` rebuilds Caplets source changes and restarts the local stdio MCP server from `dist/index.js`. Use it for local development, not as the command configured in an MCP -client, because build logs are written to stdout. +client, because build logs are written to stdout. Runtime config hot reload is built into +normal `caplets serve` and does not require `pnpm dev`. ## Product Notes diff --git a/docs/plans/2026-05-12-hot-reload.md b/docs/plans/2026-05-12-hot-reload.md new file mode 100644 index 00000000..971117b6 --- /dev/null +++ b/docs/plans/2026-05-12-hot-reload.md @@ -0,0 +1,81 @@ +# Hot Reload Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Make normal `caplets serve` reload config and Caplet file changes without restarting the Caplets MCP process. + +**Architecture:** Introduce an in-process runtime controller that owns the MCP server, registered top-level tool handles, the current registry, backend managers, and filesystem watchers. Reloads parse the effective config exactly as startup does, update registered MCP tools through `RegisteredTool` handles, invalidate only affected backend caches, and keep the last known-good config active when a reload fails. + +**Tech Stack:** TypeScript, Node.js `fs.watch`, `@modelcontextprotocol/sdk` `McpServer.registerTool`, Vitest. + +--- + +## Task 1: Runtime Controller + +**Files:** + +- Create: `src/runtime.ts` +- Modify: `src/index.ts` +- Test: `test/runtime.test.ts` + +- [ ] Add `CapletsRuntime` that creates `McpServer`, `ServerRegistry`, `DownstreamManager`, `OpenApiManager`, and `GraphQLManager`. +- [ ] Track registered top-level Caplet tools in a `Map`. +- [ ] Register initial enabled Caplets during startup. +- [ ] Keep `src/index.ts` as a small CLI/server entrypoint. + +## Task 2: Reload Reconciliation + +**Files:** + +- Modify: `src/runtime.ts` +- Modify: `src/downstream.ts` +- Modify: `src/openapi.ts` +- Modify: `src/graphql.ts` +- Test: `test/runtime.test.ts`, `test/downstream.test.ts`, `test/openapi.test.ts`, `test/graphql.test.ts` + +- [ ] On reload, call existing `loadConfig()` with the same resolved paths used at startup. +- [ ] Add new Caplets with `registerTool`. +- [ ] Update existing Caplets with `RegisteredTool.update`. +- [ ] Remove missing or disabled Caplets with `RegisteredTool.remove`. +- [ ] Add `DownstreamManager.closeServer(serverId)`. +- [ ] Add `OpenApiManager.invalidate(serverId)` and `GraphQLManager.invalidate(serverId)`. +- [ ] Invalidate or close only backends whose normalized config changed. +- [ ] Keep serving the previous registry and tools if reload parsing or validation fails. + +## Task 3: Filesystem Watching + +**Files:** + +- Modify: `src/runtime.ts` +- Test: `test/runtime.test.ts` + +- [ ] Watch the effective user config file, project config file, user Caplets root, and trusted project Caplets root. +- [ ] Use dependency-free `fs.watch` with a 250 ms debounce. +- [ ] Watch directories so new, renamed, and deleted Markdown Caplet files are detected. +- [ ] Recreate watchers after every successful reload because roots can change with `CAPLETS_CONFIG`. +- [ ] Close all watchers during runtime shutdown. + +## Task 4: Documentation + +**Files:** + +- Modify: `README.md` + +- [ ] Document that `caplets serve` hot reloads config and Caplet file changes by default. +- [ ] Document last known-good behavior for invalid edits. +- [ ] Document that changed or removed MCP-backed Caplets close only their affected downstream connection. +- [ ] Keep `pnpm dev` documented as source rebuild/restart tooling, separate from runtime config hot reload. + +## Task 5: Verification + +**Files:** + +- Modify tests as needed. + +- [ ] Run targeted tests for runtime reload and manager invalidation. +- [ ] Run `pnpm format:check`. +- [ ] Run `pnpm lint`. +- [ ] Run `pnpm typecheck`. +- [ ] Run `pnpm schema:check`. +- [ ] Run `pnpm test`. +- [ ] Run `pnpm build`. diff --git a/src/downstream.ts b/src/downstream.ts index 8de4a32a..8f8f9c5c 100644 --- a/src/downstream.ts +++ b/src/downstream.ts @@ -28,25 +28,46 @@ export type CompactTool = { type ManagedConnection = { client: Client; transport: { close(): Promise; onclose?: () => void; onerror?: (error: Error) => void }; + configFingerprint: string; tools?: Tool[]; toolsFetchedAt?: number; restartingAfterDeath?: boolean; + closing?: boolean; }; export class DownstreamManager { private readonly connections = new Map(); + private readonly connecting = new Map(); private readonly restartState = new Map(); constructor( - private readonly registry: ServerRegistry, + private registry: ServerRegistry, private readonly options: { authDir?: string } = {}, ) {} + updateRegistry(registry: ServerRegistry): void { + this.registry = registry; + } + async close(): Promise { - await Promise.allSettled( - [...this.connections.values()].map((connection) => connection.transport.close()), - ); + const connections = [...this.connections.values(), ...this.connecting.values()]; + for (const connection of connections) { + connection.closing = true; + } + await Promise.allSettled(connections.map((connection) => connection.transport.close())); this.connections.clear(); + this.connecting.clear(); + } + + async closeServer(serverId: string): Promise { + const connection = this.connections.get(serverId) ?? this.connecting.get(serverId); + this.connections.delete(serverId); + this.connecting.delete(serverId); + this.restartState.delete(serverId); + if (connection) { + connection.closing = true; + await connection.transport.close(); + } } async checkServer(server: CapletServerConfig): Promise<{ @@ -186,9 +207,23 @@ export class DownstreamManager { } private async connect(server: CapletServerConfig): Promise { + const expectedFingerprint = this.currentServerFingerprint(server); const existing = this.connections.get(server.server); if (existing) { - return existing; + if (existing.configFingerprint !== expectedFingerprint) { + this.connections.delete(server.server); + existing.closing = true; + await existing.transport.close(); + } else { + return existing; + } + } + if (this.currentServerFingerprint(server) !== expectedFingerprint) { + throw staleServerConfigError(server.server); + } + const currentServer = this.currentServer(server.server); + if (!sameServerConfig(currentServer, server)) { + throw staleServerConfigError(server.server); } const restart = this.restartState.get(server.server); if (restart && restart.restartUsed && Date.now() < restart.backoffUntil) { @@ -196,11 +231,28 @@ export class DownstreamManager { } this.registry.setStatus(server.server, "starting"); + let pendingConnection: ManagedConnection | undefined; try { const client = new Client({ name: "caplets", version: "1.0.0" }, { capabilities: {} }); const transport = this.createTransport(server); + const connection: ManagedConnection = { + client, + transport, + configFingerprint: expectedFingerprint, + }; + pendingConnection = connection; + this.connecting.set(server.server, connection); transport.onclose = () => { - this.connections.delete(server.server); + const current = this.connections.get(server.server); + if (current === connection) { + this.connections.delete(server.server); + } + if (connection.closing) { + return; + } + if (current !== connection) { + return; + } this.restartState.set(server.server, { restartUsed: true, backoffUntil: Date.now() + 1_000, @@ -212,6 +264,12 @@ export class DownstreamManager { ); }; transport.onerror = (error: Error) => { + if (connection.closing) { + return; + } + if (this.connections.get(server.server) !== connection) { + return; + } this.registry.setStatus( server.server, "unavailable", @@ -219,11 +277,28 @@ export class DownstreamManager { ); }; await client.connect(transport, { timeout: server.startupTimeoutMs }); - const connection: ManagedConnection = { client, transport }; + if (connection.closing) { + await transport.close(); + throw new CapletsError("SERVER_UNAVAILABLE", `${server.server} connection was closed`); + } + if (this.currentServerFingerprint(server) !== expectedFingerprint) { + connection.closing = true; + await transport.close(); + throw staleServerConfigError(server.server); + } + if (this.connecting.get(server.server) !== connection) { + connection.closing = true; + await transport.close(); + throw new CapletsError("SERVER_UNAVAILABLE", `${server.server} connection was replaced`); + } + this.connecting.delete(server.server); this.connections.set(server.server, connection); this.registry.setStatus(server.server, "available"); return connection; } catch (error) { + if (pendingConnection && this.connecting.get(server.server) === pendingConnection) { + this.connecting.delete(server.server); + } const code = isTimeoutLike(error) ? "SERVER_START_TIMEOUT" : "SERVER_UNAVAILABLE"; const safe = toSafeError(error, code); this.registry.setStatus(server.server, "unavailable", safe); @@ -327,6 +402,34 @@ export class DownstreamManager { this.options.authDir, ); } + + private currentServer(serverId: string): CapletServerConfig { + const current = this.registry.require(serverId); + if (current.backend !== "mcp") { + throw staleServerConfigError(serverId); + } + return current; + } + + private currentServerFingerprint(server: CapletServerConfig): string { + const current = this.currentServer(server.server); + if (!sameServerConfig(current, server)) { + throw staleServerConfigError(server.server); + } + return serializeServerConfig(current); + } +} + +function sameServerConfig(left: CapletServerConfig, right: CapletServerConfig): boolean { + return serializeServerConfig(left) === serializeServerConfig(right); +} + +function serializeServerConfig(server: CapletServerConfig): string { + return JSON.stringify(server); +} + +function staleServerConfigError(serverId: string): CapletsError { + return new CapletsError("SERVER_UNAVAILABLE", `${serverId} configuration changed; retry request`); } function nearbyToolNames(tools: Tool[], needle: string): string[] { diff --git a/src/graphql.ts b/src/graphql.ts index 2857d99a..8a63caa1 100644 --- a/src/graphql.ts +++ b/src/graphql.ts @@ -66,10 +66,18 @@ export class GraphQLManager { private readonly cache = new Map(); constructor( - private readonly registry: ServerRegistry, + private registry: ServerRegistry, private readonly options: { authDir?: string } = {}, ) {} + updateRegistry(registry: ServerRegistry): void { + this.registry = registry; + } + + invalidate(serverId: string): void { + this.cache.delete(serverId); + } + async checkEndpoint(endpoint: GraphQlEndpointConfig): Promise<{ server: string; status: string; diff --git a/src/index.ts b/src/index.ts index b989db39..a20b079d 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,14 +1,6 @@ -import { McpServer } from "@modelcontextprotocol/sdk/server/mcp"; -import { version as packageJsonVersion } from "../package.json"; import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio"; -import { loadConfig } from "./config.js"; -import { DownstreamManager } from "./downstream.js"; -import { errorResult } from "./errors.js"; -import { GraphQLManager } from "./graphql.js"; -import { OpenApiManager } from "./openapi.js"; -import { capabilityDescription, ServerRegistry } from "./registry.js"; -import { generatedToolInputSchema, handleServerTool } from "./tools.js"; import { runCli } from "./cli.js"; +import { CapletsRuntime } from "./runtime.js"; async function main() { if (process.argv[2] && process.argv[2] !== "serve") { @@ -16,50 +8,18 @@ async function main() { return; } - const config = loadConfig(process.env.CAPLETS_CONFIG); - const registry = new ServerRegistry(config); - const downstream = new DownstreamManager(registry); - const openapi = new OpenApiManager(registry); - const graphql = new GraphQLManager(registry); - const server = new McpServer({ - name: "caplets", - version: packageJsonVersion, - }); - - for (const capletServer of registry.enabledServers()) { - server.registerTool( - capletServer.server, - { - title: capletServer.name, - description: capabilityDescription(capletServer), - inputSchema: generatedToolInputSchema, - }, - async (request) => { - try { - return await handleServerTool( - capletServer, - request, - registry, - downstream, - openapi, - graphql, - ); - } catch (error) { - return errorResult(error); - } - }, - ); - } + const runtime = process.env.CAPLETS_CONFIG + ? new CapletsRuntime({ configPath: process.env.CAPLETS_CONFIG }) + : new CapletsRuntime(); const shutdown = async () => { - await downstream.close(); - await server.close(); + await runtime.close(); }; process.once("SIGINT", () => void shutdown().finally(() => process.exit(130))); process.once("SIGTERM", () => void shutdown().finally(() => process.exit(143))); const transport = new StdioServerTransport(); - await server.connect(transport); + await runtime.connect(transport); } main().catch((error) => { diff --git a/src/openapi.ts b/src/openapi.ts index 78b0ddfd..a13c415d 100644 --- a/src/openapi.ts +++ b/src/openapi.ts @@ -49,10 +49,18 @@ export class OpenApiManager { private readonly cache = new Map(); constructor( - private readonly registry: ServerRegistry, + private registry: ServerRegistry, private readonly options: { authDir?: string } = {}, ) {} + updateRegistry(registry: ServerRegistry): void { + this.registry = registry; + } + + invalidate(serverId: string): void { + this.cache.delete(serverId); + } + async checkEndpoint(endpoint: OpenApiEndpointConfig): Promise<{ server: string; status: string; diff --git a/src/runtime.ts b/src/runtime.ts new file mode 100644 index 00000000..91fca95b --- /dev/null +++ b/src/runtime.ts @@ -0,0 +1,452 @@ +import { existsSync, readdirSync, statSync, watch, type FSWatcher } from "node:fs"; +import { dirname, parse } from "node:path"; +import { McpServer, type RegisteredTool } from "@modelcontextprotocol/sdk/server/mcp"; +import type { Transport } from "@modelcontextprotocol/sdk/shared/transport.js"; +import { version as packageJsonVersion } from "../package.json"; +import { + type CapletConfig, + type CapletsConfig, + loadConfig, + resolveCapletsRoot, + resolveConfigPath, + resolveProjectConfigPath, + TRUST_PROJECT_CAPLETS_ENV, + isTrustedEnvEnabled, +} from "./config.js"; +import { DownstreamManager } from "./downstream.js"; +import { errorResult, toSafeError } from "./errors.js"; +import { GraphQLManager } from "./graphql.js"; +import { OpenApiManager } from "./openapi.js"; +import { capabilityDescription, ServerRegistry } from "./registry.js"; +import { generatedToolInputSchema, handleServerTool } from "./tools.js"; + +type ToolServer = Pick; + +type RuntimePaths = { + configPath: string; + projectConfigPath: string; +}; + +type CapletsRuntimeOptions = { + configPath?: string; + projectConfigPath?: string; + authDir?: string; + watchDebounceMs?: number; + server?: ToolServer; + writeErr?: (value: string) => void; +}; + +type WatchedPath = { + path: string; + reason: "config" | "caplets"; +}; + +export class CapletsRuntime { + readonly server: ToolServer; + private registry: ServerRegistry; + private readonly downstream: DownstreamManager; + private readonly openapi: OpenApiManager; + private readonly graphql: GraphQLManager; + private readonly tools = new Map(); + private readonly paths: RuntimePaths; + private readonly watchDebounceMs: number; + private readonly writeErr: (value: string) => void; + private watchers: FSWatcher[] = []; + private reloadTimer: NodeJS.Timeout | undefined; + private watcherRefreshTimer: NodeJS.Timeout | undefined; + private reloading: Promise | undefined; + private pendingReload = false; + private closed = false; + + constructor(options: CapletsRuntimeOptions = {}) { + this.paths = { + configPath: resolveConfigPath(options.configPath), + projectConfigPath: options.projectConfigPath ?? resolveProjectConfigPath(), + }; + const config = loadConfig(this.paths.configPath, this.paths.projectConfigPath); + this.registry = new ServerRegistry(config); + this.downstream = new DownstreamManager(this.registry, selectAuthOptions(options.authDir)); + this.openapi = new OpenApiManager(this.registry, selectAuthOptions(options.authDir)); + this.graphql = new GraphQLManager(this.registry, selectAuthOptions(options.authDir)); + this.server = + options.server ?? + new McpServer({ + name: "caplets", + version: packageJsonVersion, + }); + this.watchDebounceMs = options.watchDebounceMs ?? 250; + this.writeErr = options.writeErr ?? ((value: string) => process.stderr.write(value)); + this.reconcileTools(undefined, config); + this.resetWatchers(); + } + + async connect(transport: Transport): Promise { + await this.server.connect(transport); + } + + scheduleReload(): void { + if (this.closed) { + return; + } + if (this.reloadTimer) { + clearTimeout(this.reloadTimer); + } + this.reloadTimer = setTimeout(() => { + this.reloadTimer = undefined; + void this.reload(); + }, this.watchDebounceMs); + } + + async reload(): Promise { + if (this.closed) { + return false; + } + if (this.reloading) { + this.pendingReload = true; + return await this.reloading; + } + this.reloading = this.reloadUntilSettled().finally(() => { + this.reloading = undefined; + }); + return await this.reloading; + } + + async close(): Promise { + this.closed = true; + try { + if (this.reloadTimer) { + clearTimeout(this.reloadTimer); + this.reloadTimer = undefined; + } + if (this.watcherRefreshTimer) { + clearTimeout(this.watcherRefreshTimer); + this.watcherRefreshTimer = undefined; + } + if (this.reloading) { + await this.reloading; + } + } finally { + this.closeWatchers(); + await this.downstream.close(); + await this.server.close(); + } + } + + currentConfig(): CapletsConfig { + return this.registry.config; + } + + registeredToolIds(): string[] { + return [...this.tools.keys()].sort(); + } + + watchedPaths(): string[] { + return [...new Set(watchedPaths(this.paths).map((entry) => entry.path))].sort(); + } + + private async reloadOnce(): Promise { + if (this.closed) { + return false; + } + let nextConfig: CapletsConfig; + try { + nextConfig = loadConfig(this.paths.configPath, this.paths.projectConfigPath); + } catch (error) { + this.writeErr(`Caplets config reload failed; keeping last known-good config.\n`); + this.writeErr(`${JSON.stringify(toSafeError(error, "CONFIG_INVALID"), null, 2)}\n`); + return false; + } + + if (this.closed) { + return false; + } + const previousConfig = this.registry.config; + const nextRegistry = new ServerRegistry(nextConfig); + this.registry = nextRegistry; + this.downstream.updateRegistry(nextRegistry); + this.openapi.updateRegistry(nextRegistry); + this.graphql.updateRegistry(nextRegistry); + let invalidated = true; + try { + await this.invalidateChangedBackends(previousConfig, nextConfig); + } catch (error) { + invalidated = false; + this.writeErr(`Caplets backend invalidation failed; continuing reload.\n`); + this.writeErr(`${JSON.stringify(toSafeError(error, "INTERNAL_ERROR"), null, 2)}\n`); + } + if (this.closed) { + return false; + } + this.reconcileTools(previousConfig, nextConfig); + this.resetWatchers(); + return invalidated; + } + + private async reloadUntilSettled(): Promise { + let succeeded = true; + do { + this.pendingReload = false; + try { + succeeded = (await this.reloadOnce()) && succeeded; + } catch (err) { + this.writeErr(`Caplets reload failed.\n`); + this.writeErr(`${JSON.stringify(toSafeError(err, "INTERNAL_ERROR"), null, 2)}\n`); + succeeded = false; + } + } while (this.pendingReload && !this.closed); + return succeeded && !this.closed; + } + + private reconcileTools(previous: CapletsConfig | undefined, next: CapletsConfig): void { + const enabled = new Map(nextEnabledServers(next).map((server) => [server.server, server])); + + for (const [serverId, tool] of this.tools) { + const caplet = enabled.get(serverId); + if (!caplet) { + tool.remove(); + this.tools.delete(serverId); + continue; + } + + const previousCaplet = previous ? capletById(previous, serverId) : undefined; + if (!previousCaplet || serializeCaplet(previousCaplet) !== serializeCaplet(caplet)) { + tool.update({ + title: caplet.name, + description: capabilityDescription(caplet), + callback: async (request) => this.handleTool(serverId, request), + enabled: true, + }); + } + } + + for (const caplet of enabled.values()) { + if (this.tools.has(caplet.server)) { + continue; + } + this.tools.set(caplet.server, this.registerCapletTool(caplet)); + } + } + + private registerCapletTool(caplet: CapletConfig): RegisteredTool { + return this.server.registerTool( + caplet.server, + { + title: caplet.name, + description: capabilityDescription(caplet), + inputSchema: generatedToolInputSchema, + }, + async (request) => this.handleTool(caplet.server, request), + ); + } + + private async handleTool(serverId: string, request: unknown): Promise { + try { + const caplet = this.registry.require(serverId); + return await handleServerTool( + caplet, + request, + this.registry, + this.downstream, + this.openapi, + this.graphql, + ); + } catch (error) { + return errorResult(error); + } + } + + private async invalidateChangedBackends( + previous: CapletsConfig, + next: CapletsConfig, + ): Promise { + const previousCaplets = new Map(allCaplets(previous).map((server) => [server.server, server])); + const nextCaplets = new Map(allCaplets(next).map((server) => [server.server, server])); + const changedIds = new Set([...previousCaplets.keys(), ...nextCaplets.keys()]); + + for (const serverId of changedIds) { + const before = previousCaplets.get(serverId); + const after = nextCaplets.get(serverId); + const changed = serializeCaplet(before) !== serializeCaplet(after); + if (!changed) { + continue; + } + if (before?.backend === "mcp") { + await this.downstream.closeServer(serverId); + } + if (before?.backend === "openapi" || after?.backend === "openapi" || !after) { + this.openapi.invalidate(serverId); + } + if (before?.backend === "graphql" || after?.backend === "graphql" || !after) { + this.graphql.invalidate(serverId); + } + } + } + + private resetWatchers(): void { + this.closeWatchers(); + const watched = new Set(); + for (const entry of watchedPaths(this.paths)) { + const watchPath = existsSync(entry.path) ? entry.path : nearestExistingParent(entry.path); + const watchKey = `${entry.reason}:${watchPath}`; + if (!watchPath || watched.has(watchKey)) { + continue; + } + watched.add(watchKey); + try { + this.watchers.push(...this.watchEntry(entry, watchPath)); + } catch (error) { + this.writeErr(`Caplets could not watch ${entry.reason} path ${entry.path}.\n`); + this.writeErr(`${JSON.stringify(toSafeError(error, "SERVER_UNAVAILABLE"), null, 2)}\n`); + } + } + } + + private closeWatchers(): void { + for (const watcher of this.watchers) { + watcher.close(); + } + this.watchers = []; + } + + private watchEntry(entry: WatchedPath, watchPath: string): FSWatcher[] { + if (entry.reason === "caplets" && existsSync(entry.path) && isDirectory(watchPath)) { + return this.watchDirectoryTree(watchPath); + } + + return [ + watch(watchPath, { persistent: true }, (eventType) => { + this.scheduleReload(); + if (eventType === "rename" && entry.reason === "caplets" && existsSync(entry.path)) { + this.scheduleWatcherRefresh(); + } + }), + ]; + } + + private watchDirectoryTree(root: string): FSWatcher[] { + const watchers: FSWatcher[] = []; + const directories = discoverDirectories(root); + for (const directory of directories) { + try { + watchers.push( + watch(directory, { persistent: true }, (eventType) => { + this.scheduleReload(); + if (eventType === "rename") { + this.scheduleWatcherRefresh(); + } + }), + ); + } catch (error) { + // Clean up any watchers created so far before rethrowing + for (const watcher of watchers) { + watcher.close(); + } + throw error; + } + } + return watchers; + } + + private scheduleWatcherRefresh(): void { + if (this.closed) { + return; + } + if (this.watcherRefreshTimer) { + clearTimeout(this.watcherRefreshTimer); + } + this.watcherRefreshTimer = setTimeout(() => { + this.watcherRefreshTimer = undefined; + if (!this.closed) { + this.resetWatchers(); + } + }, this.watchDebounceMs); + } +} + +function selectAuthOptions(authDir: string | undefined): { authDir?: string } { + return authDir ? { authDir } : {}; +} + +function watchedPaths(paths: RuntimePaths): WatchedPath[] { + const entries: WatchedPath[] = [ + { path: dirname(paths.configPath), reason: "config" }, + { path: dirname(paths.projectConfigPath), reason: "config" }, + { path: resolveCapletsRoot(paths.configPath), reason: "caplets" }, + ]; + if (isTrustedEnvEnabled(process.env[TRUST_PROJECT_CAPLETS_ENV])) { + entries.push({ path: dirname(paths.projectConfigPath), reason: "caplets" }); + } + + return uniqueWatchedPaths(entries); +} + +function uniqueWatchedPaths(entries: WatchedPath[]): WatchedPath[] { + const seen = new Set(); + const unique: WatchedPath[] = []; + for (const entry of entries) { + const key = `${entry.reason}:${entry.path}`; + if (seen.has(key)) { + continue; + } + seen.add(key); + unique.push(entry); + } + return unique; +} + +function allCaplets(config: CapletsConfig): CapletConfig[] { + return [ + ...Object.values(config.mcpServers), + ...Object.values(config.openapiEndpoints), + ...Object.values(config.graphqlEndpoints), + ]; +} + +function nextEnabledServers(config: CapletsConfig): CapletConfig[] { + return allCaplets(config).filter((server) => !server.disabled); +} + +function capletById(config: CapletsConfig, serverId: string): CapletConfig | undefined { + return ( + config.mcpServers[serverId] ?? + config.openapiEndpoints[serverId] ?? + config.graphqlEndpoints[serverId] + ); +} + +function serializeCaplet(caplet: CapletConfig | undefined): string { + return JSON.stringify(caplet ?? null); +} + +function nearestExistingParent(path: string): string | undefined { + let candidate = dirname(path); + const root = parse(candidate).root; + while (candidate && candidate !== root) { + if (existsSync(candidate)) { + return candidate; + } + candidate = dirname(candidate); + } + return existsSync(root) ? root : undefined; +} + +function discoverDirectories(root: string): string[] { + if (!isDirectory(root)) { + return []; + } + const directories = [root]; + for (const entry of readdirSync(root, { withFileTypes: true })) { + if (entry.isDirectory()) { + directories.push(...discoverDirectories(`${root}/${entry.name}`)); + } + } + return directories; +} + +function isDirectory(path: string): boolean { + try { + return statSync(path).isDirectory(); + } catch { + return false; + } +} diff --git a/test/downstream.test.ts b/test/downstream.test.ts index b6b6b096..a32e51e0 100644 --- a/test/downstream.test.ts +++ b/test/downstream.test.ts @@ -46,6 +46,171 @@ describe("downstream stdio lifecycle", () => { await manager.close(); } }); + + it("closes one managed stdio server so the next operation reconnects", async () => { + const fixture = join(process.cwd(), "test", "fixtures", "stdio-server.mjs"); + const config = parseConfig({ + mcpServers: { + fixture: { + name: "Fixture", + description: "A useful fixture server.", + command: process.execPath, + args: [fixture], + toolCacheTtlMs: 30_000, + }, + }, + }); + const registry = new ServerRegistry(config); + const manager = new DownstreamManager(registry); + const server = config.mcpServers.fixture!; + + try { + const first = await manager.listTools(server); + expect(first.map((tool) => tool.name).sort()).toEqual(["duplicate", "echo"]); + + await manager.closeServer("fixture"); + + const second = await manager.listTools(server); + expect(second.map((tool) => tool.name).sort()).toEqual(["duplicate", "echo"]); + expect(registry.getStatus("fixture")).toBe("available"); + } finally { + await manager.close(); + } + }); + + it("refuses stale server configs after the registry changes", async () => { + const fixture = join(process.cwd(), "test", "fixtures", "stdio-server.mjs"); + const initialConfig = parseConfig({ + mcpServers: { + fixture: { + name: "Fixture", + description: "A useful fixture server.", + command: process.execPath, + args: [fixture], + toolCacheTtlMs: 30_000, + }, + }, + }); + const nextConfig = parseConfig({ + mcpServers: { + fixture: { + name: "Fixture Reloaded", + description: "A reloaded fixture server.", + command: process.execPath, + args: [fixture], + toolCacheTtlMs: 30_000, + }, + }, + }); + const registry = new ServerRegistry(initialConfig); + const manager = new DownstreamManager(registry); + const staleServer = initialConfig.mcpServers.fixture!; + + try { + manager.updateRegistry(new ServerRegistry(nextConfig)); + + await expect(manager.listTools(staleServer)).rejects.toMatchObject({ + code: "SERVER_UNAVAILABLE", + } satisfies Partial); + + const listed = await manager.listTools(nextConfig.mcpServers.fixture!); + expect(listed.map((tool) => tool.name).sort()).toEqual(["duplicate", "echo"]); + } finally { + await manager.close(); + } + }); + + it("closes an in-flight remote connection before it can be cached", async () => { + const firstInitialize = deferred(); + const releaseFirstInitialize = deferred(); + let initializeCount = 0; + const server = createServer((request: IncomingMessage, response: ServerResponse) => { + let body = ""; + request.setEncoding("utf8"); + request.on("data", (chunk) => { + body += chunk; + }); + request.on("end", () => { + void (async () => { + if (!body) { + response.statusCode = 202; + response.end(); + return; + } + const message = JSON.parse(body) as { id?: number; method?: string }; + response.setHeader("content-type", "application/json"); + if (message.method === "initialize") { + initializeCount += 1; + if (initializeCount === 1) { + firstInitialize.resolve(); + await releaseFirstInitialize.promise; + } + response.end( + JSON.stringify({ + jsonrpc: "2.0", + id: message.id, + result: { + protocolVersion: "2025-06-18", + capabilities: { tools: {} }, + serverInfo: { name: "fixture-remote", version: "1.0.0" }, + }, + }), + ); + return; + } + if (message.method === "tools/list") { + response.end( + JSON.stringify({ + jsonrpc: "2.0", + id: message.id, + result: { tools: [{ name: "remote_echo", inputSchema: {} }] }, + }), + ); + return; + } + response.statusCode = 202; + response.end(); + })().catch((error) => { + response.statusCode = 500; + response.end(String(error)); + }); + }); + }); + + try { + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const address = server.address(); + if (!address || typeof address === "string") { + throw new Error("Could not bind fixture server"); + } + const config = parseConfig({ + mcpServers: { + remote: { + name: "Remote", + description: "A useful remote server.", + transport: "http", + url: `http://127.0.0.1:${address.port}/mcp`, + }, + }, + }); + const registry = new ServerRegistry(config); + const manager = new DownstreamManager(registry); + const firstList = manager.listTools(config.mcpServers.remote!); + + await firstInitialize.promise; + await manager.closeServer("remote"); + releaseFirstInitialize.resolve(); + + await expect(firstList).rejects.toMatchObject({ + code: "SERVER_UNAVAILABLE", + } satisfies Partial); + + await manager.close(); + } finally { + releaseFirstInitialize.resolve(); + await new Promise((resolve) => server.close(() => resolve())); + } + }); }); describe("downstream remote OAuth lifecycle", () => { @@ -262,3 +427,17 @@ describe("downstream remote OAuth lifecycle", () => { } }); }); + +function deferred(): { + promise: Promise; + resolve: (value: T | PromiseLike) => void; + reject: (reason?: unknown) => void; +} { + let resolve!: (value: T | PromiseLike) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((promiseResolve, promiseReject) => { + resolve = promiseResolve; + reject = promiseReject; + }); + return { promise, resolve, reject }; +} diff --git a/test/graphql.test.ts b/test/graphql.test.ts index 80116515..38271206 100644 --- a/test/graphql.test.ts +++ b/test/graphql.test.ts @@ -293,6 +293,46 @@ describe("GraphQLManager", () => { ), ).rejects.toMatchObject({ code: "DOWNSTREAM_PROTOCOL_ERROR" }); }); + + it("invalidates cached operations for one endpoint", async () => { + const dir = mkdtempSync(join(tmpdir(), "caplets-graphql-invalidate-")); + const schemaPath = join(dir, "schema.graphql"); + writeFileSync( + schemaPath, + ` + type Query { + first: String + } + `, + ); + const manager = new GraphQLManager(registry()); + const endpoint = graphqlEndpoint({ + schemaPath, + operationCacheTtlMs: 30_000, + }); + + try { + expect((await manager.listTools(endpoint)).map((tool) => tool.name)).toEqual(["query_first"]); + + writeFileSync( + schemaPath, + ` + type Query { + second: String + } + `, + ); + expect((await manager.listTools(endpoint)).map((tool) => tool.name)).toEqual(["query_first"]); + + manager.invalidate("users"); + + expect((await manager.listTools(endpoint)).map((tool) => tool.name)).toEqual([ + "query_second", + ]); + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }); }); const schemaSdl = ` diff --git a/test/openapi.test.ts b/test/openapi.test.ts index 0e2b0838..3aeac783 100644 --- a/test/openapi.test.ts +++ b/test/openapi.test.ts @@ -601,6 +601,39 @@ describe("native OpenAPI Caplets", () => { rmSync(dir, { recursive: true, force: true }); } }); + + it("invalidates cached operations for one endpoint", async () => { + const dir = mkdtempSync(join(tmpdir(), "caplets-openapi-invalidate-")); + const specPath = join(dir, "openapi.json"); + writeFileSync(specPath, JSON.stringify(singleOperationSpec("first"))); + const config = parseConfig({ + openapiEndpoints: { + users: { + name: "Users API", + description: "Manage users through the internal HTTP API.", + specPath, + baseUrl, + auth: { type: "none" }, + }, + }, + }); + const registry = new ServerRegistry(config); + const openapi = new OpenApiManager(registry); + const endpoint = config.openapiEndpoints.users!; + + try { + expect((await openapi.listTools(endpoint)).map((tool) => tool.name)).toEqual(["first"]); + + writeFileSync(specPath, JSON.stringify(singleOperationSpec("second"))); + expect((await openapi.listTools(endpoint)).map((tool) => tool.name)).toEqual(["first"]); + + openapi.invalidate("users"); + + expect((await openapi.listTools(endpoint)).map((tool) => tool.name)).toEqual(["second"]); + } finally { + rmSync(dir, { recursive: true, force: true }); + } + }); }); function openApiSpec(baseUrl: string) { @@ -650,3 +683,18 @@ function openApiSpec(baseUrl: string) { }, }; } + +function singleOperationSpec(operationId: string) { + return { + openapi: "3.0.3", + info: { title: "Users API", version: "1.0.0" }, + paths: { + "/users": { + get: { + operationId, + responses: { "200": { description: "OK" } }, + }, + }, + }, + }; +} diff --git a/test/runtime.test.ts b/test/runtime.test.ts new file mode 100644 index 00000000..5a002c97 --- /dev/null +++ b/test/runtime.test.ts @@ -0,0 +1,298 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { RegisteredTool } from "@modelcontextprotocol/sdk/server/mcp"; +import { CapletsRuntime } from "../src/runtime.js"; + +describe("CapletsRuntime", () => { + const dirs: string[] = []; + + afterEach(() => { + for (const dir of dirs.splice(0)) { + rmSync(dir, { recursive: true, force: true }); + } + }); + + it("registers initial enabled Caplets only", async () => { + const { dir, configPath, projectConfigPath } = tempConfig({ + mcpServers: { + alpha: { + name: "Alpha", + description: "Search alpha project documents.", + command: "node", + }, + beta: { + name: "Beta", + description: "Search beta project documents.", + command: "node", + disabled: true, + }, + }, + }); + dirs.push(dir); + const server = mockServer(); + const runtime = new CapletsRuntime({ configPath, projectConfigPath, server }); + + expect(runtime.registeredToolIds()).toEqual(["alpha"]); + expect(server.registerTool).toHaveBeenCalledTimes(1); + + await runtime.close(); + }); + + it("adds, updates, and removes tools across successful reloads", async () => { + const { dir, configPath, projectConfigPath } = tempConfig({ + mcpServers: { + alpha: { + name: "Alpha", + description: "Search alpha project documents.", + command: "node", + }, + }, + }); + dirs.push(dir); + const server = mockServer(); + const runtime = new CapletsRuntime({ configPath, projectConfigPath, server }); + const alpha = server.registered.get("alpha")!; + + writeConfig(configPath, { + mcpServers: { + alpha: { + name: "Alpha Updated", + description: "Search updated alpha project documents.", + command: "node", + }, + gamma: { + name: "Gamma", + description: "Search gamma project documents.", + command: "node", + }, + }, + }); + await runtime.reload(); + + expect(runtime.registeredToolIds()).toEqual(["alpha", "gamma"]); + expect(alpha.update).toHaveBeenCalledWith( + expect.objectContaining({ + title: "Alpha Updated", + description: expect.stringContaining("Search updated alpha project documents."), + }), + ); + expect(server.registered.get("gamma")).toBeDefined(); + + writeConfig(configPath, { + mcpServers: { + gamma: { + name: "Gamma", + description: "Search gamma project documents.", + command: "node", + }, + }, + }); + await runtime.reload(); + + expect(alpha.remove).toHaveBeenCalledTimes(1); + expect(runtime.registeredToolIds()).toEqual(["gamma"]); + + await runtime.close(); + }); + + it("keeps the last known-good tools when reload validation fails", async () => { + const { dir, configPath, projectConfigPath } = tempConfig({ + mcpServers: { + alpha: { + name: "Alpha", + description: "Search alpha project documents.", + command: "node", + }, + }, + }); + dirs.push(dir); + const server = mockServer(); + const errors: string[] = []; + const runtime = new CapletsRuntime({ + configPath, + projectConfigPath, + server, + writeErr: (value) => errors.push(value), + }); + + writeFileSync(configPath, "{not json"); + const reloaded = await runtime.reload(); + + expect(runtime.registeredToolIds()).toEqual(["alpha"]); + expect(server.registered.get("alpha")?.remove).not.toHaveBeenCalled(); + expect(reloaded).toBe(false); + expect(errors.join("")).toContain("keeping last known-good config"); + + await runtime.close(); + }); + + it("reconciles tools when backend invalidation fails", async () => { + const { dir, configPath, projectConfigPath } = tempConfig({ + mcpServers: { + alpha: { + name: "Alpha", + description: "Search alpha project documents.", + command: "node", + }, + }, + }); + dirs.push(dir); + const server = mockServer(); + const errors: string[] = []; + const runtime = new CapletsRuntime({ + configPath, + projectConfigPath, + server, + writeErr: (value) => errors.push(value), + }); + const alpha = server.registered.get("alpha")!; + ( + runtime as unknown as { + invalidateChangedBackends: () => Promise; + } + ).invalidateChangedBackends = vi.fn(async () => { + throw new Error("close failed"); + }); + + writeConfig(configPath, { + mcpServers: { + gamma: { + name: "Gamma", + description: "Search gamma project documents.", + command: "node", + }, + }, + }); + const reloaded = await runtime.reload(); + + expect(reloaded).toBe(false); + expect(alpha.remove).toHaveBeenCalledTimes(1); + expect(runtime.registeredToolIds()).toEqual(["gamma"]); + expect(server.registered.get("gamma")).toBeDefined(); + expect(errors.join("")).toContain("backend invalidation failed"); + + await runtime.close(); + }); + + it("watches nested Caplet files when the config dir is also the Caplets root", async () => { + const { dir, configPath, projectConfigPath } = tempConfig({ + mcpServers: { + alpha: { + name: "Alpha", + description: "Search alpha project documents.", + command: "node", + }, + }, + }); + dirs.push(dir); + const nestedFile = join(dir, "user", "nested", "notes.md"); + mkdirSync(join(dir, "user", "nested"), { recursive: true }); + writeFileSync(nestedFile, "before"); + const runtime = new CapletsRuntime({ + configPath, + projectConfigPath, + server: mockServer(), + watchDebounceMs: 10, + }); + let reloads = 0; + (runtime as unknown as { reload: () => Promise }).reload = vi.fn(async () => { + reloads += 1; + return true; + }); + + writeFileSync(nestedFile, "after"); + await eventually(() => expect(reloads).toBeGreaterThan(0)); + + await runtime.close(); + }); + + it("runs a follow-up reload when another reload is requested mid-flight", async () => { + const { dir, configPath, projectConfigPath } = tempConfig({ + mcpServers: { + alpha: { + name: "Alpha", + description: "Search alpha project documents.", + command: "node", + }, + }, + }); + dirs.push(dir); + const runtime = new CapletsRuntime({ configPath, projectConfigPath, server: mockServer() }); + let calls = 0; + + (runtime as unknown as { reloadOnce: () => Promise }).reloadOnce = vi.fn(async () => { + calls += 1; + if (calls === 1) { + void runtime.reload(); + } + }); + + await runtime.reload(); + + expect(calls).toBe(2); + + await runtime.close(); + }); + + function tempConfig(config: unknown): { + dir: string; + configPath: string; + projectConfigPath: string; + } { + const dir = mkdtempSync(join(tmpdir(), "caplets-runtime-")); + const userRoot = join(dir, "user"); + const projectRoot = join(dir, "project", ".caplets"); + mkdirSync(userRoot, { recursive: true }); + mkdirSync(projectRoot, { recursive: true }); + const configPath = join(userRoot, "config.json"); + const projectConfigPath = join(projectRoot, "config.json"); + writeConfig(configPath, config); + return { dir, configPath, projectConfigPath }; + } +}); + +function writeConfig(path: string, config: unknown): void { + writeFileSync(path, JSON.stringify(config)); +} + +function mockServer() { + const registered = new Map(); + return { + registered, + registerTool: vi.fn((name: string) => { + const tool = { + update: vi.fn(), + remove: vi.fn(() => registered.delete(name)), + enable: vi.fn(), + disable: vi.fn(), + enabled: true, + handler: vi.fn(), + } as unknown as RegisteredTool; + registered.set(name, tool); + return tool; + }), + connect: vi.fn(async () => {}), + close: vi.fn(async () => {}), + }; +} + +async function eventually(assertion: () => void): Promise { + const deadline = Date.now() + 1_000; + let lastError: unknown; + while (Date.now() < deadline) { + try { + assertion(); + return; + } catch (error) { + lastError = error; + await new Promise((resolve) => setTimeout(resolve, 20)); + } + } + try { + assertion(); + } catch { + throw lastError; + } +}