fix(mcp): own cancellation and raw input throughout startup

docs/mcp-contracts.md

@@ -666,6 +666,15 @@ error's reported resources.
 
 ## Transports
 
+The eager stdio runtime owns SIGINT, SIGTERM and stdin EOF before opening an
+initial target or loading optional providers. Shutdown cancels the initial open,
+waits for late startup handles, and closes them through the same retryable owner
+as a running server, retaining provider documents. Incoming stdio bytes during
+startup are staged without parsing, within the installed SDK's stdio buffer
+capacity, then handed unchanged to its single framing owner. Exceeding that
+startup capacity ends startup with a transport resource constraint; it has no
+runtime override. HTTP startup never reads stdin.
+
 stdio is the default transport and the only one that client registrations,
 `rea mcp doctor` and the MCP registry entry describe. `rea mcp --http [--host H]
 [--port N]` is an opt-in alternative that serves the same catalog over the

src/main.ts

@@ -29,6 +29,7 @@ import {
 
 const runtimeDependencies = (): RuntimeDependencies => ({
   env: process.env,
+  stdin: process.stdin,
   serve: serveStdio,
   writeStderr: (text) => process.stderr.write(text),
   setExitCode: (code) => {

src/main/StdioStartupInput.ts

@@ -0,0 +1,117 @@
+import type { Readable } from "node:stream";
+import { STDIO_DEFAULT_MAX_BUFFER_SIZE } from "@modelcontextprotocol/server";
+import type { AnalysisError } from "../domain/analysisErrorBase.js";
+import {
+  AnalysisOutputError,
+  AnalysisResourceConstraintError,
+} from "../domain/analysisErrorCore.js";
+
+/** Raw startup bytes remain owned until the SDK's stdio transport can consume them. */
+export const ownStdioStartupInput = (
+  input: Readable,
+  stop: (failure?: AnalysisError) => void,
+): { handoff(): void; resume(): void; close(): void } => {
+  const chunks: Buffer[] = [];
+  let retainedBytes = 0;
+  let tail: Buffer | undefined;
+  let tailBytes = 0;
+  let handedOff = false;
+  let closed = false;
+  const detach = (): void => {
+    input.off("data", onData);
+    input.off("end", onEnd);
+    input.off("close", onEnd);
+    input.off("error", onError);
+  };
+  const onEnd = (): void => stop();
+  const onError = (cause: Error): void =>
+    stop(
+      new AnalysisOutputError(
+        "mcp_startup",
+        "MCP stdin could not be read during startup.",
+        { cause, userMessage: "MCP stdin could not be read during startup." },
+      ),
+    );
+  const onData = (chunk: Buffer | string): void => {
+    const incomingBytes = Buffer.isBuffer(chunk)
+      ? chunk.byteLength
+      : Buffer.byteLength(chunk);
+    if (retainedBytes + incomingBytes > STDIO_DEFAULT_MAX_BUFFER_SIZE) {
+      input.pause();
+      detach();
+      stop(
+        new AnalysisResourceConstraintError(
+          "mcp_startup",
+          "transport",
+          "MCP input exceeded the stdio buffer capacity before startup completed.",
+          {
+            max_buffer_bytes: STDIO_DEFAULT_MAX_BUFFER_SIZE,
+            retained_bytes: retainedBytes,
+            incoming_bytes: incomingBytes,
+          },
+          {
+            remediationAction:
+              "Wait for the MCP server to finish startup before sending more requests. The stdio buffer capacity has no runtime override.",
+          },
+        ),
+      );
+      return;
+    }
+    const bytes = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
+    let offset = 0;
+    while (offset < bytes.byteLength) {
+      // Bound chunk bookkeeping too, even when stdin delivers one byte per event.
+      tail ??= Buffer.allocUnsafe(
+        Math.min(64 * 1024, STDIO_DEFAULT_MAX_BUFFER_SIZE - retainedBytes),
+      );
+      const copied = Math.min(
+        tail.byteLength - tailBytes,
+        bytes.byteLength - offset,
+      );
+      bytes.copy(tail, tailBytes, offset, offset + copied);
+      offset += copied;
+      retainedBytes += copied;
+      tailBytes += copied;
+      if (tailBytes === tail.byteLength) {
+        chunks.push(tail);
+        tail = undefined;
+        tailBytes = 0;
+      }
+    }
+  };
+  input.on("data", onData);
+  input.on("end", onEnd);
+  input.on("close", onEnd);
+  input.on("error", onError);
+  if (input.readableEnded || input.destroyed) queueMicrotask(onEnd);
+  return {
+    handoff: () => {
+      if (closed || handedOff) return;
+      input.pause();
+      detach();
+      if (tail !== undefined && tailBytes > 0)
+        chunks.push(tail.subarray(0, tailBytes));
+      tail = undefined;
+      tailBytes = 0;
+      // Unshift in reverse chunk order; the SDK remains the only frame parser.
+      for (let index = chunks.length - 1; index >= 0; index -= 1)
+        input.unshift(chunks[index]!);
+      chunks.length = 0;
+      retainedBytes = 0;
+      handedOff = true;
+    },
+    resume: () => {
+      if (handedOff && !closed) input.resume();
+    },
+    close: () => {
+      if (closed) return;
+      closed = true;
+      detach();
+      chunks.length = 0;
+      tail = undefined;
+      tailBytes = 0;
+      retainedBytes = 0;
+      if (!handedOff) input.pause();
+    },
+  };
+};

src/main/deferredRuntime.ts

@@ -79,6 +79,9 @@ const initializeRuntime = async (
   const result = await runEager(
     {
       ...dependencies,
+      // The outer SDK transport already owns and frames this stream. Eager
+      // materialization owns providers only, never another copy of raw input.
+      stdin: undefined,
       registerShutdown: () => () => undefined,
       captureShutdown: (close) => {
         shutdown = close;

src/main/eagerRuntime.ts

@@ -14,7 +14,9 @@ import {
   loadOptionalObservationProviders,
 } from "../composition/optionalObservationProviders.js";
 import { createToolResultDelivery } from "../server/toolResult.js";
-import { createShutdown } from "./shutdown.js";
+import { createShutdown, reportMcpShutdownFailure } from "./shutdown.js";
+import { ownStdioStartupInput } from "./StdioStartupInput.js";
+import type { AnalysisError } from "../domain/analysisErrorBase.js";
 
 /**
  * Start the long-lived MCP adapter and install idempotent shutdown handlers.
@@ -37,34 +39,120 @@ export const runEager = async (
   const logger = createLogger("mcp", config.value.logLevel);
   const serverLogger = logger.child({ layer: "server" });
   const session = createBinarySession(config.value, logger, environment);
-  const opened = await openInitialTarget(
-    session,
-    config.value,
-    serverLogger,
-    dependencies.writeStderr,
-  );
-  if (!opened.ok) return opened.exitCode;
-  const transport = await startMcpTransport(dependencies, session, {
-    logger,
-    serverLogger,
-    environment,
-    delivery: createToolResultDelivery(config.value.mcpMaxResponseBytes),
-    loadOptionalProviders:
-      dependencies.loadOptionalProviders ??
-      (() =>
-        loadOptionalObservationProviders(
-          createOptionalObservationFactories(environment),
-        )),
+  const startup = new AbortController();
+  let transport: Awaited<ReturnType<typeof startMcpTransport>> | undefined;
+  let settleStartup!: () => void;
+  const startupSettled = new Promise<void>((resolve) => {
+    settleStartup = resolve;
   });
-  if (!transport.ok) return 1;
+  let inputFailure: AnalysisError | undefined;
   const shutdown = createShutdown({
-    handle: transport.handle,
-    closeAndroid: transport.closeAndroid,
-    closeOptionalProviders: transport.closeOptionalProviders,
+    // A shutdown request seals session admission immediately, then waits for
+    // startup to return every late handle before any resource is released.
+    handle: {
+      prepareShutdown: async () => {
+        await startupSettled;
+        await transport?.handle.prepareShutdown?.();
+      },
+      close: async () => {
+        await transport?.handle.close();
+      },
+    },
+    closeAndroid: async () => {
+      await transport?.closeAndroid();
+    },
+    closeOptionalProviders: async () => {
+      await transport?.closeOptionalProviders();
+    },
     session,
-    dependencies,
+    dependencies: {
+      ...dependencies,
+      registerShutdown: (handler) =>
+        dependencies.registerShutdown(() => {
+          startup.abort();
+          handler();
+        }),
+    },
     serverLogger,
   });
-  dependencies.captureShutdown?.(shutdown.shutdown);
-  return 0;
+  const startupInput =
+    dependencies.transport !== "http" &&
+    dependencies.stdin !== undefined &&
+    !startup.signal.aborted
+      ? ownStdioStartupInput(dependencies.stdin, (failure) => {
+          inputFailure ??= failure;
+          startup.abort(failure);
+          shutdown.request();
+        })
+      : undefined;
+  dependencies.captureShutdown?.(async () => {
+    startup.abort();
+    await shutdown.shutdown();
+  });
+  let ready = false;
+  let startupRejected = false;
+  let startupCause: unknown;
+  let exitCode = 0;
+  try {
+    const opened = await openInitialTarget(
+      session,
+      config.value,
+      serverLogger,
+      dependencies.writeStderr,
+      startup.signal,
+    );
+    if (!opened.ok) exitCode = opened.exitCode;
+    else if (!startup.signal.aborted) {
+      const started = await startMcpTransport(
+        {
+          ...dependencies,
+          serve: async (factory, options) => {
+            if (startup.signal.aborted) throw startup.signal.reason;
+            startupInput?.handoff();
+            const handle = await dependencies.serve(factory, options);
+            if (!startup.signal.aborted) startupInput?.resume();
+            return handle;
+          },
+        },
+        session,
+        {
+          logger,
+          serverLogger,
+          environment,
+          signal: startup.signal,
+          delivery: createToolResultDelivery(config.value.mcpMaxResponseBytes),
+          loadOptionalProviders:
+            dependencies.loadOptionalProviders ??
+            (() =>
+              loadOptionalObservationProviders(
+                createOptionalObservationFactories(environment),
+              )),
+        },
+      );
+      transport = started;
+      if (started.ok) ready = true;
+      else exitCode = started.cancelled ? 0 : 1;
+    }
+  } catch (cause: unknown) {
+    startupRejected = true;
+    startupCause = cause;
+  } finally {
+    settleStartup();
+    startupInput?.close();
+  }
+  if (startupRejected || !ready || startup.signal.aborted) {
+    if (inputFailure !== undefined)
+      dependencies.writeStderr(
+        `${projectAnalysisError(inputFailure).message}\n`,
+      );
+    try {
+      await shutdown.shutdown();
+    } catch (cause: unknown) {
+      reportMcpShutdownFailure(cause, dependencies, serverLogger);
+      return 1;
+    }
+    if (inputFailure !== undefined) return 1;
+    if (startupRejected) throw startupCause;
+  }
+  return exitCode;
 };

src/main/startup.ts

@@ -1,6 +1,7 @@
 import type { BinarySession } from "../application/binary/BinarySession.js";
 import type { AppConfig } from "../config/types.js";
 import type { Logger } from "pino";
+import { AnalysisCancelledError } from "../domain/analysisErrorCore.js";
 import { projectAnalysisError } from "../domain/analysisErrorProjection.js";
 
 export const openInitialTarget = async (
@@ -8,26 +9,25 @@ export const openInitialTarget = async (
   config: AppConfig,
   serverLogger: Logger,
   writeStderr: (text: string) => void,
+  signal?: AbortSignal,
 ): Promise<
-  { readonly ok: true } | { readonly ok: false; readonly exitCode: 1 }
+  { readonly ok: true } | { readonly ok: false; readonly exitCode: 0 | 1 }
 > => {
+  if (signal?.aborted) return { ok: false, exitCode: 0 };
   if (config.hopperTargetPath === undefined) return { ok: true };
   const opened = await session.open(config.hopperTargetPath, {
     targetKind: config.hopperTargetKind,
+    ...(signal === undefined ? {} : { signal }),
   });
   if (opened.ok) return { ok: true };
-  const closed = await session.close();
+  if (opened.error instanceof AnalysisCancelledError && signal?.aborted)
+    return { ok: false, exitCode: 0 };
   serverLogger.error(
     {
       error: projectAnalysisError(opened.error),
-      ...(closed.ok
-        ? {}
-        : { cleanup_error: projectAnalysisError(closed.error) }),
     },
     "Initial target failed to open",
   );
   writeStderr(`${projectAnalysisError(opened.error).message}\n`);
-  if (!closed.ok)
-    writeStderr(`${projectAnalysisError(closed.error).message}\n`);
   return { ok: false, exitCode: 1 };
 };

src/main/transport.ts

@@ -9,7 +9,6 @@ import type { ToolResultDelivery } from "../server/toolResult.js";
 import type { McpServerHandle, RuntimeDependencies } from "./types.js";
 import type { OptionalProviderLoadResult } from "../application/OptionalObservationProviders.js";
 import { OwnedMcpServers } from "./OwnedMcpServers.js";
-import { projectAnalysisError } from "../domain/analysisErrorProjection.js";
 import {
   MCP_CONNECTION_LOST,
   MCP_CONNECTION_START_FAILED,
@@ -22,6 +21,7 @@ interface ServerContext {
   readonly logger: Logger;
   readonly serverLogger: Logger;
   readonly loadOptionalProviders: () => Promise<OptionalProviderLoadResult>;
+  readonly signal?: AbortSignal;
 }
 
 type ClosableOptionalProvider = {
@@ -40,7 +40,13 @@ export const startMcpTransport = async (
       readonly closeAndroid: () => Promise<void>;
       readonly closeOptionalProviders: () => Promise<void>;
     }
-  | { readonly ok: false }
+  | {
+      readonly ok: false;
+      readonly cancelled: boolean;
+      readonly handle: McpServerHandle;
+      readonly closeAndroid: () => Promise<void>;
+      readonly closeOptionalProviders: () => Promise<void>;
+    }
 > => {
   const { serverLogger } = serverContext;
   let optionalProviders: OptionalProviderLoadResult = {};
@@ -62,7 +68,7 @@ export const startMcpTransport = async (
       "Optional MCP adapter could not load; its peers remain available",
     );
   }
-  const androidProviders: AndroidAnalysisPort[] = [];
+  const androidProviders = new Set<AndroidAnalysisPort>();
   const servers = new OwnedMcpServers<ReturnType<typeof createServer>>();
   const closeServers = async (): Promise<void> => {
     const results = await servers.closeAll();
@@ -74,7 +80,10 @@ export const startMcpTransport = async (
   };
   const closeAndroid = async (): Promise<void> => {
     const results = await Promise.allSettled(
-      androidProviders.map((provider) => provider.close()),
+      [...androidProviders].map(async (provider) => {
+        await provider.close();
+        androidProviders.delete(provider);
+      }),
     );
     for (const result of results)
       if (result.status === "rejected") throw result.reason;
@@ -108,6 +117,7 @@ export const startMcpTransport = async (
   };
   let handle: McpServerHandle;
   try {
+    if (serverContext.signal?.aborted) throw serverContext.signal.reason;
     handle = await dependencies.serve(
       () =>
         servers.create(() => {
@@ -115,7 +125,7 @@ export const startMcpTransport = async (
           const android = createAndroidAnalysisProvider(
             serverContext.environment,
           );
-          androidProviders.push(android);
+          androidProviders.add(android);
           const server = (dependencies.createServer ?? createServer)(
             { kind: "session", session },
             {
@@ -135,41 +145,35 @@ export const startMcpTransport = async (
           dependencies.writeStderr(`${MCP_CONNECTION_LOST}\n`);
         },
         logger: serverLogger,
+        ...(serverContext.signal === undefined
+          ? {}
+          : { signal: serverContext.signal }),
       },
     );
   } catch (cause: unknown) {
-    const [
-      binaryCleanup,
-      androidCleanup,
-      serverCleanup,
-      optionalProviderCleanup,
-    ] = await Promise.allSettled([
-      session.close(),
-      closeAndroid(),
-      closeServers(),
-      closeOptionalProviders(),
-    ]);
-    const startFailure =
-      cause instanceof StreamableHttpListenError
-        ? MCP_HTTP_LISTEN_FAILED
-        : MCP_CONNECTION_START_FAILED;
-    serverLogger.error(
-      {
-        error: cause instanceof Error ? cause.message : String(cause),
-      },
-      startFailure,
-    );
-    dependencies.writeStderr(`${startFailure}\n`);
-    if (binaryCleanup.status === "fulfilled" && !binaryCleanup.value.ok) {
-      const cleanupError = projectAnalysisError(binaryCleanup.value.error);
+    const cancelled =
+      serverContext.signal?.aborted === true &&
+      cause === serverContext.signal.reason;
+    const [androidCleanup, serverCleanup, optionalProviderCleanup] =
+      await Promise.allSettled([
+        closeAndroid(),
+        closeServers(),
+        closeOptionalProviders(),
+      ]);
+    if (!cancelled) {
+      const startFailure =
+        cause instanceof StreamableHttpListenError
+          ? MCP_HTTP_LISTEN_FAILED
+          : MCP_CONNECTION_START_FAILED;
       serverLogger.error(
-        { cleanup_error: cleanupError },
-        "Binary provider cleanup failed after MCP startup",
+        {
+          error: cause instanceof Error ? cause.message : String(cause),
+        },
+        startFailure,
       );
-      dependencies.writeStderr(`${cleanupError.message}\n`);
+      dependencies.writeStderr(`${startFailure}\n`);
     }
     for (const cleanup of [
-      binaryCleanup,
       androidCleanup,
       serverCleanup,
       optionalProviderCleanup,
@@ -186,7 +190,15 @@ export const startMcpTransport = async (
         dependencies.writeStderr(`Provider cleanup failed: ${reason}\n`);
       }
     }
-    return { ok: false };
+    // Return the same partial owners for retry; the caller owns the binary
+    // session independently of transport admission.
+    return {
+      ok: false,
+      cancelled,
+      handle: { close: closeServers },
+      closeAndroid,
+      closeOptionalProviders,
+    };
   }
   return {
     ok: true,

src/main/types.ts

@@ -1,3 +1,4 @@
+import type { Readable } from "node:stream";
 import type { McpServerFactory } from "@modelcontextprotocol/server";
 import type { Logger } from "pino";
 
@@ -14,6 +15,7 @@ export interface McpServerHandle {
 export interface McpServeOptions {
   readonly onerror?: (error: Error) => void;
   readonly logger?: Logger;
+  readonly signal?: AbortSignal;
 }
 
 /** Start one transport around the server factory; a listener binds asynchronously. */
@@ -27,6 +29,8 @@ export interface RuntimeDependencies {
   readonly transport?: "stdio" | "http";
   readonly captureShutdown?: (shutdown: () => Promise<void>) => void;
   readonly env: NodeJS.ProcessEnv;
+  /** Stdio startup owns this raw input until the SDK starts; HTTP never reads it. */
+  readonly stdin?: Readable | undefined;
   readonly serve: McpServe;
   readonly writeStderr: (text: string) => void;
   readonly setExitCode: (code: number) => void;

tests/boundary/mcp/deferredRuntime.test.ts

@@ -794,3 +794,77 @@ it("rejects a corrupted build-owned catalog before starting the connection", asy
   expect(child.stdout).toBe("");
   expect(child.stderr).toBe(`${SERVER_START_FAILED}\n`);
 });
+
+it("keeps the outer SDK as the only raw-input owner during deferred materialization", () => {
+  const child = spawnSync(
+    process.execPath,
+    [
+      "--input-type=module",
+      "--eval",
+      `
+import assert from "node:assert/strict";
+import { PassThrough } from "node:stream";
+import { serveStdio, StdioServerTransport } from "@modelcontextprotocol/server/stdio";
+import { runDeferred } from "./dist/main/deferredRuntime.js";
+import { parseConfig } from "./dist/config/parseConfig.js";
+const parsed = parseConfig({ REA_LOG_LEVEL: "silent" });
+assert.equal(parsed.ok, true);
+const input = new PassThrough();
+const output = new PassThrough();
+const received = [];
+input.on("data", chunk => received.push(Buffer.from(chunk)));
+const writes = [];
+const responses = [];
+let pendingOutput = "";
+output.on("data", chunk => {
+  pendingOutput += chunk.toString("utf8");
+  for (;;) {
+    const boundary = pendingOutput.indexOf("\\n");
+    if (boundary < 0) break;
+    responses.push(JSON.parse(pendingOutput.slice(0, boundary)));
+    pendingOutput = pendingOutput.slice(boundary + 1);
+  }
+});
+const until = async (predicate) => {
+  const deadline = Date.now() + 10000;
+  while (!predicate()) {
+    if (Date.now() >= deadline) throw new Error("Deferred SDK control did not settle");
+    await new Promise(resolve => setTimeout(resolve, 5));
+  }
+};
+let entered = false;
+let release;
+const held = new Promise(resolve => { release = resolve; });
+let close;
+const result = await runDeferred({
+  env: { REA_LOG_LEVEL: "silent" }, stdin: input,
+  serve: (factory, options) => serveStdio(factory, { ...options, transport: new StdioServerTransport(input, output) }),
+  loadOptionalProviders: async () => { entered = true; await held; return {}; },
+  captureShutdown: owned => { close = owned; },
+  registerShutdown: () => () => {}, writeStderr: text => { throw new Error(text); }, setExitCode: () => {},
+}, parsed.value);
+assert.equal(result, 0);
+const send = value => { const bytes = Buffer.from(JSON.stringify(value) + "\\n"); writes.push(bytes); input.write(bytes); };
+try {
+  send({ jsonrpc: "2.0", id: 1, method: "initialize", params: { protocolVersion: "2025-11-25", capabilities: {}, clientInfo: { name: "deferred-input-owner", version: "1" } } });
+  await until(() => responses.some(value => value.id === 1));
+  send({ jsonrpc: "2.0", method: "notifications/initialized" });
+  send({ jsonrpc: "2.0", id: 2, method: "tools/call", params: { name: "list_unknowns", arguments: {} } });
+  await until(() => entered);
+  send({ jsonrpc: "2.0", id: 3, method: "tools/call", params: { name: "list_unknowns", arguments: { limit: 1 } } });
+  release();
+  await until(() => responses.some(value => value.id === 2) && responses.some(value => value.id === 3));
+  await close();
+  assert.deepEqual(Buffer.concat(received), Buffer.concat(writes));
+  assert.equal(responses.filter(value => value.id === 2).length, 1);
+  assert.equal(responses.filter(value => value.id === 3).length, 1);
+  console.log("ONE_INPUT_OWNER");
+} finally { release(); await close?.(); input.destroy(); output.destroy(); }
+`,
+    ],
+    { cwd: process.cwd(), encoding: "utf8", timeout: 15000 },
+  );
+  expect(child.error).toBeUndefined();
+  expect(child.status, child.stderr).toBe(0);
+  expect(child.stdout).toBe("ONE_INPUT_OWNER\n");
+});

tests/process-global/mainStartupCancellation.test.ts

@@ -0,0 +1,252 @@
+import { EventEmitter } from "node:events";
+import { PassThrough } from "node:stream";
+import { afterEach, expect, it, vi } from "vitest";
+import { StdioServerTransport } from "@modelcontextprotocol/server/stdio";
+import { STDIO_DEFAULT_MAX_BUFFER_SIZE } from "@modelcontextprotocol/server";
+import type { BinarySession } from "../../src/application/binary/BinarySession.js";
+import type { BinaryTarget } from "../../src/domain/binaryTargetTypes.js";
+import { AnalysisCancelledError } from "../../src/domain/analysisErrorCore.js";
+import { ProviderCleanupError } from "../../src/domain/providerCleanupError.js";
+import { err, ok } from "../../src/domain/result.js";
+import { run } from "../../src/main.js";
+import { registerProcessShutdown } from "../../src/main/processShutdown.js";
+import { ownStdioStartupInput } from "../../src/main/StdioStartupInput.js";
+import { MCP_SHUTDOWN_FAILED } from "../../src/main/messages.js";
+import {
+  createDeferred,
+  createTestBinarySession,
+} from "../fixtures/binarySession.js";
+
+const control = vi.hoisted(() => ({
+  session: undefined as BinarySession | undefined,
+}));
+vi.mock("../../src/composition/binary.js", async (importOriginal) => {
+  const actual =
+    await importOriginal<typeof import("../../src/composition/binary.js")>();
+  return {
+    ...actual,
+    createBinarySession: () => {
+      const session = control.session;
+      if (session === undefined)
+        throw new Error("Missing startup session owner");
+      return session;
+    },
+  };
+});
+afterEach(() => {
+  vi.restoreAllMocks();
+  control.session = undefined;
+});
+const target: BinaryTarget = {
+  path: "/fixture",
+  sha256: "0".repeat(64),
+  kind: "executable",
+  format: "elf",
+  architecture: "x86_64",
+  availableArchitectures: ["x86_64"],
+};
+const sessionOwner = () => {
+  const session = createTestBinarySession(() => {
+    throw new Error("No real provider may start");
+  });
+  control.session = session;
+  return session;
+};
+
+it.each([
+  { trigger: "SIGINT", failedClose: false },
+  { trigger: "SIGTERM", failedClose: false },
+  { trigger: "EOF", failedClose: false },
+  { trigger: "SIGINT", failedClose: true },
+  { trigger: "CAPACITY", failedClose: true },
+  { trigger: "READ_ERROR", failedClose: true },
+] as const)(
+  "owns initial open before $trigger (cleanup failure=$failedClose)",
+  async ({ trigger, failedClose }) => {
+    const session = sessionOwner();
+    const input = new PassThrough();
+    const signals = new EventEmitter();
+    const entered = createDeferred<AbortSignal>();
+    vi.spyOn(session, "open").mockImplementation(async (_path, options) => {
+      const signal = options?.signal;
+      if (signal === undefined)
+        throw new Error("Initial open has no cancellation owner");
+      entered.resolve(signal);
+      await new Promise<void>((resolve) =>
+        signal.addEventListener("abort", () => resolve(), { once: true }),
+      );
+      return err(
+        new AnalysisCancelledError("open_binary", { cause: signal.reason }),
+      );
+    });
+    let closeFails = failedClose;
+    const close = vi.spyOn(session, "close").mockImplementation(async () =>
+      closeFails
+        ? err(
+            new ProviderCleanupError("fixture", ["retained provider"], {
+              reason: "close denied",
+            }),
+          )
+        : ok(null),
+    );
+    const serve = vi.fn(() => ({ close: async () => {} }));
+    const output: string[] = [];
+    const pending = run({
+      env: { HOPPER_TARGET_PATH: target.path, REA_LOG_LEVEL: "silent" },
+      stdin: input,
+      serve,
+      loadOptionalProviders: async () => ({}),
+      writeStderr: (text) => output.push(text),
+      setExitCode: () => {},
+      registerShutdown: (handler) =>
+        registerProcessShutdown(handler, { signals, stdin: input }),
+    });
+    const signal = await entered.promise;
+    expect(signals.listenerCount("SIGINT")).toBe(1);
+    expect(signals.listenerCount("SIGTERM")).toBe(1);
+    if (trigger === "EOF") input.end();
+    else if (trigger === "CAPACITY")
+      input.write(Buffer.alloc(STDIO_DEFAULT_MAX_BUFFER_SIZE + 1));
+    else if (trigger === "READ_ERROR")
+      input.emit("error", new Error("controlled stdin read failure"));
+    else signals.emit(trigger);
+    expect(await pending).toBe(failedClose ? 1 : 0);
+    expect(signal.aborted).toBe(true);
+    expect(serve).not.toHaveBeenCalled();
+    expect(close).toHaveBeenCalledWith({ retainProviderDocuments: true });
+    if (failedClose) {
+      expect(output.join("")).toContain(MCP_SHUTDOWN_FAILED);
+      if (trigger === "CAPACITY" || trigger === "READ_ERROR") {
+        const primary =
+          trigger === "CAPACITY"
+            ? "MCP input exceeded"
+            : "MCP stdin could not be read";
+        expect(output.join("")).toContain(primary);
+        expect(output.join("").indexOf(primary)).toBeLessThan(
+          output.join("").indexOf(MCP_SHUTDOWN_FAILED),
+        );
+      }
+      closeFails = false;
+      signals.emit("SIGINT");
+      await expect.poll(() => signals.listenerCount("SIGINT")).toBe(0);
+    } else expect(output).toEqual([]);
+    expect(signals.listenerCount("SIGTERM")).toBe(0);
+    input.destroy();
+  },
+);
+
+it.each(["optional", "late-handle"] as const)(
+  "closes resources settling after startup cancellation at %s",
+  async (stage) => {
+    const session = sessionOwner();
+    vi.spyOn(session, "open").mockResolvedValue(ok(target));
+    const closeSession = vi.spyOn(session, "close").mockResolvedValue(ok(null));
+    const input = new PassThrough();
+    const signals = new EventEmitter();
+    const entered = createDeferred<void>();
+    const release = createDeferred<void>();
+    const closeHandle = vi.fn(async () => {});
+    const serve = vi.fn(async () => {
+      entered.resolve();
+      await release.promise;
+      return { close: closeHandle };
+    });
+    const pending = run({
+      env: { HOPPER_TARGET_PATH: target.path, REA_LOG_LEVEL: "silent" },
+      stdin: input,
+      serve,
+      loadOptionalProviders: async () => {
+        if (stage === "optional") {
+          entered.resolve();
+          await release.promise;
+        }
+        return {};
+      },
+      writeStderr: () => {},
+      setExitCode: () => {},
+      registerShutdown: (handler) =>
+        registerProcessShutdown(handler, { signals, stdin: input }),
+    });
+    await entered.promise;
+    if (stage === "optional") input.end();
+    else signals.emit("SIGTERM");
+    expect(closeHandle).not.toHaveBeenCalled();
+    expect(closeSession).not.toHaveBeenCalled();
+    release.resolve();
+    expect(await pending).toBe(0);
+    expect(serve).toHaveBeenCalledTimes(stage === "optional" ? 0 : 1);
+    expect(closeHandle).toHaveBeenCalledTimes(stage === "optional" ? 0 : 1);
+    expect(closeSession).toHaveBeenCalledTimes(1);
+    expect(closeSession).toHaveBeenCalledWith({
+      retainProviderDocuments: true,
+    });
+    expect(signals.listenerCount("SIGINT")).toBe(0);
+    input.destroy();
+  },
+);
+
+it("hands exact startup bytes to the installed SDK framing owner", async () => {
+  const input = new PassThrough();
+  const output = new PassThrough();
+  const stopped = vi.fn();
+  const owner = ownStdioStartupInput(input, stopped);
+  const request = Buffer.from(
+    JSON.stringify({
+      jsonrpc: "2.0",
+      id: 1,
+      method: "initialize",
+      params: {
+        protocolVersion: "2025-11-25",
+        capabilities: {},
+        clientInfo: { name: "é🚀", version: "1" },
+      },
+    }) + "\n",
+  );
+  const ping = Buffer.from('{"jsonrpc":"2.0","id":2,"method":"ping"}\n');
+  input.write(request.subarray(0, request.indexOf(Buffer.from("🚀")) + 1));
+  input.write(request.subarray(request.indexOf(Buffer.from("🚀")) + 1));
+  input.write(ping.subarray(0, 11));
+  await new Promise<void>((resolve) => setImmediate(resolve));
+  owner.handoff();
+  const raw: Buffer[] = [];
+  input.on("data", (chunk: Buffer) => raw.push(chunk));
+  const messages: unknown[] = [];
+  const transport = new StdioServerTransport(input, output);
+  transport.onmessage = (message) => messages.push(message);
+  try {
+    await transport.start();
+    owner.resume();
+    input.write(ping.subarray(11));
+    await expect.poll(() => messages.length).toBe(2);
+    expect(Buffer.concat(raw)).toEqual(Buffer.concat([request, ping]));
+    expect(messages[0]).toMatchObject({
+      params: { clientInfo: { name: "é🚀" } },
+    });
+    expect(messages[1]).toMatchObject({ id: 2, method: "ping" });
+    expect(stopped).not.toHaveBeenCalled();
+  } finally {
+    owner.close();
+    await transport.close();
+    input.destroy();
+    output.destroy();
+  }
+});
+
+it("bounds raw startup retention by the installed SDK stdio capacity", async () => {
+  const input = new PassThrough();
+  const stopped = vi.fn();
+  const owner = ownStdioStartupInput(input, stopped);
+  try {
+    input.write(Buffer.alloc(STDIO_DEFAULT_MAX_BUFFER_SIZE + 1));
+    await expect.poll(() => stopped.mock.calls.length).toBe(1);
+    expect(stopped).toHaveBeenCalledWith(
+      expect.objectContaining({
+        _tag: "AnalysisResourceConstraintError",
+        resource: "transport",
+      }),
+    );
+  } finally {
+    owner.close();
+    input.destroy();
+  }
+});