diff --git a/docs/mcp-contracts.md b/docs/mcp-contracts.md index 2ac744b2ae..2089e32064 100644 --- a/docs/mcp-contracts.md +++ b/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 diff --git a/src/main.ts b/src/main.ts index 1efe4f0788..ceefd957ab 100644 --- a/src/main.ts +++ b/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) => { diff --git a/src/main/StdioStartupInput.ts b/src/main/StdioStartupInput.ts new file mode 100644 index 0000000000..5f637ab0d3 --- /dev/null +++ b/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(); + }, + }; +}; diff --git a/src/main/deferredRuntime.ts b/src/main/deferredRuntime.ts index 6ca59fc2c3..aa9f682b54 100644 --- a/src/main/deferredRuntime.ts +++ b/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; diff --git a/src/main/eagerRuntime.ts b/src/main/eagerRuntime.ts index 874d8007d3..fb6353380a 100644 --- a/src/main/eagerRuntime.ts +++ b/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> | undefined; + let settleStartup!: () => void; + const startupSettled = new Promise((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; }; diff --git a/src/main/startup.ts b/src/main/startup.ts index d71ff19cc7..e46c8f14f1 100644 --- a/src/main/startup.ts +++ b/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 }; }; diff --git a/src/main/transport.ts b/src/main/transport.ts index 871880f224..9199576894 100644 --- a/src/main/transport.ts +++ b/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; + readonly signal?: AbortSignal; } type ClosableOptionalProvider = { @@ -40,7 +40,13 @@ export const startMcpTransport = async ( readonly closeAndroid: () => Promise; readonly closeOptionalProviders: () => Promise; } - | { readonly ok: false } + | { + readonly ok: false; + readonly cancelled: boolean; + readonly handle: McpServerHandle; + readonly closeAndroid: () => Promise; + readonly closeOptionalProviders: () => Promise; + } > => { 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(); const servers = new OwnedMcpServers>(); const closeServers = async (): Promise => { const results = await servers.closeAll(); @@ -74,7 +80,10 @@ export const startMcpTransport = async ( }; const closeAndroid = async (): Promise => { 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, diff --git a/src/main/types.ts b/src/main/types.ts index 4cde88d1a8..4ff380322c 100644 --- a/src/main/types.ts +++ b/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; 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; diff --git a/tests/boundary/mcp/deferredRuntime.test.ts b/tests/boundary/mcp/deferredRuntime.test.ts index 602cfee4b5..3718f20650 100644 --- a/tests/boundary/mcp/deferredRuntime.test.ts +++ b/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"); +}); diff --git a/tests/process-global/mainStartupCancellation.test.ts b/tests/process-global/mainStartupCancellation.test.ts new file mode 100644 index 0000000000..f934e9c53d --- /dev/null +++ b/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(); + 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(); + 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((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(); + const release = createDeferred(); + 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((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(); + } +});