// Cloud: the org identity a session needs at init travels in the session props// the worker already resolved — it is NOT re-read from Postgres inside the// session Durable Object.//// The defect this pins: on EVERY cold DO init the DO opened a brand-new// Postgres connection purely to re-read the `organizations` row the worker had// just read microseconds earlier on the same request (the worker threw it away:// the auth principal hardcoded an empty organization name). When that fresh// connection could not be established the whole `initialize` died — a hard,// client-visible failure on a healthy database, because the only unhealthy// thing was a socket nothing needed to open.//// Two contracts, in the order a user meets them:////   1. the session opens and WORKS — initialize mints a session id and the same//      id then serves `tools/list`, all off the identity the props carried;//   2. the whole request performs exactly ONE organization read (the worker's//      own authorization check) and the DO opens no database connection of its//      own to name the org.//// (2) is asserted on the EXPORTED spans, and the two planes — worker and// Durable Object — export independently, so the count is only taken once a// worker-plane span for this same request has landed. Without that wait a// still-pending worker batch would make a two-read request look like a one-read// request, and the assertion would pass for the wrong reason.//// The failure half of the fix (an unreachable directory becomes a bounded retry// and a retryable 503 rather than a bare 500) is not reachable from here: the// harness runs one single-process PGlite shared by the worker and the DO, and// the worker's own authorization reads that same row on the same request — so// freezing the database fails the request before init identically on both sides// of the fix. Those branches are pinned in// `apps/cloud/src/mcp/session-meta.node.test.ts`.import { randomBytes } from "node:crypto"; import { expect } from "@effect/vitest";import { Effect } from "effect"; import { scenario } from "../src/scenario";import { Mcp, Target, Telemetry } from "../src/services";import type { Identity } from "../src/target"; const JSON_AND_SSE = "application/json, text/event-stream"; const INITIALIZE_REQUEST = {  jsonrpc: "2.0" as const,  id: 1,  method: "initialize",  params: {    protocolVersion: "2025-03-26",    capabilities: {},    clientInfo: { name: "executor-e2e-session-cold-init", 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; /** A client-supplied W3C trace context, so every span this one request produces *  — worker plane and Durable Object alike — is addressable by one trace id. */const newTraceContext = (): { readonly traceId: string; readonly traceparent: string } => {  const traceId = randomBytes(16).toString("hex");  const spanId = randomBytes(8).toString("hex");  return { traceId, traceparent: `00-${traceId}-${spanId}-01` };}; const mcpPost = (  url: string,  init: {    readonly bearer: string;    readonly sessionId?: string;    readonly traceparent?: 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 } : {}),      ...(init.traceparent ? { traceparent: init.traceparent } : {}),    },    body: JSON.stringify(init.body),  }); scenario(  "MCP session cold init · the org identity rides in the session props instead of a second Postgres read",  { timeout: 120_000 },  Effect.gen(function* () {    const target = yield* Target;    const mcp = yield* Mcp;    const telemetry = yield* Telemetry;     const identity = yield* target.newIdentity();    const bearer = yield* mcp.mintBearer(emailOf(identity));    const trace = newTraceContext();     // ---- 1. the session opens, and it works -------------------------------    const response = yield* Effect.promise(() =>      mcpPost(target.mcpUrl, {        bearer,        traceparent: trace.traceparent,        body: INITIALIZE_REQUEST,      }),    );    yield* Effect.promise(() => response.text());    expect(response.status, "initialize opens a session").toBe(200);    const sessionId = response.headers.get("mcp-session-id");    expect(sessionId, "the session id is minted").toBeTruthy();     const initialized = yield* Effect.promise(() =>      mcpPost(target.mcpUrl, {        bearer,        sessionId: sessionId ?? "",        body: INITIALIZED_NOTIFICATION,      }),    );    yield* Effect.promise(() => initialized.text());    expect(initialized.status, "the client completes the handshake").toBe(202);     // The session built from the props-carried identity actually serves work —    // the production symptom was an `initialize` that never got this far.    const tools = yield* Effect.promise(() =>      mcpPost(target.mcpUrl, {        bearer,        sessionId: sessionId ?? "",        body: TOOLS_LIST_REQUEST,      }),    );    const toolsBody = yield* Effect.promise(() => tools.text());    expect(tools.status, "the session serves requests once open").toBe(200);    expect(toolsBody, "the session advertises the execute tool").toContain("execute");     // ---- 2. one organization read for the whole init request --------------    // The DO really did resolve meta on this request (a cold init), and it    // resolved it from the props the worker handed over.    const resolveSpan = yield* telemetry.expectSpan({      traceId: trace.traceId,      operation: "McpSessionDOSqlite.resolveSessionMeta",    });    expect(resolveSpan.span.status, "the cold init resolved its meta without failing").toBe("ok");     // The worker plane exports on its own batch, independently of the DO's.    // Wait for a worker-plane span from this same request before counting, or    // an unflushed worker batch would hide the very read being counted.    yield* telemetry.expectSpan({ traceId: trace.traceId, operation: "mcp.request" });     // One org read for the whole request: the worker's own authorization    // lookup. A second one means the DO reopened a connection to re-read a row    // the request already had.    const orgReads = yield* telemetry.searchSpans({      traceId: trace.traceId,      operation: "user_store.getOrganization",    });    expect(      orgReads.length,      "only the worker's authorization check reads the organization row",    ).toBe(1);     // …and the DO's own database step never ran at all. `resolveSessionMeta`    // has already landed and this span is its child, so its absence here is    // absence, not lag.    const doDatabaseReads = yield* telemetry.searchSpans({      traceId: trace.traceId,      operation: "mcp.session.resolve_organization",    });    expect(      doDatabaseReads.length,      "the session DO opens no connection of its own to name the organization",    ).toBe(0);     // …because the identity it used is the one the worker handed it.    expect(      resolveSpan.span.tags["mcp.session.meta_source"],      "the session meta comes from the props the worker already resolved",    ).toBe("props");  }),);