mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
fix(server): keep tool questions and classify every remote call failure
- A deliberate stop of a remote tool call (elicitation_required, provider_interaction_required, oauth_insufficient_scope) keeps its own reason when the call deadline passes while it is recorded. Before, the deadline check came first and turned a tool question into tool_timeout. - An error that is already a ToolGatewayHttpError now gets a failure kind: tool_timeout is invocation_timeout, a malformed, invalid or oversized response is protocol_failure, and remote_http_* and mcp_remote_fetch_failed are transport_failure. - A non-success response body is read as a stream, so a body that fails partway records the bytes that arrived. The reader also stops at the size limit and cancels the body, as the tools/call reader does. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
717e6c77cd
commit
00150e59c1
2 files changed
+134
-5
No files matched your search
@@ -4651,6 +4651,76 @@ rl.on("line", (line) => {
|
||||
});
|
||||
}
|
||||
|
||||
async function startRemoteCallWithControlledBody(respond: (body: ReadableStream<Uint8Array>) => Response, waitForRead = true) {
|
||||
const company = await createCompany(db);
|
||||
const agent = await createAgent(db, company.id);
|
||||
const { run } = await createIssueAndRun(db, company.id, agent.id);
|
||||
await createRemoteMcpTool(db, company.id);
|
||||
await allowAllToolsForAgent(db, company.id, agent.id);
|
||||
const reading = Promise.withResolvers<void>();
|
||||
let bodyController!: ReadableStreamDefaultController<Uint8Array>;
|
||||
let requestId!: string;
|
||||
const gateway = createTestToolGatewayService(db, {
|
||||
remoteHttpRequest: async (_url, init) => {
|
||||
requestId = JSON.parse(String(init.body)).id;
|
||||
return respond(new ReadableStream<Uint8Array>({
|
||||
start(controller) { bodyController = controller; },
|
||||
pull() { reading.resolve(); },
|
||||
}, { highWaterMark: 0 }));
|
||||
},
|
||||
});
|
||||
const session = await gateway.createSession({ companyId: company.id, agentId: agent.id, runId: run.id });
|
||||
const tool = (await gateway.listToolsForSession(session.token)).find((entry) => entry.providerType === "mcp_remote_http")!;
|
||||
vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] });
|
||||
const call = gateway.executeTool({ sessionToken: session.token, tool: tool.name, parameters: { key: "a", value: "b" }, timeoutMs: 1500 })
|
||||
.then((value) => ({ value, error: null }), (error: unknown) => ({ value: null, error }));
|
||||
if (waitForRead) await reading.promise;
|
||||
const failedEvent = async () => (await db.select().from(toolCallEvents).where(eq(toolCallEvents.eventType, "call_failed")))[0];
|
||||
return { call, failedEvent, body: () => bodyController, requestId: () => requestId };
|
||||
}
|
||||
|
||||
it("keeps a tool question when the call deadline passes while the question is recorded", async () => {
|
||||
const remote = await startRemoteCallWithControlledBody((body) => new Response(body, { headers: { "content-type": "text/event-stream" } }));
|
||||
try {
|
||||
await vi.advanceTimersByTimeAsync(1500);
|
||||
const question = { jsonrpc: "2.0", id: "question-1", method: "elicitation/create", params: { message: "Which folder?", requestedSchema: { type: "object" } } };
|
||||
remote.body().enqueue(new TextEncoder().encode(`data: ${JSON.stringify(question)}\n\n`));
|
||||
expectGatewayError((await remote.call).error, 409, "elicitation_required");
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
remote.body().error(new Error("fixture cleanup"));
|
||||
await remote.call;
|
||||
}
|
||||
});
|
||||
|
||||
it("records a transport failure kind when the remote hostname does not resolve", async () => {
|
||||
const remote = await startRemoteCallWithControlledBody(() => {
|
||||
throw new ToolGatewayHttpError(502, "Remote MCP connection hostname did not resolve", "remote_http_dns_failed");
|
||||
}, false);
|
||||
try {
|
||||
expectGatewayError((await remote.call).error, 502, "remote_http_dns_failed");
|
||||
expect((await remote.failedEvent()).metadata).toMatchObject({ execution: { failureKind: "transport_failure" } });
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("records the bytes of an error response body that fails partway", async () => {
|
||||
const remote = await startRemoteCallWithControlledBody((body) => new Response(body, { status: 503, headers: { "content-type": "text/plain" } }));
|
||||
try {
|
||||
const partial = "upstream overloaded, partial";
|
||||
remote.body().enqueue(new TextEncoder().encode(partial));
|
||||
remote.body().error(Object.assign(new Error("aborted"), { code: "ECONNRESET" }));
|
||||
expectGatewayError((await remote.call).error, 502, "mcp_remote_fetch_failed");
|
||||
expect((await remote.failedEvent()).metadata).toMatchObject({ execution: {
|
||||
failureKind: "transport_failure",
|
||||
response: { httpStatus: 503, bodySizeBytes: Buffer.byteLength(partial) },
|
||||
} });
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
const remoteFailureCases = [
|
||||
{
|
||||
name: "HTTP status",
|
||||
|
||||
@@ -507,6 +507,27 @@ type RemoteHttpErrorEvidence = {
|
||||
cause: RemoteHttpErrorEvidence | null;
|
||||
};
|
||||
|
||||
// Deliberate stops of a remote tool call. They are not failures, and a deadline
|
||||
// that passes while they are recorded must not turn them into a timeout.
|
||||
const REMOTE_TOOL_CALL_WORKFLOW_REASON_CODES = new Set([
|
||||
"elicitation_required",
|
||||
"provider_interaction_required",
|
||||
"oauth_insufficient_scope",
|
||||
]);
|
||||
|
||||
const REMOTE_TOOL_CALL_PROTOCOL_REASON_CODES = new Set([
|
||||
"mcp_remote_invalid_json",
|
||||
"remote_mcp_malformed_response",
|
||||
"mcp_remote_response_too_large",
|
||||
]);
|
||||
|
||||
function remoteToolCallFailureKind(reasonCode: string): RemoteHttpExecutionAudit["failureKind"] {
|
||||
if (reasonCode === "tool_timeout") return "invocation_timeout";
|
||||
if (REMOTE_TOOL_CALL_PROTOCOL_REASON_CODES.has(reasonCode)) return "protocol_failure";
|
||||
if (reasonCode === "mcp_remote_fetch_failed" || reasonCode.startsWith("remote_http_")) return "transport_failure";
|
||||
return undefined;
|
||||
}
|
||||
|
||||
function remoteHttpErrorEvidence(error: unknown, seen = new Set<unknown>()): RemoteHttpErrorEvidence {
|
||||
if (seen.has(error)) return { name: null, message: "[Circular cause]", code: null, stack: null, cause: null };
|
||||
const record = asRecord(error);
|
||||
@@ -5942,6 +5963,7 @@ export function createToolGatewayService(
|
||||
|
||||
async function readBoundedRemoteResponse(
|
||||
response: Response,
|
||||
onBytes?: (bytes: number) => void,
|
||||
): Promise<string> {
|
||||
const contentLength = response.headers.get("content-length");
|
||||
if (
|
||||
@@ -5950,11 +5972,36 @@ export function createToolGatewayService(
|
||||
) {
|
||||
throw responseTooLargeError();
|
||||
}
|
||||
const body = await response.text();
|
||||
if (Buffer.byteLength(body, "utf8") > MAX_REMOTE_MCP_RESPONSE_BYTES) {
|
||||
throw responseTooLargeError();
|
||||
const reader = response.body?.getReader();
|
||||
// Injected HTTP transports can expose a buffered text response rather than a
|
||||
// Web ReadableStream. Keep the same size check for that form.
|
||||
if (!reader) {
|
||||
const body = await response.text();
|
||||
onBytes?.(Buffer.byteLength(body, "utf8"));
|
||||
if (Buffer.byteLength(body, "utf8") > MAX_REMOTE_MCP_RESPONSE_BYTES) {
|
||||
throw responseTooLargeError();
|
||||
}
|
||||
return body;
|
||||
}
|
||||
// Count each chunk as it arrives, so a body that fails partway still
|
||||
// reports the bytes that were read.
|
||||
const decoder = new TextDecoder();
|
||||
let body = "";
|
||||
let bytes = 0;
|
||||
try {
|
||||
while (true) {
|
||||
const { value, done } = await reader.read();
|
||||
if (done) break;
|
||||
bytes += value.byteLength;
|
||||
onBytes?.(bytes);
|
||||
if (bytes > MAX_REMOTE_MCP_RESPONSE_BYTES) throw responseTooLargeError();
|
||||
body += decoder.decode(value, { stream: true });
|
||||
}
|
||||
return body + decoder.decode();
|
||||
} finally {
|
||||
await reader.cancel().catch(() => undefined);
|
||||
reader.releaseLock();
|
||||
}
|
||||
return body;
|
||||
}
|
||||
|
||||
function malformedRemoteMcpResponse(): ToolGatewayHttpError {
|
||||
@@ -6552,7 +6599,9 @@ export function createToolGatewayService(
|
||||
if (request) await requestElicitationForRecordedToolCall({ session, tool, invocationId, request });
|
||||
},
|
||||
}))
|
||||
: await readBoundedRemoteResponse(response);
|
||||
: await readBoundedRemoteResponse(response, (bytes) => {
|
||||
execution.response!.bodySizeBytes = bytes;
|
||||
});
|
||||
execution.response.bodySizeBytes = Buffer.byteLength(body, "utf8");
|
||||
if (isInsufficientConnectionScope(response, body)) {
|
||||
await markCallHealth("degraded", INSUFFICIENT_CONNECTION_SCOPE_MESSAGE);
|
||||
@@ -6657,6 +6706,14 @@ export function createToolGatewayService(
|
||||
execution.error = remoteHttpErrorEvidence(error);
|
||||
execution.deadline = { timeoutMs: ms, aborted: controller.signal.aborted };
|
||||
if (execution.response && execution.stream) execution.response.bodySizeBytes = execution.stream.bytesReceived;
|
||||
// A deliberate workflow stop, such as a tool question for the user, keeps
|
||||
// its own reason even when the deadline passes while it is recorded.
|
||||
if (error instanceof ToolGatewayHttpError && REMOTE_TOOL_CALL_WORKFLOW_REASON_CODES.has(error.reasonCode)) {
|
||||
throw new ToolGatewayHttpError(error.status, error.message, error.reasonCode, {
|
||||
...error.details,
|
||||
execution: error.details.execution ?? execution,
|
||||
});
|
||||
}
|
||||
if (controller.signal.aborted || (error instanceof Error && error.name === "AbortError")) {
|
||||
if (mcpSession) forgetMcpHttpSession(mcpSession);
|
||||
execution.failureKind = "invocation_timeout";
|
||||
@@ -6683,6 +6740,8 @@ export function createToolGatewayService(
|
||||
throw new ToolGatewayHttpError(error.status, error.message, error.code, { connectionId: connection.id, catalogEntryId: entry.id, execution });
|
||||
}
|
||||
if (error instanceof ToolGatewayHttpError) {
|
||||
const failureKind = remoteToolCallFailureKind(error.reasonCode);
|
||||
if (failureKind) execution.failureKind = failureKind;
|
||||
throw new ToolGatewayHttpError(
|
||||
error.status,
|
||||
error.message,
|
||||
|
||||
Reference in new issue
Block a user