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.