mirror of
https://github.com/morluto/rea
synced 2026-10-27 22:45:19 -04:00
fix(ida): bound readiness requests by the remaining deadline
This commit is contained in:
@@ -131,7 +131,7 @@ Use a package version that actually includes this adapter; repository main can b
|
||||
{"name":"close_binary","arguments":{}}
|
||||
```
|
||||
|
||||
Opening an MCP target starts its selected provider session; headless opening waits for auto-analysis. Set the agent's MCP request deadline long enough for the target. The registration's optional `timeoutMs` defaults to 300000 for upstream requests. An upstream SDK request that reaches this deadline returns `provider_timeout` with its operation and configured budget. A freshly opened headless worker can briefly report `status: "busy"` from `server_health`; REA re-reads the health every 100 ms until the worker answers, the caller cancels, or `timeoutMs` has elapsed since the first read. Expiry of this polling deadline while the worker is still busy also produces a retryable `provider_timeout`; no further read is started once the deadline has passed. That error names the busy tool of the last observation, or `unknown` when the worker reported none or a value that is not tool-name-shaped. Cancellation during the wait produces `cancelled`. Any other unexpected health payload, including a busy payload with an invalid shape, remains `unreadable_output`. Cancellation drains the current upstream request before cleanup; it does not force-stop GUI analysis. An open request that times out without a completed response retains the private workspace because worker inventory alone cannot prove the request has stopped. `binary_session` reports actual tool availability. Repeated doctor/setup calls are unnecessary when the registration already works.
|
||||
Opening an MCP target starts its selected provider session; headless opening waits for auto-analysis. Set the agent's MCP request deadline long enough for the target. The registration's optional `timeoutMs` defaults to 300000 for upstream requests. An upstream SDK request that reaches this limit returns `provider_timeout` with its operation and actual request budget. A freshly opened headless worker can briefly report `status: "busy"` from `server_health`; REA re-reads health every 100 ms until the worker answers, the caller cancels, or `timeoutMs` has elapsed since the first read. Each health request is capped by the time remaining in that overall wait, so a stalled read cannot extend the readiness deadline by another full request timeout. Expiry while the worker is still busy produces a retryable `provider_timeout`; no further read is started after the deadline. That error names the busy tool from the last observation, or `unknown` when the worker reported none or a value that is not tool-name-shaped. If a health request itself consumes the remaining wait budget, the same last-busy timeout is returned with the stalled request's captured output. Cancellation during the wait produces `cancelled`. Any other unexpected health payload, including a busy payload with an invalid shape, remains `unreadable_output`. Cancellation drains the current upstream request before cleanup; it does not force-stop GUI analysis. An open request that times out without a completed response retains the private workspace because worker inventory alone cannot prove the request has stopped. `binary_session` reports actual tool availability. Repeated doctor/setup calls are unnecessary when the registration already works.
|
||||
|
||||
For guided registration, set `REA_IDA_MCP_CONFIG` before `rea setup --client <client>`. Setup includes that file reference in its reviewed plan and preserves it after approval; it does not copy upstream credentials into the agent configuration. `rea doctor --provider ida --json` validates the registration without starting IDA and explicitly leaves runtime compatibility and target binding for analysis.
|
||||
|
||||
|
||||
@@ -61,6 +61,34 @@ it("sleeps the fixed retry delay between busy reads and returns the settled answ
|
||||
expect(sleeps, "two fixed 100 ms pauses").toEqual([100, 100]);
|
||||
});
|
||||
|
||||
it("returns an expired readiness timeout without starting a health read", async () => {
|
||||
const timeoutMs = 10;
|
||||
const { connection, reads } = scripted([ok]);
|
||||
let clockReads = 0;
|
||||
const clock = {
|
||||
now: () => (clockReads++ === 0 ? 0 : 11),
|
||||
delay: async () => {
|
||||
throw new Error("An expired wait must not schedule a pause");
|
||||
},
|
||||
};
|
||||
|
||||
const failure = await readSettledHealth(connection, {
|
||||
database: "db",
|
||||
timeoutMs,
|
||||
clock,
|
||||
}).catch((cause: unknown) => cause);
|
||||
|
||||
expect(failure).toMatchObject({
|
||||
_tag: "AnalysisTimeoutError",
|
||||
operation: "server_health",
|
||||
timeoutMs,
|
||||
});
|
||||
expect(
|
||||
reads,
|
||||
"an expired budget never becomes an SDK timeout of zero",
|
||||
).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("clamps the last pause to the remaining budget and stops without another read at expiry", async () => {
|
||||
const { connection, reads } = scripted([busy("config_json_get")]);
|
||||
const { clock, sleeps } = virtualClock();
|
||||
|
||||
@@ -29,6 +29,11 @@ const isBusyStatus = (raw: JsonValue): boolean =>
|
||||
!Array.isArray(raw) &&
|
||||
raw["status"] === "busy";
|
||||
|
||||
interface LastBusyHealth {
|
||||
readonly raw: JsonValue;
|
||||
readonly operation: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Reads `server_health` until the worker stops reporting a transient busy
|
||||
* state. The wait ends when the worker answers, the caller aborts, or the
|
||||
@@ -53,22 +58,62 @@ export const readSettledHealth = async (
|
||||
): Promise<JsonValue> => {
|
||||
const started = clock.now();
|
||||
const remaining = (): number => timeoutMs - (clock.now() - started);
|
||||
let lastBusy: LastBusyHealth | undefined;
|
||||
for (;;) {
|
||||
const raw = await connection.call("server_health", { database });
|
||||
const budgetMs = remaining();
|
||||
if (budgetMs <= 0)
|
||||
throw lastBusy === undefined
|
||||
? new AnalysisTimeoutError("server_health", timeoutMs)
|
||||
: busyTimeout(lastBusy, timeoutMs);
|
||||
const requestTimeoutMs = Math.min(timeoutMs, Math.ceil(budgetMs));
|
||||
let raw: JsonValue;
|
||||
try {
|
||||
raw = await connection.call(
|
||||
"server_health",
|
||||
{ database },
|
||||
{ timeoutMs: requestTimeoutMs },
|
||||
);
|
||||
} catch (cause: unknown) {
|
||||
if (
|
||||
lastBusy !== undefined &&
|
||||
cause instanceof AnalysisTimeoutError &&
|
||||
cause.timeoutMs === requestTimeoutMs &&
|
||||
requestTimeoutMs < timeoutMs
|
||||
) {
|
||||
throw new AnalysisTimeoutError(lastBusy.operation, timeoutMs, {
|
||||
cause: lastBusy.raw,
|
||||
capturedOutput: cause.capturedOutput,
|
||||
diagnostics: {
|
||||
health_request_operation: cause.operation,
|
||||
health_request_timeout_ms: cause.timeoutMs,
|
||||
},
|
||||
});
|
||||
}
|
||||
throw cause;
|
||||
}
|
||||
const busy = busyHealthSchema.safeParse(raw);
|
||||
if (!busy.success) {
|
||||
if (isBusyStatus(raw)) throw busy.error;
|
||||
return raw;
|
||||
}
|
||||
const observedBusy = {
|
||||
raw,
|
||||
operation: `server_health: headless worker still busy in ${busyToolName(busy.data.busy_tool)}`,
|
||||
};
|
||||
lastBusy = observedBusy;
|
||||
const expired = (): AnalysisTimeoutError =>
|
||||
new AnalysisTimeoutError(
|
||||
`server_health: headless worker still busy in ${busyToolName(busy.data.busy_tool)}`,
|
||||
timeoutMs,
|
||||
{ cause: raw },
|
||||
);
|
||||
busyTimeout(observedBusy, timeoutMs);
|
||||
if (remaining() <= 0) throw expired();
|
||||
await clock.delay(Math.min(BUSY_HEALTH_RETRY_MS, remaining()), signal);
|
||||
signal?.throwIfAborted();
|
||||
if (remaining() <= 0) throw expired();
|
||||
}
|
||||
};
|
||||
|
||||
const busyTimeout = (
|
||||
lastBusy: LastBusyHealth,
|
||||
timeoutMs: number,
|
||||
): AnalysisTimeoutError =>
|
||||
new AnalysisTimeoutError(lastBusy.operation, timeoutMs, {
|
||||
cause: lastBusy.raw,
|
||||
});
|
||||
|
||||
@@ -7,6 +7,7 @@ import {
|
||||
} from "./IdaMcpConnection.js";
|
||||
import { createIdaDiagnosticRedactor } from "./IdaDiagnosticRedactor.js";
|
||||
import { AnalysisTimeoutError } from "../domain/analysisErrorCore.js";
|
||||
import { readSettledHealth } from "./IdaBusyHealth.js";
|
||||
|
||||
describe("upstream IDA tool result decoding", () => {
|
||||
it("leaves successful legacy, structured, and native evidence unchanged", () => {
|
||||
@@ -144,60 +145,88 @@ const failingHttpConnection = (
|
||||
);
|
||||
};
|
||||
|
||||
it("preserves the SDK request timeout as a typed IDA timeout", async () => {
|
||||
const timeoutMs = 40;
|
||||
const pendingCall = vi.fn();
|
||||
interface MockedIdaHttpRequest {
|
||||
readonly id?: string | number;
|
||||
readonly method: string;
|
||||
readonly params?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
const jsonRpcResponse = (value: unknown): Response =>
|
||||
new Response(JSON.stringify(value), {
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
|
||||
const stalledResponse = (signal: AbortSignal | null | undefined) =>
|
||||
new Promise<Response>((_resolve, reject) => {
|
||||
const aborted = () =>
|
||||
reject(new DOMException("The request was aborted", "AbortError"));
|
||||
if (signal?.aborted === true) {
|
||||
aborted();
|
||||
return;
|
||||
}
|
||||
signal?.addEventListener("abort", aborted, { once: true });
|
||||
});
|
||||
|
||||
const mockedHttpBoundary = ({
|
||||
serverName,
|
||||
tools,
|
||||
call,
|
||||
}: {
|
||||
readonly serverName: string;
|
||||
readonly tools: readonly {
|
||||
readonly name: string;
|
||||
readonly inputSchema: object;
|
||||
}[];
|
||||
readonly call: (
|
||||
request: MockedIdaHttpRequest,
|
||||
signal: AbortSignal | null | undefined,
|
||||
) => Response | Promise<Response>;
|
||||
}): void => {
|
||||
vi.stubGlobal(
|
||||
"fetch",
|
||||
vi.fn(async (_input: Parameters<typeof fetch>[0], init?: RequestInit) => {
|
||||
const message = JSON.parse(String(init?.body ?? "")) as {
|
||||
readonly id?: string | number;
|
||||
readonly method: string;
|
||||
readonly params?: Record<string, unknown>;
|
||||
};
|
||||
const json = (value: unknown) =>
|
||||
new Response(JSON.stringify(value), {
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
if (message.method === "initialize")
|
||||
return json({
|
||||
const request = JSON.parse(
|
||||
String(init?.body ?? ""),
|
||||
) as MockedIdaHttpRequest;
|
||||
if (request.method === "initialize")
|
||||
return jsonRpcResponse({
|
||||
jsonrpc: "2.0",
|
||||
id: message.id,
|
||||
id: request.id,
|
||||
result: {
|
||||
protocolVersion: message.params?.protocolVersion,
|
||||
protocolVersion: request.params?.protocolVersion,
|
||||
capabilities: { tools: {} },
|
||||
serverInfo: { name: "ida-timeout-fixture", version: "1" },
|
||||
serverInfo: { name: serverName, version: "1" },
|
||||
},
|
||||
});
|
||||
if (
|
||||
message.method === "notifications/initialized" ||
|
||||
message.method === "notifications/cancelled"
|
||||
request.method === "notifications/initialized" ||
|
||||
request.method === "notifications/cancelled"
|
||||
)
|
||||
return new Response(null, { status: 202 });
|
||||
if (message.method === "tools/list")
|
||||
return json({
|
||||
if (request.method === "tools/list")
|
||||
return jsonRpcResponse({
|
||||
jsonrpc: "2.0",
|
||||
id: message.id,
|
||||
result: { tools: [] },
|
||||
id: request.id,
|
||||
result: { tools },
|
||||
});
|
||||
if (message.method !== "tools/call")
|
||||
throw new Error(`Unexpected fixture request: ${message.method}`);
|
||||
pendingCall();
|
||||
return new Promise<Response>((_resolve, reject) => {
|
||||
const signal = init?.signal;
|
||||
if (signal?.aborted === true) {
|
||||
reject(signal.reason);
|
||||
return;
|
||||
}
|
||||
signal?.addEventListener(
|
||||
"abort",
|
||||
() =>
|
||||
reject(new DOMException("The request was aborted", "AbortError")),
|
||||
{ once: true },
|
||||
);
|
||||
});
|
||||
if (request.method === "tools/call") return call(request, init?.signal);
|
||||
throw new Error(`Unexpected fixture request: ${request.method}`);
|
||||
}),
|
||||
);
|
||||
};
|
||||
|
||||
it("preserves the SDK request timeout as a typed IDA timeout", async () => {
|
||||
const timeoutMs = 1000;
|
||||
const requestTimeoutMs = 40;
|
||||
const pendingCall = vi.fn();
|
||||
mockedHttpBoundary({
|
||||
serverName: "ida-timeout-fixture",
|
||||
tools: [],
|
||||
call: (_request, signal) => {
|
||||
pendingCall();
|
||||
return stalledResponse(signal);
|
||||
},
|
||||
});
|
||||
const connection = createIdaMcpConnection(
|
||||
{
|
||||
url: "http://127.0.0.1:12345/mcp",
|
||||
@@ -210,11 +239,14 @@ it("preserves the SDK request timeout as a typed IDA timeout", async () => {
|
||||
try {
|
||||
await expect(connection.connect()).resolves.toEqual([]);
|
||||
const failure = await connection
|
||||
.call("decompile", {})
|
||||
.call("decompile", {}, { timeoutMs: requestTimeoutMs })
|
||||
.catch((cause: unknown) => cause);
|
||||
expect(pendingCall).toHaveBeenCalledOnce();
|
||||
expect(failure).toBeInstanceOf(AnalysisTimeoutError);
|
||||
expect(failure).toMatchObject({ operation: "decompile", timeoutMs });
|
||||
expect(failure).toMatchObject({
|
||||
operation: "decompile",
|
||||
timeoutMs: requestTimeoutMs,
|
||||
});
|
||||
if (!(failure instanceof AnalysisTimeoutError))
|
||||
throw new Error("Expected a typed IDA request timeout");
|
||||
expect(failure.cause).toBeInstanceOf(Error);
|
||||
@@ -224,6 +256,75 @@ it("preserves the SDK request timeout as a typed IDA timeout", async () => {
|
||||
}
|
||||
});
|
||||
|
||||
it("bounds a stalled health poll by the remaining readiness deadline", async () => {
|
||||
const timeoutMs = 800;
|
||||
const pendingCall = vi.fn();
|
||||
let clockNow = 0;
|
||||
const clock = {
|
||||
now: () => clockNow,
|
||||
delay: async (ms: number) => {
|
||||
clockNow += ms;
|
||||
},
|
||||
};
|
||||
let pendingCalls = 0;
|
||||
const busy = {
|
||||
status: "busy",
|
||||
busy_tool: "config_json_get",
|
||||
busy_sec: 1,
|
||||
queued_calls: 0,
|
||||
};
|
||||
mockedHttpBoundary({
|
||||
serverName: "ida-health-deadline-fixture",
|
||||
tools: [{ name: "server_health", inputSchema: { type: "object" } }],
|
||||
call: (request, signal) => {
|
||||
pendingCalls += 1;
|
||||
if (pendingCalls === 1) {
|
||||
clockNow += 450;
|
||||
const result = {
|
||||
content: [{ type: "text", text: JSON.stringify(busy) }],
|
||||
structuredContent: busy,
|
||||
};
|
||||
return jsonRpcResponse({ jsonrpc: "2.0", id: request.id, result });
|
||||
}
|
||||
return stalledResponse(signal);
|
||||
},
|
||||
});
|
||||
const connection = createIdaMcpConnection(
|
||||
{
|
||||
url: "http://127.0.0.1:12345/mcp",
|
||||
headers: {},
|
||||
mode: "headless",
|
||||
timeoutMs,
|
||||
},
|
||||
{},
|
||||
);
|
||||
try {
|
||||
await connection.connect();
|
||||
const failure = await readSettledHealth(connection, {
|
||||
database: "fixture-db",
|
||||
timeoutMs,
|
||||
clock,
|
||||
}).catch((cause: unknown) => cause);
|
||||
|
||||
expect(pendingCalls).toBe(2);
|
||||
expect(failure).toBeInstanceOf(AnalysisTimeoutError);
|
||||
expect(failure).toMatchObject({
|
||||
operation: "server_health: headless worker still busy in config_json_get",
|
||||
timeoutMs,
|
||||
cause: busy,
|
||||
diagnostics: {
|
||||
health_request_operation: "server_health",
|
||||
health_request_timeout_ms: 250,
|
||||
},
|
||||
});
|
||||
if (!(failure instanceof AnalysisTimeoutError))
|
||||
throw new Error("Expected the readiness deadline to expire");
|
||||
expect(failure.capturedOutput).toBeDefined();
|
||||
} finally {
|
||||
await connection.close();
|
||||
}
|
||||
});
|
||||
|
||||
const inspectRetainedError = (error: unknown): string =>
|
||||
inspect(error, { depth: null, maxStringLength: null, breakLength: Infinity });
|
||||
|
||||
|
||||
@@ -35,6 +35,7 @@ export interface IdaMcpConnection {
|
||||
call(
|
||||
name: string,
|
||||
args: Readonly<Record<string, JsonValue>>,
|
||||
options?: { readonly timeoutMs?: number | undefined },
|
||||
): Promise<JsonValue>;
|
||||
readResource?(uri: string): Promise<JsonValue>;
|
||||
serverInfo(): JsonValue;
|
||||
@@ -349,6 +350,7 @@ export const createIdaMcpConnection = (
|
||||
const transportFailure = (
|
||||
cause: unknown,
|
||||
operation = "ida_mcp_response",
|
||||
requestTimeoutMs = config.timeoutMs,
|
||||
): never => {
|
||||
const bufferLimit =
|
||||
cause instanceof Error
|
||||
@@ -392,7 +394,7 @@ export const createIdaMcpConnection = (
|
||||
stderr_bytes: stderrBytes,
|
||||
};
|
||||
if (cause instanceof SdkError && cause.code === SdkErrorCode.RequestTimeout)
|
||||
throw new AnalysisTimeoutError(operation, config.timeoutMs, {
|
||||
throw new AnalysisTimeoutError(operation, requestTimeoutMs, {
|
||||
cause: failure,
|
||||
capturedOutput,
|
||||
});
|
||||
@@ -458,11 +460,17 @@ export const createIdaMcpConnection = (
|
||||
.catch((cause: unknown) => transportFailure(cause, "ida_mcp_catalog"));
|
||||
return catalog.tools.map(({ name }) => name);
|
||||
},
|
||||
async call(name, args) {
|
||||
async call(name, args, options) {
|
||||
const requestTimeoutMs = Math.min(
|
||||
config.timeoutMs,
|
||||
options?.timeoutMs ?? config.timeoutMs,
|
||||
);
|
||||
return decodeIdaToolResult(
|
||||
await client
|
||||
.callTool({ name, arguments: args }, { timeout: config.timeoutMs })
|
||||
.catch((cause: unknown) => transportFailure(cause, name)),
|
||||
.callTool({ name, arguments: args }, { timeout: requestTimeoutMs })
|
||||
.catch((cause: unknown) =>
|
||||
transportFailure(cause, name, requestTimeoutMs),
|
||||
),
|
||||
config.protocol,
|
||||
redactor,
|
||||
);
|
||||
|
||||
Reference in New Issue
Block a user