diff --git a/changelog.d/mcpl-close-down-server.fixed.md b/changelog.d/mcpl-close-down-server.fixed.md new file mode 100644 index 000000000..635a3d0b4 --- /dev/null +++ b/changelog.d/mcpl-close-down-server.fixed.md @@ -0,0 +1,36 @@ +- A `reconnect: true` MCPL server that the framework closes for good now + leaves the registry even when it's down at the time, so a later connect of + its id isn't refused as already registered. The framework closes a server + for good when a runtime `connectMcplServer` or `restartMcplServer` fails + its Discord awareness accounting or its policy registration, and when the + server refuses the host's policy with the `"close"` fallback. The registry + dropped a closed server only when its connection reported the close, and a + connection that's down between retries reports none. So the stub a failed + first connect leaves stayed registered after such a close, and a later + `connectMcplServer` of its id failed as already registered until + `disconnectMcplServer` or `restartMcplServer` dropped it. A close that + fails still keeps the server registered, so a later disconnect, restart or + shutdown can close it again, and the failed connect now logs it. +- A runtime `connectMcplServer` or `restartMcplServer` that fails after + pausing the other MCPL servers' data planes no longer leaves their data + held. The connect paused them behind a gate that it settled only when it + succeeded, so their pushes, incoming messages and inference requests + waited until some other gate completed. + - When its Discord awareness accounting fails, it now fails that gate, as + a failed reconnect, tools-list reconciliation, + `syncDiscordAwarenessMarkers()` or `resume()` already does: each running + server is recycled, and its reconnect's own gate holds its data until + that reconnect's accounting succeeds. A server without `reconnect: true` + is closed for good there, as on those paths. While the host is stopping, + the gate is released instead. + - When its policy registration throws after accounting succeeded, it now + closes the new server for good, where it stayed connected and half + registered, and completes the gate, with no recycle. +- `McplServerConnection.ready()` and `readyControlPlane()` now do nothing once + the connection is closed for good: closed, with reconnect off, as after + `close()` or a drop with reconnect off. What it held stays held. Any gate's + completion used to pass the part of it that needs no grant to listeners, + as for a connection whose close failed and so stayed registered. While the + host stops, a connection closed for good, such as one being closed, no + longer passes what it held to the stopping refusal, whose answers can't be + sent on a closed connection. diff --git a/src/framework.ts b/src/framework.ts index 3455acb3b..0f4e79003 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -17520,7 +17520,8 @@ export class AgentFramework { // would start reconnects only for stop() to close them again. Release the // barrier instead, as a completed one is released: nothing waits on a // rejected barrier, and unless the host is quiesced, what the pause held - // back reaches the listeners' stopping refusal. + // back reaches the listeners' stopping refusal, except what a connection + // closed for good held, which stays held. if (this.queue?.isClosed) { this.completeMcplDataPlaneGate(barrier); return; @@ -17535,6 +17536,25 @@ export class AgentFramework { })); } + /** + * Close a runtime connect's new connection for good, through the registry, + * after a step of that connect failed. A close that fails keeps the entry, + * so a later disconnect, restart or shutdown can close it again, and is + * logged here rather than replacing the failure the connect reports. + */ + private async closeFailedRuntimeConnect( + connection: McplServerConnection, + failedStep: string, + ): Promise { + await this.mcplServerRegistry!.closeServer(connection.id, connection).catch((closeError) => { + console.error( + `[mcpl] could not close MCPL server "${connection.id}" after its ${failedStep} failed; ` + + 'it stays registered until a disconnect, restart or shutdown closes it:', + closeError instanceof Error ? closeError.message : closeError, + ); + }); + } + /** * Connect a single MCPL server: register routing entries, open the * connection, wire events, and initialize feature sets / scopes / @@ -17627,13 +17647,27 @@ export class AgentFramework { // the runtime connect or restart that asked for it. (A startup connect // defers awareness and returns before this gate; a startup that fails // ends in create()'s release, which closes its servers.) + // That close goes through the registry, which drops the entry itself + // rather than waiting for a 'close' event: close() emits none for a + // connection that's down between retries, such as a stub waiting for + // its next retry, and a stub left registered would refuse a later + // connect of its id as already registered. + // Then fail the gate, as every other runtime gate does when its + // accounting fails. Installing it paused every registered server's + // data plane, and left standing it would hold their data until some + // other gate completed. While serving, failMcplDataPlaneGate recycles + // those servers, so a server without `reconnect: true` stays closed; + // while stopping, it releases the gate. It runs after the close, which + // has taken this connection out of the registry unless the close + // failed, which closeFailedRuntimeConnect logs. // Ordering matters: awareness accounting settles BEFORE the §5.3 // policy round-trip below, so an accounting failure aborts with the // grant still empty and every buffered event still behind the gate — // a failed connect never reports a released data plane. (The reverse // order let a push arrive during the policy await and get flushed by // teardown.) - await connection.close().catch(() => {}); + await this.closeFailedRuntimeConnect(connection, 'awareness accounting'); + await this.failMcplDataPlaneGate(awarenessBarrier, `connect of MCPL server "${config.id}"`, error); throw error; } @@ -17641,7 +17675,22 @@ export class AgentFramework { // completes: the grant is established (or knowingly left empty) before // ANY buffered traffic (control or data) flows. complete() then // ready()s both planes through the admission gate. - await this.registerMcplServerFeatures(config, connection); + try { + await this.registerMcplServerFeatures(config, connection); + } catch (error) { + // A policy registration that throws would leave the new connection + // live and half registered, and the gate standing over every other + // server's data. So close the connection for good through the + // registry, as the catch above does, then complete the gate rather + // than fail it: accounting succeeded, so nothing calls for a recycle. + // A close that fails keeps the connection registered, but a connection + // closed for good readies nothing, so this completion, and any later + // one, leaves it paused. + // Every exit of a runtime connect settles the gate it installed. + await this.closeFailedRuntimeConnect(connection, 'policy registration'); + this.completeMcplDataPlaneGate(awarenessBarrier); + throw error; + } this.completeMcplDataPlaneGate(awarenessBarrier); this.emitTrace({ type: 'module:added', moduleName: `mcpl:${config.id}` }); @@ -17797,7 +17846,7 @@ export class AgentFramework { ...AgentFramework.policyRefusalFields(receipt), }); if (fallback === 'close') { - await connection.close(); + await this.mcplServerRegistry!.closeServer(connection.id, connection); return; } connection.establishGrant(CapabilityGrant.empty()); @@ -17915,7 +17964,7 @@ export class AgentFramework { ...AgentFramework.policyRefusalFields(receipt), }); if (fallback === 'close') { - await connection.close(); + await this.mcplServerRegistry!.closeServer(connection.id, connection); return; } // mcp-only (§3.2): grant stays empty, MCPL privileged surface stays diff --git a/src/mcpl/server-connection.ts b/src/mcpl/server-connection.ts index 16bcbbeb4..bb105f53b 100644 --- a/src/mcpl/server-connection.ts +++ b/src/mcpl/server-connection.ts @@ -315,6 +315,13 @@ export class McplServerConnection extends EventEmitter { return this.reconnectEnabled; } + /** Closed with no reconnect to come: after close(), whether or not its + * transport close succeeded, or a drop with reconnect off. A connection + * between retries keeps reconnectEnabled, so it isn't closed for good. */ + private get closedForGood(): boolean { + return this.closed && !this.reconnectEnabled; + } + /** * Private constructor. Use `McplServerConnection.connect()` instead. */ @@ -342,8 +349,17 @@ export class McplServerConnection extends EventEmitter { * Mark the connection as ready — flushes any events that arrived between * construction (when setupMessageRouting starts emitting) and now (when * the caller has attached listeners via wireMcplEvents). + * + * Does nothing once the connection is closed for good. Its policy was reset + * when it closed, so what it held that needs a grant would be refused, and a + * closed connection sends no response, so nothing it held could be answered. + * Only what needs no grant would pass, to listeners of a server that won't + * be served again. A connection between retries isn't closed for good: what + * it held, such as a reconnect stub's connect-failed, still reaches its + * listeners. */ ready(): void { + if (this.closedForGood) return; this.controlPlaneReady = true; this.dataPlaneReady = true; const pending = this.bufferedEvents; @@ -375,9 +391,11 @@ export class McplServerConnection extends EventEmitter { * Release only server-establishment and lifecycle/control traffic. Responses * to host requests are never event-buffered, so tools/list and tools/call * responses remain live as well. Inference-bearing requests stay queued in - * their original relative order until {@link ready} is called. + * their original relative order until {@link ready} is called. Like + * {@link ready}, it does nothing once the connection is closed for good. */ readyControlPlane(): void { + if (this.closedForGood) return; this.controlPlaneReady = true; const pending = this.bufferedEvents; this.bufferedEvents = []; @@ -955,7 +973,8 @@ export class McplServerConnection extends EventEmitter { const attempt = Promise.resolve().then(async () => { // An already opened handshake owns a transport too. It hands that // transport back without admission once reconnect has been disabled. - // The registry bounds this wait; retry reuses this real close attempt. + // Only the registry's closeAll bounds its caller's wait for this; retry + // reuses this real close attempt. if (this.reconnectPromise) await this.reconnectPromise; if (this.transport) await this.transport.close(); this.closeIncomplete = false; diff --git a/src/mcpl/server-registry.ts b/src/mcpl/server-registry.ts index 7aed41e47..f4dd246f7 100644 --- a/src/mcpl/server-registry.ts +++ b/src/mcpl/server-registry.ts @@ -148,7 +148,16 @@ export class McplServerRegistry { })), timeoutMs); } - private closeServer(id: string, connection: McplServerConnection): Promise { + /** + * Close a server for good and drop its entry, if this connection is still + * the one registered under `id`. The entry goes when close() resolves, + * whatever the connection reports: close() emits no 'close' for a + * connection that's down between retries, such as a reconnect stub waiting + * for its next retry, so the 'close' listener alone would leave it + * registered. A close that fails keeps the entry and rejects; a close + * already under way is shared. + */ + closeServer(id: string, connection: McplServerConnection): Promise { const existing = this.closing.get(connection); if (existing) return existing; const attempt = Promise.resolve().then(() => connection.close()).then(() => { diff --git a/test/mcpl-awareness-barrier.test.ts b/test/mcpl-awareness-barrier.test.ts index 6530bee6a..6277c0797 100644 --- a/test/mcpl-awareness-barrier.test.ts +++ b/test/mcpl-awareness-barrier.test.ts @@ -14,6 +14,7 @@ import { EventEmitter } from 'node:events'; import { AgentFramework } from '../src/framework.js'; import { McplServerConnection } from '../src/mcpl/server-connection.js'; import { CapabilityGrant } from '../src/mcpl/capability-grant.js'; +import { FeatureSetManager } from '../src/mcpl/feature-set-manager.js'; import type { McplServerConfig } from '../src/mcpl/types.js'; import { DiscordAwarenessOutbox } from '../src/recovery/discord-awareness-outbox.js'; import { MockMembrane } from './helpers/mock-membrane.js'; @@ -700,6 +701,261 @@ for (const writeMethod of ['recordSuccess', 'recordFailure'] as const) test( }, ); +/** A second fixture server, `heartbeat`, for a framework to serve from create(). */ +function heartbeatConfig(dir: string, overrides: Partial = {}): McplServerConfig { + return { + id: 'heartbeat', + command: process.execPath, + args: [FIXTURE], + requestTimeoutMs: 1_000, + env: { + SERVER_ID: 'heartbeat', + STATUS_PATH: join(dir, 'heartbeat-status.jsonl'), + GENERATION_PATH: join(dir, 'heartbeat-generation'), + }, + enabledFeatureSets: ['chat'], + ...overrides, + }; +} + +/** + * A framework serving `heartbeat`, then a runtime connect of `discord` whose + * first connect fails, so the registry holds a reconnect stub, and whose + * awareness accounting then fails (recordFailure throws). Serving another + * server brings MCPL up at create(), so the runtime connect is the one that + * accounts for the pending work: brought up lazily by that connect, MCPL + * would account for it in its own startup barrier first. A failure after + * create() stops the framework, so no child outlives the test. + */ +async function failRuntimeConnectWhileServing(dir: string, heartbeatOverrides: Partial = {}) { + const statusPath = join(dir, 'status.jsonl'); + const heartbeatStatusPath = join(dir, 'heartbeat-status.jsonl'); + const { outboxPath, config } = frameworkConfig(dir, { + STATUS_PATH: statusPath, + LEDGER_PATH: join(dir, 'awareness.json'), + GENERATION_PATH: join(dir, 'generation'), + FAIL_INITIAL_GENERATION: '1', + }, 1_000, { + reconnect: true, + // A stub waiting a minute for its retry is down between retries when the + // accounting failure closes it: the close that emits no 'close'. + reconnectIntervalMs: 60_000, + reconnectMaxIntervalMs: 60_000, + }); + const discord = config.mcplServers[0]!; + const heartbeat = heartbeatConfig(dir, heartbeatOverrides); + const framework = await AgentFramework.create({ ...config, mcplServers: [heartbeat] }); + try { + const traces: Array<{ type: string; serverId?: string; error?: string }> = []; + framework.onTrace((event) => traces.push(event as never)); + preparePending(outboxPath); + const prototype = DiscordAwarenessOutbox.prototype as any; + const original = prototype.recordFailure; + prototype.recordFailure = () => { throw new Error('injected recordFailure write failure'); }; + try { + await assert.rejects( + framework.connectMcplServer(discord), + /awareness accounting failed.*injected recordFailure write failure/i, + ); + } finally { + prototype.recordFailure = original; + } + assert.equal( + records(statusPath).filter((record) => record.event === 'initial-connect-failure').length, + 1, + 'the first connect failed, so the registry held a reconnect stub', + ); + // The catch closes the refused connection before it fails the gate, so + // the gate's failure finds it gone from the registry and doesn't trace or + // recycle it. + assert.equal( + traces.some((trace) => trace.type === 'mcpl:server-error' + && trace.serverId === discord.id + && /accounting unhealthy/.test(trace.error ?? '')), + false, + 'the refused connection was out of the registry when the gate failed', + ); + return { framework, discord, heartbeatStatusPath, traces }; + } catch (error) { + await framework.stop(); + throw error; + } +} + +test('a runtime connect whose awareness accounting fails frees the id of the reconnect stub it closed, and closes a running server without reconnect: true for good', async () => { + mkdirSync(TEST_ROOT, { recursive: true }); + const dir = mkdtempSync(join(TEST_ROOT, 'awareness-runtime-stub-')); + let framework: AgentFramework | undefined; + + try { + const failed = await failRuntimeConnectWhileServing(dir); + framework = failed.framework; + + // The running server has no reconnect: true, so the failed gate's + // recycle closed it for good: neither connected and held, nor retrying. + await waitFor('the running server closed for good', () => { + const heartbeat = failed.framework.listMcplServers().find((server) => server.id === 'heartbeat'); + return heartbeat !== undefined && !heartbeat.connected && !heartbeat.retrying; + }, 10_000); + + // Connecting the id again reaches the server: a stub left registered + // would fail this connect as already registered. + await framework.connectMcplServer(failed.discord); + assert.equal(framework.listMcplServers().find((server) => server.id === 'discord')?.connected, true); + } finally { + await framework?.stop(); + rmSync(dir, { recursive: true, force: true }); + } +}); + +test('a runtime connect whose awareness accounting fails fails its gate, so a running server is recycled rather than held', async () => { + mkdirSync(TEST_ROOT, { recursive: true }); + const dir = mkdtempSync(join(TEST_ROOT, 'awareness-runtime-gate-')); + let framework: AgentFramework | undefined; + + try { + const failed = await failRuntimeConnectWhileServing(dir, { + reconnect: true, + reconnectIntervalMs: 20, + reconnectMaxIntervalMs: 20, + }); + framework = failed.framework; + assert.ok( + failed.traces.some((trace) => trace.type === 'mcpl:server-error' + && trace.serverId === 'heartbeat' + && /accounting unhealthy/.test(trace.error ?? '')), + 'the gate failed for the running server, as a failed gate on the other runtime paths does', + ); + + // Recycled, the server reconnects, and its data flows once its + // reconnect's accounting succeeds: the new process's startup push is + // answered. + await waitFor('the recycled server\'s startup push answered', () => + records(failed.heartbeatStatusPath).some((record) => + record.generation === 2 && record.event === 'push-response'), 10_000); + } finally { + await framework?.stop(); + rmSync(dir, { recursive: true, force: true }); + } +}); + +/** + * A framework serving `heartbeat`, then a runtime connect of `discord` whose + * awareness accounting succeeds and whose §5.3 policy registration throws + * (initializeServer made to throw for it). With `closeFails`, the close that + * follows fails once in its transport close, after close() has marked the + * connection closed for good, as a real one fails. Returns the framework with + * every patch undone; a failure before then stops it. + */ +async function failPolicyRegistrationWhileServing(dir: string, closeFails: boolean) { + const { config } = frameworkConfig(dir, { + STATUS_PATH: join(dir, 'status.jsonl'), + LEDGER_PATH: join(dir, 'awareness.json'), + GENERATION_PATH: join(dir, 'generation'), + }); + const discord = config.mcplServers[0]!; + const framework = await AgentFramework.create({ ...config, mcplServers: [heartbeatConfig(dir)] }); + try { + const traces: Array<{ type: string; serverId?: string; error?: string }> = []; + framework.onTrace((event) => traces.push(event as never)); + const features = FeatureSetManager.prototype as any; + const initializeServer = features.initializeServer; + features.initializeServer = function (this: unknown, serverId: string, ...rest: unknown[]) { + if (serverId === discord.id) throw new Error('injected policy registration failure'); + return initializeServer.call(this, serverId, ...rest); + }; + const connections = McplServerConnection.prototype as any; + const close = connections.close; + let closeFailed = false; + connections.close = function (this: any) { + if (closeFails && this.id === discord.id && !closeFailed && this.transport) { + closeFailed = true; + const transport = this.transport; + const transportClose = transport.close; + transport.close = () => { + transport.close = transportClose; + return Promise.reject(new Error('injected close failure')); + }; + } + return close.call(this); + }; + try { + // The connect reports its own failure, not the close's. + await assert.rejects(framework.connectMcplServer(discord), /injected policy registration failure/); + } finally { + features.initializeServer = initializeServer; + connections.close = close; + } + assert.equal(closeFailed, closeFails, 'the close failed only when asked to'); + const internals = framework as unknown as { + discordAwarenessBarrier: unknown; + mcplServerRegistry: { getServer(id: string): unknown }; + }; + assert.equal(internals.discordAwarenessBarrier, null, 'the failed connect completed the gate it installed'); + const running = internals.mcplServerRegistry.getServer('heartbeat') as { isConnected: boolean; dataPlaneReady: boolean }; + assert.equal(running.isConnected, true, 'accounting succeeded, so the running server was not recycled'); + assert.equal(running.dataPlaneReady, true, 'the running server\'s data flows'); + assert.equal( + traces.some((trace) => trace.type === 'mcpl:server-error' && /accounting unhealthy/.test(trace.error ?? '')), + false, + 'no gate failed', + ); + return { framework, discord, registry: internals.mcplServerRegistry }; + } catch (error) { + await framework.stop(); + throw error; + } +} + +test('a runtime connect whose policy registration throws closes its connection for good and completes its gate, so a running server keeps its data', async () => { + mkdirSync(TEST_ROOT, { recursive: true }); + const dir = mkdtempSync(join(TEST_ROOT, 'awareness-runtime-policy-')); + let framework: AgentFramework | undefined; + + try { + const failed = await failPolicyRegistrationWhileServing(dir, false); + framework = failed.framework; + assert.equal(failed.registry.getServer(failed.discord.id), null, 'the half-registered connection was closed for good'); + + // Connecting the id again reaches the server. + await framework.connectMcplServer(failed.discord); + assert.equal(framework.listMcplServers().find((server) => server.id === failed.discord.id)?.connected, true); + } finally { + await framework?.stop(); + rmSync(dir, { recursive: true, force: true }); + } +}); + +test('a runtime connect whose policy registration throws and whose close then fails keeps that connection registered and paused through later gates, and logs the close', async (t) => { + mkdirSync(TEST_ROOT, { recursive: true }); + const dir = mkdtempSync(join(TEST_ROOT, 'awareness-runtime-policy-close-')); + const logged: string[] = []; + t.mock.method(console, 'error', (...args: unknown[]) => { logged.push(args.map(String).join(' ')); }); + let framework: AgentFramework | undefined; + + try { + const failed = await failPolicyRegistrationWhileServing(dir, true); + framework = failed.framework; + // A close that fails keeps the entry, so a later disconnect, restart or + // shutdown can close it again. Closed for good, it readies nothing: not + // at this connect's completion, and not at a later gate's. + const refused = failed.registry.getServer(failed.discord.id) as { dataPlaneReady: boolean; willReconnect: boolean } | null; + assert.ok(refused, 'the connection whose close failed is still registered'); + assert.equal(refused.willReconnect, false, 'its close had disarmed it before the transport close failed'); + assert.equal(refused.dataPlaneReady, false, 'the gate\'s completion left the refused connection paused'); + await failed.framework.syncDiscordAwarenessMarkers(); + assert.equal(refused.dataPlaneReady, false, 'a later gate\'s completion leaves it paused too'); + assert.ok( + logged.some((line) => line.includes(`could not close MCPL server "${failed.discord.id}" after its policy registration failed`) + && line.includes('injected close failure')), + 'the failed close was logged', + ); + } finally { + await framework?.stop(); + rmSync(dir, { recursive: true, force: true }); + } +}); + test('runtime reconciliation failure recycles the connection and keeps fresh data gated', async () => { mkdirSync(TEST_ROOT, { recursive: true }); const dir = mkdtempSync(join(TEST_ROOT, 'awareness-runtime-accounting-')); @@ -817,6 +1073,43 @@ test('a control flush that installs a newer generation cannot be released by the assert.equal(readyCalls, 1); }); +test('while the host stops, a failed gate lets out what a running connection held, and nothing of one closed for good', async () => { + const framework = Object.create(AgentFramework.prototype) as any; + framework.discordAwarenessBarrier = null; + framework.discordAwarenessBarrierGeneration = 0; + framework.beginDiscordAwarenessBarrier = () => ({ requiresBarrier: true, promise: new Promise(() => {}) }); + framework.queue = { isClosed: true }; // stop() has closed the queue + framework.emitTrace = () => {}; + const running = Reflect.construct(McplServerConnection, ['running', null, null]) as McplServerConnection; + const closing = Reflect.construct(McplServerConnection, ['closing', null, null]) as McplServerConnection; + framework.mcplServerRegistry = { getAllServers: () => [running, closing] }; + const delivered: string[] = []; + for (const connection of [running, closing]) { + connection.on('stderr', () => delivered.push(`${connection.id} control`)); + connection.on('held-data', () => delivered.push(`${connection.id} data`)); + } + + // Each connection holds a control event and a data event behind the gate. + const barrier = framework.installMcplDataPlaneGate(); + for (const connection of [running, closing]) { + connection.emit('stderr', 'a line'); + connection.emit('held-data', {}); + } + assert.deepEqual(delivered, []); + + // closeAll has begun closing one of them, and that close is still under + // way while the gate is released and then failed: close() closes the + // connection for good as it starts. + const closed = closing.close(); + framework.releaseMcplDataPlaneGate(barrier); + assert.deepEqual(delivered, ['running control'], 'only the running connection lets out its control traffic'); + await framework.failMcplDataPlaneGate(barrier, 'test sync', new Error('accounting failed')); + assert.equal(framework.discordAwarenessBarrier, null, 'the stopping host released the failed gate'); + assert.deepEqual(delivered, ['running control', 'running data'], + 'the running connection\'s held data reaches its listeners, and the closing one lets out nothing'); + await closed; +}); + test('overlapping reconnect/list-change generations release only the newest barrier', async () => { const reconnectDrain = deferred(); const listChangeDrain = deferred(); diff --git a/test/mcpl-config-diagnostics.test.ts b/test/mcpl-config-diagnostics.test.ts index a49292a87..a820d86dc 100644 --- a/test/mcpl-config-diagnostics.test.ts +++ b/test/mcpl-config-diagnostics.test.ts @@ -18,6 +18,7 @@ import { join } from 'node:path'; import { AgentFramework } from '../src/index.js'; import type { EventResponse, Module, ModuleContext, ToolDefinition, ToolResult, TraceEvent } from '../src/index.js'; import { FeatureSetManager } from '../src/mcpl/feature-set-manager.js'; +import { McplServerRegistry } from '../src/mcpl/server-registry.js'; import { CapabilityGrant, computeGrant, expandAdvertisementShorthand } from '../src/mcpl/capability-grant.js'; import { checkToolPattern } from '../src/mcpl/tool-pattern-check.js'; import type { McplCapabilities, McplServerConfig } from '../src/mcpl/types.js'; @@ -333,7 +334,7 @@ test('FeatureSetManager: each §6.4 disablement is traced beside its console lin assert.equal(quiet.lines.length, 3, 'the console line stays'); }); -test('registerMcplServerFeatures: a refused initial policy is traced, and fallback close closes', async () => { +test('registerMcplServerFeatures: a refused initial policy is traced, and fallback close closes the server and unregisters it', async () => { for (const fallback of ['close', 'mcp-only'] as const) { const fw = Object.create(AgentFramework.prototype) as any; fw.traceListeners = []; @@ -343,6 +344,8 @@ test('registerMcplServerFeatures: a refused initial policy is traced, and fallba fw.onTrace((e: Traced) => traces.push(e)); let closed = 0; let established = 0; + // Its close() emits no 'close', as a server's doesn't once it's down + // between retries, so only a close through the registry unregisters it. const connection = { id: 'srv', capabilities: { version: '0.5', pushEvents: true } as unknown as McplCapabilities, @@ -351,6 +354,9 @@ test('registerMcplServerFeatures: a refused initial policy is traced, and fallba sendFeatureSetsUpdateRequest: async () => ({ accepted: false, fallback, reason: 'policy too narrow' }), close: async () => { closed++; }, }; + const registry = new McplServerRegistry(); + (registry as any).servers.set('srv', connection); + fw.mcplServerRegistry = registry; const capture = captureConsoleError(); try { await fw.registerMcplServerFeatures({ id: 'srv', command: 'unused' } as McplServerConfig, connection); @@ -358,6 +364,7 @@ test('registerMcplServerFeatures: a refused initial policy is traced, and fallba capture.stop(); } assert.equal(closed, fallback === 'close' ? 1 : 0, fallback); + assert.equal(registry.getServer('srv'), fallback === 'close' ? null : connection, `${fallback}: registered afterwards`); assert.equal(established, 0, 'a refusal never establishes the grant'); assert.deepEqual( traces.filter((t) => t.type === 'mcpl:policy-refused').map(withoutTimestamp), @@ -414,7 +421,9 @@ function manifestHarness(receipt: Record) { const config: McplServerConfig = { id: 'srv', command: 'unused', enabledFeatureSets: ['chat', 'bare'] }; fw.mcplServerConfigs = new Map([[config.id, config]]); const connection = new ManifestFake(receipt); - fw.mcplServerRegistry = { getAllServers: () => [connection], getServer: (id: string) => (id === 'srv' ? connection : null) }; + const registry = new McplServerRegistry(); + (registry as any).servers.set('srv', connection); + fw.mcplServerRegistry = registry; // Post-§5.3 state: initial policy done, full grant active. const grant = computeGrant(connection.capabilities!, config, { mcpToolsAdvertised: false }); const quiet = captureConsoleError(); diff --git a/test/mcpl-manifest-changed.test.ts b/test/mcpl-manifest-changed.test.ts index 82e12aec9..eeffc5383 100644 --- a/test/mcpl-manifest-changed.test.ts +++ b/test/mcpl-manifest-changed.test.ts @@ -16,6 +16,7 @@ import { EventEmitter } from 'node:events'; import { AgentFramework } from '../src/framework.js'; import { FeatureSetManager } from '../src/mcpl/feature-set-manager.js'; +import { McplServerRegistry } from '../src/mcpl/server-registry.js'; import { CapabilityGrant, computeGrant, expandAdvertisementShorthand } from '../src/mcpl/capability-grant.js'; import type { McplCapabilities, McplServerConfig } from '../src/mcpl/types.js'; @@ -87,7 +88,9 @@ function makeHarness() { const connection = new ManifestFake(); connection.capabilities = expandAdvertisementShorthand(INITIAL); - fw.mcplServerRegistry = { getAllServers: () => [connection], getServer: (id: string) => (id === 'srv' ? connection : null) }; + const registry = new McplServerRegistry(); + (registry as any).servers.set('srv', connection); + fw.mcplServerRegistry = registry; // Model post-§5.3 state: initial policy done, full grant active. const grant = computeGrant(connection.capabilities, config, { mcpToolsAdvertised: false }); @@ -95,7 +98,7 @@ function makeHarness() { connection.establishGrant(grant); connection.events = []; // discard setup noise - return { fw, connection, traces }; + return { fw, connection, traces, registry }; } const settle = () => new Promise((r) => setImmediate(r)); @@ -176,14 +179,17 @@ test('refusal with fallback mcp-only: empty grant, empty-derived feature state', assert.equal(connection.closed, false); }); -test('refusal with fallback close: the transport is closed', async () => { - const { fw, connection } = makeHarness(); +test('refusal with fallback close: the transport is closed and the server unregistered', async () => { + const { fw, connection, registry } = makeHarness(); connection.answerPolicy = 'refuse-close'; connection.nextManifest = () => INITIAL; await fw.handleManifestChanged(connection, { revision: 'sha256:r2', domains: ['capabilities'] }); assert.equal(connection.closed, true, 'fallback close must be honored (§6.7)'); + // The fake's close() emits no 'close', as a server's doesn't once it's down + // between retries, so only a close through the registry unregisters it. + assert.equal(registry.getServer('srv'), null); }); test('failed fetch: previous manifest and grant stand untouched', async () => { diff --git a/test/shutdown-errors.test.ts b/test/shutdown-errors.test.ts index bfbc2433b..7fe241dcc 100644 --- a/test/shutdown-errors.test.ts +++ b/test/shutdown-errors.test.ts @@ -464,6 +464,20 @@ test('MCPL registry retains failed and pending servers and bounds concurrent att assert.equal(registry.getAllServers().length, 0); }); +test('MCPL registry closeServer leaves an entry that a newer connection holds under the same id', async () => { + const registry = new McplServerRegistry(); + const servers = (registry as unknown as { servers: Map }).servers; + let replacedCloses = 0; + const replaced = { close: async () => { replacedCloses++; } } as unknown as McplServerConnection; + const current = { close: async () => { throw new Error('the newer connection was closed'); } } as unknown as McplServerConnection; + // As when a restart has replaced the connection that a policy fallback or + // a failed connect is closing for good. + servers.set('restarted', current); + await registry.closeServer('restarted', replaced); + assert.equal(replacedCloses, 1, 'the connection asked about is closed'); + assert.equal(registry.getServer('restarted'), current, 'the newer connection keeps its entry'); +}); + test('public MCPL recovery serializes overlapping reconnects and close waits for late cleanup', async t => { t.mock.timers.enable({ apis: ['setTimeout'] });