diff --git a/docs/jeb-analysis.md b/docs/jeb-analysis.md index 8840243141..986701e779 100644 --- a/docs/jeb-analysis.md +++ b/docs/jeb-analysis.md @@ -22,6 +22,15 @@ inspection and decompilation while refusing `open_jeb_project`. Catalog discover uses one 60-second deadline across connection and pagination; a responsive server cannot keep discovery running indefinitely by returning more pages. +Each upstream tool call has an absolute 60-second request deadline by default. +Set `REA_JEB_MCP_TIMEOUT_MS` to a positive decimal number of milliseconds to +select a different per-request budget, for example `REA_JEB_MCP_TIMEOUT_MS=300000` +for a five-minute decompilation. This budget starts after discovery and applies +to each individual `tools/call`; it does not extend the separate catalog deadline +or establish one end-to-end deadline across a whole CLI or MCP operation. A +request that reaches its budget returns a typed timeout, while caller +cancellation remains a cancellation. + Verified against JEB 5.48.0 (Linux x64, licensed build) serving `jeb-mcp-server` 1.3.0 over streamable HTTP. ## Operations diff --git a/src/composition/jeb.ts b/src/composition/jeb.ts index 38f5dae5d1..e0c26551e4 100644 --- a/src/composition/jeb.ts +++ b/src/composition/jeb.ts @@ -1,8 +1,9 @@ import type { JebAnalysisPort } from "../application/jeb/JebAnalysisPort.js"; +import { parseJebMcpTimeoutMs } from "../config/environment.js"; import { ConfigurationError } from "../domain/configurationErrors.js"; +import { AnalysisInputError } from "../domain/analysisErrorCore.js"; import { err } from "../domain/result.js"; import type { JebMcpConnectionFactory } from "../jeb/JebMcpConnection.js"; -import { parseJebMcpEndpoint } from "../jeb/JebMcpEndpoint.js"; import { JebProvider } from "../jeb/JebProvider.js"; /** @@ -14,35 +15,44 @@ export const createJebAnalysisProvider = ( environment: Readonly> = process.env, connectionFactory?: JebMcpConnectionFactory, ): JebAnalysisPort => { - const endpoint = parseJebMcpEndpoint(environment.REA_JEB_MCP_URL); - if (!endpoint.ok) { + const requestTimeout = parseJebMcpTimeoutMs( + environment.REA_JEB_MCP_TIMEOUT_MS, + ); + if (!requestTimeout.ok) return unconfiguredJebProvider(requestTimeout.error); + try { + return new JebProvider({ + environment, + ...(requestTimeout.value === undefined + ? {} + : { requestTimeoutMs: requestTimeout.value }), + ...(connectionFactory === undefined ? {} : { connectionFactory }), + }); + } catch (cause: unknown) { + if (!(cause instanceof AnalysisInputError)) throw cause; const constraint = - typeof endpoint.error.cause === "string" - ? endpoint.error.cause - : endpoint.error.message; + typeof cause.cause === "string" ? cause.cause : cause.message; return unconfiguredJebProvider( new ConfigurationError("Invalid REA environment configuration", { - cause: endpoint.error, + cause, settings: [{ setting: "REA_JEB_MCP_URL", constraint }], }), ); } - return new JebProvider({ - environment, - ...(connectionFactory === undefined ? {} : { connectionFactory }), - }); }; const unconfiguredJebProvider = ( error: ConfigurationError, -): JebAnalysisPort => ({ - inspectAvailability: () => - Promise.resolve({ - status: "unavailable", - code: "not_configured", - reason: `REA_JEB_MCP_URL is invalid: ${error.settings[0]?.constraint ?? error.message}`, - diagnostics: { setting: "REA_JEB_MCP_URL" }, - }), - close: () => Promise.resolve(), - execute: () => Promise.resolve(err(error)), -}); +): JebAnalysisPort => { + const setting = error.settings[0]?.setting ?? "REA_JEB_MCP_URL"; + return { + inspectAvailability: () => + Promise.resolve({ + status: "unavailable", + code: "not_configured", + reason: `${setting} is invalid: ${error.settings[0]?.constraint ?? error.message}`, + diagnostics: { setting }, + }), + close: () => Promise.resolve(), + execute: () => Promise.resolve(err(error)), + }; +}; diff --git a/src/config.test.ts b/src/config.test.ts index 5fc1712560..8f105d044a 100644 --- a/src/config.test.ts +++ b/src/config.test.ts @@ -4,6 +4,10 @@ import { defaultHopperLauncherPath, parseConfig, } from "./config/parseConfig.js"; +import { + parseEnvironmentValues, + parseJebMcpTimeoutMs, +} from "./config/environment.js"; import { projectAnalysisError } from "./domain/analysisErrorProjection.js"; it("reports a configuration failure for a NUL in Hopper loader arguments", () => { @@ -328,6 +332,40 @@ describe("configuration failure diagnostics", () => { }); }); +describe("JEB MCP timeout configuration", () => { + it("stays optional and validates the caller-selected environment value", () => { + expect(parseJebMcpTimeoutMs(undefined)).toEqual({ + ok: true, + value: undefined, + }); + expect(parseJebMcpTimeoutMs("900000")).toEqual({ + ok: true, + value: 900_000, + }); + expect( + parseEnvironmentValues({ REA_JEB_MCP_TIMEOUT_MS: "900000" }), + ).toMatchObject({ ok: true, value: { REA_JEB_MCP_TIMEOUT_MS: 900_000 } }); + }); + + it.each(["", "0", "-1", "1.5", "1e3", "invalid", "2147483648"])( + "rejects malformed timeout %j with its setting", + (value) => { + const result = parseJebMcpTimeoutMs(value); + if (result.ok) throw new Error("Expected invalid JEB MCP timeout"); + expect(projectAnalysisError(result.error)).toMatchObject({ + details: { + settings: [ + { + setting: "REA_JEB_MCP_TIMEOUT_MS", + constraint: expect.any(String), + }, + ], + }, + }); + }, + ); +}); + describe("default Linux Hopper launcher", () => { const home = "/fixture/home"; const withExecutables = diff --git a/src/config/environment.ts b/src/config/environment.ts index 69e0f0ed85..181df6d33a 100644 --- a/src/config/environment.ts +++ b/src/config/environment.ts @@ -30,6 +30,7 @@ const environmentSchema = z.object({ .min(1) .refine(isAbsolute, "REA_IDA_MCP_CONFIG must be absolute") .optional(), + REA_JEB_MCP_TIMEOUT_MS: providerStartupTimeoutSchema.optional(), GHIDRA_INSTALL_DIR: z .string() .min(1) @@ -77,6 +78,27 @@ const environmentSchema = z.object({ REA_REFERENCE_SECRET_PATTERNS_JSON: z.string().default("[]"), }); +/** Parse the JEB-only timeout at its isolated provider-composition boundary. */ +export const parseJebMcpTimeoutMs = ( + value: string | undefined, +): Result => { + if (value === undefined) return ok(undefined); + const parsed = providerStartupTimeoutSchema.safeParse(value); + if (parsed.success) return ok(parsed.data); + return err( + new ConfigurationError("Invalid REA environment configuration", { + cause: parsed.error, + settings: [ + { + setting: "REA_JEB_MCP_TIMEOUT_MS", + constraint: + parsed.error.issues[0]?.message ?? "must be a valid timeout", + }, + ], + }), + ); +}; + type Environment = Omit< z.infer, "HOPPER_LOADER_ARGS_JSON" | "REA_REFERENCE_SECRET_PATTERNS_JSON" diff --git a/src/jeb/JebMcpConnection.test.ts b/src/jeb/JebMcpConnection.test.ts index 284f04f07d..b5bc163d46 100644 --- a/src/jeb/JebMcpConnection.test.ts +++ b/src/jeb/JebMcpConnection.test.ts @@ -5,7 +5,171 @@ import { decodeJebToolResult, createStreamableHttpJebMcpConnection, } from "./JebMcpConnection.js"; -import { AnalysisProtocolError } from "../domain/analysisErrorCore.js"; +import { + AnalysisCancelledError, + AnalysisProtocolError, + AnalysisTimeoutError, +} from "../domain/analysisErrorCore.js"; + +const jebTools = [ + "get_client_information", + "open_project", + "list_units", + "decompile_code_item", +] as const; + +const waitForRequestCount = async ( + readCount: () => number, + expected: number, + method: string, +): Promise => { + const deadline = performance.now() + 1_000; + while (readCount() < expected && performance.now() < deadline) + await new Promise((wake) => setTimeout(wake, 5)); + if (readCount() < expected) + throw new Error(`Expected ${expected} ${method} requests`); +}; + +const startJebHttpFixture = ( + responseDelayMs: number, + catalogRefreshDelayMs = 0, +) => { + let callCount = 0; + let catalogCount = 0; + const pendingTimers = new Set(); + const server = createServer((request, response) => { + const chunks: Buffer[] = []; + request.on("data", (chunk: Buffer) => chunks.push(chunk)); + request.on("end", () => { + if (chunks.length === 0) { + response.writeHead(202).end(); + return; + } + const message: { + readonly id?: string | number; + readonly method: string; + readonly params?: Record; + } = JSON.parse(Buffer.concat(chunks).toString("utf8")); + if (message.method === "notifications/initialized") { + response.writeHead(202).end(); + return; + } + if (message.method === "notifications/cancelled") { + response.writeHead(202).end(); + return; + } + if (message.method === "initialize") { + response.writeHead(200, { "Content-Type": "application/json" }).end( + JSON.stringify({ + jsonrpc: "2.0", + id: message.id, + result: { + protocolVersion: message.params?.protocolVersion, + capabilities: { tools: {} }, + serverInfo: { name: "jeb-timeout-fixture", version: "1" }, + }, + }), + ); + return; + } + if (message.method === "tools/list") { + catalogCount++; + let responseClosed = false; + response.on("close", () => { + if (!response.writableEnded) responseClosed = true; + }); + const sendCatalog = () => { + if (responseClosed) return; + response.writeHead(200, { "Content-Type": "application/json" }).end( + JSON.stringify({ + jsonrpc: "2.0", + id: message.id, + result: { + tools: jebTools.map((name) => ({ + name, + inputSchema: { type: "object", properties: {} }, + })), + }, + }), + ); + }; + if (catalogCount > 1 && catalogRefreshDelayMs > 0) { + const timer = setTimeout(() => { + pendingTimers.delete(timer); + sendCatalog(); + }, catalogRefreshDelayMs); + pendingTimers.add(timer); + } else { + sendCatalog(); + } + return; + } + if (message.method !== "tools/call") { + response.writeHead(202).end(); + return; + } + callCount++; + let responseClosed = false; + response.on("close", () => { + if (!response.writableEnded) { + responseClosed = true; + } + }); + const timer = setTimeout(() => { + pendingTimers.delete(timer); + if (responseClosed) return; + response.writeHead(200, { "Content-Type": "application/json" }).end( + JSON.stringify({ + jsonrpc: "2.0", + id: message.id, + result: { + structuredContent: { success: true, text: "finished" }, + content: [{ type: "text", text: '{"success":true}' }], + }, + }), + ); + }, responseDelayMs); + pendingTimers.add(timer); + }); + }); + + return new Promise<{ + readonly endpoint: URL; + readonly callCount: () => number; + readonly waitForCall: (count?: number) => Promise; + readonly waitForCatalog: (count?: number) => Promise; + readonly close: () => Promise; + }>((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", () => { + const address = server.address(); + if (address === null || typeof address === "string") { + reject(new Error("Expected TCP listener")); + return; + } + server.removeListener("error", reject); + resolve({ + endpoint: new URL(`http://127.0.0.1:${address.port}/mcp`), + callCount: () => callCount, + waitForCall: (count = 1) => + waitForRequestCount(() => callCount, count, "tools/call"), + waitForCatalog: (count = 1) => + waitForRequestCount(() => catalogCount, count, "tools/list"), + close: async () => { + for (const timer of pendingTimers) clearTimeout(timer); + pendingTimers.clear(); + const closed = new Promise((done, fail) => + server.close((cause) => + cause === undefined ? done() : fail(cause), + ), + ); + server.closeAllConnections(); + await closed; + }, + }); + }); + }); +}; describe("decodeJebToolResult", () => { it("passes refusal envelopes through so the provider preserves the engine reason", () => { @@ -118,3 +282,93 @@ it("refuses HTTP redirects before sending requests to another endpoint", async ( ); } }); + +it("applies the selected timeout to real MCP tool calls", async () => { + const fixture = await startJebHttpFixture(180); + const short = createStreamableHttpJebMcpConnection(fixture.endpoint, { + requestTimeoutMs: 50, + }); + const extended = createStreamableHttpJebMcpConnection(fixture.endpoint, { + requestTimeoutMs: 500, + }); + try { + await short.connect(); + await expect(short.call("decompile_code_item", {})).rejects.toBeInstanceOf( + AnalysisTimeoutError, + ); + expect(fixture.callCount()).toBe(1); + + await extended.connect(); + await expect( + extended.call("decompile_code_item", {}), + ).resolves.toMatchObject({ success: true, text: "finished" }); + expect(fixture.callCount()).toBe(2); + } finally { + try { + await Promise.all([short.close(), extended.close()]); + } finally { + await fixture.close(); + } + } +}); + +it("keeps caller abort and connection close as cancellations for active HTTP calls", async () => { + const fixture = await startJebHttpFixture(1_000); + const connection = createStreamableHttpJebMcpConnection(fixture.endpoint, { + requestTimeoutMs: 2_000, + }); + try { + await connection.connect(); + const alreadyAborted = new AbortController(); + alreadyAborted.abort(); + await expect( + connection.call("decompile_code_item", {}, alreadyAborted.signal), + ).rejects.toBeInstanceOf(AnalysisCancelledError); + expect(fixture.callCount()).toBe(0); + + const caller = new AbortController(); + const callerCall = connection.call( + "decompile_code_item", + {}, + caller.signal, + ); + await fixture.waitForCall(1); + caller.abort(); + await expect(callerCall).rejects.toBeInstanceOf(AnalysisCancelledError); + + const closingCall = connection.call("decompile_code_item", {}); + await fixture.waitForCall(2); + await connection.close(); + await expect(closingCall).rejects.toBeInstanceOf(AnalysisCancelledError); + } finally { + try { + await connection.close(); + } finally { + await fixture.close(); + } + } +}); + +it("keeps caller abort and connection close as cancellations during catalog refresh", async () => { + const fixture = await startJebHttpFixture(0, 1_000); + const connection = createStreamableHttpJebMcpConnection(fixture.endpoint); + try { + await connection.connect(); + const caller = new AbortController(); + const callerRefresh = connection.listTools(caller.signal); + await fixture.waitForCatalog(2); + caller.abort(); + await expect(callerRefresh).rejects.toBeInstanceOf(AnalysisCancelledError); + + const closingRefresh = connection.listTools(); + await fixture.waitForCatalog(3); + await connection.close(); + await expect(closingRefresh).rejects.toBeInstanceOf(AnalysisCancelledError); + } finally { + try { + await connection.close(); + } finally { + await fixture.close(); + } + } +}); diff --git a/src/jeb/JebMcpConnection.ts b/src/jeb/JebMcpConnection.ts index afa177f5cf..41c4a4fc36 100644 --- a/src/jeb/JebMcpConnection.ts +++ b/src/jeb/JebMcpConnection.ts @@ -1,6 +1,8 @@ import { Client, DEFAULT_REQUEST_TIMEOUT_MSEC, + SdkError, + SdkErrorCode, StreamableHTTPClientTransport, } from "@modelcontextprotocol/client"; import { z } from "zod"; @@ -82,16 +84,27 @@ const isJebFailureEnvelope = (value: JsonValue): boolean => !Array.isArray(value) && value.success === false; +export interface JebMcpConnectionTimeouts { + /** Absolute catalog connection and pagination deadline. */ + readonly discoveryTimeoutMs?: number; + /** Absolute timeout for each upstream tools/call request. */ + readonly requestTimeoutMs?: number; +} + export type JebMcpConnectionFactory = ( endpoint: URL, - timeoutMs?: number, + timeouts?: JebMcpConnectionTimeouts, ) => JebMcpConnection; /** Serve JEB requests from a running engine client over streamable HTTP. */ export const createStreamableHttpJebMcpConnection: JebMcpConnectionFactory = ( endpoint, - timeoutMs = DEFAULT_REQUEST_TIMEOUT_MSEC, + timeouts, ) => { + const discoveryTimeoutMs = + timeouts?.discoveryTimeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MSEC; + const requestTimeoutMs = + timeouts?.requestTimeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MSEC; interface ConnectionState { readonly client: Client; readonly transport: StreamableHTTPClientTransport; @@ -184,7 +197,7 @@ export const createStreamableHttpJebMcpConnection: JebMcpConnectionFactory = ( state = owner; const acquisition = (async () => { const deadline = new ProviderStartupDeadline( - timeoutMs, + discoveryTimeoutMs, owner.signal.signal, ); try { @@ -197,7 +210,9 @@ export const createStreamableHttpJebMcpConnection: JebMcpConnectionFactory = ( } catch (cause: unknown) { const failure = deadline.interruption === "timeout" - ? new AnalysisTimeoutError("jeb_mcp_catalog", timeoutMs, { cause }) + ? new AnalysisTimeoutError("jeb_mcp_catalog", discoveryTimeoutMs, { + cause, + }) : cause; owner.signal.abort(failure); try { @@ -234,19 +249,30 @@ export const createStreamableHttpJebMcpConnection: JebMcpConnectionFactory = ( // could detach this waiter before that owner has disposed a failed client. const tools = await connect(signal); if (!existing) return tools; - const deadline = new ProviderStartupDeadline(timeoutMs, signal); + const owner = state; + if (owner === undefined) + throw new AnalysisProtocolError("JEB MCP client failed to connect"); + const discoverySignal = + signal === undefined + ? owner.signal.signal + : AbortSignal.any([signal, owner.signal.signal]); + const deadline = new ProviderStartupDeadline( + discoveryTimeoutMs, + discoverySignal, + ); try { - const owner = state; - if (owner === undefined) - throw new AnalysisProtocolError("JEB MCP client failed to connect"); const refreshed = await readToolNames(owner.client, deadline.signal); owner.tools = refreshed; return refreshed; } catch (cause: unknown) { if (deadline.interruption === "timeout") - throw new AnalysisTimeoutError("jeb_mcp_catalog", timeoutMs, { - cause, - }); + throw new AnalysisTimeoutError( + "jeb_mcp_catalog", + discoveryTimeoutMs, + { cause }, + ); + if (deadline.interruption === "cancelled") + throw new AnalysisCancelledError("jeb_mcp_catalog", { cause }); throw cause; } finally { deadline.dispose(); @@ -257,11 +283,34 @@ export const createStreamableHttpJebMcpConnection: JebMcpConnectionFactory = ( const owner = state; if (owner === undefined || owner.tools === undefined) throw new AnalysisProtocolError("JEB MCP client failed to connect"); - const result = await owner.client.callTool( - { name, arguments: args }, - signal === undefined ? {} : { signal }, + const requestSignal = + signal === undefined + ? owner.signal.signal + : AbortSignal.any([signal, owner.signal.signal]); + const deadline = new ProviderStartupDeadline( + requestTimeoutMs, + requestSignal, ); - return decodeJebToolResult(result); + try { + const result = await owner.client.callTool( + { name, arguments: args }, + { timeout: deadline.remainingMs(), signal: deadline.signal }, + ); + return decodeJebToolResult(result); + } catch (cause: unknown) { + if (deadline.interruption === "timeout") + throw new AnalysisTimeoutError(name, requestTimeoutMs, { cause }); + if (requestSignal.aborted) + throw new AnalysisCancelledError(name, { cause }); + if ( + cause instanceof SdkError && + cause.code === SdkErrorCode.RequestTimeout + ) + throw new AnalysisTimeoutError(name, requestTimeoutMs, { cause }); + throw cause; + } finally { + deadline.dispose(); + } }, async close() { closed = true; diff --git a/src/jeb/JebProvider.test.ts b/src/jeb/JebProvider.test.ts index 83cee039dc..e8b46d7a66 100644 --- a/src/jeb/JebProvider.test.ts +++ b/src/jeb/JebProvider.test.ts @@ -1,7 +1,11 @@ import { describe, expect, it } from "vitest"; +import { createJebAnalysisProvider } from "../composition/jeb.js"; import { JebProvider } from "./JebProvider.js"; -import type { JebMcpConnection } from "./JebMcpConnection.js"; +import type { + JebMcpConnection, + JebMcpConnectionTimeouts, +} from "./JebMcpConnection.js"; import { AnalysisCapabilityUnavailableError, AnalysisInputError, @@ -42,6 +46,36 @@ const providerWith = (connection: JebMcpConnection) => }); describe("JebProvider live availability", () => { + it("passes its validated environment timeout to the production connection", async () => { + let configured: JebMcpConnectionTimeouts | undefined; + const provider = createJebAnalysisProvider( + { REA_JEB_MCP_TIMEOUT_MS: "900000" }, + (_endpoint, timeouts) => { + configured = timeouts; + return scriptedConnection({}); + }, + ); + expect(configured).toEqual({ requestTimeoutMs: 900_000 }); + await provider.close(); + }); + + it("rejects an invalid JEB timeout before constructing its MCP connection", async () => { + let constructed = false; + const provider = createJebAnalysisProvider( + { REA_JEB_MCP_TIMEOUT_MS: "0" }, + () => { + constructed = true; + return scriptedConnection({}); + }, + ); + expect(constructed).toBe(false); + await expect(provider.inspectAvailability()).resolves.toMatchObject({ + status: "unavailable", + diagnostics: { setting: "REA_JEB_MCP_TIMEOUT_MS" }, + }); + await provider.close(); + }); + it("reports only the operations advertised by the live engine", async () => { const connection = scriptedConnection({ get_client_information: { diff --git a/src/jeb/JebProvider.ts b/src/jeb/JebProvider.ts index 3f37c4be30..797060d597 100644 --- a/src/jeb/JebProvider.ts +++ b/src/jeb/JebProvider.ts @@ -39,6 +39,8 @@ export const JEB_PROVIDER_IDENTITY: ProviderIdentity = { export interface JebProviderOptions { /** Caller environment carrying REA_JEB_MCP_URL. */ readonly environment?: Readonly>; + /** Per-tools/call budget selected by REA_JEB_MCP_TIMEOUT_MS. */ + readonly requestTimeoutMs?: number; /** Injectable MCP boundary; defaults to the streamable HTTP client. */ readonly connectionFactory?: JebMcpConnectionFactory; } @@ -111,10 +113,14 @@ export class JebProvider implements JebAnalysisPort { const endpoint = parseJebMcpEndpoint(environment.REA_JEB_MCP_URL); if (!endpoint.ok) throw endpoint.error; this.#endpoint = endpoint.value; - this.#connection = ( - options.connectionFactory ?? - ((url: URL) => createStreamableHttpJebMcpConnection(url)) - )(this.#endpoint); + const createConnection = + options.connectionFactory ?? createStreamableHttpJebMcpConnection; + this.#connection = + options.requestTimeoutMs === undefined + ? createConnection(this.#endpoint) + : createConnection(this.#endpoint, { + requestTimeoutMs: options.requestTimeoutMs, + }); } async inspectAvailability( diff --git a/tests/boundary/providers/jebMcpCatalog.test.ts b/tests/boundary/providers/jebMcpCatalog.test.ts index e61f4bec13..2ba02f2a11 100644 --- a/tests/boundary/providers/jebMcpCatalog.test.ts +++ b/tests/boundary/providers/jebMcpCatalog.test.ts @@ -76,7 +76,7 @@ it("reads every JEB catalog page and refreshes the connected engine catalog", as throw new Error("Expected an ephemeral TCP listener"); const connection = createStreamableHttpJebMcpConnection( new URL(`http://127.0.0.1:${address.port}/mcp`), - 1_000, + { discoveryTimeoutMs: 1_000 }, ); try { expect(await connection.listTools()).toEqual([ @@ -109,7 +109,7 @@ it("reads every JEB catalog page and refreshes the connected engine catalog", as // Shared initial acquisition also owns a finite deadline and can recover. const acquiring = createStreamableHttpJebMcpConnection( new URL(`http://127.0.0.1:${address.port}/mcp`), - 1_000, + { discoveryTimeoutMs: 1_000 }, ); try { infinitePages = true;