import { request as httpRequest, createServer, type ClientRequest, type ServerResponse,} from "node:http";import { randomUUID } from "node:crypto";import type { AddressInfo } from "node:net"; import { expect } from "@effect/vitest";import { Effect } from "effect"; import { Client } from "@modelcontextprotocol/sdk/client/index.js";import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; import { scenario } from "../src/scenario";import { Mcp, Target } from "../src/services";import type { Identity } from "../src/target"; const emailOf = (identity: Identity): string => identity.credentials?.email ?? identity.label; const delay = (ms: number): Promise<void> => new Promise((resolve) => setTimeout(resolve, ms)); /** * A minimal HTTP relay that forwards to `origin` and, on `sever()`, destroys * the in-flight POST tools/call socket. This models a real network cut of the * POST tools/call SSE stream that the SDK observes as a stream read error, NOT * a client-initiated fetch abort. The relay itself stays up so the SDK's own * reconnect GET has something to connect back to. * * The cut is armed on a real event — the tools/call POST response headers — so * the trigger is identical whether or not the server emits a priming event. * `severPostMidCall()` waits for that stream to open, then cuts it a bounded * grace later: after any priming event has flushed and been read, but well * before the multi-second tool result. */const startSeverableRelay = ( originUrl: string,): Promise<{ readonly url: (path: string) => string; readonly severPostMidCall: () => Promise<void>; readonly close: () => void;}> => { const origin = new URL(originUrl); // The tools/call POST hop, captured the moment its response headers arrive // (before any body: this is independent of whether the server emits a priming // event, so the trigger is identical on fixed and unfixed servers). The sever // must land AFTER the stream opened but BEFORE the tool result, so it fires a // bounded grace after the headers — long enough for a priming event to have // flushed and been read by the SDK, far short of the multi-second tool call. const SEVER_GRACE_MS = 2_000; let toolCallHop: { readonly upstream: ClientRequest; readonly res: ServerResponse } | null = null; let onToolCallOpen: (() => void) | null = null; const toolCallOpened = new Promise<void>((resolve) => { onToolCallOpen = resolve; }); const server = createServer((req, res) => { // Detect the tools/call POST by its request body (initialize also returns // SSE, so a plain content-type check would arm on the wrong stream). const bodyChunks: Buffer[] = []; let isToolCall = false; req.on("data", (chunk: Buffer) => { bodyChunks.push(chunk); if (!isToolCall && Buffer.concat(bodyChunks).toString("utf8").includes('"tools/call"')) { isToolCall = true; } }); const upstream = httpRequest( { host: origin.hostname, port: Number(origin.port), path: req.url, method: req.method, headers: { ...req.headers, host: origin.host }, }, (upstreamRes) => { res.writeHead(upstreamRes.statusCode ?? 502, upstreamRes.headers); if ( isToolCall && (upstreamRes.headers["content-type"] ?? "").includes("text/event-stream") ) { toolCallHop = { upstream, res }; onToolCallOpen?.(); } upstreamRes.pipe(res); }, ); upstream.on("error", () => { try { res.destroy(); } catch { /* already torn down */ } }); req.pipe(upstream); req.on("aborted", () => { try { upstream.destroy(); } catch { /* already torn down */ } }); }); const severPostMidCall = async (): Promise<void> => { await toolCallOpened; await delay(SEVER_GRACE_MS); const hop = toolCallHop; if (!hop) return; // Destroy BOTH ends so every fetch client (undici and bun) observes a hard // stream error: the client-facing response socket (what the SDK's reader is // attached to) AND the upstream to the origin. for (const destroy of [ () => hop.res.socket?.destroy(), () => hop.res.destroy(), () => hop.upstream.socket?.destroy(), () => hop.upstream.destroy(), ]) { try { destroy(); } catch { /* already gone */ } } }; return new Promise((resolve) => { server.listen(0, "127.0.0.1", () => { const port = (server.address() as AddressInfo).port; resolve({ url: (path: string) => `http://127.0.0.1:${port}${path}`, severPostMidCall, close: () => server.close(), }); }); });}; scenario( "MCP streamable HTTP · a stock SDK client recovers a tool result after its POST stream is severed mid-call", { timeout: 160_000 }, Effect.gen(function* () { const target = yield* Target; const mcp = yield* Mcp; const identity = yield* target.newIdentity(); const bearer = yield* mcp.mintBearer(emailOf(identity)); const mcpPath = new URL(target.mcpUrl).pathname; const origin = new URL(target.mcpUrl).origin; const relay = yield* Effect.promise(() => startSeverableRelay(origin)); // A fetch wrapper for the SDK transport that (1) records the SDK's own // reconnect GET (with its last-event-id) so the test can assert the SDK // actually reconnected, and (2) STRIPS the SDK's abort signal so a severed // relay socket surfaces as a stream read error (the real churn scenario), // not a client-side abort. Auth rides on the transport's requestInit. const reconnectGetEventIds: string[] = []; const drivenFetch: typeof fetch = (input, init) => { const url = typeof input === "string" ? input : input instanceof URL ? input.href : input.url; const method = init?.method ?? (input instanceof Request ? input.method : "GET"); const headers = new Headers( init?.headers ?? (input instanceof Request ? input.headers : undefined), ); const lastEventId = headers.get("last-event-id"); if (method === "GET" && lastEventId) reconnectGetEventIds.push(lastEventId); const { signal: _stripped, ...rest } = init ?? {}; return fetch(url, rest); }; const transport = new StreamableHTTPClientTransport(new URL(relay.url(mcpPath)), { requestInit: { headers: { authorization: `Bearer ${bearer}` } }, fetch: drivenFetch, }); const client = new Client({ name: "executor-e2e-priming-reconnect", version: "0.0.1" }); const result = yield* Effect.promise(async () => { try { await client.connect(transport); const marker = `MARKER_PRIMING_RECONNECT_${randomUUID()}`; const code = [ `const marker = ${JSON.stringify(marker)};`, "await new Promise((resolve) => setTimeout(resolve, 15000));", "return marker;", ].join("\n"); // Arm the sever: the relay cuts the POST tools/call socket a short // grace after the stream opens (see SEVER_GRACE_MS) — after any priming // event has flushed and been read by the SDK, but ~13s before the tool // result. The trigger (stream open) is identical on a fixed and an // unfixed server, so the ONLY variable is whether the client saw a // priming id to reconnect from. void relay.severPostMidCall(); // Explicit request timeout > (call duration + reconnect backoff) so a // successful recover resolves well inside it; the scenario timeout is // the real ceiling. const call = await client.callTool({ name: "execute", arguments: { code } }, undefined, { timeout: 90_000, }); return { marker, call, reconnected: reconnectGetEventIds.length > 0 }; } finally { await client.close().catch(() => undefined); await delay(200); relay.close(); } }); expect( reconnectGetEventIds.length, "the stock SDK issued a reconnect GET carrying a last-event-id (it had a priming id to resume from)", ).toBeGreaterThan(0); expect( JSON.stringify(result.call), "callTool resolves with the completed tool result after the mid-call sever", ).toContain(result.marker); }),);
import {
request as httpRequest,
createServer,
type ClientRequest,
type ServerResponse,
} from "node:http";
import { randomUUID } from "node:crypto";
import type { AddressInfo } from "node:net";
import { expect } from "@effect/vitest";
import { Effect } from "effect";
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
import { scenario } from "../src/scenario";
import { Mcp, Target } from "../src/services";
import type { Identity } from "../src/target";
const emailOf = (identity: Identity): string => identity.credentials?.email ?? identity.label;
const delay = (ms: number): Promise<void> => new Promise((resolve) => setTimeout(resolve, ms));
/**
* A minimal HTTP relay that forwards to `origin` and, on `sever()`, destroys
* the in-flight POST tools/call socket. This models a real network cut of the
* POST tools/call SSE stream that the SDK observes as a stream read error, NOT
* a client-initiated fetch abort. The relay itself stays up so the SDK's own
* reconnect GET has something to connect back to.
*
* The cut is armed on a real event — the tools/call POST response headers — so
* the trigger is identical whether or not the server emits a priming event.
* `severPostMidCall()` waits for that stream to open, then cuts it a bounded
* grace later: after any priming event has flushed and been read, but well
* before the multi-second tool result.
*/
const startSeverableRelay = (
originUrl: string,
): Promise<{
readonly url: (path: string) => string;
readonly severPostMidCall: () => Promise<void>;
readonly close: () => void;
}> => {
const origin = new URL(originUrl);
// The tools/call POST hop, captured the moment its response headers arrive
// (before any body: this is independent of whether the server emits a priming
// event, so the trigger is identical on fixed and unfixed servers). The sever
// must land AFTER the stream opened but BEFORE the tool result, so it fires a
// bounded grace after the headers — long enough for a priming event to have
// flushed and been read by the SDK, far short of the multi-second tool call.
const SEVER_GRACE_MS = 2_000;
let toolCallHop: { readonly upstream: ClientRequest; readonly res: ServerResponse } | null = null;
let onToolCallOpen: (() => void) | null = null;
const toolCallOpened = new Promise<void>((resolve) => {
onToolCallOpen = resolve;
});
const server = createServer((req, res) => {
// Detect the tools/call POST by its request body (initialize also returns
// SSE, so a plain content-type check would arm on the wrong stream).
const bodyChunks: Buffer[] = [];
let isToolCall = false;
req.on("data", (chunk: Buffer) => {
bodyChunks.push(chunk);
if (!isToolCall && Buffer.concat(bodyChunks).toString("utf8").includes('"tools/call"')) {
isToolCall = true;
}
});
const upstream = httpRequest(
{
host: origin.hostname,
port: Number(origin.port),
path: req.url,
method: req.method,
headers: { ...req.headers, host: origin.host },
},
(upstreamRes) => {
res.writeHead(upstreamRes.statusCode ?? 502, upstreamRes.headers);
if (
isToolCall &&
(upstreamRes.headers["content-type"] ?? "").includes("text/event-stream")
) {
toolCallHop = { upstream, res };
onToolCallOpen?.();
}
upstreamRes.pipe(res);
},
);
upstream.on("error", () => {
try {
res.destroy();
} catch {
/* already torn down */
}
});
req.pipe(upstream);
req.on("aborted", () => {
try {
upstream.destroy();
} catch {
/* already torn down */
}
});
});
const severPostMidCall = async (): Promise<void> => {
await toolCallOpened;
await delay(SEVER_GRACE_MS);
const hop = toolCallHop;
if (!hop) return;
// Destroy BOTH ends so every fetch client (undici and bun) observes a hard
// stream error: the client-facing response socket (what the SDK's reader is
// attached to) AND the upstream to the origin.
for (const destroy of [
() => hop.res.socket?.destroy(),
() => hop.res.destroy(),
() => hop.upstream.socket?.destroy(),
() => hop.upstream.destroy(),
]) {
try {
destroy();
} catch {
/* already gone */
}
}
};
return new Promise((resolve) => {
server.listen(0, "127.0.0.1", () => {
const port = (server.address() as AddressInfo).port;
resolve({
url: (path: string) => `http://127.0.0.1:${port}${path}`,
severPostMidCall,
close: () => server.close(),
});
});
});
};
scenario(
"MCP streamable HTTP · a stock SDK client recovers a tool result after its POST stream is severed mid-call",
{ timeout: 160_000 },
Effect.gen(function* () {
const target = yield* Target;
const mcp = yield* Mcp;
const identity = yield* target.newIdentity();
const bearer = yield* mcp.mintBearer(emailOf(identity));
const mcpPath = new URL(target.mcpUrl).pathname;
const origin = new URL(target.mcpUrl).origin;
const relay = yield* Effect.promise(() => startSeverableRelay(origin));
// A fetch wrapper for the SDK transport that (1) records the SDK's own
// reconnect GET (with its last-event-id) so the test can assert the SDK
// actually reconnected, and (2) STRIPS the SDK's abort signal so a severed
// relay socket surfaces as a stream read error (the real churn scenario),
// not a client-side abort. Auth rides on the transport's requestInit.
const reconnectGetEventIds: string[] = [];
const drivenFetch: typeof fetch = (input, init) => {
const url = typeof input === "string" ? input : input instanceof URL ? input.href : input.url;
const method = init?.method ?? (input instanceof Request ? input.method : "GET");
const headers = new Headers(
init?.headers ?? (input instanceof Request ? input.headers : undefined),
);
const lastEventId = headers.get("last-event-id");
if (method === "GET" && lastEventId) reconnectGetEventIds.push(lastEventId);
const { signal: _stripped, ...rest } = init ?? {};
return fetch(url, rest);
};
const transport = new StreamableHTTPClientTransport(new URL(relay.url(mcpPath)), {
requestInit: { headers: { authorization: `Bearer ${bearer}` } },
fetch: drivenFetch,
});
const client = new Client({ name: "executor-e2e-priming-reconnect", version: "0.0.1" });
const result = yield* Effect.promise(async () => {
try {
await client.connect(transport);
const marker = `MARKER_PRIMING_RECONNECT_${randomUUID()}`;
const code = [
`const marker = ${JSON.stringify(marker)};`,
"await new Promise((resolve) => setTimeout(resolve, 15000));",
"return marker;",
].join("\n");
// Arm the sever: the relay cuts the POST tools/call socket a short
// grace after the stream opens (see SEVER_GRACE_MS) — after any priming
// event has flushed and been read by the SDK, but ~13s before the tool
// result. The trigger (stream open) is identical on a fixed and an
// unfixed server, so the ONLY variable is whether the client saw a
// priming id to reconnect from.
void relay.severPostMidCall();
// Explicit request timeout > (call duration + reconnect backoff) so a
// successful recover resolves well inside it; the scenario timeout is
// the real ceiling.
const call = await client.callTool({ name: "execute", arguments: { code } }, undefined, {
timeout: 90_000,
});
return { marker, call, reconnected: reconnectGetEventIds.length > 0 };
} finally {
await client.close().catch(() => undefined);
await delay(200);
relay.close();
}
});
expect(
reconnectGetEventIds.length,
"the stock SDK issued a reconnect GET carrying a last-event-id (it had a priming id to resume from)",
).toBeGreaterThan(0);
expect(
JSON.stringify(result.call),
"callTool resolves with the completed tool result after the mid-call sever",
).toContain(result.marker);
}),
);