From 1502181a71a9e0a546c273e04fe12c38b1f89da9 Mon Sep 17 00:00:00 2001 From: Crispin-2406 Date: Mon, 19 Oct 2026 04:52:23 -0400 Subject: [PATCH] fix(mcp): a request JSON can't write reads not-sent, an HTTP failure is reported once, and an unreaped launch is held first Anarchid's review of #540 found four gaps in the two MCP connection classes (https://github.com/anima-research/agent-framework/pull/540#pullrequestreview-5566557817). #556 holds the two it set aside; the review itself gave R1-1 and R1-2. #556's first: a call whose arguments JSON can't write (a BigInt, a cyclic object) put nothing on the wire, but the modern engine read it as no-response, "whether it took effect is unknown". ObservedTransport.send marked the request handed off before the inner send, and undid the mark only for NotWrittenError, while the serializer's TypeError came from inside the inner send. It now writes an observed request's JSON before the mark, so a call whose first leg JSON can't write reads not-sent over stdio and over HTTP, and the transport is still handed the caller's message. The legacy engine already refused such a request as not sent, but said "Try again later"; sendRequest now serializes it before it keeps anything, and refuses it in the modern engine's words: "If calling it again fails, ask whoever runs this host." #556's second: the SDK's HTTP send reports a failure through the transport's onerror and then throws the same object. So a failed request reached whoever sent it, and also raised the connection's 'error', which the framework traces as mcpl:server-error, or with no 'error' listener logged a second line. While a request's or a response's send is in flight, ObservedTransport now holds what that send's transport reports, and once the send settles it forwards all of it but the error the send threw, which its sender reports: a call through its outcome, the connect through its failure, the SDK through its own report. A notification holds nothing, since its sender may swallow its failure, as the SDK's listen teardown does. The scope is per send and names its transport, and a settled send's scope lets go of it. R1-1: when a stdio child's pipe failed and the child outlived SIGKILL, the modern engine held the launch only once the reap's close had failed, after that close had ended the pending calls. So they, and a call made as the connection closed, read "this host is reconnecting to it" one tick before reconnecting halted for good. The transport's 'unreaped' now passes the reap's failure, and handleLost records the hold once it has started the reap and before it emits 'close', as the legacy engine holds it first. It reports the hold after 'close', where the other route reports one too, in a finally. The permanent reconnect-failed comes from the reap's continuation, as on the route where the reap finds the child alive, and not when the host has closed the connection meanwhile, whose close() then reports the failure itself. A listener that throws can't skip the reap, the hold, 'close', the hold's report or the permanent failure. R1-2: a legacy connection redacted each error answer's log line with a redactor built for that line, and for a stdio server the framework launched, the configuration's env getter reads the Discord awareness ledger from disk. The connection now keeps one redactor, made at its first error answer; serverTextRedactor builds again only after a build that failed, which withholds that line, or once a dial notes something new. Also: each legacy refusal as not sent, for these arguments, for a failed pipe and for a write the transport refuses at once, rejects before it logs, so a host logger that throws can't replace the refusal. holdsUnreapedChild's doc says it may not read true yet while a reap runs. Co-Authored-By: Claude Opus 5.5 --- .../mcp-540-review-follow-ups.fixed.md | 37 +++ src/mcpl/modern-connection.ts | 109 +++++-- src/mcpl/server-connection.ts | 68 ++++- test/mcp-modern-http.test.ts | 45 +++ test/mcp-modern-outcomes.test.ts | 283 +++++++++++++++++- test/mcp-recorded-text.test.ts | 14 +- test/mcpl-agent-words.test.ts | 80 +++++ 7 files changed, 596 insertions(+), 40 deletions(-) create mode 100644 changelog.d/mcp-540-review-follow-ups.fixed.md diff --git a/changelog.d/mcp-540-review-follow-ups.fixed.md b/changelog.d/mcp-540-review-follow-ups.fixed.md new file mode 100644 index 000000000..c39211629 --- /dev/null +++ b/changelog.d/mcp-540-review-follow-ups.fixed.md @@ -0,0 +1,37 @@ +- **A tool call whose arguments JSON can't write is refused as not sent + (`not-sent`), in both engines** (#556). Such arguments, a BigInt or a + cyclic object, can come only from in-process code. The modern engine + reported the call as possibly executed (`no-response`: "whether it took + effect is unknown"), though nothing reached the server. The legacy engine + already refused it, but said to try again later, which fails the same way. + Both now say "If calling it again fails, ask whoever runs this host", and + the host's log line names the serializer's error. +- **A legacy request refused as not sent stays a `not-sent` refusal when the + host's `console.error` throws:** for these arguments, for a stdio pipe + that has failed, and for a write the transport refuses at once. Before, + the call threw the logger's error, or failed with it. The refusal's log + line is then lost. +- **A request that fails over HTTP is reported once** (#556), by whatever + made it: a tool call or a listing to its caller and in the host's own log + line. Its failure no longer also raises the modern connection's `'error'` + event, and with it the framework's `mcpl:server-error` trace. A host that + logs that trace, as connectome-host does in each server's `mcpl-stderr` + log, no longer shows a failed tool call there; its HTTP status and the + server's words stay in the host's own log line. A failed listing still + shows there once, as the framework's `tools/list failed` trace, and so + does a failure the SDK reports itself, such as a list-change + subscription's. A connection with no `'error'` listener of the host's no + longer logs a second line for it. A failed notification is reported as + before, since what sent it may report nothing of it. +- **A stdio MCP server whose pipe fails and whose child outlives SIGKILL no + longer has calls told that the host is reconnecting to it** (from the + review of #540). The modern engine now holds the launch before the calls + the failure ends are answered, as the legacy engine does, so they, and a + call made as the connection closes, read that the host won't reconnect to + it on its own. +- **A legacy stdio server the framework launched no longer has the Discord + awareness ledger read from disk for each JSON-RPC error answer's log + line** (from the review of #540). The connection builds that line's + redaction at its first error answer and keeps it. If the ledger can't be + read when it builds, that line is withheld, and the next answer builds + again. diff --git a/src/mcpl/modern-connection.ts b/src/mcpl/modern-connection.ts index 022b48954..c9f6ccb42 100644 --- a/src/mcpl/modern-connection.ts +++ b/src/mcpl/modern-connection.ts @@ -132,6 +132,12 @@ class NotWrittenError extends Error { * transport inside the caller's async chain, so the scope reaches `send`. */ const wireScope = new AsyncLocalStorage(); +/** A send in flight, its transport, and what that transport reported + * meanwhile, held until the send settles (ObservedTransport.send). A + * scope that names no transport, a notification's or a settled send's, + * holds nothing. */ +const sendScope = new AsyncLocalStorage<{ wire: ObservedTransport | null; held: Error[] }>(); + /** * The boundary between the SDK and the actual transport. It records, for the * logical request in scope, each request of the observed method that it @@ -172,15 +178,35 @@ class ObservedTransport implements Transport { if (record) wireScope.run(record, () => this.onmessage?.(message, extra)); else this.onmessage?.(message, extra); }; - this.inner.onerror = (error) => this.onerror?.(error); + // An error this transport reports while one of its sends is in flight + // waits for that send to settle: the SDK's HTTP send reports a failure + // here and then throws it, and whoever sent it reports what it throws + // (#556). Anything else is reported at once, another transport's report + // made in a send's async context too. + this.inner.onerror = (error) => { + const sending = sendScope.getStore(); + if (sending?.wire === this) sending.held.push(error); + else this.onerror?.(error); + }; this.inner.onclose = () => this.onclose?.(); await this.inner.start(); } async send(message: JSONRPCMessage, options?: TransportSendOptions): Promise { + if ('method' in message && !('id' in message)) { + // A notification's sender can swallow what its send throws (the SDK's + // listen teardown does), so what its transport reports is reported at + // once, whatever send it was made in. + await sendScope.run({ wire: null, held: [] }, () => this.inner.send(message, options)); + return; + } const record = wireScope.getStore(); let undo: (() => void) | undefined; if (record && 'method' in message && 'id' in message && message.method === record.method) { + // Every transport writes a message as JSON, so one JSON can't write (a + // BigInt, a cycle) never leaves this host: it's refused here, before + // it's marked handed off (#556). + JSON.stringify(message); const prior = { handedOff: record.handedOff, currentId: record.currentId, answer: record.answer, error: record.error }; // Handed off from here on: past this point nothing proves it unsent, // unless the transport says it wrote nothing. @@ -194,14 +220,29 @@ class ObservedTransport implements Transport { Object.assign(record, prior); }; } + // A request's or a response's sender reports what its send throws: a + // call through its outcome, the connect through its failure, the SDK + // through its own report. So what the transport reports meanwhile is held + // here, and the failure the send threw isn't reported a second time (#556). + const sending: { wire: ObservedTransport | null; held: Error[] } = { wire: this, held: [] }; + let thrown: unknown; try { - await this.inner.send(message, options); + await sendScope.run(sending, () => this.inner.send(message, options)); } catch (error) { + thrown = error; // Refused before any write: this leg never reached the server, so the // record is as it was before it. Any other throw (an HTTP send can fail // after part of the request went out) leaves it handed off. if (error instanceof NotWrittenError) undo?.(); throw error; + } finally { + // Settled: what was held goes on, but for the failure the send threw, + // and the scope lets go of this transport, which work the send started + // still carries. + const { held } = sending; + sending.wire = null; + sending.held = []; + for (const error of held) if (error !== thrown) this.onerror?.(error); } } @@ -278,8 +319,8 @@ class SpawnerStdioTransport implements Transport { private readonly onStderr: (line: string) => void, /** The child's pipe failed and it outlived SIGKILL too, so the line * stays open, owning it (the line's 'unreaped'): `reason` names the - * pipe's error. */ - private readonly onUnreaped: (reason: string) => void, + * pipe's error, and `failure` is the reap's. */ + private readonly onUnreaped: (reason: string, failure: Error) => void, ) { this.recorded = serverTextRedactor(config); } @@ -313,7 +354,7 @@ class SpawnerStdioTransport implements Transport { line.on('stderr', (text: string) => this.onStderr(text)); line.on('error', (error: Error) => this.onerror?.(error)); line.on('close', () => this.onclose?.()); - line.on('unreaped', () => this.onUnreaped(`its stdio pipe failed with ${line.pipeFailure?.message ?? 'an error'}`)); + line.on('unreaped', (failure: Error) => this.onUnreaped(`its stdio pipe failed with ${line.pipeFailure?.message ?? 'an error'}`, failure)); } async send(message: JSONRPCMessage): Promise { @@ -470,7 +511,8 @@ export class ModernMcpConnection extends EventEmitter { * While it's true the connection owns that child, so whatever registered * it keeps its id held; close() retries the reap. It's settled only once * a reap has finished: while a lost session's reap or a close runs, it - * reads false, so read it once lossReaped() or close() has settled. */ + * may not read true yet, so read it once lossReaped() or close() has + * settled. */ get holdsUnreapedChild(): boolean { return this.unreaped.length > 0; } @@ -483,10 +525,15 @@ export class ModernMcpConnection extends EventEmitter { } /** A failed generation's cleanup couldn't reap its child: keep owning it, - * and halt reconnecting until close() settles it. */ - private holdUnreaped(session: Session, failure: Error): void { + * and halt reconnecting until close() settles it. It's reported as an + * 'error' at once, unless the caller reports it later itself. */ + private holdUnreaped(session: Session, failure: Error, report = true): void { if (this.unreaped.some((held) => held.session === session)) return; this.unreaped.push({ session, failure }); + if (report) this.reportUnreaped(failure); + } + + private reportUnreaped(failure: Error): void { this.emit('error', new Error( `MCP server "${this.id}": ${failure.message}. Not starting another launch while it may still be running; close the connection to retry the reap.`, )); @@ -584,11 +631,11 @@ export class ModernMcpConnection extends EventEmitter { ? new SpawnerStdioTransport( this.config, (line) => { if (this.isCurrent(generation)) this.deliverStderr(line); }, - // A child whose pipe failed and that outlived SIGKILL: a loss, whose - // reap finds the child still running and holds the launch, as for any - // launch it can't reap. During a connect there is no session to lose + // A child whose pipe failed and that outlived SIGKILL: a loss whose + // launch is already known unreaped, so it's held as the loss is + // handled (handleLost). During a connect there is no session to lose // yet: the connect's own cleanup holds it, once the connect fails. - (reason) => { if (this.isCurrent(generation)) this.handleLost(generation, reason); }, + (reason, failure) => { if (this.isCurrent(generation)) this.handleLost(generation, reason, failure); }, ) : new StreamableHTTPClientTransport(new URL(this.config.url!), { authProvider: bearerAuth(this.config) }); const wire = new ObservedTransport(inner); @@ -762,8 +809,10 @@ export class ModernMcpConnection extends EventEmitter { } } - /** The transport went away (child exit, or a closed HTTP client). */ - private handleLost(generation: number, reason: string): void { + /** The transport went away (child exit, or a closed HTTP client). + * `unreaped`: the child's pipe failed and it outlived SIGKILL, so its + * launch is already known unreaped. */ + private handleLost(generation: number, reason: string, unreaped?: Error): void { const session = this.session; if (session?.generation !== generation) return; this.session = null; @@ -775,18 +824,34 @@ export class ModernMcpConnection extends EventEmitter { // reconnect's does, and is a permanent failure, as that one is: no // later attempt will come. The reap starts, and is recorded, before // 'close' is emitted, so a listener that throws can't skip it. + const permanent = () => this.emit('reconnect-failed', { error: `${reason}; ${this.unreapedFailure()}`, attempt: Math.max(1, this.reconnectAttempts), permanent: true }); const done = ModernMcpConnection.closeSession(session).then((failure) => { if (this.losing?.session === session) this.losing = null; + // A close() the host made meanwhile reports the failure itself. + if (unreaped) { + // Held as the loss was handled, whatever the reap found now. + if (!this.closedByHost) permanent(); + return; + } if (failure) { this.holdUnreaped(session, failure); - // A close() the host made meanwhile reports the failure itself. - if (!this.closedByHost) { - this.emit('reconnect-failed', { error: `${reason}; ${this.unreapedFailure()}`, attempt: Math.max(1, this.reconnectAttempts), permanent: true }); - } + if (!this.closedByHost) permanent(); } else if (this.willReconnect) this.scheduleReconnect(); }); this.losing = { session, done }; - this.emit('close', { reason }); + // A launch already known unreaped is held now, before 'close', so the + // calls the reap's close ends, which classify their failure only once + // this returns, and any made meanwhile, read that this host won't + // reconnect, as the legacy engine has it (#540's review, R1-1). It's + // reported after 'close', which is where the other route reports a hold + // too, whatever the 'close' listeners do, so a listener that throws can + // skip neither 'close' nor the report. + if (unreaped) this.holdUnreaped(session, unreaped, false); + try { + this.emit('close', { reason }); + } finally { + if (unreaped) this.reportUnreaped(unreaped); + } } /** The legacy engine's backoff: doubling from the base interval, capped, @@ -1013,9 +1078,9 @@ export class ModernMcpConnection extends EventEmitter { * The rules apply in this order. The call's answer is the server's final * one: an error or a `complete` result. * - nothing handed to the transport: `not-sent`, whatever was raised (a - * pre-dispatch validation, a signal already aborted, no transport, a send - * Connectome's stdio line refused before writing, an HTTP refusal before - * any request went out); + * pre-dispatch validation, a signal already aborted, no transport, a + * message JSON can't write, a send Connectome's stdio line refused + * before writing); * - the most recent leg answered with a JSON-RPC error: `error-response` * with that error's data; * - the most recent leg answered with a complete result: the server diff --git a/src/mcpl/server-connection.ts b/src/mcpl/server-connection.ts index 4b4d39337..6b26cd414 100644 --- a/src/mcpl/server-connection.ts +++ b/src/mcpl/server-connection.ts @@ -21,6 +21,7 @@ import { MODERN_MCP_PROTOCOL_VERSION, checkServerConfig, redactServerText, + serverTextRedactor, } from './protocol-family.js'; import type { @@ -568,6 +569,13 @@ export class McplServerConnection extends EventEmitter { // Reconnect state (adapted from Anarchid/agent-framework@mcpl-module-proto) private config: McplServerConfig | null = null; + /** The redactor of the server's text for `config`, made at its first error + * answer, by when `config` is set, and kept: a framework-launched child's + * `env` is a getter that reads the awareness ledger, so a redactor built + * per answer would read it per answer (#540's review, R1-2). It builds + * again only after a build that failed, which withholds that line, or once + * a dial notes something new (serverTextRedactor). */ + private redactor: ((text: unknown) => string) | null = null; private hostCapabilities: McplHostCapabilities | null = null; private reconnectEnabled = false; private reconnectIntervalMs = 5000; @@ -1921,6 +1929,18 @@ export class McplServerConnection extends EventEmitter { // Private: message sending // ========================================================================== + /** + * A request refused before anything is kept for it, as `error`, with its + * operator's line logged after the refusal: a host logger that throws then + * loses that line, and can't turn the refusal into a throw from the call. + */ + private static refused(error: McplRequestError, line: string): Promise { + return new Promise((_, reject) => { + reject(error); + console.error(line); + }); + } + /** * Send a JSON-RPC request and return a promise that resolves with the result * or rejects with a JSON-RPC error. @@ -1958,14 +1978,16 @@ export class McplServerConnection extends EventEmitter { // pipe's error are the operator's, so they go to the log. const pipeFailure = this.transport?.pipeFailure; if (pipeFailure) { - console.error(`MCPL server "${this.id}": ${method} not sent, since its stdio pipe failed: ${logDetail(pipeFailure.message)}`); - return Promise.reject(new McplRequestError( - `The request wasn't sent: this host can no longer write to the "${this.id}" server, so nothing ran. ` + - (this.willReconnect - ? 'This host is reconnecting to it; try again later.' - : "This host won't reconnect to it on its own. If you need it, ask whoever runs this host."), - 'not-sent', - )); + return McplServerConnection.refused( + new McplRequestError( + `The request wasn't sent: this host can no longer write to the "${this.id}" server, so nothing ran. ` + + (this.willReconnect + ? 'This host is reconnecting to it; try again later.' + : "This host won't reconnect to it on its own. If you need it, ask whoever runs this host."), + 'not-sent', + ), + `MCPL server "${this.id}": ${method} not sent, since its stdio pipe failed: ${logDetail(pipeFailure.message)}`, + ); } const id = this.nextRequestId++; @@ -1975,6 +1997,23 @@ export class McplServerConnection extends EventEmitter { id, params, }; + // A request JSON can't write (a BigInt, a cycle) never leaves this host, + // and calling again fails the same way, so it's refused before anything + // is kept for it, in the modern engine's words (#556). The method, id and + // the serializer's words are the operator's. + let line: string; + try { + line = JSON.stringify(request); + } catch (err) { + return McplServerConnection.refused( + new McplRequestError( + `The request wasn't sent: this host couldn't send it to the "${this.id}" server, so nothing ran. ` + + 'If calling it again fails, ask whoever runs this host.', + 'not-sent', + ), + `MCPL server "${this.id}": ${method} (id=${id}) not sent, since it can't be written as JSON: ${logDetail(err instanceof Error ? err.message : err)}`, + ); + } return new Promise((resolve, reject) => { const pending: PendingRequest = { resolve, reject, method }; @@ -2017,20 +2056,21 @@ export class McplServerConnection extends EventEmitter { } this.pendingRequests.set(id, pending); try { - this.transport!.writeLine(JSON.stringify(request)); + this.transport!.writeLine(line); } catch (err) { // A synchronous write rejection sent no request. Remove its pending // state and watchdog before reporting that it is safe to retry. The - // method, id and the transport's words are the operator's. + // method, id and the transport's words are the operator's, logged + // after the refusal, as refused() logs. this.pendingRequests.delete(id); if (pending.timer) clearTimeout(pending.timer); - console.error( - `MCPL server "${this.id}": writing ${method} (id=${id}) failed: ${logDetail(err instanceof Error ? err.message : err)}`, - ); reject(new McplRequestError( `The request wasn't sent: this host couldn't write it to the "${this.id}" server, so nothing ran. Try again later.`, 'not-sent', )); + console.error( + `MCPL server "${this.id}": writing ${method} (id=${id}) failed: ${logDetail(err instanceof Error ? err.message : err)}`, + ); } }); } @@ -2414,7 +2454,7 @@ export class McplServerConnection extends EventEmitter { )); console.error( `MCPL server "${this.id}" answered ${pending.method} (id=${response.id}) with error ${logValue(response.error.code)}: ` + - logValueRedacted(response.error.message, (text) => redactServerText(text, this.config ?? {})), + logValueRedacted(response.error.message, (this.redactor ??= serverTextRedactor(this.config ?? {}))), ); } else { pending.resolve(response.result); diff --git a/test/mcp-modern-http.test.ts b/test/mcp-modern-http.test.ts index aef71c105..debfc427d 100644 --- a/test/mcp-modern-http.test.ts +++ b/test/mcp-modern-http.test.ts @@ -131,6 +131,51 @@ test('HTTP: a deadline aborts the request stream, and the server sees the cancel assert.ok(events.includes('slow-aborted'), `server events: ${events.join(', ')}`); }); +test('HTTP: an argument JSON cannot write is refused before the wire, and raises no connection error (#556)', async (t) => { + t.mock.method(console, 'error', () => {}); + const { url, seen, server } = await startServer(() => true); + cleanups.push(() => new Promise((resolve) => { server.closeAllConnections?.(); server.close(() => resolve()); })); + const connection = await ModernMcpConnection.connect({ id: 'remote', url, token: 't' }); + cleanups.push(() => connection.close()); + const errors: string[] = []; + connection.on('error', (error: Error) => errors.push(error.message)); + const toolCalls = () => seen.filter((s) => s.method === 'POST' && s.mcpMethod === 'tools/call').length; + const before = toolCalls(); + await assert.rejects(connection.callTool('echo', { text: 'x', n: 1n }), (err: unknown) => { + assert.ok(err instanceof McplRequestError, String(err)); + assert.equal(err.outcome, 'not-sent'); + return true; + }); + assert.equal(toolCalls(), before, 'no tools/call reached the server'); + assert.deepEqual(errors, []); +}); + +test('HTTP: a request the server answers with HTTP 500 is reported once, to its caller and in one log line, not as a connection error (#556)', async (t) => { + const lines: string[] = []; + t.mock.method(console, 'error', (...args: unknown[]) => lines.push(args.map(String).join(' '))); + const { url, server } = await startServer(() => true, (message, res) => { + if (message.method !== 'tools/call') return false; + res.writeHead(500, { 'content-type': 'text/plain' }).end('boom'); + return true; + }); + cleanups.push(() => new Promise((resolve) => { server.closeAllConnections?.(); server.close(() => resolve()); })); + const connection = await ModernMcpConnection.connect({ id: 'remote', url, token: 't' }); + cleanups.push(() => connection.close()); + const failed = (err: unknown) => { + assert.ok(err instanceof McplRequestError, String(err)); + assert.equal(err.outcome, 'no-response'); + return true; + }; + // With no listener of the host's, the connection logs an 'error' itself, + // so a second report would be a second line. + await assert.rejects(connection.callTool('echo', { text: 'x' }), failed); + assert.deepEqual(lines.filter((line) => line.includes('boom')), ['MCP server "remote" tools/call echo failed with HTTP 500: "Error POSTing to endpoint: boom"']); + const errors: string[] = []; + connection.on('error', (error: Error) => errors.push(error.message)); + await assert.rejects(connection.callTool('echo', { text: 'x' }), failed); + assert.deepEqual(errors, []); +}); + test('HTTP: a credential the server keeps refusing fails the connect, naming 401', async () => { const { url, server } = await startServer(() => false); cleanups.push(() => new Promise((resolve) => { server.closeAllConnections?.(); server.close(() => resolve()); })); diff --git a/test/mcp-modern-outcomes.test.ts b/test/mcp-modern-outcomes.test.ts index 2e698c523..f2ab07915 100644 --- a/test/mcp-modern-outcomes.test.ts +++ b/test/mcp-modern-outcomes.test.ts @@ -53,6 +53,11 @@ function hostLog(): string[] { /** Whether one of `lines` matches `pattern`, for an assertion's message too. */ const loggedLine = (lines: string[], pattern: RegExp) => assert.ok(lines.some((line) => pattern.test(line)), `${pattern} in:\n${lines.join('\n')}`); +/** A live connection's transport under the observed boundary, to stand in + * for its sends and its reports. */ +const innerOf = (connection: ModernMcpConnection) => + (connection as unknown as { session: { wire: { inner: { onerror: (error: Error) => void; send: (...args: unknown[]) => Promise } } } }).session.wire.inner; + function scratchLog(): string { const dir = mkdtempSync(join(tmpdir(), 'modern-raw-')); cleanups.push(() => rmSync(dir, { recursive: true, force: true })); @@ -111,6 +116,195 @@ test('an outputSchema the SDK cannot compile rejects before dispatch: not-sent, assert.equal(calls(log, 'op'), 0); }); +test('an argument JSON cannot write is refused before the wire: not-sent, nothing on the wire (#556)', async () => { + const logged = hostLog(); + const log = scratchLog(); + const connection = await connect('', log); + await connection.listTools(); + const cyclic: Record = {}; + cyclic.self = cyclic; + for (const [args, why] of [[{ n: 1n }, /serialize a BigInt/], [cyclic, /circular structure/]] as const) { + await assert.rejects(connection.callTool('op', args), (err: unknown) => { + assert.ok(err instanceof McplRequestError, String(err)); + assert.equal(err.outcome, 'not-sent'); + assert.equal(err.message, 'The request wasn\'t sent: this host couldn\'t send it to the "raw" server, so nothing ran. ' + + 'If calling it again fails, ask whoever runs this host.'); + return true; + }); + loggedLine(logged, new RegExp(`^MCP server "raw" tools/call op not sent: .*${why.source}`)); + } + assert.equal(calls(log, 'op'), 0); +}); + +test("a send's own failure is its call's to report, and anything else its transport reports still reaches 'error' (#556)", async () => { + hostLog(); + const log = scratchLog(); + const connection = await connect('', log); + const errors: string[] = []; + connection.on('error', (error: Error) => errors.push(error.message)); + const inner = innerOf(connection); + const realSend = inner.send.bind(inner); + // Out of band, with no send in flight: reported at once. + inner.onerror(new Error('out of band')); + assert.deepEqual(errors, ['out of band']); + // As the SDK's HTTP send does: the failure reported, then thrown. The call + // reports what its send threw; whatever else came meanwhile is reported + // once the send has settled, and what comes later, as from a stream the + // send opened, at once. + const thrown = new Error('the send failed'); + inner.send = async () => { + setTimeout(() => inner.onerror(new Error('after the send')), 20); + inner.onerror(new Error('beside the send')); + inner.onerror(thrown); + throw thrown; + }; + await assert.rejects(connection.callTool('plain', {}), (err: unknown) => { + assert.ok(err instanceof McplRequestError, String(err)); + assert.equal(err.outcome, 'no-response'); + return true; + }); + assert.deepEqual(errors, ['out of band', 'beside the send']); + await until(() => errors.length === 3, 'the later report'); + assert.deepEqual(errors, ['out of band', 'beside the send', 'after the send']); + // A send that goes out: what its transport reported meanwhile is reported + // once it has. + inner.send = async (...args: unknown[]) => { + inner.onerror(new Error('beside a send that went out')); + return realSend(...args); + }; + assert.deepEqual((await connection.callTool('plain', {})).content, [{ type: 'text', text: 'plain' }]); + assert.deepEqual(errors.slice(3), ['beside a send that went out']); +}); + +/** A live connection's observed boundary, to send through it directly. */ +const observedOf = (connection: ModernMcpConnection) => + (connection as unknown as { session: { wire: { send: (message: unknown) => Promise } } }).session.wire; + +test("a request or a response that isn't a call's leaves what its send threw to its sender too, as the connect and the SDK report it themselves (#556)", async () => { + hostLog(); + const connection = await connect('', scratchLog()); + const errors: string[] = []; + connection.on('error', (error: Error) => errors.push(error.message)); + const inner = innerOf(connection); + // As the SDK's HTTP send does: the failure reported, then thrown. + inner.send = async (message: unknown) => { + const what = (message as { method?: string }).method ?? 'the response'; + inner.onerror(new Error(`beside ${what}`)); + const failed = new Error(`${what} failed`); + inner.onerror(failed); + throw failed; + }; + await assert.rejects(observedOf(connection).send({ jsonrpc: '2.0', id: 'd1', method: 'server/discover', params: {} }), /^Error: server\/discover failed$/); + await assert.rejects(observedOf(connection).send({ jsonrpc: '2.0', id: 7, result: {} }), /^Error: the response failed$/); + assert.deepEqual(errors, ['beside server/discover', 'beside the response']); +}); + +test("a notification's send keeps what its transport reports, its own failure included, at once even during a call's send, since its sender can swallow what it throws (#556)", async () => { + hostLog(); + const connection = await connect('', scratchLog()); + const errors: string[] = []; + connection.on('error', (error: Error) => errors.push(error.message)); + const observed = observedOf(connection); + const inner = innerOf(connection); + const cancel = { jsonrpc: '2.0', method: 'notifications/cancelled', params: { requestId: 1 } }; + let notifications = 0; + let whileInFlight: string[] | undefined; + inner.send = async (message: unknown) => { + if ((message as { id?: unknown }).id === undefined) { + // As the SDK's HTTP send does: the failure reported, then thrown, and + // later a report from work the send started, as from a stream. + const n = ++notifications; + setTimeout(() => inner.onerror(new Error(`after notification ${n}`)), 20); + const failed = new Error(`notification ${n} failed`); + inner.onerror(failed); + throw failed; + } + // A call's send, during which a notification fails, its sender + // swallowing what it throws, as the SDK's listen teardown does. + await observed.send(cancel).catch(() => {}); + await until(() => errors.includes('after notification 2'), "the later report of the call's notification"); + whileInFlight = [...errors]; + throw new Error("the call's send failed"); + }; + await assert.rejects(observed.send(cancel), /^Error: notification 1 failed$/); + await until(() => errors.includes('after notification 1'), "the notification's later report"); + assert.deepEqual(errors, ['notification 1 failed', 'after notification 1']); + await assert.rejects(connection.callTool('plain', {}), (err: unknown) => err instanceof McplRequestError && err.outcome === 'no-response'); + const all = ['notification 1 failed', 'after notification 1', 'notification 2 failed', 'after notification 2']; + assert.deepEqual(whileInFlight, all, "reported while the call's send was still in flight"); + assert.deepEqual(errors, all); +}); + +test('a call\'s arguments go out as the caller wrote them, JSON.rawJSON numbers included (#556)', { skip: typeof (JSON as unknown as { rawJSON?: unknown }).rawJSON !== 'function' && 'JSON.rawJSON needs Node 21 or later' }, async () => { + hostLog(); + const connection = await connect('', scratchLog()); + const inner = innerOf(connection); + const realSend = inner.send.bind(inner); + const written: string[] = []; + inner.send = async (...args: unknown[]) => { + written.push(JSON.stringify(args[0])); + return realSend(...args); + }; + const rawJSON = (JSON as unknown as { rawJSON: (text: string) => unknown }).rawJSON; + assert.deepEqual((await connection.callTool('plain', { id: rawJSON('9007199254740993'), huge: rawJSON('1e400') })).content, [{ type: 'text', text: 'plain' }]); + const call = written.find((line) => line.includes('"tools/call"')); + assert.ok(call?.includes('"arguments":{"id":9007199254740993,"huge":1e400}'), String(call)); +}); + +test("another connection's report, made during this connection's send, is that connection's, at once (#556)", async () => { + hostLog(); + const a = await connect('', scratchLog()); + const b = await connect('', scratchLog()); + const aErrors: string[] = []; + const bErrors: string[] = []; + a.on('error', (error: Error) => aErrors.push(error.message)); + b.on('error', (error: Error) => bErrors.push(error.message)); + const aInner = innerOf(a); + const bInner = innerOf(b); + let whileInFlight: string[] | undefined; + aInner.send = async () => { + bInner.onerror(new Error("b's report")); + whileInFlight = [...bErrors]; + throw new Error("a's send failed"); + }; + await assert.rejects(a.callTool('plain', {})); + assert.deepEqual(whileInFlight, ["b's report"], "reported while a's send was still in flight"); + assert.deepEqual(bErrors, ["b's report"]); + assert.deepEqual(aErrors, []); +}); + +test('each send holds only what its own transport reported during it, so overlapping sends each leave their own failure to their call (#556)', async () => { + hostLog(); + const connection = await connect('', scratchLog()); + const errors: string[] = []; + connection.on('error', (error: Error) => errors.push(error.message)); + const inner = innerOf(connection); + let release!: () => void; + const gate = new Promise((resolve) => { release = resolve; }); + let sends = 0; + inner.send = async () => { + if (++sends === 1) { + inner.onerror(new Error('beside the first')); + await gate; + const first = new Error('the first send failed'); + inner.onerror(first); + throw first; + } + const second = new Error('the second send failed'); + inner.onerror(second); + throw second; + }; + const first = connection.callTool('plain', {}).then(() => 'answered', (error: unknown) => error); + await until(() => sends === 1, 'the first send'); + // The second send starts and ends while the first is still in flight. + await assert.rejects(connection.callTool('plain', {}), (err: unknown) => err instanceof McplRequestError && err.outcome === 'no-response'); + assert.deepEqual(errors, [], "the second's failure is its call's, and the first's report waits for the first"); + release(); + const ended = await first; + assert.ok(ended instanceof McplRequestError && ended.outcome === 'no-response', String(ended)); + assert.deepEqual(errors, ['beside the first']); +}); + test('a result that fails its outputSchema was answered: a plain error, not a server error', async () => { const logged = hostLog(); const log = scratchLog(); @@ -984,9 +1178,13 @@ test("a server that closes its stdin and outlives SIGKILL too is held as a launc realKill('SIGKILL'); await until(() => !alive(line.child.pid), 'the held child killed'); }); + // A call made as the connection closes, in the same turn as the loss. + let duringClose: Promise | undefined; + connection.on('close', () => { duringClose ??= connection.callTool('plain', {}).then(() => 'answered', (error: unknown) => error); }); await connection.callTool('deafen', {}); await until(() => wire(log).some((e) => e.event === 'deaf'), 'the server to close its stdin'); - void connection.callTool('plain', {}).catch(() => { /* its write meets EPIPE */ }); + // Its write meets EPIPE, so it's pending when the child is found unreaped. + const pending = connection.callTool('plain', {}).then(() => 'answered', (error: unknown) => error); await until(() => failures.length > 0, 'the permanent failure'); assert.equal(failures.length, 1); assert.equal(failures[0]!.permanent, true); @@ -994,6 +1192,21 @@ test("a server that closes its stdin and outlives SIGKILL too is held as a launc assert.equal(connection.isConnected, false); assert.equal(connection.willReconnect, false); assert.equal(connection.holdsUnreapedChild, true); + // Neither the call the loss ended nor one made as it closed is told this + // host is reconnecting: the launch is held before either is answered + // (#540's review, R1-1). + const ended = await pending; + assert.ok(ended instanceof McplRequestError, String(ended)); + assert.equal(ended.outcome, 'no-response'); + assert.equal(ended.message, 'The connection to the "raw" server closed before it answered, so whether the request took effect is unknown. ' + + "This host won't reconnect to it on its own. If you need it, ask whoever runs this host."); + const refused = await duringClose; + assert.ok(refused instanceof McplRequestError, String(refused)); + assert.equal(refused.outcome, 'not-sent'); + assert.equal(refused.message, 'The request wasn\'t sent: the connection to the "raw" server is closed, so nothing ran. ' + + "This host won't reconnect to it on its own. If you need it, ask whoever runs this host."); + await connection.lossReaped(); + assert.equal(failures.length, 1, "one permanent failure, not a second from the reap's verdict"); await new Promise((r) => setTimeout(r, 200)); assert.equal(wire(log).filter((e) => e.event === 'start').length, 1, 'no second launch'); // close() retries the reap: refused while the child runs, then done. @@ -1003,3 +1216,71 @@ test("a server that closes its stdin and outlives SIGKILL too is held as a launc await connection.close(); assert.equal(connection.holdsUnreapedChild, false); }); + +/** Hand a live connection the loss of its session, as the line's 'unreaped' + * does once the child outlived SIGKILL, here with a child that's well, so + * the reap ends it. It throws what a listener throws, to the test. */ +const loseUnreaped = (connection: ModernMcpConnection) => { + const internals = connection as unknown as { session: { generation: number }; handleLost: (generation: number, reason: string, unreaped?: Error) => void }; + internals.handleLost(internals.session.generation, 'its stdio pipe failed', new Error('it outlived SIGKILL')); +}; + +test("an unreaped launch's loss with a 'close' listener that throws still reaps the launch, holds it, reports the hold, and reports the permanent failure (#540's review, R1-1)", async () => { + hostLog(); + const connection = await connect('', scratchLog()); + const child = liveLine(connection).child.pid; + const seen: string[] = []; + connection.on('error', () => seen.push('error')); + connection.on('reconnect-failed', (failure: { permanent?: boolean }) => seen.push(failure.permanent ? 'permanent' : 'failed')); + const failing = () => { seen.push('close'); throw new Error('a close listener failed'); }; + connection.on('close', failing); + assert.throws(() => loseUnreaped(connection), /^Error: a close listener failed$/); + connection.off('close', failing); + assert.equal(connection.holdsUnreapedChild, true, 'held before the listener threw'); + await connection.lossReaped(); + assert.equal(alive(child), false, 'the reap ended the child'); + assert.deepEqual(seen, ['close', 'error', 'permanent']); +}); + +test("an unreaped launch's loss with an 'error' listener that throws still reaps the launch, closes the connection, and reports the permanent failure (#540's review, R1-1)", async () => { + hostLog(); + const connection = await connect('', scratchLog()); + const child = liveLine(connection).child.pid; + const seen: string[] = []; + connection.on('close', () => seen.push('close')); + connection.on('reconnect-failed', (failure: { permanent?: boolean }) => seen.push(failure.permanent ? 'permanent' : 'failed')); + const failing = () => { seen.push('error'); throw new Error('an error listener failed'); }; + connection.on('error', failing); + assert.throws(() => loseUnreaped(connection), /^Error: an error listener failed$/); + connection.off('error', failing); + connection.on('error', () => { /* the rest, reported */ }); + assert.equal(connection.holdsUnreapedChild, true); + await connection.lossReaped(); + assert.equal(alive(child), false, 'the reap ended the child'); + assert.deepEqual(seen, ['close', 'error', 'permanent']); +}); + +test("a host that closes the connection as an unreaped launch's loss closes it gets close()'s failure, with no permanent failure beside it (#540's review, R1-1)", async () => { + const log = scratchLog(); + const connection = await connect('', log, { reconnect: true, reconnectIntervalMs: 30, reconnectMaxIntervalMs: 60, requestTimeoutMs: 30_000 }); + connection.on('error', () => { /* the held launch, reported */ }); + const failures: unknown[] = []; + connection.on('reconnect-failed', (failure: unknown) => failures.push(failure)); + const line = liveLine(connection); + const realKill = line.child.kill.bind(line.child); + line.child.kill = () => true; // signals never arrive: the child outlives both bounds + // Before the connection's close(), which then reaps an exited child. + cleanups.push(async () => { + realKill('SIGKILL'); + await until(() => !alive(line.child.pid), 'the held child killed'); + }); + let closing: Promise | undefined; + connection.on('close', () => { closing ??= connection.close().then(() => 'closed', (error: unknown) => error); }); + await connection.callTool('deafen', {}); + await until(() => wire(log).some((e) => e.event === 'deaf'), 'the server to close its stdin'); + void connection.callTool('plain', {}).catch(() => { /* its write meets EPIPE, and the loss ends it */ }); + await until(() => closing !== undefined, 'the close made as the loss closes the connection'); + assert.match(String(await closing), /could not be reaped/); + await connection.lossReaped(); + assert.deepEqual(failures, [], "close()'s rejection is the report"); +}); diff --git a/test/mcp-recorded-text.test.ts b/test/mcp-recorded-text.test.ts index 88800f41b..f2b76c5e1 100644 --- a/test/mcp-recorded-text.test.ts +++ b/test/mcp-recorded-text.test.ts @@ -189,8 +189,9 @@ test("a legacy initial connect that fails while its configuration can't be read } finally { await connection.close(); } }); -test('a real discover-then-tools/list 404 is redacted in the framework error trace', async (t) => { - t.mock.method(console, 'error', () => {}); +test("a real discover-then-tools/list 404 is redacted in the host's log line, and the framework's trace of it carries no credential", async (t) => { + const lines: string[] = []; + t.mock.method(console, 'error', (...args: unknown[]) => lines.push(args.map(String).join(' '))); const dir = mkdtempSync(join(tmpdir(), 'mcp-list-text-')); const traces: Array<{ type: string; error?: string }> = []; const server = echo404Server(); @@ -203,8 +204,15 @@ test('a real discover-then-tools/list 404 is redacted in the framework error tra modules: [{ name: 'watcher', async start(ctx: ModuleContext) { ctx.onTrace(event => traces.push(event as never)); }, async stop() {}, getTools: () => [], async handleToolCall() { return { success: false }; }, async onProcess() { return {}; } }], mcplServers: [{ id: 'echo', url: `http://127.0.0.1:${(server.address() as AddressInfo).port}/api/mcp/s/${secret}/mcp` }], }); - assert.ok(traces.some(e => e.type === 'mcpl:server-error' && e.error?.includes('Cannot POST /api/mcp/s/redacted/mcp')), JSON.stringify(traces)); + // The failed request is reported once: its status and the server's words + // in the host's log line, redacted, and the listing's failure in the + // framework's trace, in the agent's words (#556). + assert.ok(lines.some(line => /^MCP server "echo" tools\/list failed with HTTP 404: .*Cannot POST \/api\/mcp\/s\/redacted\/mcp/.test(line)), lines.join('\n')); + assert.deepEqual(traces.filter(e => e.type === 'mcpl:server-error').map(e => e.error), [ + 'tools/list failed: The request to the "echo" server failed before an answer came back. Listing changes nothing on the server, so it\'s safe to try again.', + ]); assert.ok(!JSON.stringify(traces).includes(secret)); + assert.ok(!lines.join('\n').includes(secret)); } finally { await framework?.stop(); server.closeAllConnections(); diff --git a/test/mcpl-agent-words.test.ts b/test/mcpl-agent-words.test.ts index 7118280bf..cc160ccf9 100644 --- a/test/mcpl-agent-words.test.ts +++ b/test/mcpl-agent-words.test.ts @@ -25,6 +25,7 @@ import { import { convertBlock } from '../src/mcpl/convert-block.js'; import { toolResultDataToHistoryString } from '../src/tool-result-history.js'; import { McplServerConnection } from '../src/mcpl/server-connection.js'; +import { McplTransport } from '../src/mcpl/transport.js'; import { WITHHELD_TEXT } from '../src/mcpl/protocol-family.js'; import { logValue } from '../src/mcpl/log-value.js'; import type { McplHostCapabilities, McplServerConfig } from '../src/mcpl/types.js'; @@ -179,6 +180,85 @@ test("an error answer whose configuration can't be read settles its call, and it assert.ok(lines.some((line) => /^MCPL server "words" answered tools\/call \(id=\d+\) with error -32000: /.test(line) && line.endsWith(logValue(WITHHELD_TEXT))), lines.join('\n')); }); +test("a request JSON can't write is refused before anything is kept for it: not sent, in the modern engine's words (#556)", async (t) => { + const errors = t.mock.method(console, 'error', () => {}); + const connection = await McplServerConnection.connect(config(), HOST_CAPS); + t.after(() => connection.close()); + connection.ready(); + const cyclic: Record = {}; + cyclic.self = cyclic; + for (const args of [{ n: 1n }, cyclic]) { + await assert.rejects(connection.sendToolsCall('refuse', args), { + outcome: 'not-sent', + message: 'The request wasn\'t sent: this host couldn\'t send it to the "words" server, so nothing ran. If calling it again fails, ask whoever runs this host.', + }); + } + assert.equal((connection as unknown as { pendingRequests: Map }).pendingRequests.size, 0, 'nothing kept pending'); + assert.ok(logged(errors.mock.calls, /^MCPL server "words": tools\/call \(id=\d+\) not sent, since it can't be written as JSON: "Do not know how to serialize a BigInt"$/), + errors.mock.calls.map((call) => String(call.arguments[0])).join('\n')); +}); + +test("a request refused as not sent is refused through its promise when the host's logger throws (#556)", async (t) => { + t.mock.method(console, 'error', () => { throw new Error('the logger is down'); }); + class PipeTransport extends McplTransport { + constructor() { super(); } + readonly kind = 'stdio' as const; + failed: Error | undefined; + writeFails: Error | undefined; + override get pipeFailure(): Error | undefined { return this.failed; } + writeLine(): void { if (this.writeFails) throw this.writeFails; } + async close(): Promise {} + } + const transport = new PipeTransport(); + const connection = new (McplServerConnection as any)('words', null, transport) as McplServerConnection; + let refused: Promise | undefined; + // Arguments JSON can't write. + assert.doesNotThrow(() => { refused = connection.sendToolsCall('refuse', { n: 1n }); }); + await assert.rejects(refused!, { outcome: 'not-sent' }); + // A write the transport refuses at once. + transport.writeFails = new Error('the stream is destroyed'); + assert.doesNotThrow(() => { refused = connection.sendToolsCall('refuse', {}); }); + await assert.rejects(refused!, { outcome: 'not-sent' }); + // A stdio pipe that has failed, with nothing else to refuse the write. + transport.writeFails = undefined; + transport.failed = new Error('write EPIPE'); + assert.doesNotThrow(() => { refused = connection.sendToolsCall('refuse', {}); }); + await assert.rejects(refused!, { outcome: 'not-sent' }); +}); + +test("error answers' log lines read their connection's configuration until a read succeeds, and not after (#540's review, R1-2)", async (t) => { + const errors = t.mock.method(console, 'error', () => {}); + // A stdio server's env, as the framework gives it, is a getter, which reads + // the reaction baseline's ledger from disk each time, and throws while the + // ledger can't be read. + let reads = 0; + let unreadable = false; + const counted = config(); + Object.defineProperty(counted, 'env', { + enumerable: true, + get: () => { reads++; if (unreadable) throw new Error('reaction-baseline ledger unreadable'); return {}; }, + }); + const connection = await McplServerConnection.connect(counted, HOST_CAPS); + t.after(() => { unreadable = false; return connection.close(); }); + connection.ready(); + const refused = () => assert.rejects(connection.sendToolsCall('refuse', {}), { message: 'The "words" server answered with an error: Missing Access' }); + const answerLines = () => errors.mock.calls.map((call) => String(call.arguments[0])).filter((line) => / answered tools\/call /.test(line)); + // While the ledger can't be read, each answer's line is withheld, and the + // next answer reads it again. + unreadable = true; + let before = reads; + await refused(); + await refused(); + assert.equal(reads - before, 2); + assert.ok(answerLines().every((line) => line.endsWith(logValue(WITHHELD_TEXT))), answerLines().join('\n')); + // Once a read succeeds, the redaction it built is kept. + unreadable = false; + before = reads; + for (let i = 0; i < 3; i++) await refused(); + assert.equal(reads - before, 1, 'read for the first of these answers only'); + assert.ok(answerLines().at(-1)!.includes('Missing'), answerLines().join('\n')); +}); + test("an error answer settles its call when writing its log line throws", async (t) => { // A host's logger can throw. The call's entry and timer are gone by the // time the line is written, so nothing else would settle it.