diff --git a/changelog.d/mcpl-buffered-drop-count.fixed.md b/changelog.d/mcpl-buffered-drop-count.fixed.md new file mode 100644 index 000000000..35254640f --- /dev/null +++ b/changelog.d/mcpl-buffered-drop-count.fixed.md @@ -0,0 +1,4 @@ +- The MCPL connection's buffered-event cap counts each pause's drops: the + flush line that ends a pause reports that pause's drops, and the drop log + speaks at each pause's first drop. Previously the count ran for the + connection's whole life. diff --git a/changelog.d/mcpl-manifest-changed-without-params.fixed.md b/changelog.d/mcpl-manifest-changed-without-params.fixed.md new file mode 100644 index 000000000..632e7f4c2 --- /dev/null +++ b/changelog.d/mcpl-manifest-changed-without-params.fixed.md @@ -0,0 +1,7 @@ +- An `mcpl/manifestChanged` notification sent without params is handled like + any other: the fetched manifest is applied, its + `mcpl:manifest-change-receipt` is emitted naming no domains, and the tools + list is refreshed. Previously the handler threw once it had applied the + manifest, so the receipt and the refresh were skipped. The receipt's + `announcedDomains` now holds only the strings of the announcement's + `domains` array. diff --git a/changelog.d/mcpl-websocket-close-reason.fixed.md b/changelog.d/mcpl-websocket-close-reason.fixed.md new file mode 100644 index 000000000..45cc6668b --- /dev/null +++ b/changelog.d/mcpl-websocket-close-reason.fixed.md @@ -0,0 +1,10 @@ +- A WebSocket server's close reason no longer reaches the errors a close + makes. A WebSocket transport that closes during the handshake now fails the + connect with the transport's words and the close code, such as + `(websocket closed, code 4001)`, whether or not the server sent a reason, + and each request pending when the connection drops is rejected with + `reason=websocket closed`. Previously both errors carried the server's text + as it wrote it, into the host's log, the `mcpl:server-connect-failed` trace, + the `mcpl-down` alert and a tool call's result, and the connect error gave + no code. `mcpl:server-closed` already left that text out. A stdio child's + exit reads as before. diff --git a/src/framework.ts b/src/framework.ts index 5b4f3785f..be23c38de 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -530,6 +530,15 @@ interface AdmittedEnvelope { resolved?: { channelId: string } | { error: string; candidates?: string[] }; } +/** The domains an mcpl/manifestChanged announcement names, for its receipt: + * its `domains` strings, or none when its params are missing or malformed. */ +function announcedDomainsOf(announce: unknown): string[] { + const domains = typeof announce === 'object' && announce !== null + ? (announce as { domains?: unknown }).domains + : undefined; + return Array.isArray(domains) ? domains.filter((domain): domain is string => typeof domain === 'string') : []; +} + /** * Whether a stream resumes its turn's request rather than starting one: a * budget restart, or a tool-result-guard retry. Such a start takes in no new @@ -17022,8 +17031,12 @@ export class AgentFramework { */ private async handleManifestChanged( connection: McplServerConnection, - announce: { revision?: string; domains?: string[] }, + announce: unknown, ): Promise { + // The announcement's params are the server's hints, which a notification + // may leave out: read the one the receipt reports here, before anything + // is applied, so no later step can throw on them. + const announcedDomains = announcedDomainsOf(announce); const st = this.manifestRefreshState.get(connection.id) ?? { inFlight: false, queued: false, lastCompletedAt: 0 }; this.manifestRefreshState.set(connection.id, st); @@ -17178,7 +17191,7 @@ export class AgentFramework { type: 'mcpl:manifest-change-receipt', serverId: connection.id, revision: connection.manifestState.lastValidatedRevision, - announcedDomains: announce.domains ?? [], + announcedDomains, impacts, }); this.handleToolsListChanged(connection.id); @@ -17725,7 +17738,7 @@ export class AgentFramework { // §17.3: a manifest-change announcement. Rate-limited re-fetch → diff // → §6.7 consequences → one receipt. The announcement itself carries no // payload and no authority; everything below acts on FETCHED content. - connection.on('manifest-changed', (params: { revision?: string; domains?: string[] }) => { + connection.on('manifest-changed', (params: unknown) => { void this.handleManifestChanged(connection, params).catch((err) => { console.error(`[mcpl] ${connection.id} manifest refresh failed:`, err instanceof Error ? err.message : err); }); diff --git a/src/mcpl/server-connection.ts b/src/mcpl/server-connection.ts index 6c378cdae..fdc4ee4f4 100644 --- a/src/mcpl/server-connection.ts +++ b/src/mcpl/server-connection.ts @@ -95,6 +95,20 @@ export function withStderr(error: unknown): string { return '. The last of its stderr:' + tail.split('\n').map((line) => `\n stderr: ${logDetail(line)}`).join(''); } +/** A transport close's reason, as an error message gives it: the transport's + * own words. A WebSocket peer's close reason is the server's text + * (`reasonFromPeer`), so an error gives the words the transport uses for a + * close without one, as `mcpl:server-closed` leaves that text out. A + * WebSocket close's code, a number, says what the server meant (4001: a + * refused credential), with or without a reason, so each error gives it too: + * a pending request's error has `code=`, and the handshake's adds it for any + * WebSocket close, since a close before the handshake completes reaches no + * 'close' event. A stdio child's exit reads as before. */ +function closeReasonText(info: TransportCloseInfo): string { + if (info.reasonFromPeer) return 'websocket closed'; + return info.reason ?? 'unknown'; +} + /** Default per-request timeout in milliseconds (see McplServerConfig.requestTimeoutMs). * A live-but-wedged server (accepts requests, never answers, never closes its * transport) would otherwise freeze the awaiting agent turn forever — no @@ -334,10 +348,14 @@ export class McplServerConnection extends EventEmitter { this.dataPlaneReady = true; const pending = this.bufferedEvents; this.bufferedEvents = []; + // The cap's drops are counted per pause: this flush ends one, so its line + // reports that pause's drops, and the next pause counts (and logs) its own. + const dropped = this.droppedBufferedEvents; + this.droppedBufferedEvents = 0; if (pending.length > 200) { console.error( `[mcpl] ${this.id}: flushing ${pending.length} buffered event(s) synchronously ` + - `(long pause backlog${this.droppedBufferedEvents > 0 ? `; ${this.droppedBufferedEvents} dropped at cap` : ''})`, + `(long pause backlog${dropped > 0 ? `; ${dropped} dropped at cap` : ''})`, ); } this.flushDepth++; @@ -625,7 +643,8 @@ export class McplServerConnection extends EventEmitter { }; const onClose = (info: TransportCloseInfo) => { cleanup(); - reject(failed(`MCPL server "${config.id}" transport closed before handshake completed (${info.reason ?? 'unknown'})`)); + const code = transport.kind === 'websocket' && typeof info.code === 'number' ? `, code ${info.code}` : ''; + reject(failed(`MCPL server "${config.id}" transport closed before handshake completed (${closeReasonText(info)}${code})`)); }; const onError = (err: Error) => { cleanup(); @@ -1395,7 +1414,7 @@ export class McplServerConnection extends EventEmitter { if (pending.timer) clearTimeout(pending.timer); pending.reject( new Error( - `MCPL server "${this.id}" disconnected unexpectedly (code=${info.code ?? 'n/a'}, signal=${info.signal ?? 'n/a'}, reason=${info.reason ?? 'unknown'}) while awaiting ${pending.method} (id=${id})`, + `MCPL server "${this.id}" disconnected unexpectedly (code=${info.code ?? 'n/a'}, signal=${info.signal ?? 'n/a'}, reason=${closeReasonText(info)}) while awaiting ${pending.method} (id=${id})`, ), ); } diff --git a/test/mcpl-manifest-changed.test.ts b/test/mcpl-manifest-changed.test.ts index a5b80ebe6..82e12aec9 100644 --- a/test/mcpl-manifest-changed.test.ts +++ b/test/mcpl-manifest-changed.test.ts @@ -211,3 +211,39 @@ test('announcements coalesce: rapid-fire N announces cause one immediate fetch', const fetches = connection.events.filter((e) => e.kind === 'fetch').length; assert.equal(fetches, 1, `rapid announcements must coalesce (§17.8), got ${fetches} fetches`); }); + +test('an announcement without params is handled like any other: one receipt naming no domains, then the tools-list refresh', async () => { + // A notification may carry no params. The announcement's fields are hints + // the server sends; the change itself is read from the fetched manifest. + const { fw, connection, traces } = makeHarness(); + let refreshed = 0; + fw.handleToolsListChanged = () => { refreshed++; }; + connection.nextManifest = () => ({ + version: '0.5', + pushEvents: true, + channels: { incoming: true }, + featureSets: { chat: { description: 'chat', uses: ['pushEvents', 'channels.incoming', 'channels.publish'] } }, + }) as unknown as McplCapabilities; + + await fw.handleManifestChanged(connection, undefined); + + assert.ok(!connection.grant.has('channels.publish'), 'the fetched manifest was applied'); + const receipts = traces.filter((t) => t.type === 'mcpl:manifest-change-receipt'); + assert.equal(receipts.length, 1, 'exactly one receipt'); + assert.deepEqual(receipts[0].announcedDomains, []); + assert.equal(refreshed, 1, 'the tools list was refreshed'); +}); + +test("an announcement's domains are reported only as strings: anything else names none", async () => { + const { fw, traces } = makeHarness(); + const domainsOf = async (announce: unknown) => { + fw.manifestRefreshState = new Map(); // each announcement its own refresh, past the floor + const before = traces.length; + await fw.handleManifestChanged(fw.mcplServerRegistry.getServer('srv'), announce); + return traces.slice(before).filter((t) => t.type === 'mcpl:manifest-change-receipt').map((t) => t.announcedDomains); + }; + assert.deepEqual(await domainsOf({ domains: 'capabilities' }), [[]]); + assert.deepEqual(await domainsOf({ domains: ['capabilities', 7, null, 'featureSets'] }), [['capabilities', 'featureSets']]); + assert.deepEqual(await domainsOf('capabilities'), [[]]); + assert.deepEqual(await domainsOf(null), [[]], 'params: null, as a server may send them'); +}); diff --git a/test/mcpl-ws-transport.test.ts b/test/mcpl-ws-transport.test.ts index cbcb3b0e5..8e99befa7 100644 --- a/test/mcpl-ws-transport.test.ts +++ b/test/mcpl-ws-transport.test.ts @@ -312,3 +312,110 @@ test("reconnectAfterFailure() on a live transport reports byHost: true and still wss.close(); } }); + +/** A WebSocket server that closes each socket with `code` and the server's own + * close reason `reason` when it receives `method`, and otherwise answers as + * startMockMcplWsServer does. */ +function startClosingWsServer(method: string, code: number, reason: string) { + const wss = startMockMcplWsServer((ws) => { + ws.on('message', (data) => { + const msg = JSON.parse(String(data)) as { method?: string }; + if (msg.method === method) ws.close(code, reason); + }); + }); + return wss; +} + +// A close reason the server writes, made to forge a second host log line. +const PEER_REASON = 'token expired\n[mcpl] editor: forged line'; + +test("a server's close reason during the handshake stays out of the connect error, which gives the transport's words and the close code", async () => { + const wss = startClosingWsServer('initialize', 4001, PEER_REASON); + await waitListening(wss); + try { + await assert.rejects( + withDeadline(McplServerConnection.connect({ id: 'editor', url: wsUrl(wss) }, HOST_CAPS), 10_000, 'the connect to fail'), + (error: Error) => { + assert.equal(error.message, 'MCPL server "editor" transport closed before handshake completed (websocket closed, code 4001)'); + return true; + }, + ); + } finally { + wss.close(); + } +}); + +test('a server that closes during the handshake with a code and no reason gets the same connect error, the code included', async () => { + // A refused credential can come as 4001 alone: the code is what tells it + // from another close, whether or not the server sent words with it. + const wss = startClosingWsServer('initialize', 4001, ''); + await waitListening(wss); + try { + await assert.rejects( + withDeadline(McplServerConnection.connect({ id: 'editor', url: wsUrl(wss) }, HOST_CAPS), 10_000, 'the connect to fail'), + (error: Error) => { + assert.equal(error.message, 'MCPL server "editor" transport closed before handshake completed (websocket closed, code 4001)'); + return true; + }, + ); + } finally { + wss.close(); + } +}); + +test("a server's close reason stays out of the error a pending request is rejected with", async () => { + const wss = startClosingWsServer('tools/call', 4001, PEER_REASON); + await waitListening(wss); + let conn: McplServerConnection | undefined; + try { + conn = await McplServerConnection.connect({ id: 'editor', url: wsUrl(wss) }, HOST_CAPS); + conn.ready(); + await assert.rejects( + withDeadline(conn.sendToolsCall('echo', { text: 'hello' }), 10_000, 'the call to be rejected'), + (error: Error) => { + assert.match(error.message, /^MCPL server "editor" disconnected unexpectedly \(code=4001, signal=n\/a, reason=websocket closed\) while awaiting tools\/call \(id=\d+\)$/); + return true; + }, + ); + } finally { + await conn?.close(); + wss.close(); + } +}); + +test('the buffered-event cap counts the drops of each pause: a later pause reports its own, and logs its first drop', async () => { + // The second pause runs through a control-only flush and a second pause of + // the data plane before ready() ends it, and its count covers all of it. + const wss = startMockMcplWsServer(); + await waitListening(wss); + const conn = await McplServerConnection.connect({ id: 'editor', url: wsUrl(wss) }, HOST_CAPS); + const lines: string[] = []; + const consoleError = console.error; + console.error = (...args: unknown[]) => { lines.push(args.map(String).join(' ')); }; + try { + conn.ready(); + // A responder-less data-plane event while the data plane is paused is + // buffered, and past the cap (1000) the oldest such event is dropped. + const pauseAndBuffer = (count: number) => { + conn.pauseDataPlane(); + for (let i = 0; i < count; i++) conn.emit('test-data-event', { i }); + }; + pauseAndBuffer(1150); + conn.ready(); + pauseAndBuffer(1050); + // Within one pause, neither a control-only flush (a quiesce's) nor + // pausing the paused data plane again starts the count over. + conn.readyControlPlane(); + conn.pauseDataPlane(); + for (let i = 0; i < 25; i++) conn.emit('test-data-event', { late: i }); + conn.ready(); + } finally { + console.error = consoleError; + await conn.close(); + wss.close(); + } + const flushes = lines.filter((line) => line.includes('buffered event(s) synchronously')); + assert.deepEqual(flushes.map((line) => /; (\d+) dropped at cap\)$/.exec(line)?.[1]), ['150', '75']); + const dropLogs = lines.filter((line) => line.includes('buffered-event cap')); + assert.deepEqual(dropLogs.map((line) => /dropped (\d+) responder-less/.exec(line)?.[1]), ['1', '101', '1']); +});