// Cloud: a session id must keep answering in the MCP protocol's own vocabulary// for the whole time its Durable Object is being torn down — not just in the// first instant.//// `DELETE /mcp` condemns the session object with a durable marker and defers// the real teardown to an immediate alarm; that alarm wipes the object's// storage and then aborts the isolate. e2e/cloud/mcp-protocol.test.ts covers// the calm case (terminate, then ask again, get a 404 reconnect). This scenario// covers the violent middle of it: a client whose requests are ALREADY IN// FLIGHT when the termination lands, which is what a real client with an open// tool loop looks like when a session ends underneath it.//// Every one of those in-flight requests must come back as a well-formed// JSON-RPC error — 404 (this id is dead, reconnect) or 503 (the object is// restarting, retry the same id, here is how long to wait) — and never as a// bare unhandled 500 from a platform failure escaping the request handler.import { expect } from "@effect/vitest";import { Effect } from "effect"; import { scenario } from "../src/scenario";import { Mcp, Target } from "../src/services";import type { Identity } from "../src/target"; const JSON_AND_SSE = "application/json, text/event-stream";const PROTOCOL_VERSION = "2025-03-26"; const INITIALIZE_REQUEST = {  jsonrpc: "2.0" as const,  id: 1,  method: "initialize",  params: {    protocolVersion: PROTOCOL_VERSION,    capabilities: {},    clientInfo: { name: "executor-e2e-destroyed-session-envelope", version: "0.0.1" },  },}; const INITIALIZED_NOTIFICATION = {  jsonrpc: "2.0" as const,  method: "notifications/initialized",}; const TOOLS_LIST_REQUEST = {  jsonrpc: "2.0" as const,  id: 2,  method: "tools/list",  params: {},}; const emailOf = (identity: Identity): string => identity.credentials?.email ?? identity.label; const mcpPost = (  url: string | URL,  init: { readonly bearer: string; readonly sessionId?: string; readonly body: unknown },): Promise<Response> =>  fetch(url, {    method: "POST",    headers: {      accept: JSON_AND_SSE,      "content-type": "application/json",      authorization: `Bearer ${init.bearer}`,      ...(init.sessionId ? { "mcp-session-id": init.sessionId } : {}),    },    body: JSON.stringify(init.body),  }); const openSession = async (mcpUrl: string, bearer: string): Promise<string> => {  const initialize = await mcpPost(mcpUrl, { bearer, body: INITIALIZE_REQUEST });  const sessionId = initialize.headers.get("mcp-session-id");  await initialize.text();  if (initialize.status !== 200 || !sessionId) {    throw new Error(`openSession: initialize failed (${initialize.status})`);  }  const initialized = await mcpPost(mcpUrl, { bearer, sessionId, body: INITIALIZED_NOTIFICATION });  await initialized.text();  if (initialized.status !== 202) {    throw new Error(`openSession: notifications/initialized failed (${initialized.status})`);  }  return sessionId;}; type Probe = {  readonly atMs: number;  readonly status: number;  readonly body: string;  readonly retryAfter: string | null;}; /** One JSON-RPC error shape, or null when the body is not the protocol envelope. */const jsonRpcError = (body: string): { readonly code: number; readonly message: string } | null => {  // oxlint-disable-next-line executor/no-try-catch-or-throw -- test helper: a non-JSON body is itself the signal we assert on  try {    const parsed = JSON.parse(body) as {      readonly jsonrpc?: string;      readonly error?: { readonly code?: number; readonly message?: string };    };    if (parsed.jsonrpc !== "2.0" || typeof parsed.error?.code !== "number") return null;    return { code: parsed.error.code, message: parsed.error.message ?? "" };  } catch {    return null;  }}; /** * The verdict a request gets once the destroy alarm has wiped the session's * storage — i.e. the teardown this scenario is timing itself against is over. * Matched on the protocol answer, not on any server internal. */const isWipedVerdict = (probe: Probe): boolean =>  probe.status === 404 && jsonRpcError(probe.body)?.message === "Session not found"; const describeProbe = (probe: Probe): string =>  `+${probe.atMs}ms ${probe.status} ${probe.body.slice(0, 200)}`; scenario(  "MCP protocol · in-flight requests survive their session's teardown as protocol errors",  { timeout: 180_000 },  Effect.gen(function* () {    const target = yield* Target;    const mcp = yield* Mcp;     // The fatal instant is the isolate abort at the end of the teardown, and a    // session only crosses it once — so cross it many times. A single crossing    // leaks the unhandled 500 only when a `validateMcpSessionOwner` RPC is    // inside the object at exactly that instant, so the scenario has to make    // that likely rather than hope for it: `sessions` crossings, each blanketed    // by `concurrency` requests. At 6 × 8 the pre-fix bug reproduced in only    // about half of runs; the numbers below are the measured point where it    // reproduces every run without pushing the shared auth path into 403s.    const sessions = 12;    // Requests are kept continuously in flight rather than polled on a timer:    // the point is to have work ALREADY inside the session object when the    // termination lands, not to sample the window from outside it.    const concurrency = 24;    // Ceiling, not the plan: the stream normally stops as soon as the teardown    // is OBSERVED to have completed (below). This only bounds a teardown whose    // alarm never lands, so a hung platform fails the scenario instead of    // hanging it.    const streamAfterDeleteMs = 3_000;    // Once the storage wipe is visible the abort has already happened, so the    // window this scenario exists to cover is behind us. Keep going briefly —    // the abort and the wipe are not the same instant — then stop, because    // every further request is pure load on the shared auth path.    const graceAfterWipeMs = 150;    // Enough in-flight requests to be mid-teardown, without racing the DELETE    // itself before the session is fully established.    const warmupMs = 250;     const probes: Probe[] = [];    // Per teardown, so the scenario can PROVE each crossing was straddled    // rather than assume it.    const perSession: Probe[][] = [];     // One identity per teardown, all minted BEFORE any load starts. Two    // reasons, both about keeping this scenario's traffic off the shared auth    // path: a single identity carrying every teardown's requests degrades the    // org-membership lookup into a 403, and signing new users in while the    // probe stream is running fails the sign-in itself. Neither has anything to    // do with what is being tested here.    const bearers: string[] = [];    for (let index = 0; index < sessions; index += 1) {      const identity = yield* target.newIdentity();      bearers.push(yield* mcp.mintBearer(emailOf(identity)));    }     for (const bearer of bearers) {      const sessionId = yield* Effect.promise(() => openSession(target.mcpUrl, bearer));       const thisSession: Probe[] = [];      perSession.push(thisSession);       yield* Effect.promise(async () => {        const startedAt = Date.now();        let hardStopAt = Number.POSITIVE_INFINITY;        // Set the first time the wiped-storage verdict is seen, which is the        // observable end of the teardown — the stream stops on this, not a        // timer.        let stopAfterWipeAt = Number.POSITIVE_INFINITY;         const probeOnce = async (): Promise<void> => {          const at = Date.now() - startedAt;          const response = await mcpPost(target.mcpUrl, {            bearer,            sessionId,            body: TOOLS_LIST_REQUEST,          });          const probe: Probe = {            atMs: at,            status: response.status,            body: await response.text(),            retryAfter: response.headers.get("retry-after"),          };          probes.push(probe);          thisSession.push(probe);          // "Session not found" is the post-wipe verdict: the destroy alarm has          // run, storage is gone, and the object has been aborted. Anything          // after this point is a request against an already-dead id.          if (isWipedVerdict(probe) && stopAfterWipeAt === Number.POSITIVE_INFINITY) {            stopAfterWipeAt = Date.now() + graceAfterWipeMs;          }        };         // One worker replenishes its request the moment the previous one        // settles, so the session object is never idle and the DELETE has to        // land on top of real traffic.        const worker = async (): Promise<void> => {          while (Date.now() < hardStopAt && Date.now() < stopAfterWipeAt) await probeOnce();        };        const workers = Array.from({ length: concurrency }, () => worker());         await new Promise((resolve) => setTimeout(resolve, warmupMs));        // Terminate WITHOUT draining the stream: this is the whole scenario.        const terminate = await fetch(target.mcpUrl, {          method: "DELETE",          headers: { authorization: `Bearer ${bearer}`, "mcp-session-id": sessionId },        });        await terminate.text();        expect(terminate.status, "the client can terminate its session").toBe(200);         hardStopAt = Date.now() + streamAfterDeleteMs;        await Promise.all(workers);      });    }     // What the dying session actually answered, so a reviewer can see the shape    // of the teardown window and not just the verdict.    const bucket = new Map<string, number>();    for (const probe of probes) {      const key = `${probe.status} ${jsonRpcError(probe.body)?.message ?? probe.body.slice(0, 80)}`;      bucket.set(key, (bucket.get(key) ?? 0) + 1);    }    console.info(      `[destroyed-session-envelope] ${probes.length} probes across ${sessions} teardowns: ${[        ...bucket,      ]        .map(([key, count]) => `${count}× ${key}`)        .join(" | ")}`,    );     expect(probes.length, "the stream actually exercised the teardown").toBeGreaterThan(sessions);     const unhandled = probes.filter((probe) => probe.status >= 500 && probe.status !== 503);    expect(      unhandled.map(describeProbe),      "no request on a terminating session produces an unhandled server error",    ).toEqual([]);     // Only rejections are asserted on: a request the session still served    // answers 200 over SSE, which is not a JSON-RPC error body and not what    // this scenario is about.    const malformed = probes.filter(      (probe) => probe.status !== 200 && jsonRpcError(probe.body) === null,    );    expect(      malformed.map(describeProbe),      "every rejection is a JSON-RPC error envelope the client can parse",    ).toEqual([]);     // Coverage precondition, checked per teardown rather than in aggregate: a    // run in which some session's stream finished before its object was torn    // down never entered the window this scenario exists to cover, and its    // green is worth nothing. Requiring EVERY crossing to have been observed is    // what stops the assertions above from passing vacuously.    const notStraddled = perSession      .map((session, index) => ({ index, wiped: session.some(isWipedVerdict) }))      .filter((entry) => !entry.wiped)      .map((entry) => `teardown #${entry.index + 1}`);    expect(      notStraddled,      "every teardown was still under load when its session object was destroyed",    ).toEqual([]);     // The disposition half of the contract, on the one failure this scenario    // can actually produce. Every platform failure reachable here is the    // session's own `ctx.abort` after a DELETE the client itself sent, and that    // id is dead for good — so the answer has to be "reconnect", never    // "restarting, retry". A retryable verdict for a deliberately terminated    // session is worse than the 500 this PR removed: the client is told to keep    // asking, and the loop never ends.    const deadIdCalledRetryable = probes.filter((probe) => probe.status === 503);    expect(      deadIdCalledRetryable.map(describeProbe),      "a session the client terminated is never advertised as retryable",    ).toEqual([]);     // A 503 here means "the platform is mid-reset, come back". It is only    // actionable if the client is told how long to wait, so it must carry    // Retry-After AND a positive delay — a `retry-after: 0`, or a header the    // renderer dropped, is the same dead end as no answer at all.    const badBackoff = probes.filter((probe) => {      if (probe.status !== 503) return false;      const seconds = probe.retryAfter === null ? null : Number(probe.retryAfter);      return seconds === null || !Number.isFinite(seconds) || seconds <= 0;    });    expect(      badBackoff.map(describeProbe),      "every retryable rejection tells the client how long to back off",    ).toEqual([]);     // The other half of the classifier's contract, and the half a teardown    // cannot demonstrate: a DEFINITIVE refusal must stay definitive. If the    // platform-failure path ever widens far enough to swallow ordinary    // rejections, these are the two answers that would silently turn into "the    // session is restarting, retry" and send a client into a loop it can never    // exit.    const liveIdentity = yield* target.newIdentity();    const liveBearer = yield* mcp.mintBearer(emailOf(liveIdentity));    const liveSessionId = yield* Effect.promise(() => openSession(target.mcpUrl, liveBearer));     const strangerIdentity = yield* target.newIdentity();    const strangerBearer = yield* mcp.mintBearer(emailOf(strangerIdentity));     yield* Effect.gen(function* () {      const hijack = yield* Effect.promise(() =>        mcpPost(target.mcpUrl, {          bearer: strangerBearer,          sessionId: liveSessionId,          body: TOOLS_LIST_REQUEST,        }),      );      const hijackBody = yield* Effect.promise(() => hijack.text());      expect(        hijack.status,        "another account's session id is refused outright, not advertised as retryable",      ).toBe(403);      expect(hijack.headers.get("retry-after"), "a refusal carries no backoff advice").toBeNull();      expect(jsonRpcError(hijackBody)?.code, "the refusal is the session-ownership error").toBe(        -32003,      );       const unauthenticated = yield* Effect.promise(() =>        fetch(target.mcpUrl, {          method: "POST",          headers: { accept: JSON_AND_SSE, "content-type": "application/json" },          body: JSON.stringify(TOOLS_LIST_REQUEST),        }),      );      yield* Effect.promise(() => unauthenticated.text());      expect(        unauthenticated.status,        "a missing credential is refused outright, not advertised as retryable",      ).toBe(401);      expect(        unauthenticated.headers.get("retry-after"),        "a refusal carries no backoff advice",      ).toBeNull();    }).pipe(      // Runs even when an assertion above fails, so the extra session never      // outlives the scenario. `tryPromise` so a refused cleanup is a failure      // `ignore` can actually swallow — cleanup must not mask the real verdict.      Effect.ensuring(        Effect.tryPromise(async () => {          const closed = await fetch(target.mcpUrl, {            method: "DELETE",            headers: {              authorization: `Bearer ${liveBearer}`,              "mcp-session-id": liveSessionId,            },          });          await closed.text();        }).pipe(Effect.ignore),      ),    );  }),);