fix(mcp): do not retry cancelled HTTP requests after an authorization refresh

A tool call that was aborted or timed out while waiting for a shared OAuth
refresh was still retried by StreamableHttpTransport once the refresh
finished, so the server executed a call the caller had already given up on.

McpClient now passes a per-request signal to McpTransport.send(), aborted
once the request settles (cancelled, timed out, answered, or closed).
StreamableHttpTransport checks it before each dispatch. The shared refresh
itself is not aborted, so other requests waiting on it still succeed.

fixes #10598

packages/mcp/CHANGELOG.md

@@ -9,6 +9,7 @@
 ### Fixed
 
 - Fixed `authorizeMcp()` exchanging an authorization code whose `iss` parameter could not be checked because no authorization server metadata was discovered. It now throws `OAuthIssuerMismatchError`, whose `expected` is `undefined` in that case (RFC 9207). Reported by [@AizenvoltPrime](https://github.com/AizenvoltPrime)
+- Fixed `StreamableHttpTransport` retrying a request after an authorization refresh even though the caller had already cancelled it or it had timed out, so the server still executed it. `McpTransport.send()` now takes an optional `signal`, which `McpClient` aborts once a request settles ([#10598](https://github.com/earendil-works/pi/issues/10598))
 
 ## [1.1.0] - 2026-10-07
 

packages/mcp/src/client.ts

@@ -64,6 +64,8 @@ interface PendingRequest {
 	timer: ReturnType<typeof setTimeout> | undefined;
 	signal: AbortSignal | undefined;
 	onAbort: () => void;
+	/** Aborted when the request settles, so the transport does not dispatch (or retry) it afterwards. */
+	sendController: AbortController;
 	cancellable: boolean;
 	onProgress: ((progress: ProgressNotification) => void) | undefined;
 	progressToken: JsonRpcId | undefined;
@@ -426,6 +428,7 @@ export class McpClient {
 						method !== "initialize",
 						String(options.signal?.reason ?? "Aborted"),
 					),
+				sendController: new AbortController(),
 				cancellable: method !== "initialize",
 				onProgress: options.onProgress,
 				progressToken,
@@ -434,7 +437,9 @@ export class McpClient {
 			if (progressToken !== undefined) this.progressRequests.set(progressToken, id);
 			options.signal?.addEventListener("abort", entry.onAbort, { once: true });
 			this.armTimeout(id, entry);
-			transport.send(message).catch((error) => this.cancelPending(id, error, false));
+			transport
+				.send(message, { signal: entry.sendController.signal })
+				.catch((error) => this.cancelPending(id, error, false));
 		});
 	}
 
@@ -574,6 +579,7 @@ export class McpClient {
 		if (entry.timer) clearTimeout(entry.timer);
 		if (entry.progressToken !== undefined) this.progressRequests.delete(entry.progressToken);
 		entry.signal?.removeEventListener("abort", entry.onAbort);
+		entry.sendController.abort();
 	}
 
 	private rejectPending(error: unknown): void {

packages/mcp/src/index.ts

@@ -68,4 +68,5 @@ export type {
 	McpTransportCloseListener,
 	McpTransportErrorListener,
 	McpTransportMessageListener,
+	McpTransportSendOptions,
 } from "./transports/transport.ts";

packages/mcp/src/transports/streamable-http.ts

@@ -5,11 +5,17 @@ import {
 	JSON_RPC_ERROR_CODES,
 	type JsonRpcId,
 	type JsonRpcMessage,
+	McpAbortError,
 	McpConnectionClosedError,
 	parseJsonRpcMessage,
 	toError,
 } from "../protocol/jsonrpc.ts";
-import { DEFAULT_MAX_MESSAGE_BYTES, type McpTransport, TransportEvents } from "./transport.ts";
+import {
+	DEFAULT_MAX_MESSAGE_BYTES,
+	type McpTransport,
+	type McpTransportSendOptions,
+	TransportEvents,
+} from "./transport.ts";
 
 const MAX_ERROR_BODY_BYTES = 8 * 1024;
 const ERROR_MESSAGE_BODY_CHARS = 500;
@@ -222,12 +228,16 @@ export class StreamableHttpTransport extends TransportEvents implements McpTrans
 		this.protocolVersion = version;
 	}
 
-	async send(message: JsonRpcMessage): Promise<void> {
+	async send(message: JsonRpcMessage, { signal }: McpTransportSendOptions = {}): Promise<void> {
 		if (!this.started || this.closed) throw new McpConnectionClosedError();
-		const response = await this.authorizedFetch("POST", {
+		const response = await this.authorizedFetch(
-			headers: { accept: "application/json, text/event-stream", "content-type": "application/json" },
+			"POST",
-			body: JSON.stringify(message),
+			{
-		});
+				headers: { accept: "application/json, text/event-stream", "content-type": "application/json" },
+				body: JSON.stringify(message),
+			},
+			signal,
+		);
 		await this.checkResponse(response);
 		this.captureSession(response);
 
@@ -277,14 +287,20 @@ export class StreamableHttpTransport extends TransportEvents implements McpTrans
 	/**
 	 * Fetch with auth headers. A 401 (or a 403 asking for more scope) is handed to the auth provider
 	 * once, and the request is retried with whatever credentials it left behind.
+	 *
+	 * `signal` only stops dispatch: getting a token or the authorization recovery can take a while, and a
+	 * request its sender gave up on in the meantime must not reach the server. It does not abort the shared
+	 * recovery, which other requests may wait for.
 	 */
 	private async authorizedFetch(
 		method: "GET" | "POST",
 		init: { headers: Record<string, string>; body?: string },
+		signal?: AbortSignal,
 	): Promise<Response> {
 		const onUnauthorized = this.options.authProvider?.onUnauthorized?.bind(this.options.authProvider);
 		for (let attempt = 0; ; attempt++) {
 			const { headers, token } = await this.headers(init.headers);
+			if (signal?.aborted) throw new McpAbortError();
 			const response = await this.fetch(this.url, {
 				method,
 				headers,

packages/mcp/src/transports/transport.ts

@@ -6,9 +6,18 @@ export type McpTransportMessageListener = (message: JsonRpcMessage) => void;
 export type McpTransportErrorListener = (error: Error) => void;
 export type McpTransportCloseListener = () => void;
 
+export interface McpTransportSendOptions {
+	/**
+	 * Aborted once the sender no longer waits for a reply (the request was cancelled, timed out, answered,
+	 * or the client closed). A transport must not dispatch the message after that, for example when
+	 * retrying it after an authorization refresh. It does not undo a message already dispatched.
+	 */
+	signal?: AbortSignal;
+}
+
 export interface McpTransport {
 	start(): Promise<void>;
-	send(message: JsonRpcMessage): Promise<void>;
+	send(message: JsonRpcMessage, options?: McpTransportSendOptions): Promise<void>;
 	close(): Promise<void>;
 	onMessage(listener: McpTransportMessageListener): () => void;
 	onError(listener: McpTransportErrorListener): () => void;

packages/mcp/test/streamable-http.test.ts

@@ -342,6 +342,66 @@ describe("StreamableHttpTransport", () => {
 		await client.close();
 	});
 
+	// #10598: a request cancelled or timed out while waiting for a shared refresh must not be retried.
+	it("does not retry requests that were cancelled while authorization was recovering", async () => {
+		let token = "old";
+		const executed: unknown[] = [];
+		const { url } = await startServer(async (request, response, requests) => {
+			if (request.method !== "POST") return protocolHandler(request, response, requests);
+			const message = JSON.parse(await readBody(request)) as Record<string, unknown>;
+			if (message.method === "tools/call" && request.headers.authorization === "Bearer old") {
+				response.writeHead(401, { "www-authenticate": "Bearer" }).end();
+				return;
+			}
+			if (message.method === "tools/call") executed.push((message.params as { name: string }).name);
+			return protocolHandler(request, response, requests, message);
+		});
+		let releaseRefresh = () => {};
+		const refreshGate = new Promise<void>((resolve) => {
+			releaseRefresh = resolve;
+		});
+		let refresh: Promise<void> | undefined;
+		let refreshes = 0;
+		let waiting = 0;
+		let allWaiting = () => {};
+		const waitingGate = new Promise<void>((resolve) => {
+			allWaiting = resolve;
+		});
+		const client = new McpClient({ name: "http-test", version: "1.0.0" });
+		await client.connect(
+			new StreamableHttpTransport({
+				url,
+				openGetStream: false,
+				authProvider: {
+					token: async () => token,
+					onUnauthorized: async () => {
+						if (!refresh) {
+							refreshes++;
+							refresh = refreshGate.then(() => {
+								token = "new";
+							});
+						}
+						if (++waiting === 3) allWaiting();
+						await refresh;
+					},
+				},
+			}),
+		);
+		const controller = new AbortController();
+		const aborted = client.callTool("aborted", {}, { signal: controller.signal });
+		const timedOut = client.callTool("timed-out", {}, { timeoutMs: 200 });
+		const kept = client.callTool("kept");
+		await waitingGate;
+		controller.abort();
+		await expect(aborted).rejects.toMatchObject({ name: "AbortError" });
+		await expect(timedOut).rejects.toMatchObject({ name: "McpTimeoutError" });
+		releaseRefresh();
+		expect(await kept).toEqual({ content: [{ type: "text", text: "hello" }] });
+		expect(executed).toEqual(["kept"]);
+		expect(refreshes).toBe(1);
+		await client.close();
+	});
+
 	// #10565: the auth provider may refresh over the network, which closing must not wait for.
 	it("closes the session with the last request's token, without asking the auth provider", async () => {
 		const { url, requests } = await startServer(protocolHandler);