// Cloud: MCP sessions driven by the REAL @modelcontextprotocol/sdk Client over// StreamableHTTP — exactly the code path Claude/Cursor run. The dev server is// the production wrangler topology (real workerd, real McpSessionDO), so// session continuity here is real Durable Object state surviving across// client connections, not a stub.//// Ported from apps/cloud/src/mcp-miniflare.e2e.node.test.ts (unstable_dev +// test-seam bearers) onto the e2e dev server with real OAuth bearers.// Telemetry-span assertions from that file required injecting an OTLP// receiver into the worker env and were NOT carried (not black-box). import { expect } from "@effect/vitest";import { Effect } from "effect";import { composePluginApi } from "@executor-js/api/server";import { Client } from "@modelcontextprotocol/sdk/client/index.js";import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; import { scenario } from "../src/scenario";import { Api, Mcp, Target } from "../src/services";import type { Identity } from "../src/target";import {  configuredMcpPausedSessionIdleTimeoutMs,  configuredMcpSessionTimeoutMs,} from "../setup/mcp-session-timeouts"; const coreApi = composePluginApi([] as const); const emailOf = (identity: Identity): string => identity.credentials?.email ?? identity.label; interface Connected {  readonly client: Client;  readonly transport: StreamableHTTPClientTransport;} /** A real SDK client over StreamableHTTP; `sessionId` resumes an existing session. */const connectClient = async (  mcpUrl: string,  bearer: string,  sessionId?: string,): Promise<Connected> => {  const client = new Client(    { name: "executor-e2e-sessions", version: "0.0.1" },    { capabilities: {} },  );  const transport = new StreamableHTTPClientTransport(new URL(mcpUrl), {    requestInit: { headers: { authorization: `Bearer ${bearer}` } },    ...(sessionId === undefined ? {} : { sessionId }),  });  await client.connect(transport);  return { client, transport };}; const textOf = (result: { content?: unknown; toolResult?: unknown }): string =>  ((result.content ?? []) as Array<{ type: string; text?: string }>)    .filter((part) => part.type === "text")    .map((part) => part.text)    .join("\n"); const closeQuietly = (connected: Connected): Effect.Effect<void> =>  Effect.promise(() => connected.client.close().catch(() => undefined)); scenario(  "MCP sessions · a real MCP client connects, lists tools, and executes code",  {},  Effect.gen(function* () {    const target = yield* Target;    const mcp = yield* Mcp;    const identity = yield* target.newIdentity();    const bearer = yield* mcp.mintBearer(emailOf(identity));    const session = yield* Effect.promise(() => connectClient(target.mcpUrl, bearer));    yield* Effect.gen(function* () {      expect(        session.client.getServerVersion()?.name,        "the handshake reports the product server",      ).toBe("executor");      expect(session.transport.sessionId, "the transport holds a session id").toEqual(        expect.any(String),      );       const { tools } = yield* Effect.promise(() => session.client.listTools());      expect(        tools.map((tool) => tool.name),        "the execute tool is advertised",      ).toContain("execute");      expect(        tools.map((tool) => tool.name),        "the resume tool is advertised",      ).toContain("resume");       const result = yield* Effect.promise(() =>        session.client.callTool({ name: "execute", arguments: { code: "return 6 * 7;" } }),      );      expect(result.isError, "the call succeeds").not.toBe(true);      expect(textOf(result), "the sandbox returns the value").toContain("42");    }).pipe(Effect.ensuring(closeQuietly(session)));  }),); scenario(  "MCP sessions · a second client resuming the session id continues the session",  {},  Effect.gen(function* () {    const target = yield* Target;    const mcp = yield* Mcp;    const identity = yield* target.newIdentity();    const bearer = yield* mcp.mintBearer(emailOf(identity));     // First connection establishes the session, does real work, then goes    // away — the laptop-closed / process-restarted case.    const first = yield* Effect.promise(() => connectClient(target.mcpUrl, bearer));    const sessionId = first.transport.sessionId;    expect(sessionId, "the first client got a session id").toEqual(expect.any(String));    const before = yield* Effect.promise(() =>      first.client.callTool({ name: "execute", arguments: { code: 'return "before";' } }),    ).pipe(Effect.ensuring(closeQuietly(first)));    expect(textOf(before), "the first client's call succeeded").toContain("before");     // A brand-new client resumes with nothing but the session id. The    // session's Durable Object state persists across connections — this is    // the restore guarantee.    const second = yield* Effect.promise(() => connectClient(target.mcpUrl, bearer, sessionId));    yield* Effect.gen(function* () {      expect(second.transport.sessionId, "the session id is preserved, not reissued").toBe(        sessionId,      );      const { tools } = yield* Effect.promise(() => second.client.listTools());      expect(        tools.map((tool) => tool.name),        "the resumed session serves requests",      ).toContain("execute");      const after = yield* Effect.promise(() =>        second.client.callTool({ name: "execute", arguments: { code: 'return "after";' } }),      );      expect(after.isError, "the resumed session executes code").not.toBe(true);      expect(textOf(after), "the resumed session returns results").toContain("after");    }).pipe(Effect.ensuring(closeQuietly(second)));  }),); const IDLE_RUNTIME_DISPOSE_BUFFER_MS = 2_000;const IDLE_RUNTIME_DISPOSE_GAP_MS =  configuredMcpSessionTimeoutMs() + IDLE_RUNTIME_DISPOSE_BUFFER_MS;const IDLE_RUNTIME_DISPOSE_SCENARIO_TIMEOUT_MS = IDLE_RUNTIME_DISPOSE_GAP_MS + 120_000; scenario(  "MCP sessions · a resumed session after idle runtime disposal restores cleanly",  { timeout: IDLE_RUNTIME_DISPOSE_SCENARIO_TIMEOUT_MS },  Effect.gen(function* () {    const target = yield* Target;    const mcp = yield* Mcp;    const identity = yield* target.newIdentity();    const bearer = yield* mcp.mintBearer(emailOf(identity));     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");     const before = yield* Effect.promise(() =>      first.client.callTool({ name: "execute", arguments: { code: 'return "before-idle";' } }),    ).pipe(Effect.ensuring(closeQuietly(first)));    expect(before.isError, "the pre-idle call succeeds").not.toBe(true);    expect(textOf(before), "the pre-idle call returns results").toContain("before-idle");     yield* Effect.sleep(IDLE_RUNTIME_DISPOSE_GAP_MS);     const second = yield* Effect.promise(() => connectClient(target.mcpUrl, bearer, sessionId));    yield* Effect.gen(function* () {      expect(second.transport.sessionId, "the session id is preserved").toBe(sessionId);      const after = yield* Effect.promise(() =>        second.client.callTool({ name: "execute", arguments: { code: 'return "after-idle";' } }),      );      expect(after.isError, "the post-idle call succeeds").not.toBe(true);      expect(textOf(after), "the post-idle call returns results").toContain("after-idle");    }).pipe(Effect.ensuring(closeQuietly(second)));  }),); scenario(  "MCP sessions · an unknown session id fails fast with a clean error, not a hang",  {},  Effect.gen(function* () {    const target = yield* Target;    const mcp = yield* Mcp;    const identity = yield* target.newIdentity();    const bearer = yield* mcp.mintBearer(emailOf(identity));    // A session id the server never issued (right shape, never minted).    const session = yield* Effect.promise(() =>      connectClient(target.mcpUrl, bearer, "0".repeat(64)),    );    const failure = yield* Effect.flip(      Effect.tryPromise({        try: () => session.client.listTools(),        catch: (cause) => cause,      }),    ).pipe(      Effect.timeoutOrElse({        duration: "15 seconds",        orElse: () =>          Effect.die(new Error("listTools on an unknown session hung instead of failing")),      }),      Effect.ensuring(closeQuietly(session)),    );    expect(String(failure), "the server answered with a JSON-RPC error envelope").toContain(      "jsonrpc",    );  }),); scenario(  "MCP sessions · two concurrent clients hold isolated sessions that don't interfere",  {},  Effect.gen(function* () {    const target = yield* Target;    const mcp = yield* Mcp;    const identity = yield* target.newIdentity();    const bearer = yield* mcp.mintBearer(emailOf(identity));    const [alpha, beta] = yield* Effect.promise(() =>      Promise.all([connectClient(target.mcpUrl, bearer), connectClient(target.mcpUrl, bearer)]),    );    yield* Effect.gen(function* () {      expect(alpha.transport.sessionId, "each client gets its own session").not.toBe(        beta.transport.sessionId,      );       const [alphaResult, betaResult] = yield* Effect.promise(() =>        Promise.all([          alpha.client.callTool({            name: "execute",            arguments: {              code: 'await new Promise((resolve) => setTimeout(resolve, 300));\nreturn "alpha-result";',            },          }),          beta.client.callTool({            name: "execute",            arguments: { code: 'return "beta-result";' },          }),        ]),      );      expect(textOf(alphaResult), "the first session got its own answer").toContain("alpha-result");      expect(textOf(alphaResult), "no cross-talk into the first session").not.toContain(        "beta-result",      );      expect(textOf(betaResult), "the second session got its own answer").toContain("beta-result");      expect(textOf(betaResult), "no cross-talk into the second session").not.toContain(        "alpha-result",      );    }).pipe(Effect.ensuring(closeQuietly(alpha)), Effect.ensuring(closeQuietly(beta)));  }),); const APPROVAL_TARGET_TOOL = "executor.coreTools.policies.list"; const GATED_CODE = `const result = await tools.executor.coreTools.policies.list({});return JSON.stringify(result);`; const executionIdOf = (text: string): string | undefined =>  /\bexecutionId:\s*(\S+)/.exec(text)?.[1]; const IDLE_TEARDOWN_BUFFER_MS = 2_000;const IDLE_TEARDOWN_GAP_MS = configuredMcpPausedSessionIdleTimeoutMs() + IDLE_TEARDOWN_BUFFER_MS;const IDLE_TEARDOWN_SCENARIO_TIMEOUT_MS = IDLE_TEARDOWN_GAP_MS + 120_000; // The session DO tears its runtime down after the configured idle ceiling and// rebuilds it from storage on the next request. The e2e harness shortens that// ceiling when it owns the cloud dev stack. Paused approvals deliberately do// not survive this (durable pause state is out of scope); the contract is that// an expired pause fails with recovery guidance and the session keeps working. scenario(  "MCP sessions · an approval paused past the idle window expires with re-run guidance, not a dead end",  { timeout: IDLE_TEARDOWN_SCENARIO_TIMEOUT_MS },  Effect.gen(function* () {    const target = yield* Target;    const { client } = yield* Api;    const mcp = yield* Mcp;    const identity = yield* target.newIdentity();    const api = yield* client(coreApi, identity);    const bearer = yield* mcp.mintBearer(emailOf(identity));     const policy = yield* api.policies.create({      payload: { owner: "org", pattern: APPROVAL_TARGET_TOOL, action: "require_approval" },    });     yield* Effect.gen(function* () {      const first = yield* Effect.promise(() => connectClient(target.mcpUrl, bearer));      const sessionId = first.transport.sessionId;      expect(sessionId, "the first client got a session id").toEqual(expect.any(String));      if (sessionId === undefined) return yield* Effect.die("missing session id");      const paused = yield* Effect.promise(() =>        first.client.callTool({ name: "execute", arguments: { code: GATED_CODE } }),      ).pipe(Effect.ensuring(closeQuietly(first)));      const pausedText = textOf(paused);      expect(pausedText, "the gated call pauses instead of completing").toContain(        "Execution paused",      );      const executionId = executionIdOf(pausedText);      expect(executionId, "the paused result carries the executionId").toEqual(expect.any(String));      if (executionId === undefined) return yield* Effect.die("missing paused execution id");       // The user thinks the approval over for longer than the session keeps      // its runtime warm; the pause is gone when they come back.      yield* Effect.sleep(IDLE_TEARDOWN_GAP_MS);       const second = yield* Effect.promise(() => connectClient(target.mcpUrl, bearer, sessionId));      yield* Effect.gen(function* () {        const resumed = yield* Effect.promise(() =>          second.client.callTool({            name: "resume",            arguments: { executionId, action: "accept", content: "{}" },          }),        );        const resumedText = textOf(resumed);        expect(          resumedText,          "the error tells the model how to recover, not just that the pause is gone",        ).toContain("run the execute tool again");        expect(resumed.isError, "the expired resume is an error, not a silent success").toBe(true);         // The advertised recovery actually works on the same session: a fresh        // execute pauses with a NEW id (stale ids are never reused), and        // resuming that completes the gated call.        const reExecuted = yield* Effect.promise(() =>          second.client.callTool({ name: "execute", arguments: { code: GATED_CODE } }),        );        const reExecutedText = textOf(reExecuted);        expect(reExecutedText, "the re-run pauses for approval again").toContain(          "Execution paused",        );        const freshExecutionId = executionIdOf(reExecutedText);        expect(freshExecutionId, "the re-run mints a different executionId").not.toBe(executionId);        if (freshExecutionId === undefined) return yield* Effect.die("missing fresh execution id");         const resumedFresh = yield* Effect.promise(() =>          second.client.callTool({            name: "resume",            arguments: { executionId: freshExecutionId, action: "accept", content: "{}" },          }),        );        expect(resumedFresh.isError, "the fresh approval resumes to completion").not.toBe(true);        expect(textOf(resumedFresh), "the gated tool's result comes back after approval").toContain(          APPROVAL_TARGET_TOOL,        );      }).pipe(Effect.ensuring(closeQuietly(second)));    }).pipe(      Effect.ensuring(        api.policies          .remove({ params: { policyId: policy.id }, payload: { owner: "org" } })          .pipe(Effect.ignore),      ),    );  }),); scenario(  "MCP sessions · a duplicate resume replays the outcome instead of losing the approval",  {},  Effect.gen(function* () {    const target = yield* Target;    const { client } = yield* Api;    const mcp = yield* Mcp;    const identity = yield* target.newIdentity();    const api = yield* client(coreApi, identity);    const bearer = yield* mcp.mintBearer(emailOf(identity));     const policy = yield* api.policies.create({      payload: { owner: "org", pattern: APPROVAL_TARGET_TOOL, action: "require_approval" },    });     yield* Effect.gen(function* () {      const session = yield* Effect.promise(() => connectClient(target.mcpUrl, bearer));      yield* Effect.gen(function* () {        const paused = yield* Effect.promise(() =>          session.client.callTool({ name: "execute", arguments: { code: GATED_CODE } }),        );        const executionId = executionIdOf(textOf(paused));        expect(executionId, "the paused result carries the executionId").toEqual(          expect.any(String),        );         const resume = () =>          Effect.promise(() =>            session.client.callTool({              name: "resume",              arguments: { executionId: executionId ?? "", action: "accept", content: "{}" },            }),          );         const first = yield* resume();        expect(first.isError, "the first resume completes the execution").not.toBe(true);         // MCP clients retry resume when a response is lost in transit. The        // retry must replay the same completed outcome — the production        // failure mode was "No paused execution" seconds after a successful        // resume.        const retry = yield* resume();        expect(retry.isError, "the duplicate resume is not an error").not.toBe(true);        expect(textOf(retry), "the duplicate resume returns the same completed result").toContain(          APPROVAL_TARGET_TOOL,        );      }).pipe(Effect.ensuring(closeQuietly(session)));    }).pipe(      Effect.ensuring(        api.policies          .remove({ params: { policyId: policy.id }, payload: { owner: "org" } })          .pipe(Effect.ignore),      ),    );  }),); scenario(  "MCP sessions · a paused approval survives the client reconnecting and resumes",  {},  Effect.gen(function* () {    const target = yield* Target;    const { client } = yield* Api;    const mcp = yield* Mcp;    const identity = yield* target.newIdentity();    const api = yield* client(coreApi, identity);    const bearer = yield* mcp.mintBearer(emailOf(identity));     // Gate a built-in tool behind human approval; the org is fresh, so the    // gate affects no other scenario, but remove it anyway on every exit.    const policy = yield* api.policies.create({      payload: { owner: "org", pattern: APPROVAL_TARGET_TOOL, action: "require_approval" },    });     yield* Effect.gen(function* () {      const first = yield* Effect.promise(() => connectClient(target.mcpUrl, bearer));      const sessionId = first.transport.sessionId;      const paused = yield* Effect.promise(() =>        first.client.callTool({ name: "execute", arguments: { code: GATED_CODE } }),      ).pipe(Effect.ensuring(closeQuietly(first)));      const pausedText = textOf(paused);      expect(pausedText, "the gated call pauses instead of completing").toContain(        "Execution paused",      );      const executionId = /\bexecutionId:\s*(\S+)/.exec(pausedText)?.[1];      expect(executionId, "the paused result carries the executionId").toEqual(expect.any(String));       // The user answers from a NEW client on the same session — the paused      // execution lives in the session, not the connection.      const second = yield* Effect.promise(() => connectClient(target.mcpUrl, bearer, sessionId));      const resumed = yield* Effect.promise(() =>        second.client.callTool({          name: "resume",          arguments: { executionId: executionId ?? "", action: "accept", content: "{}" },        }),      ).pipe(Effect.ensuring(closeQuietly(second)));      expect(resumed.isError, "the resumed execution completes").not.toBe(true);      expect(textOf(resumed), "the gated tool's result comes back after approval").toContain(        APPROVAL_TARGET_TOOL,      );    }).pipe(      Effect.ensuring(        api.policies          .remove({ params: { policyId: policy.id }, payload: { owner: "org" } })          .pipe(Effect.ignore),      ),    );  }),);