fix(tools): handle pagination failures and propagate refreshed Composio session
- Throw HttpError on tools/list pagination errors (HTTP 4xx/5xx or JSON-RPC error) instead of silently returning truncated tool lists - Throw HttpError if a cyclic pagination cursor is returned or if nextCursor remains after reaching maxPages - Update endpoint URL and activeHeaders upon Composio 401 session refresh so paginated page requests use refreshed credentials - Add tests covering later-page errors, limit exhaustion, cyclic cursors, and refreshed Composio sessions Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
parent
ca6fecedee
commit
71714cc349
|
|
@ -4224,6 +4224,160 @@ describeEmbeddedPostgres("tool access service", () => {
|
|||
]));
|
||||
});
|
||||
|
||||
it("fails catalog refresh if remote MCP tools/list pagination encounters an HTTP error on later pages", async () => {
|
||||
const company = await createCompany(db);
|
||||
const service = createTestToolAccessService(db);
|
||||
|
||||
vi.stubGlobal("fetch", async (_input: RequestInfo | URL, init?: RequestInit) => {
|
||||
const payload = JSON.parse(String(init?.body ?? "{}"));
|
||||
if (payload.method === "tools/list") {
|
||||
if (!payload.params?.cursor) {
|
||||
return mcpHttpResponse({
|
||||
jsonrpc: "2.0",
|
||||
id: payload.id,
|
||||
result: {
|
||||
tools: [{ name: "tool_page_1", annotations: { readOnlyHint: true } }],
|
||||
nextCursor: "cursor-token-page-2",
|
||||
},
|
||||
});
|
||||
}
|
||||
return new Response("Internal Server Error", { status: 500, statusText: "Internal Server Error" });
|
||||
}
|
||||
return mcpHttpResponse({ jsonrpc: "2.0", id: payload.id, result: { tools: [] } });
|
||||
});
|
||||
|
||||
await expect(service.connectGalleryApp(company.id, {
|
||||
link: "https://error-paginated.example/mcp",
|
||||
name: "Error Paginated MCP",
|
||||
}, { actorType: "user", actorId: "board" })).rejects.toMatchObject({
|
||||
status: 502,
|
||||
});
|
||||
});
|
||||
|
||||
it("fails catalog refresh if remote MCP tools/list pagination exceeds the maximum page limit while nextCursor remains", async () => {
|
||||
const company = await createCompany(db);
|
||||
const service = createTestToolAccessService(db);
|
||||
let page = 0;
|
||||
|
||||
vi.stubGlobal("fetch", async (_input: RequestInfo | URL, init?: RequestInit) => {
|
||||
const payload = JSON.parse(String(init?.body ?? "{}"));
|
||||
if (payload.method === "tools/list") {
|
||||
page++;
|
||||
return mcpHttpResponse({
|
||||
jsonrpc: "2.0",
|
||||
id: payload.id,
|
||||
result: {
|
||||
tools: [{ name: `tool_page_${page}`, annotations: { readOnlyHint: true } }],
|
||||
nextCursor: `cursor-token-page-${page + 1}`,
|
||||
},
|
||||
});
|
||||
}
|
||||
return mcpHttpResponse({ jsonrpc: "2.0", id: payload.id, result: { tools: [] } });
|
||||
});
|
||||
|
||||
await expect(service.connectGalleryApp(company.id, {
|
||||
link: "https://infinite-paginated.example/mcp",
|
||||
name: "Infinite Paginated MCP",
|
||||
}, { actorType: "user", actorId: "board" })).rejects.toMatchObject({
|
||||
status: 502,
|
||||
});
|
||||
});
|
||||
|
||||
it("fails catalog refresh if remote MCP tools/list returns cyclic pagination cursor", async () => {
|
||||
const company = await createCompany(db);
|
||||
const service = createTestToolAccessService(db);
|
||||
|
||||
vi.stubGlobal("fetch", async (_input: RequestInfo | URL, init?: RequestInit) => {
|
||||
const payload = JSON.parse(String(init?.body ?? "{}"));
|
||||
if (payload.method === "tools/list") {
|
||||
return mcpHttpResponse({
|
||||
jsonrpc: "2.0",
|
||||
id: payload.id,
|
||||
result: {
|
||||
tools: [{ name: "tool_loop", annotations: { readOnlyHint: true } }],
|
||||
nextCursor: "same-cursor-cycle",
|
||||
},
|
||||
});
|
||||
}
|
||||
return mcpHttpResponse({ jsonrpc: "2.0", id: payload.id, result: { tools: [] } });
|
||||
});
|
||||
|
||||
await expect(service.connectGalleryApp(company.id, {
|
||||
link: "https://cyclic-paginated.example/mcp",
|
||||
name: "Cyclic Paginated MCP",
|
||||
}, { actorType: "user", actorId: "board" })).rejects.toMatchObject({
|
||||
status: 502,
|
||||
});
|
||||
});
|
||||
|
||||
it("uses refreshed Composio endpoint and headers on subsequent paginated pages after initial 401", async () => {
|
||||
const company = await createCompany(db);
|
||||
const { child } = await createComposioParentAndChild(db, company.id);
|
||||
let sessionCount = 0;
|
||||
const client = fakeComposioClient(() => "ACTIVE");
|
||||
client.createSession = vi.fn(async () => {
|
||||
sessionCount++;
|
||||
return {
|
||||
session_id: `session-${sessionCount}`,
|
||||
mcp: {
|
||||
url: `https://composio.test/mcp-${sessionCount}`,
|
||||
headers: { "x-composio-token": `token-${sessionCount}` },
|
||||
},
|
||||
};
|
||||
});
|
||||
const service = createTestToolAccessService(db, {
|
||||
composioClientFactory: () => client,
|
||||
});
|
||||
|
||||
const requests: Array<{ url: string; token: string | null; cursor: unknown }> = [];
|
||||
vi.stubGlobal("fetch", async (input: RequestInfo | URL, init?: RequestInit) => {
|
||||
const url = String(input);
|
||||
const headers = new Headers(init?.headers);
|
||||
const payload = JSON.parse(String(init?.body ?? "{}"));
|
||||
requests.push({
|
||||
url,
|
||||
token: headers.get("x-composio-token"),
|
||||
cursor: payload.params?.cursor,
|
||||
});
|
||||
|
||||
if (url === "https://composio.test/mcp-1") {
|
||||
return new Response("Unauthorized", { status: 401, statusText: "Unauthorized" });
|
||||
}
|
||||
if (url === "https://composio.test/mcp-2") {
|
||||
if (!payload.params?.cursor) {
|
||||
return mcpHttpResponse({
|
||||
jsonrpc: "2.0",
|
||||
id: payload.id,
|
||||
result: {
|
||||
tools: [{ name: "composio_tool_1", annotations: { readOnlyHint: true } }],
|
||||
nextCursor: "cursor-page-2",
|
||||
},
|
||||
});
|
||||
}
|
||||
if (payload.params.cursor === "cursor-page-2") {
|
||||
return mcpHttpResponse({
|
||||
jsonrpc: "2.0",
|
||||
id: payload.id,
|
||||
result: {
|
||||
tools: [{ name: "composio_tool_2", annotations: { readOnlyHint: true } }],
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
return mcpHttpResponse({ jsonrpc: "2.0", id: payload.id, result: { tools: [] } });
|
||||
});
|
||||
|
||||
const refresh = await service.refreshCatalog(child.id);
|
||||
expect(requests).toEqual([
|
||||
{ url: "https://composio.test/mcp-1", token: "token-1", cursor: undefined },
|
||||
{ url: "https://composio.test/mcp-2", token: "token-2", cursor: undefined },
|
||||
{ url: "https://composio.test/mcp-2", token: "token-2", cursor: "cursor-page-2" },
|
||||
]);
|
||||
expect(refresh.catalog.map((entry) => entry.toolName)).toEqual(
|
||||
expect.arrayContaining(["composio_tool_1", "composio_tool_2"]),
|
||||
);
|
||||
});
|
||||
|
||||
it("serves persisted MCP actions until the cache expires and then refreshes them", async () => {
|
||||
const company = await createCompany(db);
|
||||
let currentTime = new Date("2026-08-20T12:00:00.000Z");
|
||||
|
|
|
|||
|
|
@ -4969,7 +4969,7 @@ export function toolAccessService(db: Db, options: ToolAccessServiceOptions = {}
|
|||
let headers = composioSession?.headers
|
||||
?? credentialHeaders
|
||||
?? { ...projectedConnectionHeaders(connection), ...await resolveCredentialHeaders(connection, actor) };
|
||||
const endpoint = composioSession?.url ?? await resolvedRemoteEndpoint(connection, actor);
|
||||
let endpoint = composioSession?.url ?? await resolvedRemoteEndpoint(connection, actor);
|
||||
// Pinned to the address the guard approved: `config.url` is operator-supplied,
|
||||
// so a second DNS resolution here would reopen the rebinding window that
|
||||
const sendRemote = (init: RequestInit) => requestRemoteHttpEndpoint(new URL(endpoint), init);
|
||||
|
|
@ -5032,9 +5032,12 @@ export function toolAccessService(db: Db, options: ToolAccessServiceOptions = {}
|
|||
}
|
||||
if (response.status === 401 && composioChild) {
|
||||
const refreshed = await composioSessions.ensureSession(connection.id, { force: true });
|
||||
response = await requestRemoteHttpEndpoint(new URL(refreshed.url), {
|
||||
endpoint = refreshed.url;
|
||||
headers = refreshed.headers;
|
||||
activeHeaders = refreshed.headers;
|
||||
response = await sendRemote({
|
||||
method: "POST",
|
||||
headers: mcpHttpRequestHeaders(refreshed.headers),
|
||||
headers: mcpHttpRequestHeaders(activeHeaders),
|
||||
body: JSON.stringify({
|
||||
jsonrpc: "2.0",
|
||||
id: "paperclip-catalog-refresh-retry",
|
||||
|
|
@ -5137,17 +5140,43 @@ export function toolAccessService(db: Db, options: ToolAccessServiceOptions = {}
|
|||
let nextCursor = typeof result.nextCursor === "string" && result.nextCursor.trim() ? result.nextCursor.trim() : null;
|
||||
let pageCount = 1;
|
||||
const maxPages = 50;
|
||||
const seenCursors = new Set<string>();
|
||||
|
||||
while (nextCursor && pageCount < maxPages) {
|
||||
seenCursors.add(nextCursor);
|
||||
pageCount++;
|
||||
const pageResponse = await sendToolsList(activeHeaders, nextCursor);
|
||||
if (!pageResponse.ok) break;
|
||||
if (!pageResponse.ok) {
|
||||
throw new HttpError(502, `Remote app returned HTTP ${pageResponse.status} during tools/list pagination`, {
|
||||
status: pageResponse.status,
|
||||
});
|
||||
}
|
||||
const pagePayload = parseMcpHttpResponseBody(await pageResponse.text(), pageResponse.headers.get("content-type"));
|
||||
result = asRecord(asRecord(pagePayload).result);
|
||||
payloadTools = asRecord(pagePayload).tools;
|
||||
const pageRecord = asRecord(pagePayload);
|
||||
if (pageRecord.error !== undefined) {
|
||||
const errorDetails = asRecord(pageRecord.error);
|
||||
const errorMessage = typeof errorDetails.message === "string" ? errorDetails.message : `code ${errorDetails.code ?? "unknown"}`;
|
||||
throw new HttpError(502, `Remote app returned an error during tools/list pagination: ${errorMessage}`, {
|
||||
code: "mcp_pagination_failed",
|
||||
});
|
||||
}
|
||||
result = asRecord(pageRecord.result);
|
||||
payloadTools = pageRecord.tools;
|
||||
const pageTools: unknown[] = Array.isArray(result.tools) ? result.tools : Array.isArray(payloadTools) ? payloadTools : [];
|
||||
tools.push(...pageTools);
|
||||
nextCursor = typeof result.nextCursor === "string" && result.nextCursor.trim() ? result.nextCursor.trim() : null;
|
||||
const candidateCursor = typeof result.nextCursor === "string" && result.nextCursor.trim() ? result.nextCursor.trim() : null;
|
||||
if (candidateCursor && seenCursors.has(candidateCursor)) {
|
||||
throw new HttpError(502, "Remote app returned cyclic pagination cursor during tools/list", {
|
||||
code: "pagination_cycle_detected",
|
||||
});
|
||||
}
|
||||
nextCursor = candidateCursor;
|
||||
}
|
||||
|
||||
if (nextCursor) {
|
||||
throw new HttpError(502, `Remote app tools/list pagination stopped before all tools were discovered (exceeded limit of ${maxPages} pages)`, {
|
||||
code: "pagination_limit_exceeded",
|
||||
});
|
||||
}
|
||||
|
||||
return tools.map((tool) => normalizeToolDescriptor(tool)).filter((tool): tool is McpToolDescriptor => Boolean(tool));
|
||||
|
|
|
|||
Loading…
Reference in New Issue