Skip to content

fix(mcp): a request JSON can't write reads not-sent, an HTTP failure is reported once, and an unreaped launch is held first #566

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
nissa-seru merged 1 commit into from
Oct 19, 2026
Merged

fix(mcp): a request JSON can't write reads not-sent, an HTTP failure is reported once, and an unreaped launch is held first #566

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 37 additions & 0 deletions changelog.d/mcp-540-review-follow-ups.fixed.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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.
109 changes: 87 additions & 22 deletions src/mcpl/modern-connection.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,6 +132,12 @@ class NotWrittenError extends Error {
* transport inside the caller's async chain, so the scope reaches `send`. */
const wireScope = new AsyncLocalStorage<WireRecord>();

/** 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
Expand Down Expand Up @@ -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<void> {
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.
Expand All @@ -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);
}
}

Expand Down Expand Up @@ -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);
}
Expand Down Expand Up @@ -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<void> {
Expand Down Expand Up @@ -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;
}
Expand All @@ -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.`,
));
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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;
Expand All @@ -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,
Expand Down Expand Up @@ -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
Expand Down
68 changes: 54 additions & 14 deletions src/mcpl/server-connection.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import {
MODERN_MCP_PROTOCOL_VERSION,
checkServerConfig,
redactServerText,
serverTextRedactor,
} from './protocol-family.js';

import type {
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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<never> {
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.
Expand Down Expand Up @@ -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++;
Expand All @@ -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 };
Expand Down Expand Up @@ -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)}`,
);
}
});
}
Expand Down Expand Up @@ -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);
Expand Down
Loading
Loading