// Cloud: regression test for a session-DO startup race.//// 1. Open an MCP session (real SDK client init, capture mcp-session-id),// make a call, close the client.// 2. Idle past the session timeout so the DO's alarm runs disposeIdleRuntime// (runtime torn down, `initialized = false`, transport cleared).// 3. Fire an SSE GET (listen stream) and a POST (tools/list) concurrently on// the same session id.// 4. The session must survive: the concurrent pair and follow-up calls// should succeed.//// Mechanism guarded against: a request carrying a session id runs// validateMcpSessionOwner, which restores via onStart when the runtime is// disposed. The SDK's serve() streaming handler separately wakes the DO// through agent.fetch, and the Agent base also drives onStart. Without// serialization across both entry paths, two onStart calls interleave and the// second server.connect throws "Already connected to a transport", after// which every request on the session fails the same way. The concurrent pair// itself can answer 200/200 (one restore wins the race), so the assertions// cover the follow-up calls, which is where the failure persists.//import { writeFileSync } from "node:fs";import { join } from "node:path"; 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, RunDir, Target } from "../src/services";import type { Identity } from "../src/target";import { configuredMcpSessionTimeoutMs } from "../setup/mcp-session-timeouts"; const emailOf = (identity: Identity): string => identity.credentials?.email ?? identity.label; // Give the DO's idle alarm room to fire disposeIdleRuntime after the timeout.const IDLE_DISPOSE_BUFFER_MS = 2_000;const IDLE_DISPOSE_GAP_MS = configuredMcpSessionTimeoutMs() + IDLE_DISPOSE_BUFFER_MS;const SCENARIO_TIMEOUT_MS = IDLE_DISPOSE_GAP_MS + 120_000; interface Connected { readonly client: Client; readonly transport: StreamableHTTPClientTransport;} const connectClient = async (mcpUrl: string, bearer: string): Promise<Connected> => { const client = new Client({ name: "executor-e2e-brick", version: "0.0.1" }, { capabilities: {} }); const transport = new StreamableHTTPClientTransport(new URL(mcpUrl), { requestInit: { headers: { authorization: `Bearer ${bearer}` } }, }); await client.connect(transport); return { client, transport };}; interface RawResult { readonly status: number; readonly body: string;} /** A raw POST of a JSON-RPC request onto an existing session id. */const rawPost = async ( mcpUrl: string, bearer: string, sessionId: string, message: unknown,): Promise<RawResult> => { const res = await fetch(mcpUrl, { method: "POST", headers: { authorization: `Bearer ${bearer}`, "content-type": "application/json", // The streamable-http transport requires the client accept both. accept: "application/json, text/event-stream", "mcp-session-id": sessionId, "mcp-protocol-version": "2025-06-18", }, body: JSON.stringify(message), }); const body = await res.text().catch(() => ""); return { status: res.status, body };}; /** A raw SSE GET (the "listen" stream) on an existing session id. We only need * the response head — status + start of body — to see whether it 500s. */const rawSseGet = async (mcpUrl: string, bearer: string, sessionId: string): Promise<RawResult> => { const res = await fetch(mcpUrl, { method: "GET", headers: { authorization: `Bearer ${bearer}`, accept: "text/event-stream", "mcp-session-id": sessionId, "mcp-protocol-version": "2025-06-18", }, }); // A 200 opens a long-lived stream; don't hang reading it. A 500 has a short // (usually empty) body we can read fully. let body = ""; if (res.status !== 200) { body = await res.text().catch(() => ""); } else { // Cancel the stream immediately; we only wanted the status. await res.body?.cancel().catch(() => undefined); } return { status: res.status, body };}; const toolsListMessage = { jsonrpc: "2.0" as const, id: "brick-tools-list", method: "tools/list" }; const isBrick = (r: RawResult): boolean => r.status >= 500 || /"code"\s*:\s*-32603/.test(r.body) || /Already connected/i.test(r.body); scenario( "REGRESSION · concurrent SSE GET + POST after idle-dispose keeps the session alive", { timeout: SCENARIO_TIMEOUT_MS }, Effect.gen(function* () { const target = yield* Target; const mcp = yield* Mcp; const runDir = yield* RunDir; const identity = yield* target.newIdentity(); const bearer = yield* mcp.mintBearer(emailOf(identity)); const diagnostics: unknown[] = []; const record = (entry: Record<string, unknown>): void => { diagnostics.push(entry); console.log(JSON.stringify(entry)); // Vitest swallows worker stdout on passing runs; the artifact dir is the // durable copy of what each attempt observed. writeFileSync(join(runDir, "repro-diagnostics.json"), JSON.stringify(diagnostics, null, 2)); }; // 1. Open a real session (SDK init handshake persists the session in the // DO), do a call, close the client. const first = yield* Effect.promise(() => connectClient(target.mcpUrl, bearer)); const sessionId = first.transport.sessionId; expect(sessionId, "the client got a session id").toEqual(expect.any(String)); if (sessionId === undefined) return yield* Effect.die("missing session id"); yield* Effect.promise(() => first.client.callTool({ name: "execute", arguments: { code: "return 1;" } }), ); yield* Effect.promise(() => first.client.close().catch(() => undefined)); // 2. Idle past the timeout so the DO's alarm runs disposeIdleRuntime. yield* Effect.sleep(IDLE_DISPOSE_GAP_MS); // 3. Fire the SSE GET and the POST CONCURRENTLY on the same session id. // Raw fetch gives precise concurrency control (both start before either // resolves) that the SDK client's request queue would serialize away. const [getResult, postResult] = yield* Effect.promise(() => Promise.all([ rawSseGet(target.mcpUrl, bearer, sessionId), rawPost(target.mcpUrl, bearer, sessionId, toolsListMessage), ]), ); record({ event: "repro_concurrent_pair", sessionId, get: getResult, post: postResult, }); const concurrentBricked = isBrick(getResult) || isBrick(postResult); // 4. Prove the session survives: subsequent SEQUENTIAL POSTs also succeed. // This is the same control path as the idle-restore scenario from #1302. // // Observed locally: the concurrent pair frequently returns 200/200 (one // of the two restores wins the race and answers), but the race leaves // the DO's McpServer in the "Already connected to a transport" state, so // the very NEXT request 500s — and every request after that. The // regression gate checks the same two follow-up calls now stay healthy. const followUp = yield* Effect.promise(() => rawPost(target.mcpUrl, bearer, sessionId, { ...toolsListMessage, id: "brick-follow-up", }), ); const followUp2 = yield* Effect.promise(() => rawPost(target.mcpUrl, bearer, sessionId, { ...toolsListMessage, id: "brick-follow-up-2", }), ); record({ event: "repro_follow_up", sessionId, followUp, followUp2 }); const permanentlyBricked = isBrick(followUp) && isBrick(followUp2); expect(getResult.status, "the concurrent SSE GET opens successfully").toBe(200); expect(postResult.status, "the concurrent POST answers successfully").toBe(200); expect(isBrick(getResult), "the concurrent SSE GET does not expose the transport brick").toBe( false, ); expect(isBrick(postResult), "the concurrent POST does not expose the transport brick").toBe( false, ); expect(followUp.status, "the first follow-up POST answers successfully").toBe(200); expect(followUp2.status, "the second follow-up POST answers successfully").toBe(200); expect(isBrick(followUp), "the first follow-up POST does not expose the transport brick").toBe( false, ); expect( isBrick(followUp2), "the second follow-up POST does not expose the transport brick", ).toBe(false); expect( permanentlyBricked, "after the concurrent SSE-GET + POST post-idle, the session remains usable", ).toBe(false); // Diagnostic (not a gate): whether the collision surfaced on the concurrent // pair itself vs. only on the follow-up depends on which restore won the // race. Record it so the artifact shows the timing that occurred this run. record({ event: "repro_variant", concurrentBricked, permanentlyBricked }); }),);
// Cloud: regression test for a session-DO startup race.
//
// 1. Open an MCP session (real SDK client init, capture mcp-session-id),
// make a call, close the client.
// 2. Idle past the session timeout so the DO's alarm runs disposeIdleRuntime
// (runtime torn down, `initialized = false`, transport cleared).
// 3. Fire an SSE GET (listen stream) and a POST (tools/list) concurrently on
// the same session id.
// 4. The session must survive: the concurrent pair and follow-up calls
// should succeed.
//
// Mechanism guarded against: a request carrying a session id runs
// validateMcpSessionOwner, which restores via onStart when the runtime is
// disposed. The SDK's serve() streaming handler separately wakes the DO
// through agent.fetch, and the Agent base also drives onStart. Without
// serialization across both entry paths, two onStart calls interleave and the
// second server.connect throws "Already connected to a transport", after
// which every request on the session fails the same way. The concurrent pair
// itself can answer 200/200 (one restore wins the race), so the assertions
// cover the follow-up calls, which is where the failure persists.
//
import { writeFileSync } from "node:fs";
import { join } from "node:path";
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, RunDir, Target } from "../src/services";
import type { Identity } from "../src/target";
import { configuredMcpSessionTimeoutMs } from "../setup/mcp-session-timeouts";
const emailOf = (identity: Identity): string => identity.credentials?.email ?? identity.label;
// Give the DO's idle alarm room to fire disposeIdleRuntime after the timeout.
const IDLE_DISPOSE_BUFFER_MS = 2_000;
const IDLE_DISPOSE_GAP_MS = configuredMcpSessionTimeoutMs() + IDLE_DISPOSE_BUFFER_MS;
const SCENARIO_TIMEOUT_MS = IDLE_DISPOSE_GAP_MS + 120_000;
interface Connected {
readonly client: Client;
readonly transport: StreamableHTTPClientTransport;
}
const connectClient = async (mcpUrl: string, bearer: string): Promise<Connected> => {
const client = new Client({ name: "executor-e2e-brick", version: "0.0.1" }, { capabilities: {} });
const transport = new StreamableHTTPClientTransport(new URL(mcpUrl), {
requestInit: { headers: { authorization: `Bearer ${bearer}` } },
});
await client.connect(transport);
return { client, transport };
};
interface RawResult {
readonly status: number;
readonly body: string;
}
/** A raw POST of a JSON-RPC request onto an existing session id. */
const rawPost = async (
mcpUrl: string,
bearer: string,
sessionId: string,
message: unknown,
): Promise<RawResult> => {
const res = await fetch(mcpUrl, {
method: "POST",
headers: {
authorization: `Bearer ${bearer}`,
"content-type": "application/json",
// The streamable-http transport requires the client accept both.
accept: "application/json, text/event-stream",
"mcp-session-id": sessionId,
"mcp-protocol-version": "2025-06-18",
},
body: JSON.stringify(message),
});
const body = await res.text().catch(() => "");
return { status: res.status, body };
};
/** A raw SSE GET (the "listen" stream) on an existing session id. We only need
* the response head — status + start of body — to see whether it 500s. */
const rawSseGet = async (mcpUrl: string, bearer: string, sessionId: string): Promise<RawResult> => {
const res = await fetch(mcpUrl, {
method: "GET",
headers: {
authorization: `Bearer ${bearer}`,
accept: "text/event-stream",
"mcp-session-id": sessionId,
"mcp-protocol-version": "2025-06-18",
},
});
// A 200 opens a long-lived stream; don't hang reading it. A 500 has a short
// (usually empty) body we can read fully.
let body = "";
if (res.status !== 200) {
body = await res.text().catch(() => "");
} else {
// Cancel the stream immediately; we only wanted the status.
await res.body?.cancel().catch(() => undefined);
}
return { status: res.status, body };
};
const toolsListMessage = { jsonrpc: "2.0" as const, id: "brick-tools-list", method: "tools/list" };
const isBrick = (r: RawResult): boolean =>
r.status >= 500 || /"code"\s*:\s*-32603/.test(r.body) || /Already connected/i.test(r.body);
scenario(
"REGRESSION · concurrent SSE GET + POST after idle-dispose keeps the session alive",
{ timeout: SCENARIO_TIMEOUT_MS },
Effect.gen(function* () {
const target = yield* Target;
const mcp = yield* Mcp;
const runDir = yield* RunDir;
const identity = yield* target.newIdentity();
const bearer = yield* mcp.mintBearer(emailOf(identity));
const diagnostics: unknown[] = [];
const record = (entry: Record<string, unknown>): void => {
diagnostics.push(entry);
console.log(JSON.stringify(entry));
// Vitest swallows worker stdout on passing runs; the artifact dir is the
// durable copy of what each attempt observed.
writeFileSync(join(runDir, "repro-diagnostics.json"), JSON.stringify(diagnostics, null, 2));
};
// 1. Open a real session (SDK init handshake persists the session in the
// DO), do a call, close the client.
const first = yield* Effect.promise(() => connectClient(target.mcpUrl, bearer));
const sessionId = first.transport.sessionId;
expect(sessionId, "the client got a session id").toEqual(expect.any(String));
if (sessionId === undefined) return yield* Effect.die("missing session id");
yield* Effect.promise(() =>
first.client.callTool({ name: "execute", arguments: { code: "return 1;" } }),
);
yield* Effect.promise(() => first.client.close().catch(() => undefined));
// 2. Idle past the timeout so the DO's alarm runs disposeIdleRuntime.
yield* Effect.sleep(IDLE_DISPOSE_GAP_MS);
// 3. Fire the SSE GET and the POST CONCURRENTLY on the same session id.
// Raw fetch gives precise concurrency control (both start before either
// resolves) that the SDK client's request queue would serialize away.
const [getResult, postResult] = yield* Effect.promise(() =>
Promise.all([
rawSseGet(target.mcpUrl, bearer, sessionId),
rawPost(target.mcpUrl, bearer, sessionId, toolsListMessage),
]),
);
record({
event: "repro_concurrent_pair",
sessionId,
get: getResult,
post: postResult,
});
const concurrentBricked = isBrick(getResult) || isBrick(postResult);
// 4. Prove the session survives: subsequent SEQUENTIAL POSTs also succeed.
// This is the same control path as the idle-restore scenario from #1302.
//
// Observed locally: the concurrent pair frequently returns 200/200 (one
// of the two restores wins the race and answers), but the race leaves
// the DO's McpServer in the "Already connected to a transport" state, so
// the very NEXT request 500s — and every request after that. The
// regression gate checks the same two follow-up calls now stay healthy.
const followUp = yield* Effect.promise(() =>
rawPost(target.mcpUrl, bearer, sessionId, {
...toolsListMessage,
id: "brick-follow-up",
}),
);
const followUp2 = yield* Effect.promise(() =>
rawPost(target.mcpUrl, bearer, sessionId, {
...toolsListMessage,
id: "brick-follow-up-2",
}),
);
record({ event: "repro_follow_up", sessionId, followUp, followUp2 });
const permanentlyBricked = isBrick(followUp) && isBrick(followUp2);
expect(getResult.status, "the concurrent SSE GET opens successfully").toBe(200);
expect(postResult.status, "the concurrent POST answers successfully").toBe(200);
expect(isBrick(getResult), "the concurrent SSE GET does not expose the transport brick").toBe(
false,
);
expect(isBrick(postResult), "the concurrent POST does not expose the transport brick").toBe(
false,
);
expect(followUp.status, "the first follow-up POST answers successfully").toBe(200);
expect(followUp2.status, "the second follow-up POST answers successfully").toBe(200);
expect(isBrick(followUp), "the first follow-up POST does not expose the transport brick").toBe(
false,
);
expect(
isBrick(followUp2),
"the second follow-up POST does not expose the transport brick",
).toBe(false);
expect(
permanentlyBricked,
"after the concurrent SSE-GET + POST post-idle, the session remains usable",
).toBe(false);
// Diagnostic (not a gate): whether the collision surfaced on the concurrent
// pair itself vs. only on the follow-up depends on which restore won the
// race. Record it so the artifact shows the timing that occurred this run.
record({ event: "repro_variant", concurrentBricked, permanentlyBricked });
}),
);