Serialize MCP session DO startup (#1313)

* Add e2e coverage for concurrent SSE GET + POST reconnect after idle dispose

* Serialize MCP session startup

* Drop repro notes doc

* Neutral wording for the session startup regression test

Author: RhysSullivan

e2e/cloud/repro-transport-brick.test.ts

@@ -0,0 +1,215 @@
+// 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 });
+  }),
+);