// ---------------------------------------------------------------------------// MCP tool-catalog freshness (end-to-end).//// The persisted per-connection tool catalog must converge with the server's// live tool set. Spec inputs the executor reacts to:// - `notifications/tools/list_changed` received during a call window marks// the connection stale; the next tools read re-lists.// - An unknown-tool protocol error (`-32602`, "Tool … not found") on// `tools/call` means the catalog drifted: the call fails with a typed// `mcp_tool_unknown` ToolResult and the catalog heals on the next read.// - `tools/list` is paginated; discovery follows `nextCursor` to the end.// - A failed listing (server unreachable) is non-authoritative: the// previously persisted catalog is kept, not wiped.// --------------------------------------------------------------------------- import { describe, expect, it } from "@effect/vitest";import { Deferred, Effect, Fiber, Option, Ref, Schema } from "effect";import { HttpServerResponse } from "effect/unstable/http"; import { AuthTemplateSlug, ConnectionName, IntegrationSlug, STALE_TOOLS_SYNC_CONCURRENCY, ToolAddress, createExecutor,} from "@executor-js/sdk";import { makeTestConfig, memoryCredentialsPlugin, serveTestHttpApp,} from "@executor-js/sdk/testing"; import { mcpPlugin } from "./plugin";import { createMcpConnector } from "./connection";import { discoverTools } from "./discover";import { makeMutableCatalogMcpServer, serveMcpServer } from "../testing"; const INTEG = IntegrationSlug.make("catalog_mcp");const CONNECTION = ConnectionName.make("main");const TEMPLATE = AuthTemplateSlug.make("none"); const makeCatalogTestExecutor = ( serverUrl: string, options?: { readonly toolsSyncTtlMs?: number | null; readonly toolsSyncGraceMs?: number | null; },) => createExecutor({ ...makeTestConfig({ plugins: [memoryCredentialsPlugin(), mcpPlugin()] as const }), ...(options?.toolsSyncTtlMs === undefined ? {} : { toolsSyncTtlMs: options.toolsSyncTtlMs }), ...(options?.toolsSyncGraceMs === undefined ? {} : { toolsSyncGraceMs: options.toolsSyncGraceMs }), }).pipe( Effect.tap((executor) => Effect.gen(function* () { yield* executor.mcp.addServer({ name: "catalog-mcp", endpoint: serverUrl, slug: String(INTEG), }); yield* executor.connections.create({ owner: "org", name: CONNECTION, integration: INTEG, template: TEMPLATE, value: "", }); }), ), ); const toolNames = (tools: readonly { readonly name: unknown }[]): readonly string[] => tools.map((tool) => String(tool.name)).sort(); describe("MCP tool-catalog sync (end-to-end)", () => { it.effect( "tools/list_changed during a call marks the catalog stale and the next read re-lists", () => Effect.gen(function* () { const mutable = makeMutableCatalogMcpServer(); const server = yield* serveMcpServer(mutable.factory); const executor = yield* makeCatalogTestExecutor(server.url); expect(toolNames(yield* executor.tools.list())).toContain(mutable.initialToolName); // `rename_greet` renames the greet tool mid-call; the SDK server sends // `notifications/tools/list_changed` on the open connection, which the // invoke path records and turns into a stale mark after the call. const result = yield* executor.execute( ToolAddress.make(`tools.${String(INTEG)}.org.main.rename_greet`), {}, ); expect(result).toMatchObject({ ok: true }); // No manual refresh: the next tools read re-lists from the server. const refreshed = toolNames(yield* executor.tools.list()); expect(refreshed).toContain(mutable.renamedToolName); expect(refreshed).not.toContain(mutable.initialToolName); }), ); it.effect( "unknown-tool rejection fails typed, marks stale, and the catalog heals on the next read", () => Effect.gen(function* () { const mutable = makeMutableCatalogMcpServer(); const server = yield* serveMcpServer(mutable.factory); // TTL disabled: only the unknown-tool signal may trigger the re-list. const executor = yield* makeCatalogTestExecutor(server.url, { toolsSyncTtlMs: null }); expect(toolNames(yield* executor.tools.list())).toContain(mutable.initialToolName); // Mutate the server catalog outside any executor call window — the // executor has no notification to react to and its catalog is drifted. mutable.renameTool(); const staleAddress = ToolAddress.make( `tools.${String(INTEG)}.org.main.${mutable.initialToolName}`, ); const result = yield* executor.execute(staleAddress, { name: "world" }); expect(result).toMatchObject({ ok: false, error: { code: "mcp_tool_unknown" }, }); // The failure marked the connection stale; the next read converges. const healed = toolNames(yield* executor.tools.list()); expect(healed).toContain(mutable.renamedToolName); expect(healed).not.toContain(mutable.initialToolName); }), ); it.effect("expired catalogs re-list on read once older than the freshness TTL", () => Effect.gen(function* () { const mutable = makeMutableCatalogMcpServer(); const server = yield* serveMcpServer(mutable.factory); // Everything is instantly stale — every tools read re-lists. const executor = yield* makeCatalogTestExecutor(server.url, { toolsSyncTtlMs: 0 }); expect(toolNames(yield* executor.tools.list())).toContain(mutable.initialToolName); // Server-side change with no notification and no executor signal at all. mutable.renameTool(); const refreshed = toolNames(yield* executor.tools.list()); expect(refreshed).toContain(mutable.renamedToolName); expect(refreshed).not.toContain(mutable.initialToolName); }), ); it.effect("a fresh catalog inside the TTL is served from the persisted rows", () => Effect.gen(function* () { const mutable = makeMutableCatalogMcpServer(); const server = yield* serveMcpServer(mutable.factory); const executor = yield* makeCatalogTestExecutor(server.url, { toolsSyncTtlMs: 60 * 60 * 1000, }); expect(toolNames(yield* executor.tools.list())).toContain(mutable.initialToolName); const sessionsAfterFirstList = server.sessionCount(); mutable.renameTool(); // Within the TTL and with no stale signal, reads serve the persisted // catalog without dialing the server. expect(toolNames(yield* executor.tools.list())).toContain(mutable.initialToolName); expect(server.sessionCount()).toBe(sessionsAfterFirstList); }), ); it.effect("preserves the MCP discovery failure in degraded connection health", () => Effect.gen(function* () { const server = yield* serveTestHttpApp(() => Effect.succeed(HttpServerResponse.text("gateway unavailable", { status: 503 })), ); const executor = yield* makeCatalogTestExecutor(server.url("/mcp")); const connection = yield* executor.connections.get({ owner: "org", integration: INTEG, name: CONNECTION, }); expect(connection?.lastHealth).toMatchObject({ status: "degraded", detail: expect.stringContaining("Failed connecting to MCP server"), }); }), );}); // ---------------------------------------------------------------------------// Pagination — discovery follows `nextCursor` across tools/list pages.// --------------------------------------------------------------------------- const JsonRpcId = Schema.Union([Schema.String, Schema.Number, Schema.Null]);const JsonRpcRequest = Schema.Struct({ id: Schema.optional(JsonRpcId), method: Schema.String, params: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)),});type JsonRpcRequest = typeof JsonRpcRequest.Type; const decodeJsonRpcRequest = Schema.decodeUnknownOption(Schema.fromJsonString(JsonRpcRequest)); const jsonRpcResult = (request: JsonRpcRequest, result: unknown) => HttpServerResponse.jsonUnsafe({ jsonrpc: "2.0", id: request.id ?? null, result }); const pageTool = (name: string) => ({ name, description: `Tool ${name}`, inputSchema: { type: "object", properties: {} },}); // A minimal paginated fixture: page1 → cursor "p2" → page2 → done.const servePaginatedListServer = () => serveTestHttpApp((request) => Effect.gen(function* () { if (request.method === "GET") { return HttpServerResponse.text("SSE disabled", { status: 405 }); } const body = yield* request.text.pipe(Effect.orDie); return Option.match(decodeJsonRpcRequest(body), { onNone: () => HttpServerResponse.text("Invalid JSON-RPC fixture request", { status: 400 }), onSome: (rpc) => { if (rpc.method === "initialize") { return jsonRpcResult(rpc, { protocolVersion: "2025-06-18", capabilities: { tools: { listChanged: true } }, serverInfo: { name: "paginated-fixture", version: "1.0.0" }, }); } if (rpc.method === "notifications/initialized") { return HttpServerResponse.text("", { status: 202 }); } if (rpc.method === "tools/list") { const cursor = rpc.params?.cursor; if (cursor === undefined) { return jsonRpcResult(rpc, { tools: [pageTool("alpha"), pageTool("beta")], nextCursor: "p2", }); } if (cursor === "p2") { return jsonRpcResult(rpc, { tools: [pageTool("gamma")] }); } return HttpServerResponse.text("Unknown cursor", { status: 400 }); } return HttpServerResponse.text("Unexpected JSON-RPC method", { status: 400 }); }, }); }), ); describe("MCP tools/list pagination", () => { it.effect("discoverTools follows nextCursor across every page", () => Effect.gen(function* () { const server = yield* servePaginatedListServer(); const manifest = yield* discoverTools( createMcpConnector({ transport: "remote", endpoint: server.url("/mcp"), remoteTransport: "streamable-http", }), ); expect(manifest.tools.map((tool) => tool.toolName).sort()).toEqual([ "alpha", "beta", "gamma", ]); }), );}); // ---------------------------------------------------------------------------// Stale-catalog refresh concurrency.//// A tools read rebuilds every stale connection it finds. Each rebuild is an// independent upstream listing, so a host with several stale remote catalogs// must not pay the sum of every server's latency on the read that trips the// TTL. Nor may one read open an unbounded number of upstream listings.//// The fixture below refuses to answer any listing until the bound is reached,// which pins both edges at once: a serial refresh parks on the first listing// and never finishes, while an unbounded refresh puts more than// STALE_TOOLS_SYNC_CONCURRENCY listings in flight. The stale set is deliberately// one larger than the bound, so the last connection can only be served after an// earlier one completes.// --------------------------------------------------------------------------- const STALE_CONNECTIONS = STALE_TOOLS_SYNC_CONCURRENCY + 1; const serveLatchedListServer = () => Effect.gen(function* () { const armed = yield* Ref.make(false); const listings = yield* Ref.make(0); // Signalled when the bound is saturated; released by the test, not by the // fixture, so the test can first prove nothing beyond the bound arrives. const atLimit = yield* Deferred.make<void>(); const release = yield* Deferred.make<void>(); const server = yield* serveTestHttpApp((request) => Effect.gen(function* () { if (request.method === "GET") { return HttpServerResponse.text("SSE disabled", { status: 405 }); } const body = yield* request.text.pipe(Effect.orDie); const rpc = Option.getOrUndefined(decodeJsonRpcRequest(body)); if (!rpc) { return HttpServerResponse.text("Invalid JSON-RPC fixture request", { status: 400 }); } if (rpc.method === "initialize") { return jsonRpcResult(rpc, { protocolVersion: "2025-06-18", capabilities: { tools: { listChanged: true } }, serverInfo: { name: "latched-fixture", version: "1.0.0" }, }); } if (rpc.method === "notifications/initialized") { return HttpServerResponse.text("", { status: 202 }); } if (rpc.method !== "tools/list") { return HttpServerResponse.text("Unexpected JSON-RPC method", { status: 400 }); } // Once armed, park every listing until the test releases them. A serial // refresh parks on the first one and never reaches the bound. if (yield* Ref.get(armed)) { const arrived = yield* Ref.updateAndGet(listings, (n) => n + 1); if (arrived >= STALE_TOOLS_SYNC_CONCURRENCY) { yield* Deferred.succeed(atLimit, undefined); } yield* Deferred.await(release); } return jsonRpcResult(rpc, { tools: [pageTool("alpha")] }); }), ); return { // Distinct endpoint paths so each connection dials its own MCP session // instead of sharing one pooled client. endpoint: (index: number) => server.url(`/mcp/${index}`), arm: Ref.set(armed, true), awaitLimit: Deferred.await(atLimit), release: Deferred.succeed(release, undefined), listings: Ref.get(listings), } as const; }); describe("MCP stale-catalog refresh", () => { // `it.live` (real clock): proving that nothing beyond the bound is dialled // means giving a real HTTP round trip a real window to happen in, and the // timeouts below must actually fire. The TestClock advances neither. it.live("rebuilds stale connections concurrently up to the bound, then queues the rest", () => Effect.gen(function* () { const fixture = yield* serveLatchedListServer(); const executor = yield* createExecutor({ ...makeTestConfig({ plugins: [memoryCredentialsPlugin(), mcpPlugin()] as const }), // Everything is expired on every read, so a single tools read has the // whole set to rebuild. toolsSyncTtlMs: 0, // Strict mode: the assertions below synchronize on the read fiber // completing only after every rebuild has finished. With a grace // budget the read would return early and `Fiber.join` would no longer // order the final listing before the count assertion. toolsSyncGraceMs: null, }); for (let index = 0; index < STALE_CONNECTIONS; index++) { const slug = IntegrationSlug.make(`latched_mcp_${index}`); yield* executor.mcp.addServer({ name: `latched-mcp-${index}`, endpoint: fixture.endpoint(index), slug: String(slug), }); yield* executor.connections.create({ owner: "org", name: CONNECTION, integration: slug, template: TEMPLATE, value: "", }); } // Warm every catalog while the fixture still answers freely, so the // latched read below is purely the stale-refresh fan-out. yield* executor.tools.list(); yield* fixture.arm; const readFiber = yield* Effect.forkChild(executor.tools.list()); // Timeouts are well inside the harness limit, so a broken fan-out fails // on an assertion here rather than as an opaque test-runner timeout. // A serial refresh never saturates the bound and fails on this line. const saturated = yield* fixture.awaitLimit.pipe(Effect.timeoutOption("10 seconds")); expect(Option.isSome(saturated)).toBe(true); // The bound is reached and every one of those listings is still parked. // Give an unbounded fan-out ample time to dial the remaining connection: // it never may, because no permit has been given back yet. yield* Effect.sleep("500 millis"); expect(yield* fixture.listings).toBe(STALE_TOOLS_SYNC_CONCURRENCY); // Releasing the parked listings frees permits, and only then does the // last connection get dialled. yield* fixture.release; const refreshed = yield* Fiber.join(readFiber).pipe(Effect.timeoutOption("10 seconds")); expect(Option.isSome(refreshed)).toBe(true); expect(yield* fixture.listings).toBe(STALE_CONNECTIONS); }), );}); // ---------------------------------------------------------------------------// Stale-refresh grace budget.//// A tools read waits at most `toolsSyncGraceMs` for stale rebuilds, then// answers from the persisted catalog while the rebuilds finish detached. One// slow upstream server must bound neither the read nor convergence: the read// serves the stale-but-working rows now, and a later read reflects the// re-listed catalog once the server finally answers.// --------------------------------------------------------------------------- const serveLatchedMutableServer = () => Effect.gen(function* () { const catalog = yield* Ref.make("alpha"); const armed = yield* Ref.make(false); const release = yield* Deferred.make<void>(); const server = yield* serveTestHttpApp((request) => Effect.gen(function* () { if (request.method === "GET") { return HttpServerResponse.text("SSE disabled", { status: 405 }); } const body = yield* request.text.pipe(Effect.orDie); const rpc = Option.getOrUndefined(decodeJsonRpcRequest(body)); if (!rpc) { return HttpServerResponse.text("Invalid JSON-RPC fixture request", { status: 400 }); } if (rpc.method === "initialize") { return jsonRpcResult(rpc, { protocolVersion: "2025-06-18", capabilities: { tools: { listChanged: true } }, serverInfo: { name: "latched-mutable-fixture", version: "1.0.0" }, }); } if (rpc.method === "notifications/initialized") { return HttpServerResponse.text("", { status: 202 }); } if (rpc.method !== "tools/list") { return HttpServerResponse.text("Unexpected JSON-RPC method", { status: 400 }); } // Once armed, park every listing until released — the "slow server". if (yield* Ref.get(armed)) { yield* Deferred.await(release); } return jsonRpcResult(rpc, { tools: [pageTool(yield* Ref.get(catalog))] }); }), ); return { url: server.url("/mcp"), rename: Ref.set(catalog, "beta"), arm: Ref.set(armed, true), release: Deferred.succeed(release, undefined), } as const; }); describe("MCP stale-refresh grace budget", () => { // `it.live` (real clock): the grace timeout must actually fire while a real // HTTP listing stays parked. it.live("a read outlasting the grace serves the stored catalog, then converges", () => Effect.gen(function* () { const fixture = yield* serveLatchedMutableServer(); const executor = yield* makeCatalogTestExecutor(fixture.url, { // Every read finds the catalog expired, and waits at most 100ms. toolsSyncTtlMs: 0, toolsSyncGraceMs: 100, }); expect(toolNames(yield* executor.tools.list())).toContain("alpha"); // The server's catalog changes AND the server stops answering listings. yield* fixture.rename; yield* fixture.arm; // The re-list is parked, so only the grace path can produce an answer — // and it is the persisted (stale) catalog, not a failure or a hang. expect(toolNames(yield* executor.tools.list())).toContain("alpha"); // Once the server answers, the detached rebuild lands and a later read // reflects the re-listed catalog. yield* fixture.release; const converged = yield* Effect.gen(function* () { while (true) { const names = toolNames(yield* executor.tools.list()); if (names.includes("beta")) return names; yield* Effect.sleep("100 millis");Lines 1–500 of 506. Load more
// --------------------------------------------------------------------------- // MCP tool-catalog freshness (end-to-end). // // The persisted per-connection tool catalog must converge with the server's // live tool set. Spec inputs the executor reacts to: // - `notifications/tools/list_changed` received during a call window marks // the connection stale; the next tools read re-lists. // - An unknown-tool protocol error (`-32602`, "Tool … not found") on // `tools/call` means the catalog drifted: the call fails with a typed // `mcp_tool_unknown` ToolResult and the catalog heals on the next read. // - `tools/list` is paginated; discovery follows `nextCursor` to the end. // - A failed listing (server unreachable) is non-authoritative: the // previously persisted catalog is kept, not wiped. // --------------------------------------------------------------------------- import { describe, expect, it } from "@effect/vitest"; import { Deferred, Effect, Fiber, Option, Ref, Schema } from "effect"; import { HttpServerResponse } from "effect/unstable/http"; import { AuthTemplateSlug, ConnectionName, IntegrationSlug, STALE_TOOLS_SYNC_CONCURRENCY, ToolAddress, createExecutor, } from "@executor-js/sdk"; import { makeTestConfig, memoryCredentialsPlugin, serveTestHttpApp, } from "@executor-js/sdk/testing"; import { mcpPlugin } from "./plugin"; import { createMcpConnector } from "./connection"; import { discoverTools } from "./discover"; import { makeMutableCatalogMcpServer, serveMcpServer } from "../testing"; const INTEG = IntegrationSlug.make("catalog_mcp"); const CONNECTION = ConnectionName.make("main"); const TEMPLATE = AuthTemplateSlug.make("none"); const makeCatalogTestExecutor = ( serverUrl: string, options?: { readonly toolsSyncTtlMs?: number | null; readonly toolsSyncGraceMs?: number | null; }, ) => createExecutor({ ...makeTestConfig({ plugins: [memoryCredentialsPlugin(), mcpPlugin()] as const }), ...(options?.toolsSyncTtlMs === undefined ? {} : { toolsSyncTtlMs: options.toolsSyncTtlMs }), ...(options?.toolsSyncGraceMs === undefined ? {} : { toolsSyncGraceMs: options.toolsSyncGraceMs }), }).pipe( Effect.tap((executor) => Effect.gen(function* () { yield* executor.mcp.addServer({ name: "catalog-mcp", endpoint: serverUrl, slug: String(INTEG), }); yield* executor.connections.create({ owner: "org", name: CONNECTION, integration: INTEG, template: TEMPLATE, value: "", }); }), ), ); const toolNames = (tools: readonly { readonly name: unknown }[]): readonly string[] => tools.map((tool) => String(tool.name)).sort(); describe("MCP tool-catalog sync (end-to-end)", () => { it.effect( "tools/list_changed during a call marks the catalog stale and the next read re-lists", () => Effect.gen(function* () { const mutable = makeMutableCatalogMcpServer(); const server = yield* serveMcpServer(mutable.factory); const executor = yield* makeCatalogTestExecutor(server.url); expect(toolNames(yield* executor.tools.list())).toContain(mutable.initialToolName); // `rename_greet` renames the greet tool mid-call; the SDK server sends // `notifications/tools/list_changed` on the open connection, which the // invoke path records and turns into a stale mark after the call. const result = yield* executor.execute( ToolAddress.make(`tools.${String(INTEG)}.org.main.rename_greet`), {}, ); expect(result).toMatchObject({ ok: true }); // No manual refresh: the next tools read re-lists from the server. const refreshed = toolNames(yield* executor.tools.list()); expect(refreshed).toContain(mutable.renamedToolName); expect(refreshed).not.toContain(mutable.initialToolName); }), ); it.effect( "unknown-tool rejection fails typed, marks stale, and the catalog heals on the next read", () => Effect.gen(function* () { const mutable = makeMutableCatalogMcpServer(); const server = yield* serveMcpServer(mutable.factory); // TTL disabled: only the unknown-tool signal may trigger the re-list. const executor = yield* makeCatalogTestExecutor(server.url, { toolsSyncTtlMs: null }); expect(toolNames(yield* executor.tools.list())).toContain(mutable.initialToolName); // Mutate the server catalog outside any executor call window — the // executor has no notification to react to and its catalog is drifted. mutable.renameTool(); const staleAddress = ToolAddress.make( `tools.${String(INTEG)}.org.main.${mutable.initialToolName}`, ); const result = yield* executor.execute(staleAddress, { name: "world" }); expect(result).toMatchObject({ ok: false, error: { code: "mcp_tool_unknown" }, }); // The failure marked the connection stale; the next read converges. const healed = toolNames(yield* executor.tools.list()); expect(healed).toContain(mutable.renamedToolName); expect(healed).not.toContain(mutable.initialToolName); }), ); it.effect("expired catalogs re-list on read once older than the freshness TTL", () => Effect.gen(function* () { const mutable = makeMutableCatalogMcpServer(); const server = yield* serveMcpServer(mutable.factory); // Everything is instantly stale — every tools read re-lists. const executor = yield* makeCatalogTestExecutor(server.url, { toolsSyncTtlMs: 0 }); expect(toolNames(yield* executor.tools.list())).toContain(mutable.initialToolName); // Server-side change with no notification and no executor signal at all. mutable.renameTool(); const refreshed = toolNames(yield* executor.tools.list()); expect(refreshed).toContain(mutable.renamedToolName); expect(refreshed).not.toContain(mutable.initialToolName); }), ); it.effect("a fresh catalog inside the TTL is served from the persisted rows", () => Effect.gen(function* () { const mutable = makeMutableCatalogMcpServer(); const server = yield* serveMcpServer(mutable.factory); const executor = yield* makeCatalogTestExecutor(server.url, { toolsSyncTtlMs: 60 * 60 * 1000, }); expect(toolNames(yield* executor.tools.list())).toContain(mutable.initialToolName); const sessionsAfterFirstList = server.sessionCount(); mutable.renameTool(); // Within the TTL and with no stale signal, reads serve the persisted // catalog without dialing the server. expect(toolNames(yield* executor.tools.list())).toContain(mutable.initialToolName); expect(server.sessionCount()).toBe(sessionsAfterFirstList); }), ); it.effect("preserves the MCP discovery failure in degraded connection health", () => Effect.gen(function* () { const server = yield* serveTestHttpApp(() => Effect.succeed(HttpServerResponse.text("gateway unavailable", { status: 503 })), ); const executor = yield* makeCatalogTestExecutor(server.url("/mcp")); const connection = yield* executor.connections.get({ owner: "org", integration: INTEG, name: CONNECTION, }); expect(connection?.lastHealth).toMatchObject({ status: "degraded", detail: expect.stringContaining("Failed connecting to MCP server"), }); }), ); }); // --------------------------------------------------------------------------- // Pagination — discovery follows `nextCursor` across tools/list pages. // --------------------------------------------------------------------------- const JsonRpcId = Schema.Union([Schema.String, Schema.Number, Schema.Null]); const JsonRpcRequest = Schema.Struct({ id: Schema.optional(JsonRpcId), method: Schema.String, params: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)), }); type JsonRpcRequest = typeof JsonRpcRequest.Type; const decodeJsonRpcRequest = Schema.decodeUnknownOption(Schema.fromJsonString(JsonRpcRequest)); const jsonRpcResult = (request: JsonRpcRequest, result: unknown) => HttpServerResponse.jsonUnsafe({ jsonrpc: "2.0", id: request.id ?? null, result }); const pageTool = (name: string) => ({ name, description: `Tool ${name}`, inputSchema: { type: "object", properties: {} }, }); // A minimal paginated fixture: page1 → cursor "p2" → page2 → done. const servePaginatedListServer = () => serveTestHttpApp((request) => Effect.gen(function* () { if (request.method === "GET") { return HttpServerResponse.text("SSE disabled", { status: 405 }); } const body = yield* request.text.pipe(Effect.orDie); return Option.match(decodeJsonRpcRequest(body), { onNone: () => HttpServerResponse.text("Invalid JSON-RPC fixture request", { status: 400 }), onSome: (rpc) => { if (rpc.method === "initialize") { return jsonRpcResult(rpc, { protocolVersion: "2025-06-18", capabilities: { tools: { listChanged: true } }, serverInfo: { name: "paginated-fixture", version: "1.0.0" }, }); } if (rpc.method === "notifications/initialized") { return HttpServerResponse.text("", { status: 202 }); } if (rpc.method === "tools/list") { const cursor = rpc.params?.cursor; if (cursor === undefined) { return jsonRpcResult(rpc, { tools: [pageTool("alpha"), pageTool("beta")], nextCursor: "p2", }); } if (cursor === "p2") { return jsonRpcResult(rpc, { tools: [pageTool("gamma")] }); } return HttpServerResponse.text("Unknown cursor", { status: 400 }); } return HttpServerResponse.text("Unexpected JSON-RPC method", { status: 400 }); }, }); }), ); describe("MCP tools/list pagination", () => { it.effect("discoverTools follows nextCursor across every page", () => Effect.gen(function* () { const server = yield* servePaginatedListServer(); const manifest = yield* discoverTools( createMcpConnector({ transport: "remote", endpoint: server.url("/mcp"), remoteTransport: "streamable-http", }), ); expect(manifest.tools.map((tool) => tool.toolName).sort()).toEqual([ "alpha", "beta", "gamma", ]); }), ); }); // --------------------------------------------------------------------------- // Stale-catalog refresh concurrency. // // A tools read rebuilds every stale connection it finds. Each rebuild is an // independent upstream listing, so a host with several stale remote catalogs // must not pay the sum of every server's latency on the read that trips the // TTL. Nor may one read open an unbounded number of upstream listings. // // The fixture below refuses to answer any listing until the bound is reached, // which pins both edges at once: a serial refresh parks on the first listing // and never finishes, while an unbounded refresh puts more than // STALE_TOOLS_SYNC_CONCURRENCY listings in flight. The stale set is deliberately // one larger than the bound, so the last connection can only be served after an // earlier one completes. // --------------------------------------------------------------------------- const STALE_CONNECTIONS = STALE_TOOLS_SYNC_CONCURRENCY + 1; const serveLatchedListServer = () => Effect.gen(function* () { const armed = yield* Ref.make(false); const listings = yield* Ref.make(0); // Signalled when the bound is saturated; released by the test, not by the // fixture, so the test can first prove nothing beyond the bound arrives. const atLimit = yield* Deferred.make<void>(); const release = yield* Deferred.make<void>(); const server = yield* serveTestHttpApp((request) => Effect.gen(function* () { if (request.method === "GET") { return HttpServerResponse.text("SSE disabled", { status: 405 }); } const body = yield* request.text.pipe(Effect.orDie); const rpc = Option.getOrUndefined(decodeJsonRpcRequest(body)); if (!rpc) { return HttpServerResponse.text("Invalid JSON-RPC fixture request", { status: 400 }); } if (rpc.method === "initialize") { return jsonRpcResult(rpc, { protocolVersion: "2025-06-18", capabilities: { tools: { listChanged: true } }, serverInfo: { name: "latched-fixture", version: "1.0.0" }, }); } if (rpc.method === "notifications/initialized") { return HttpServerResponse.text("", { status: 202 }); } if (rpc.method !== "tools/list") { return HttpServerResponse.text("Unexpected JSON-RPC method", { status: 400 }); } // Once armed, park every listing until the test releases them. A serial // refresh parks on the first one and never reaches the bound. if (yield* Ref.get(armed)) { const arrived = yield* Ref.updateAndGet(listings, (n) => n + 1); if (arrived >= STALE_TOOLS_SYNC_CONCURRENCY) { yield* Deferred.succeed(atLimit, undefined); } yield* Deferred.await(release); } return jsonRpcResult(rpc, { tools: [pageTool("alpha")] }); }), ); return { // Distinct endpoint paths so each connection dials its own MCP session // instead of sharing one pooled client. endpoint: (index: number) => server.url(`/mcp/${index}`), arm: Ref.set(armed, true), awaitLimit: Deferred.await(atLimit), release: Deferred.succeed(release, undefined), listings: Ref.get(listings), } as const; }); describe("MCP stale-catalog refresh", () => { // `it.live` (real clock): proving that nothing beyond the bound is dialled // means giving a real HTTP round trip a real window to happen in, and the // timeouts below must actually fire. The TestClock advances neither. it.live("rebuilds stale connections concurrently up to the bound, then queues the rest", () => Effect.gen(function* () { const fixture = yield* serveLatchedListServer(); const executor = yield* createExecutor({ ...makeTestConfig({ plugins: [memoryCredentialsPlugin(), mcpPlugin()] as const }), // Everything is expired on every read, so a single tools read has the // whole set to rebuild. toolsSyncTtlMs: 0, // Strict mode: the assertions below synchronize on the read fiber // completing only after every rebuild has finished. With a grace // budget the read would return early and `Fiber.join` would no longer // order the final listing before the count assertion. toolsSyncGraceMs: null, }); for (let index = 0; index < STALE_CONNECTIONS; index++) { const slug = IntegrationSlug.make(`latched_mcp_${index}`); yield* executor.mcp.addServer({ name: `latched-mcp-${index}`, endpoint: fixture.endpoint(index), slug: String(slug), }); yield* executor.connections.create({ owner: "org", name: CONNECTION, integration: slug, template: TEMPLATE, value: "", }); } // Warm every catalog while the fixture still answers freely, so the // latched read below is purely the stale-refresh fan-out. yield* executor.tools.list(); yield* fixture.arm; const readFiber = yield* Effect.forkChild(executor.tools.list()); // Timeouts are well inside the harness limit, so a broken fan-out fails // on an assertion here rather than as an opaque test-runner timeout. // A serial refresh never saturates the bound and fails on this line. const saturated = yield* fixture.awaitLimit.pipe(Effect.timeoutOption("10 seconds")); expect(Option.isSome(saturated)).toBe(true); // The bound is reached and every one of those listings is still parked. // Give an unbounded fan-out ample time to dial the remaining connection: // it never may, because no permit has been given back yet. yield* Effect.sleep("500 millis"); expect(yield* fixture.listings).toBe(STALE_TOOLS_SYNC_CONCURRENCY); // Releasing the parked listings frees permits, and only then does the // last connection get dialled. yield* fixture.release; const refreshed = yield* Fiber.join(readFiber).pipe(Effect.timeoutOption("10 seconds")); expect(Option.isSome(refreshed)).toBe(true); expect(yield* fixture.listings).toBe(STALE_CONNECTIONS); }), ); }); // --------------------------------------------------------------------------- // Stale-refresh grace budget. // // A tools read waits at most `toolsSyncGraceMs` for stale rebuilds, then // answers from the persisted catalog while the rebuilds finish detached. One // slow upstream server must bound neither the read nor convergence: the read // serves the stale-but-working rows now, and a later read reflects the // re-listed catalog once the server finally answers. // --------------------------------------------------------------------------- const serveLatchedMutableServer = () => Effect.gen(function* () { const catalog = yield* Ref.make("alpha"); const armed = yield* Ref.make(false); const release = yield* Deferred.make<void>(); const server = yield* serveTestHttpApp((request) => Effect.gen(function* () { if (request.method === "GET") { return HttpServerResponse.text("SSE disabled", { status: 405 }); } const body = yield* request.text.pipe(Effect.orDie); const rpc = Option.getOrUndefined(decodeJsonRpcRequest(body)); if (!rpc) { return HttpServerResponse.text("Invalid JSON-RPC fixture request", { status: 400 }); } if (rpc.method === "initialize") { return jsonRpcResult(rpc, { protocolVersion: "2025-06-18", capabilities: { tools: { listChanged: true } }, serverInfo: { name: "latched-mutable-fixture", version: "1.0.0" }, }); } if (rpc.method === "notifications/initialized") { return HttpServerResponse.text("", { status: 202 }); } if (rpc.method !== "tools/list") { return HttpServerResponse.text("Unexpected JSON-RPC method", { status: 400 }); } // Once armed, park every listing until released — the "slow server". if (yield* Ref.get(armed)) { yield* Deferred.await(release); } return jsonRpcResult(rpc, { tools: [pageTool(yield* Ref.get(catalog))] }); }), ); return { url: server.url("/mcp"), rename: Ref.set(catalog, "beta"), arm: Ref.set(armed, true), release: Deferred.succeed(release, undefined), } as const; }); describe("MCP stale-refresh grace budget", () => { // `it.live` (real clock): the grace timeout must actually fire while a real // HTTP listing stays parked. it.live("a read outlasting the grace serves the stored catalog, then converges", () => Effect.gen(function* () { const fixture = yield* serveLatchedMutableServer(); const executor = yield* makeCatalogTestExecutor(fixture.url, { // Every read finds the catalog expired, and waits at most 100ms. toolsSyncTtlMs: 0, toolsSyncGraceMs: 100, }); expect(toolNames(yield* executor.tools.list())).toContain("alpha"); // The server's catalog changes AND the server stops answering listings. yield* fixture.rename; yield* fixture.arm; // The re-list is parked, so only the grace path can produce an answer — // and it is the persisted (stale) catalog, not a failure or a hang. expect(toolNames(yield* executor.tools.list())).toContain("alpha"); // Once the server answers, the detached rebuild lands and a later read // reflects the re-listed catalog. yield* fixture.release; const converged = yield* Effect.gen(function* () { while (true) { const names = toolNames(yield* executor.tools.list()); if (names.includes("beta")) return names; yield* Effect.sleep("100 millis");