From 44311d57811138272198ef609cf30d15c0a1cd18 Mon Sep 17 00:00:00 2001 From: Ian Pascoe Date: Tue, 12 May 2026 18:46:44 -0400 Subject: [PATCH 01/10] feat: hot reload caplets serve config --- .changeset/quiet-rockets-serve.md | 5 + README.md | 17 +- docs/plans/2026-05-12-hot-reload.md | 81 +++++++ src/downstream.ts | 25 ++- src/graphql.ts | 10 +- src/index.ts | 52 +---- src/openapi.ts | 10 +- src/runtime.ts | 326 ++++++++++++++++++++++++++++ test/downstream.test.ts | 31 +++ test/graphql.test.ts | 40 ++++ test/openapi.test.ts | 48 ++++ test/runtime.test.ts | 170 +++++++++++++++ 12 files changed, 761 insertions(+), 54 deletions(-) create mode 100644 .changeset/quiet-rockets-serve.md create mode 100644 docs/plans/2026-05-12-hot-reload.md create mode 100644 src/runtime.ts create mode 100644 test/runtime.test.ts diff --git a/.changeset/quiet-rockets-serve.md b/.changeset/quiet-rockets-serve.md new file mode 100644 index 00000000..e6186bfe --- /dev/null +++ b/.changeset/quiet-rockets-serve.md @@ -0,0 +1,5 @@ +--- +"caplets": minor +--- + +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..de4f5d28 --- /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..f2e33cc7 100644 --- a/src/downstream.ts +++ b/src/downstream.ts @@ -31,6 +31,7 @@ type ManagedConnection = { tools?: Tool[]; toolsFetchedAt?: number; restartingAfterDeath?: boolean; + closing?: boolean; }; export class DownstreamManager { @@ -38,17 +39,34 @@ export class DownstreamManager { 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 { + for (const connection of this.connections.values()) { + connection.closing = true; + } await Promise.allSettled( [...this.connections.values()].map((connection) => connection.transport.close()), ); this.connections.clear(); } + async closeServer(serverId: string): Promise { + const connection = this.connections.get(serverId); + this.connections.delete(serverId); + this.restartState.delete(serverId); + if (connection) { + connection.closing = true; + await connection.transport.close(); + } + } + async checkServer(server: CapletServerConfig): Promise<{ server: string; status: string; @@ -199,8 +217,12 @@ export class DownstreamManager { try { const client = new Client({ name: "caplets", version: "1.0.0" }, { capabilities: {} }); const transport = this.createTransport(server); + const connection: ManagedConnection = { client, transport }; transport.onclose = () => { this.connections.delete(server.server); + if (connection.closing) { + return; + } this.restartState.set(server.server, { restartUsed: true, backoffUntil: Date.now() + 1_000, @@ -219,7 +241,6 @@ export class DownstreamManager { ); }; await client.connect(transport, { timeout: server.startupTimeoutMs }); - const connection: ManagedConnection = { client, transport }; this.connections.set(server.server, connection); this.registry.setStatus(server.server, "available"); return connection; 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..c321972f --- /dev/null +++ b/src/runtime.ts @@ -0,0 +1,326 @@ +import { existsSync, watch, type FSWatcher } from "node:fs"; +import { dirname } 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 reloading: Promise | undefined; + 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) { + await this.reloading; + } + this.reloading = this.reloadOnce().finally(() => { + this.reloading = undefined; + }); + await this.reloading; + return true; + } + + async close(): Promise { + this.closed = true; + if (this.reloadTimer) { + clearTimeout(this.reloadTimer); + this.reloadTimer = undefined; + } + 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 watchedPaths(this.paths) + .map((entry) => entry.path) + .sort(); + } + + private async reloadOnce(): Promise { + 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; + } + + const previousConfig = this.registry.config; + const nextRegistry = new ServerRegistry(nextConfig); + await this.invalidateChangedBackends(previousConfig, nextConfig); + this.registry = nextRegistry; + this.downstream.updateRegistry(nextRegistry); + this.openapi.updateRegistry(nextRegistry); + this.graphql.updateRegistry(nextRegistry); + this.reconcileTools(previousConfig, nextConfig); + this.resetWatchers(); + } + + 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); + if ( + before && + before.backend === "mcp" && + serializeCaplet(before) !== serializeCaplet(after) + ) { + await this.downstream.closeServer(serverId); + } + if (after?.backend === "openapi" && serializeCaplet(before) !== serializeCaplet(after)) { + this.openapi.invalidate(serverId); + } + if (after?.backend === "graphql" && serializeCaplet(before) !== serializeCaplet(after)) { + this.graphql.invalidate(serverId); + } + if (!after) { + this.openapi.invalidate(serverId); + this.graphql.invalidate(serverId); + } + } + } + + private resetWatchers(): void { + this.closeWatchers(); + for (const entry of watchedPaths(this.paths)) { + if (!existsSync(entry.path)) { + continue; + } + try { + this.watchers.push( + watch(entry.path, { persistent: true }, () => { + this.scheduleReload(); + }), + ); + } 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 = []; + } +} + +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) { + if (seen.has(entry.path)) { + continue; + } + seen.add(entry.path); + 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); +} diff --git a/test/downstream.test.ts b/test/downstream.test.ts index b6b6b096..ae746afe 100644 --- a/test/downstream.test.ts +++ b/test/downstream.test.ts @@ -46,6 +46,37 @@ 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(); + } + }); }); describe("downstream remote OAuth lifecycle", () => { 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..36308e75 --- /dev/null +++ b/test/runtime.test.ts @@ -0,0 +1,170 @@ +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"); + await runtime.reload(); + + expect(runtime.registeredToolIds()).toEqual(["alpha"]); + expect(server.registered.get("alpha")?.remove).not.toHaveBeenCalled(); + expect(errors.join("")).toContain("keeping last known-good config"); + + 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 () => {}), + }; +} From 3c6f24774295c2158a3563b9b00e34a1fb196f0d Mon Sep 17 00:00:00 2001 From: Ian Pascoe Date: Tue, 12 May 2026 20:15:46 -0400 Subject: [PATCH 02/10] fix: address hot reload review feedback --- .changeset/quiet-rockets-serve.md | 2 + docs/plans/2026-05-12-hot-reload.md | 10 +-- src/downstream.ts | 3 + src/runtime.ts | 128 +++++++++++++++++++++++----- 4 files changed, 119 insertions(+), 24 deletions(-) diff --git a/.changeset/quiet-rockets-serve.md b/.changeset/quiet-rockets-serve.md index e6186bfe..9593cbf9 100644 --- a/.changeset/quiet-rockets-serve.md +++ b/.changeset/quiet-rockets-serve.md @@ -2,4 +2,6 @@ "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/docs/plans/2026-05-12-hot-reload.md b/docs/plans/2026-05-12-hot-reload.md index de4f5d28..971117b6 100644 --- a/docs/plans/2026-05-12-hot-reload.md +++ b/docs/plans/2026-05-12-hot-reload.md @@ -10,7 +10,7 @@ --- -### Task 1: Runtime Controller +## Task 1: Runtime Controller **Files:** @@ -23,7 +23,7 @@ - [ ] Register initial enabled Caplets during startup. - [ ] Keep `src/index.ts` as a small CLI/server entrypoint. -### Task 2: Reload Reconciliation +## Task 2: Reload Reconciliation **Files:** @@ -42,7 +42,7 @@ - [ ] 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 +## Task 3: Filesystem Watching **Files:** @@ -55,7 +55,7 @@ - [ ] Recreate watchers after every successful reload because roots can change with `CAPLETS_CONFIG`. - [ ] Close all watchers during runtime shutdown. -### Task 4: Documentation +## Task 4: Documentation **Files:** @@ -66,7 +66,7 @@ - [ ] 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 +## Task 5: Verification **Files:** diff --git a/src/downstream.ts b/src/downstream.ts index f2e33cc7..12af51af 100644 --- a/src/downstream.ts +++ b/src/downstream.ts @@ -234,6 +234,9 @@ export class DownstreamManager { ); }; transport.onerror = (error: Error) => { + if (connection.closing) { + return; + } this.registry.setStatus( server.server, "unavailable", diff --git a/src/runtime.ts b/src/runtime.ts index c321972f..55691d77 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -1,5 +1,5 @@ -import { existsSync, watch, type FSWatcher } from "node:fs"; -import { dirname } from "node:path"; +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"; @@ -53,6 +53,7 @@ export class CapletsRuntime { private readonly writeErr: (value: string) => void; private watchers: FSWatcher[] = []; private reloadTimer: NodeJS.Timeout | undefined; + private watcherRefreshTimer: NodeJS.Timeout | undefined; private reloading: Promise | undefined; private closed = false; @@ -101,6 +102,7 @@ export class CapletsRuntime { } if (this.reloading) { await this.reloading; + return !this.closed; } this.reloading = this.reloadOnce().finally(() => { this.reloading = undefined; @@ -115,6 +117,13 @@ export class CapletsRuntime { clearTimeout(this.reloadTimer); this.reloadTimer = undefined; } + if (this.watcherRefreshTimer) { + clearTimeout(this.watcherRefreshTimer); + this.watcherRefreshTimer = undefined; + } + if (this.reloading) { + await this.reloading; + } this.closeWatchers(); await this.downstream.close(); await this.server.close(); @@ -135,6 +144,9 @@ export class CapletsRuntime { } private async reloadOnce(): Promise { + if (this.closed) { + return; + } let nextConfig: CapletsConfig; try { nextConfig = loadConfig(this.paths.configPath, this.paths.projectConfigPath); @@ -144,9 +156,15 @@ export class CapletsRuntime { return; } + if (this.closed) { + return; + } const previousConfig = this.registry.config; const nextRegistry = new ServerRegistry(nextConfig); await this.invalidateChangedBackends(previousConfig, nextConfig); + if (this.closed) { + return; + } this.registry = nextRegistry; this.downstream.updateRegistry(nextRegistry); this.openapi.updateRegistry(nextRegistry); @@ -224,21 +242,17 @@ export class CapletsRuntime { for (const serverId of changedIds) { const before = previousCaplets.get(serverId); const after = nextCaplets.get(serverId); - if ( - before && - before.backend === "mcp" && - serializeCaplet(before) !== serializeCaplet(after) - ) { + const changed = serializeCaplet(before) !== serializeCaplet(after); + if (!changed) { + continue; + } + if (before?.backend === "mcp") { await this.downstream.closeServer(serverId); } - if (after?.backend === "openapi" && serializeCaplet(before) !== serializeCaplet(after)) { + if (before?.backend === "openapi" || after?.backend === "openapi" || !after) { this.openapi.invalidate(serverId); } - if (after?.backend === "graphql" && serializeCaplet(before) !== serializeCaplet(after)) { - this.graphql.invalidate(serverId); - } - if (!after) { - this.openapi.invalidate(serverId); + if (before?.backend === "graphql" || after?.backend === "graphql" || !after) { this.graphql.invalidate(serverId); } } @@ -246,16 +260,15 @@ export class CapletsRuntime { private resetWatchers(): void { this.closeWatchers(); + const watched = new Set(); for (const entry of watchedPaths(this.paths)) { - if (!existsSync(entry.path)) { + const watchPath = existsSync(entry.path) ? entry.path : nearestExistingParent(entry.path); + if (!watchPath || watched.has(watchPath)) { continue; } + watched.add(watchPath); try { - this.watchers.push( - watch(entry.path, { persistent: true }, () => { - this.scheduleReload(); - }), - ); + this.watchers.push(...this.watchPathTree(watchPath, entry.path)); } 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`); @@ -269,6 +282,50 @@ export class CapletsRuntime { } this.watchers = []; } + + private watchPathTree(watchPath: string, targetPath: string): FSWatcher[] { + if (isDirectory(watchPath)) { + return this.watchDirectoryTree(watchPath); + } + + return [ + watch(watchPath, { persistent: true }, () => { + this.scheduleReload(); + if (existsSync(targetPath)) { + this.scheduleWatcherRefresh(); + } + }), + ]; + } + + private watchDirectoryTree(root: string): FSWatcher[] { + const watchers: FSWatcher[] = []; + const directories = discoverDirectories(root); + for (const directory of directories) { + watchers.push( + watch(directory, { persistent: true }, () => { + this.scheduleReload(); + this.scheduleWatcherRefresh(); + }), + ); + } + 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 } { @@ -324,3 +381,36 @@ function capletById(config: CapletsConfig, serverId: string): CapletConfig | und 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; + } +} From 835f24d021fbd961e0528de3698c01774536b4b9 Mon Sep 17 00:00:00 2001 From: "coderabbitai[bot]" <136622811+coderabbitai[bot]@users.noreply.github.com> Date: Wed, 13 May 2026 01:01:47 +0000 Subject: [PATCH 03/10] fix: apply CodeRabbit auto-fixes Fixed 1 file(s) based on 2 unresolved review comments. Co-authored-by: CodeRabbit --- src/runtime.ts | 28 ++++++++++++++++++++-------- 1 file changed, 20 insertions(+), 8 deletions(-) diff --git a/src/runtime.ts b/src/runtime.ts index 55691d77..38722ad0 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -54,7 +54,7 @@ export class CapletsRuntime { private watchers: FSWatcher[] = []; private reloadTimer: NodeJS.Timeout | undefined; private watcherRefreshTimer: NodeJS.Timeout | undefined; - private reloading: Promise | undefined; + private reloading: Promise | undefined; private closed = false; constructor(options: CapletsRuntimeOptions = {}) { @@ -104,7 +104,11 @@ export class CapletsRuntime { await this.reloading; return !this.closed; } - this.reloading = this.reloadOnce().finally(() => { + this.reloading = this.reloadOnce().catch(err => { + this.writeErr(`Caplets reload failed.\n`); + this.writeErr(`${JSON.stringify(toSafeError(err, "INTERNAL_ERROR"), null, 2)}\n`); + return false; + }).finally(() => { this.reloading = undefined; }); await this.reloading; @@ -302,12 +306,20 @@ export class CapletsRuntime { const watchers: FSWatcher[] = []; const directories = discoverDirectories(root); for (const directory of directories) { - watchers.push( - watch(directory, { persistent: true }, () => { - this.scheduleReload(); - this.scheduleWatcherRefresh(); - }), - ); + try { + watchers.push( + watch(directory, { persistent: true }, () => { + this.scheduleReload(); + this.scheduleWatcherRefresh(); + }), + ); + } catch (error) { + // Clean up any watchers created so far before rethrowing + for (const watcher of watchers) { + watcher.close(); + } + throw error; + } } return watchers; } From 7a4c46cd84d7492f756718818cb88888e9929fdb Mon Sep 17 00:00:00 2001 From: Ian Pascoe Date: Tue, 12 May 2026 23:12:42 -0400 Subject: [PATCH 04/10] fix: formatting --- src/runtime.ts | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/src/runtime.ts b/src/runtime.ts index 38722ad0..9c738cde 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -104,13 +104,15 @@ export class CapletsRuntime { await this.reloading; return !this.closed; } - this.reloading = this.reloadOnce().catch(err => { - this.writeErr(`Caplets reload failed.\n`); - this.writeErr(`${JSON.stringify(toSafeError(err, "INTERNAL_ERROR"), null, 2)}\n`); - return false; - }).finally(() => { - this.reloading = undefined; - }); + this.reloading = this.reloadOnce() + .catch((err) => { + this.writeErr(`Caplets reload failed.\n`); + this.writeErr(`${JSON.stringify(toSafeError(err, "INTERNAL_ERROR"), null, 2)}\n`); + return false; + }) + .finally(() => { + this.reloading = undefined; + }); await this.reloading; return true; } From b1ecf0d55265c226783f2dfdae9d6c8e1d63c024 Mon Sep 17 00:00:00 2001 From: Ian Pascoe Date: Tue, 12 May 2026 23:15:57 -0400 Subject: [PATCH 05/10] fix: drain pending hot reload events --- src/runtime.ts | 40 +++++++++++++++++++++++++--------------- test/runtime.test.ts | 28 ++++++++++++++++++++++++++++ 2 files changed, 53 insertions(+), 15 deletions(-) diff --git a/src/runtime.ts b/src/runtime.ts index 9c738cde..b27176a1 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -54,7 +54,8 @@ export class CapletsRuntime { private watchers: FSWatcher[] = []; private reloadTimer: NodeJS.Timeout | undefined; private watcherRefreshTimer: NodeJS.Timeout | undefined; - private reloading: Promise | undefined; + private reloading: Promise | undefined; + private pendingReload = false; private closed = false; constructor(options: CapletsRuntimeOptions = {}) { @@ -101,18 +102,13 @@ export class CapletsRuntime { return false; } if (this.reloading) { + this.pendingReload = true; await this.reloading; - return !this.closed; + return !this.closed && !this.reloading; } - this.reloading = this.reloadOnce() - .catch((err) => { - this.writeErr(`Caplets reload failed.\n`); - this.writeErr(`${JSON.stringify(toSafeError(err, "INTERNAL_ERROR"), null, 2)}\n`); - return false; - }) - .finally(() => { - this.reloading = undefined; - }); + this.reloading = this.reloadUntilSettled().finally(() => { + this.reloading = undefined; + }); await this.reloading; return true; } @@ -179,6 +175,18 @@ export class CapletsRuntime { this.resetWatchers(); } + private async reloadUntilSettled(): Promise { + do { + this.pendingReload = false; + try { + await this.reloadOnce(); + } catch (err) { + this.writeErr(`Caplets reload failed.\n`); + this.writeErr(`${JSON.stringify(toSafeError(err, "INTERNAL_ERROR"), null, 2)}\n`); + } + } while (this.pendingReload && !this.closed); + } + private reconcileTools(previous: CapletsConfig | undefined, next: CapletsConfig): void { const enabled = new Map(nextEnabledServers(next).map((server) => [server.server, server])); @@ -295,9 +303,9 @@ export class CapletsRuntime { } return [ - watch(watchPath, { persistent: true }, () => { + watch(watchPath, { persistent: true }, (eventType) => { this.scheduleReload(); - if (existsSync(targetPath)) { + if (eventType === "rename" && existsSync(targetPath)) { this.scheduleWatcherRefresh(); } }), @@ -310,9 +318,11 @@ export class CapletsRuntime { for (const directory of directories) { try { watchers.push( - watch(directory, { persistent: true }, () => { + watch(directory, { persistent: true }, (eventType) => { this.scheduleReload(); - this.scheduleWatcherRefresh(); + if (eventType === "rename") { + this.scheduleWatcherRefresh(); + } }), ); } catch (error) { diff --git a/test/runtime.test.ts b/test/runtime.test.ts index 36308e75..023a067a 100644 --- a/test/runtime.test.ts +++ b/test/runtime.test.ts @@ -127,6 +127,34 @@ describe("CapletsRuntime", () => { 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; From c7f40e4f05c86ac16480fb3d6eb9ef0ade3f9d8a Mon Sep 17 00:00:00 2001 From: Ian Pascoe Date: Tue, 12 May 2026 23:28:41 -0400 Subject: [PATCH 06/10] fix: tighten hot reload review edge cases --- src/runtime.ts | 40 +++++++++++++++++++++------------------- test/runtime.test.ts | 3 ++- 2 files changed, 23 insertions(+), 20 deletions(-) diff --git a/src/runtime.ts b/src/runtime.ts index b27176a1..a0718832 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -54,7 +54,7 @@ export class CapletsRuntime { private watchers: FSWatcher[] = []; private reloadTimer: NodeJS.Timeout | undefined; private watcherRefreshTimer: NodeJS.Timeout | undefined; - private reloading: Promise | undefined; + private reloading: Promise | undefined; private pendingReload = false; private closed = false; @@ -103,14 +103,12 @@ export class CapletsRuntime { } if (this.reloading) { this.pendingReload = true; - await this.reloading; - return !this.closed && !this.reloading; + return await this.reloading; } this.reloading = this.reloadUntilSettled().finally(() => { this.reloading = undefined; }); - await this.reloading; - return true; + return await this.reloading; } async close(): Promise { @@ -145,9 +143,9 @@ export class CapletsRuntime { .sort(); } - private async reloadOnce(): Promise { + private async reloadOnce(): Promise { if (this.closed) { - return; + return false; } let nextConfig: CapletsConfig; try { @@ -155,36 +153,40 @@ export class CapletsRuntime { } 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; + return false; } if (this.closed) { - return; + return false; } const previousConfig = this.registry.config; const nextRegistry = new ServerRegistry(nextConfig); - await this.invalidateChangedBackends(previousConfig, nextConfig); - if (this.closed) { - return; - } this.registry = nextRegistry; this.downstream.updateRegistry(nextRegistry); this.openapi.updateRegistry(nextRegistry); this.graphql.updateRegistry(nextRegistry); + await this.invalidateChangedBackends(previousConfig, nextConfig); + if (this.closed) { + return false; + } this.reconcileTools(previousConfig, nextConfig); this.resetWatchers(); + return true; } - private async reloadUntilSettled(): Promise { + private async reloadUntilSettled(): Promise { + let succeeded = true; do { this.pendingReload = false; try { - await this.reloadOnce(); + 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 { @@ -282,7 +284,7 @@ export class CapletsRuntime { } watched.add(watchPath); try { - this.watchers.push(...this.watchPathTree(watchPath, entry.path)); + 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`); @@ -297,15 +299,15 @@ export class CapletsRuntime { this.watchers = []; } - private watchPathTree(watchPath: string, targetPath: string): FSWatcher[] { - if (isDirectory(watchPath)) { + 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" && existsSync(targetPath)) { + if (eventType === "rename" && entry.reason === "caplets" && existsSync(entry.path)) { this.scheduleWatcherRefresh(); } }), diff --git a/test/runtime.test.ts b/test/runtime.test.ts index 023a067a..4bc358fd 100644 --- a/test/runtime.test.ts +++ b/test/runtime.test.ts @@ -118,10 +118,11 @@ describe("CapletsRuntime", () => { }); writeFileSync(configPath, "{not json"); - await runtime.reload(); + 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(); From 60c0f5ebef81420e18e84f90798e1edebce32f9e Mon Sep 17 00:00:00 2001 From: Ian Pascoe Date: Wed, 13 May 2026 07:32:50 -0400 Subject: [PATCH 07/10] fix: ignore stale downstream callbacks --- src/downstream.ts | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/src/downstream.ts b/src/downstream.ts index 12af51af..62fcabd8 100644 --- a/src/downstream.ts +++ b/src/downstream.ts @@ -219,10 +219,16 @@ export class DownstreamManager { const transport = this.createTransport(server); const connection: ManagedConnection = { client, transport }; 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, @@ -237,6 +243,9 @@ export class DownstreamManager { if (connection.closing) { return; } + if (this.connections.get(server.server) !== connection) { + return; + } this.registry.setStatus( server.server, "unavailable", From 319e0380b87c463732a116093a9fe9c5140e24d8 Mon Sep 17 00:00:00 2001 From: Ian Pascoe Date: Wed, 13 May 2026 07:37:39 -0400 Subject: [PATCH 08/10] fix: reject stale downstream configs --- src/downstream.ts | 56 +++++++++++++++++++++++++++++++++++++++-- src/runtime.ts | 29 +++++++++++---------- test/downstream.test.ts | 42 +++++++++++++++++++++++++++++++ 3 files changed, 112 insertions(+), 15 deletions(-) diff --git a/src/downstream.ts b/src/downstream.ts index 62fcabd8..5038c530 100644 --- a/src/downstream.ts +++ b/src/downstream.ts @@ -28,6 +28,7 @@ export type CompactTool = { type ManagedConnection = { client: Client; transport: { close(): Promise; onclose?: () => void; onerror?: (error: Error) => void }; + configFingerprint: string; tools?: Tool[]; toolsFetchedAt?: number; restartingAfterDeath?: boolean; @@ -204,9 +205,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) { @@ -217,7 +232,11 @@ export class DownstreamManager { try { const client = new Client({ name: "caplets", version: "1.0.0" }, { capabilities: {} }); const transport = this.createTransport(server); - const connection: ManagedConnection = { client, transport }; + const connection: ManagedConnection = { + client, + transport, + configFingerprint: expectedFingerprint, + }; transport.onclose = () => { const current = this.connections.get(server.server); if (current === connection) { @@ -253,6 +272,11 @@ export class DownstreamManager { ); }; await client.connect(transport, { timeout: server.startupTimeoutMs }); + if (this.currentServerFingerprint(server) !== expectedFingerprint) { + connection.closing = true; + await transport.close(); + throw staleServerConfigError(server.server); + } this.connections.set(server.server, connection); this.registry.setStatus(server.server, "available"); return connection; @@ -360,6 +384,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/runtime.ts b/src/runtime.ts index a0718832..370f5af7 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -113,20 +113,23 @@ export class CapletsRuntime { async close(): Promise { this.closed = true; - if (this.reloadTimer) { - clearTimeout(this.reloadTimer); - this.reloadTimer = undefined; - } - if (this.watcherRefreshTimer) { - clearTimeout(this.watcherRefreshTimer); - this.watcherRefreshTimer = undefined; - } - if (this.reloading) { - await this.reloading; + 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(); } - this.closeWatchers(); - await this.downstream.close(); - await this.server.close(); } currentConfig(): CapletsConfig { diff --git a/test/downstream.test.ts b/test/downstream.test.ts index ae746afe..03807bc5 100644 --- a/test/downstream.test.ts +++ b/test/downstream.test.ts @@ -77,6 +77,48 @@ describe("downstream stdio lifecycle", () => { 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(); + } + }); }); describe("downstream remote OAuth lifecycle", () => { From a0eb44bf2d83dfaefff3dee526dea1643dd2cab2 Mon Sep 17 00:00:00 2001 From: Ian Pascoe Date: Wed, 13 May 2026 08:12:25 -0400 Subject: [PATCH 09/10] fix: keep reload reconciliation after invalidation errors --- src/runtime.ts | 11 ++++++++-- test/runtime.test.ts | 48 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 57 insertions(+), 2 deletions(-) diff --git a/src/runtime.ts b/src/runtime.ts index 370f5af7..fe8730b1 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -168,13 +168,20 @@ export class CapletsRuntime { this.downstream.updateRegistry(nextRegistry); this.openapi.updateRegistry(nextRegistry); this.graphql.updateRegistry(nextRegistry); - await this.invalidateChangedBackends(previousConfig, nextConfig); + 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 true; + return invalidated; } private async reloadUntilSettled(): Promise { diff --git a/test/runtime.test.ts b/test/runtime.test.ts index 4bc358fd..b0d2a646 100644 --- a/test/runtime.test.ts +++ b/test/runtime.test.ts @@ -128,6 +128,54 @@ describe("CapletsRuntime", () => { 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("runs a follow-up reload when another reload is requested mid-flight", async () => { const { dir, configPath, projectConfigPath } = tempConfig({ mcpServers: { From ec278be5b28a7e82410fa3c3b365f5d19f87c054 Mon Sep 17 00:00:00 2001 From: Ian Pascoe Date: Wed, 13 May 2026 08:32:00 -0400 Subject: [PATCH 10/10] fix: keep caplet watchers and close pending connects --- src/downstream.ts | 28 +++++++++-- src/runtime.ts | 14 +++--- test/downstream.test.ts | 106 ++++++++++++++++++++++++++++++++++++++++ test/runtime.test.ts | 51 +++++++++++++++++++ 4 files changed, 187 insertions(+), 12 deletions(-) diff --git a/src/downstream.ts b/src/downstream.ts index 5038c530..8f8f9c5c 100644 --- a/src/downstream.ts +++ b/src/downstream.ts @@ -37,6 +37,7 @@ type ManagedConnection = { export class DownstreamManager { private readonly connections = new Map(); + private readonly connecting = new Map(); private readonly restartState = new Map(); constructor( @@ -49,18 +50,19 @@ export class DownstreamManager { } async close(): Promise { - for (const connection of this.connections.values()) { + const connections = [...this.connections.values(), ...this.connecting.values()]; + for (const connection of connections) { connection.closing = true; } - await Promise.allSettled( - [...this.connections.values()].map((connection) => connection.transport.close()), - ); + 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); + 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; @@ -229,6 +231,7 @@ 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); @@ -237,6 +240,8 @@ export class DownstreamManager { transport, configFingerprint: expectedFingerprint, }; + pendingConnection = connection; + this.connecting.set(server.server, connection); transport.onclose = () => { const current = this.connections.get(server.server); if (current === connection) { @@ -272,15 +277,28 @@ export class DownstreamManager { ); }; await client.connect(transport, { timeout: server.startupTimeoutMs }); + 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); diff --git a/src/runtime.ts b/src/runtime.ts index fe8730b1..91fca95b 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -141,9 +141,7 @@ export class CapletsRuntime { } watchedPaths(): string[] { - return watchedPaths(this.paths) - .map((entry) => entry.path) - .sort(); + return [...new Set(watchedPaths(this.paths).map((entry) => entry.path))].sort(); } private async reloadOnce(): Promise { @@ -289,10 +287,11 @@ export class CapletsRuntime { const watched = new Set(); for (const entry of watchedPaths(this.paths)) { const watchPath = existsSync(entry.path) ? entry.path : nearestExistingParent(entry.path); - if (!watchPath || watched.has(watchPath)) { + const watchKey = `${entry.reason}:${watchPath}`; + if (!watchPath || watched.has(watchKey)) { continue; } - watched.add(watchPath); + watched.add(watchKey); try { this.watchers.push(...this.watchEntry(entry, watchPath)); } catch (error) { @@ -385,10 +384,11 @@ function uniqueWatchedPaths(entries: WatchedPath[]): WatchedPath[] { const seen = new Set(); const unique: WatchedPath[] = []; for (const entry of entries) { - if (seen.has(entry.path)) { + const key = `${entry.reason}:${entry.path}`; + if (seen.has(key)) { continue; } - seen.add(entry.path); + seen.add(key); unique.push(entry); } return unique; diff --git a/test/downstream.test.ts b/test/downstream.test.ts index 03807bc5..a32e51e0 100644 --- a/test/downstream.test.ts +++ b/test/downstream.test.ts @@ -119,6 +119,98 @@ describe("downstream stdio lifecycle", () => { 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", () => { @@ -335,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/runtime.test.ts b/test/runtime.test.ts index b0d2a646..5a002c97 100644 --- a/test/runtime.test.ts +++ b/test/runtime.test.ts @@ -176,6 +176,38 @@ describe("CapletsRuntime", () => { 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: { @@ -245,3 +277,22 @@ function mockServer() { 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; + } +}