feat(apps): add experimental MCP aggregator connectors (#13755)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Its tool gateway applies company access rules and approval controls
to connected apps.
> - MCP aggregators expose many apps through one provider endpoint.
> - Each aggregator needs its own credential, catalog, grants, and
lifecycle in Paperclip.
> - This pull request adds independent Zapier, Arcade, Composio Connect,
and Executor setup with a common Access → Connect layout.
> - A default-off MCP aggregators flag lets operators opt in while we
complete provider acceptance tests.
> - Agents use the normal Paperclip permissions, Test screen, and
gateway after setup.

## Linked Issues or Issue Description

**Subsystem affected**

Apps, connection setup, shared contracts, and the remote MCP gateway.

**Problem or motivation**

Aggregator endpoints need clear provider setup and correct MCP sessions.
Generic setup does not explain each provider's authentication or broad
execution tools. Provider approval must preserve the original execution
instead of replaying a write.

**Proposed solution**

Add four separate connectors behind Settings → Experimental → MCP
aggregators. Start with human and agent access, then connect the
endpoint and read its tools. Enable tools by default. Use the existing
Permissions and Test screens after setup. Keep legacy Composio API-key
and child connections intact.

**Alternatives considered**

A shared connection for all providers would mix credentials and access
rules. Separate provider-specific permission and test screens would
duplicate existing controls. Vercel Connect is outside this change.

**Roadmap alignment**

Extends the existing MCP Tool Gateway & Apps capability and the
Connected Apps roadmap area. This work was requested and reviewed by the
maintainer.

Related work: #11894, #12630, #12632, #12634, and #12906 concern the
legacy Composio broker. #13102 also covers remote MCP pagination. This
change preserves the broker path and adds initialized sessions, response
matching, and provider resume handling alongside pagination.

## What Changed

- Add branded setup and interactive Storybooks for Zapier, Arcade,
Composio Connect, and Executor. Use the existing access controls and
normal action tests. Do not request a connection name or action choices
during setup.
- Add the default-off `enableMcpAggregators` flag to settings, managed
feature metadata, the catalog, and setup guards. Hidden connections keep
running. Legacy Composio connections remain unchanged.
- Reuse the vault, grants, policy, and catalog models. Support OAuth
discovery, bearer tokens, custom headers, and credential-bearing URLs.
Add no database tables or migrations.
- Initialize and retain Streamable HTTP sessions by connection and
effective credentials. Read paginated catalogs and match streaming
responses to request IDs.
- Classify unfamiliar aggregator tools as writes despite upstream
read-only hints; only exact reviewed read capabilities enter the
read-only allowlist. Legacy Composio child behavior is preserved.
- Preserve provider authorization links and execution IDs. Support
Executor approve/resume, decline, and cancel without automatic replay of
uncertain writes.
- Preserve Off and Ask first choices during refresh and reconnect. Allow
new tools and retire removed tools. Keep agent access updates atomic and
preserve an empty agent selection.
- Document connector UX rules, provider branding sources, and live
acceptance results.
- Stabilize the existing Sentry release fixture after its repeated CI
failure by reusing one module mock; production Sentry behavior is
unchanged.

## Verification

- Final head `d11781970`: [CI
run](https://github.com/paperclipai/paperclip/actions/runs/35633534900)
passed, including broad typecheck, test shards, build, and E2E. All 54
checks pass; 2 optional checks are skipped. Greptile is 5/5, Security
Scan passes, and all review threads are resolved.

- Passed 27 focused connector Vitest checks and 18 connector-only
Storybook browser checks before the flag change. All 85 stories rendered
at desktop and narrow widths.
- Passed 5 connector lifecycle/server checks and 7 selected flag checks
after adding the flag. The latter cover settings, managed defaults,
cached catalog visibility, and all four setup routes.
- Review fixes passed 13 risk/handoff/lifecycle checks, dedicated
session-expiration and transport regressions, 13 selected
connector/gateway CI cases, and 10 selected setup/reconnect UI cases. A
real Composio connection-list call also succeeded through the refreshed
UI on `9ab115f71`.
- UI and server TypeScript checks passed. UI build, Storybook build,
token gates, and diff whitespace checks passed during implementation.
- Real browser and real Paperclip agent tests passed for Arcade,
Composio, and Executor. Tested action permissions, denied agent access,
reconnect, disconnect, and isolation. Tested Arcade catalog
additions/removal and Executor provider approve/resume, decline, and
cancel.
- Zapier live acceptance is incomplete. Its dedicated provider server is
configured, but its credential-copy dialog returned an empty clipboard
through browser automation. No live Zapier action is claimed.
- The three isolated Sentry release cases pass after the CI fixture fix.
- Local verification is deliberately narrow at the maintainer's request.
The full local suite, recursive typecheck, and repository-wide build
were not run. CI provides the broader checks.

## Risks

- Shared MCP transport changes affect other remote MCP servers. Protocol
fixtures cover initialized sessions, streaming response matching,
pagination, and isolation.
- Broad execution tools remain broad permissions. The provider governs
actions inside those tools.
- Provider handoff links are retained briefly in memory. After a server
restart, a one-time link may require reopening the provider dashboard.
Paperclip does not replay the original call.
- Zapier remains unproven live. Custom-header imports and self-hosted
endpoints have fixture coverage rather than a separate live account for
every variant.
- Turning the experimental flag off hides setup; it does not revoke
existing credentials or stop existing connections.

## Model Used

OpenAI GPT-6 through Codex, with reasoning, repository tools, shell
execution, and browser automation. The exact runtime model ID and
context-window size are not exposed in this session. A separate
Anthropic-backed Paperclip agent performed live gateway acceptance
tasks.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>

server/src/__tests__/generic-mcp-connection.test.ts

@@ -39,6 +39,7 @@ import {
   startEmbeddedPostgresTestDatabase,
 } from "./helpers/embedded-postgres.js";
 import { toolAccessService } from "../services/tool-access.js";
+import { instanceSettingsService } from "../services/instance-settings.js";
 import { ComposioApiError, type ComposioClient } from "../services/composio.js";
 import { createComposioSessionManager } from "../services/composio-session-manager.js";
 import { toolAccessPolicyService } from "../services/tool-access-policy.js";
@@ -468,6 +469,7 @@ describeEmbeddedPostgres("generic remote MCP connections", () => {
   });
 
   it("keeps a generated Zapier URL attached to the curated Zapier identity", async () => {
+    await instanceSettingsService(db).updateExperimental({ enableMcpAggregators: true });
     const secretUrl = "https://mcp.zapier.com/api/v1/connect?token=zapier-secret";
     const publicUrl = "https://mcp.zapier.com/api/v1/connect";
     const company = await createCompany(db);
@@ -477,10 +479,12 @@ describeEmbeddedPostgres("generic remote MCP connections", () => {
       remoteHttpEndpointLookup: async () => [{ address: "8.8.8.8", family: 4 }],
       remoteHttpRequest: async (url, init) => {
         if (url === secretUrl && (init.method ?? "GET").toUpperCase() === "POST") {
+          const body = JSON.parse(String(init.body));
+          if (body.method === "notifications/initialized") return new Response(null, { status: 202 });
           return jsonResponse({
             jsonrpc: "2.0",
-            id: "paperclip-catalog-refresh",
-            result: { tools: FIXTURE_TOOLS },
+            id: body.id,
+            result: body.method === "initialize" ? { protocolVersion: "2025-06-18" } : { tools: FIXTURE_TOOLS },
           });
         }
         return jsonResponse({ error: "not_found" }, 404);

server/src/__tests__/instance-settings-service.test.ts

@@ -43,6 +43,7 @@ describe("instance settings service", () => {
       enableApps: true,
       enableAgentChat: false,
       enableChatConnectors: false,
+      enableMcpAggregators: false,
       enableConferenceRoomChat: false,
       enableClassicTaskInterface: false,
       enableExternalObjects: false,

server/src/__tests__/remote-mcp-connectors.test.ts

@@ -0,0 +1,251 @@
+import { randomUUID } from "node:crypto";
+import { mkdtemp, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+import { agents, heartbeatRuns, issues, toolCatalogEntries, toolConnectionInstalls, toolInvocations, companies, companyMemberships, createDb, toolConnections, toolPolicies, toolProfileEntries } from "@paperclipai/db";
+import { eq } from "drizzle-orm";
+import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
+import { toolAccessService } from "../services/tool-access.js";
+import { createToolGatewayService } from "../services/tool-gateway.js";
+import { instanceSettingsService, normalizeExperimentalSettings } from "../services/instance-settings.js";
+import express from "express";
+import request from "supertest";
+import { toolAccessRoutes } from "../routes/tool-access.js";
+const actor = { actorType: "user" as const, actorId: "mcp-test-user", actorSource: "local_implicit" as const };
+const tool = (name: string) => ({ name, description: name, inputSchema: { type: "object", properties: {} }, annotations: { readOnlyHint: true } });
+describe("remote connector lifecycle", () => {
+  let fixture: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+  let db: ReturnType<typeof createDb>;
+  let keyDir: string;
+  beforeAll(async () => {
+    keyDir = await mkdtemp(join(tmpdir(), "mcp-connector-secrets-"));
+    vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY_FILE", join(keyDir, "key"));
+    fixture = await startEmbeddedPostgresTestDatabase("mcp-connectors-test-");
+    db = createDb(fixture.connectionString);
+    await instanceSettingsService(db).updateExperimental({ enableMcpAggregators: true });
+  });
+  afterAll(async () => { await fixture?.cleanup(); vi.unstubAllEnvs(); if (keyDir) await rm(keyDir, { recursive: true, force: true }); });
+  async function company() {
+    const [row] = await db.insert(companies).values({ name: `MCP ${randomUUID()}`, issuePrefix: `M${randomUUID().slice(0, 5)}` }).returning();
+    await db.insert(companyMemberships).values({ companyId: row.id, principalType: "user", principalId: actor.actorId, status: "active", membershipRole: "admin" });
+    return row;
+  }
+  function remoteFixture() {
+    const requests: { url: string; headers: Headers }[] = [];
+    let added = false;
+    let removed = false;
+    const service = toolAccessService(db, { deploymentMode: "local_trusted", deploymentExposure: "private",
+      remoteHttpEndpointLookup: async () => [{ address: "8.8.8.8", family: 4 }],
+      remoteHttpRequest: async (url, init) => {
+        const body = JSON.parse(String(init.body)); const headers = new Headers(init.headers);
+        requests.push({ url, headers });
+        if (body.method === "initialize") return Response.json({ id: body.id, jsonrpc: "2.0", result: { protocolVersion: "2025-06-18", capabilities: {} } }, { headers: { "Mcp-Session-Id": "session" } });
+        expect(headers.get("mcp-session-id")).toBe("session");
+        if (body.method === "notifications/initialized") return new Response(null, { status: 202 });
+        expect(body.method).toBe("tools/list");
+        return Response.json({ id: body.id, jsonrpc: "2.0", result: body.params?.cursor
+          ? { tools: [tool("ask"), ...(added ? [tool("new_tool")] : [])] }
+          : { tools: [tool("read"), ...(!removed ? [tool("off")] : [])], nextCursor: "page-two" } });
+      },
+    });
+    return { service, requests, add: () => { added = true; }, remove: () => { removed = true; }, restore: () => { removed = false; } };
+  }
+  it("defaults MCP aggregators off and rejects direct setup without provider requests or credential writes", async () => {
+    expect(normalizeExperimentalSettings({}).enableMcpAggregators).toBe(false);
+    const org = await company(); const remote = remoteFixture();
+    await instanceSettingsService(db).updateExperimental({ enableMcpAggregators: false });
+    try {
+      const app = express();
+      app.use((req, _res, next) => { req.actor = { type: "board", userId: actor.actorId, source: "local_implicit", isInstanceAdmin: true }; next(); });
+      app.use("/api", toolAccessRoutes(db, { paperclipCloudConnector: null }));
+      const gallery = await request(app).get(`/api/companies/${org.id}/tools/gallery`);
+      expect(gallery.status).toBe(200);
+      expect(gallery.body.apps.some((entry: { slug: string }) => ["zapier", "arcade", "composio", "executor"].includes(entry.slug))).toBe(false);
+      expect(gallery.body.apps.some((entry: { slug: string }) => entry.slug === "notion")).toBe(true);
+      for (const [galleryKey, connectionMethodKey] of [["zapier", "generated-url"], ["arcade", "mcp"], ["composio", "mcp"], ["executor", "mcp"]]) {
+        await expect(remote.service.connectGalleryApp(org.id, { galleryKey, connectionMethodKey, saveDraft: true }, actor))
+          .rejects.toMatchObject({ status: 403, details: { code: "mcp_aggregators_disabled" } });
+      }
+      await expect(remote.service.preflightGalleryAppMetadata("composio", "mcp"))
+        .rejects.toMatchObject({ status: 403 });
+      expect(remote.requests).toHaveLength(0);
+      expect(await db.select().from(toolConnections).where(eq(toolConnections.companyId, org.id))).toHaveLength(0);
+    } finally {
+      await instanceSettingsService(db).updateExperimental({ enableMcpAggregators: true });
+    }
+  });
+  it("vaults generated URLs, paginates discovery, preserves Off/Ask during refresh and reconnect, and isolates companies", async () => {
+    const org = await company(); const other = await company(); const remote = remoteFixture();
+    const input = { galleryKey: "zapier", connectionMethodKey: "generated-url", link: "https://mcp.zapier.com/api/v1/connect?token=fixture-secret", authMode: "none" as const };
+    const connected = await remote.service.connectGalleryApp(org.id, input, actor);
+    expect(connected.catalog).toHaveLength(3);
+    expect(JSON.stringify(connected)).not.toContain("fixture-secret");
+    expect(remote.requests.every((r) => r.url.includes("token=fixture-secret"))).toBe(true);
+    const ids = Object.fromEntries(connected.catalog.map((entry) => [entry.toolName, entry.id]));
+    const finished = await remote.service.finishGalleryAppConnection(org.id, connected.connectionId, { enabledCatalogEntryIds: [ids.read, ids.ask], askFirstCatalogEntryIds: [ids.ask], access: "all_agents" }, actor);
+    remote.add();
+    const refreshed = await remote.service.refreshCatalog(connected.connectionId, actor, { enableAllByDefault: true });
+    const newId = refreshed.catalog.find((entry) => entry.toolName === "new_tool")!.id;
+    const entries = await db.select().from(toolProfileEntries).where(eq(toolProfileEntries.profileId, finished.profile.id));
+    expect(entries.map((entry) => entry.catalogEntryId)).toEqual(expect.arrayContaining([ids.read, ids.ask, newId]));
+    expect(entries.map((entry) => entry.catalogEntryId)).not.toContain(ids.off);
+    await expect(remote.service.connectGalleryApp(other.id, { ...input, reconnectConnectionId: connected.connectionId }, actor)).rejects.toThrow("not found");
+    const reconnected = await remote.service.connectGalleryApp(org.id, { ...input, reconnectConnectionId: connected.connectionId }, actor);
+    const restored = await remote.service.finishGalleryAppConnection(org.id, connected.connectionId, { enabledCatalogEntryIds: reconnected.catalog.map((entry) => entry.id), askFirstCatalogEntryIds: [], access: "all_agents" }, actor);
+    expect(restored.connection.status).toBe("active");
+    expect(restored.profileEntries.map((entry) => entry.catalogEntryId)).not.toContain(ids.off);
+    const policies = await db.select().from(toolPolicies).where(eq(toolPolicies.companyId, org.id));
+    expect(policies.some((policy) => policy.enabled && policy.config.catalogEntryId === ids.ask)).toBe(true);
+    remote.remove();
+    const afterRemoval = await remote.service.refreshCatalog(connected.connectionId, actor);
+    expect(afterRemoval.catalog.find((entry) => entry.toolName === "off")?.status).not.toBe("active");
+    const [retired] = await db.select().from(toolCatalogEntries).where(eq(toolCatalogEntries.id, ids.off));
+    expect(retired.status).toBe("disabled"); // Check persisted discovery, not just this refresh response.
+    remote.restore();
+    const afterReturn = await remote.service.refreshCatalog(connected.connectionId, actor);
+    expect(afterReturn.catalog.find((entry) => entry.toolName === "off")?.status).toBe("active");
+    const retainedEntries = await db.select().from(toolProfileEntries).where(eq(toolProfileEntries.profileId, finished.profile.id));
+    expect(retainedEntries.map((entry) => entry.catalogEntryId)).not.toContain(ids.off);
+    await remote.service.archiveConnection(connected.connectionId, org.id, actor);
+    const [removed] = await db.select().from(toolConnections).where(eq(toolConnections.id, connected.connectionId));
+    expect(removed.enabled).toBe(false);
+    expect(removed.status).toBe("archived");
+  });
+  it("renews expired discovery sessions and requires an explicit retry after an expired execution session", async () => {
+    const org = await company();
+    const [agent] = await db.insert(agents).values({ companyId: org.id, name: "Session tester", role: "engineer", adapterType: "process", adapterConfig: {}, runtimeConfig: {} }).returning();
+    let initialized = 0;
+    const expired = new Set<string>();
+    const calls: string[] = [];
+    const send = async (_url: string, init: RequestInit) => {
+      const body = JSON.parse(String(init.body));
+      if (body.method === "initialize") return Response.json({ id: body.id, result: { protocolVersion: "2025-06-18" } }, { headers: { "Mcp-Session-Id": `session-${++initialized}` } });
+      if (body.method === "notifications/initialized") return new Response(null, { status: 202 });
+      const session = new Headers(init.headers).get("mcp-session-id")!;
+      if (body.method === "tools/call") calls.push(session);
+      if (expired.has(session)) return new Response(null, { status: 404 });
+      return Response.json({ id: body.id, result: body.method === "tools/list" ? { tools: [tool("read")] } : { content: [{ type: "text", text: "ok" }] } });
+    };
+    const service = toolAccessService(db, { deploymentMode: "local_trusted", deploymentExposure: "private", remoteHttpEndpointLookup: async () => [{ address: "8.8.8.8", family: 4 }], remoteHttpRequest: send });
+    const connected = await service.connectGalleryApp(org.id, { galleryKey: "arcade", connectionMethodKey: "mcp", link: "https://api.arcade.dev/mcp/expiration", authMode: "none" }, actor);
+    const beforeRefresh = initialized;
+    expired.add(`session-${initialized}`);
+    expect((await service.refreshCatalog(connected.connectionId, actor)).catalog).toHaveLength(1);
+    expect(initialized).toBe(beforeRefresh + 1);
+    await service.finishGalleryAppConnection(org.id, connected.connectionId, { enabledCatalogEntryIds: connected.catalog.map((entry) => entry.id), askFirstCatalogEntryIds: [], access: "all_agents" }, actor);
+    const gateway = createToolGatewayService(db, { deploymentMode: "local_trusted", deploymentExposure: "private", remoteHttpRequest: send });
+    const call = () => gateway.executeTestCall({ companyId: org.id, connectionId: connected.connectionId, agentId: agent.id, userId: actor.actorId, toolName: "read", parameters: {} });
+    expect(await call()).toMatchObject({ decision: "allowed", result: { data: { isError: false } } });
+    const executionSession = calls[0];
+    expired.add(executionSession);
+    const failed = await call();
+    expect(failed).toMatchObject({ error: { reasonCode: "mcp_remote_status" } });
+    expect(calls).toEqual([executionSession, executionSession]);
+    expect(await call()).toMatchObject({ decision: "allowed", result: { data: { isError: false } } });
+    expect(calls).toHaveLength(3);
+    expect(calls[2]).not.toBe(executionSession);
+  });
+  it("saves a vaulted draft without contacting the provider and resumes with custom headers", async () => {
+    const org = await company(); const remote = remoteFixture();
+    const input = { galleryKey: "arcade", connectionMethodKey: "mcp", link: "https://api.arcade.dev/mcp/fixture", authMode: "bearer" as const };
+    const draft = await remote.service.connectGalleryApp(org.id, { ...input, saveDraft: true, credentialValues: { "credentials.authorization": "fixture-key", "headers.Arcade-User-ID": "test-user" } }, actor);
+    expect(remote.requests).toHaveLength(0);
+    expect(JSON.stringify(draft)).not.toContain("fixture-key");
+    const connected = await remote.service.connectGalleryApp(org.id, { ...input, resumeConnectionId: draft.connectionId }, actor);
+    expect(connected.catalog).toHaveLength(3);
+    expect(remote.requests.every((request) => request.headers.get("authorization") === "Bearer fixture-key" && request.headers.get("arcade-user-id") === "test-user")).toBe(true);
+  });
+  it("does not widen an empty agent selection after an OAuth callback or reconnect", async () => {
+    const org = await company();
+    const endpoint = "https://api.arcade.dev/mcp/fixture-oauth";
+    const send = async (url: string, init: RequestInit) => {
+      if (url.includes(".well-known/oauth-protected-resource")) return Response.json({ resource: endpoint, authorization_servers: ["https://auth.arcade.dev"] });
+      if (url.includes(".well-known/")) return Response.json({ issuer: "https://auth.arcade.dev", authorization_endpoint: "https://auth.arcade.dev/authorize", token_endpoint: "https://auth.arcade.dev/token", response_types_supported: ["code"], code_challenge_methods_supported: ["S256"], token_endpoint_auth_methods_supported: ["none"] });
+      if (url === "https://auth.arcade.dev/token") return Response.json({ access_token: "fixture-access", token_type: "Bearer", expires_in: 3600 });
+      if (url !== endpoint) throw new Error(`Unexpected fixture URL: ${url}`);
+      if (!new Headers(init.headers).has("authorization")) return new Response(null, { status: 401, headers: { "WWW-Authenticate": 'Bearer resource_metadata="https://api.arcade.dev/.well-known/oauth-protected-resource"' } });
+      const body = JSON.parse(String(init.body));
+      if (body.method === "notifications/initialized") return new Response(null, { status: 202 });
+      return Response.json({ id: body.id, result: body.method === "initialize" ? { protocolVersion: "2025-06-18" } : { tools: [tool("read")] } });
+    };
+    const service = toolAccessService(db, { deploymentMode: "local_trusted", deploymentExposure: "private", remoteHttpEndpointLookup: async () => [{ address: "8.8.8.8", family: 4 }], remoteHttpRequest: send });
+    const input = { galleryKey: "arcade", connectionMethodKey: "mcp", link: endpoint, authMode: "auto" as const, oauthClient: { clientId: "fixture-client" } };
+    const connected = await service.connectGalleryApp(org.id, input, actor);
+    const callback = async () => {
+      const redirectUri = "http://127.0.0.1:3116/api/tools/oauth/callback";
+      const started = await service.startOAuth(org.id, connected.connectionId, { redirectUri, actor });
+      return service.completeOAuthCallback({ state: new URL(started.authorizationUrl).searchParams.get("state")!, code: "fixture-code", redirectUri, actor });
+    };
+    await service.putConnectionInstalls(connected.connectionId, { installs: [] }, actor);
+    const first = await callback();
+    expect(first.connection.status).toBe("active");
+    expect((await service.listConnectionInstalls(connected.connectionId))).toHaveLength(0);
+    await service.connectGalleryApp(org.id, { ...input, reconnectConnectionId: connected.connectionId }, actor);
+    await callback();
+    expect((await service.listConnectionInstalls(connected.connectionId))).toHaveLength(0);
+  });
+  it("enforces agent and action permissions before dispatch, and preserves provider resume after Paperclip approval", async () => {
+    const org = await company();
+    const [allowed, denied] = await db.insert(agents).values(["Allowed", "Denied"].map((name) => ({ companyId: org.id, name, role: "engineer", adapterType: "process", adapterConfig: {}, runtimeConfig: {} }))).returning();
+    const calls: { name: string; arguments: unknown; session: string | null }[] = [];
+    const send = async (_url: string, init: RequestInit) => {
+      const body = JSON.parse(String(init.body));
+      if (body.method === "initialize") return Response.json({ id: body.id, result: { protocolVersion: "2025-06-18" } }, { headers: { "Mcp-Session-Id": "execution-session" } });
+      if (body.method === "notifications/initialized") return new Response(null, { status: 202 });
+      if (body.method === "tools/list") return Response.json({ id: body.id, result: { tools: [tool("execute"), tool("resume"), tool("off")] } });
+      calls.push({ ...body.params, session: new Headers(init.headers).get("mcp-session-id") });
+      return Response.json({ id: body.id, result: { structuredContent: body.params.name === "execute"
+        ? { status: "waiting_for_interaction", executionId: "fixture-execution", interaction: { kind: "form", message: "Approve read?", requestedSchema: { type: "object", properties: {} } } }
+        : { ok: true, resumed: body.params.arguments.executionId }, content: [] } });
+    };
+    const service = toolAccessService(db, { deploymentMode: "local_trusted", deploymentExposure: "private", remoteHttpEndpointLookup: async () => [{ address: "8.8.8.8", family: 4 }], remoteHttpRequest: send });
+    const connection = await service.connectGalleryApp(org.id, { galleryKey: "executor", connectionMethodKey: "mcp", link: "https://executor.sh/test/mcp", authMode: "none" }, actor);
+    const ids = Object.fromEntries(connection.catalog.map((entry) => [entry.toolName, entry.id]));
+    await service.finishGalleryAppConnection(org.id, connection.connectionId, { enabledCatalogEntryIds: [ids.execute, ids.resume], askFirstCatalogEntryIds: [ids.execute], access: { agentIds: [allowed.id] } }, actor);
+    const gateway = createToolGatewayService(db, { deploymentMode: "local_trusted", deploymentExposure: "private", remoteHttpRequest: send, toolActionSigningSecret: "fixture-signing-key" });
+    const call = (agentId: string, toolName: string, parameters: Record<string, unknown> = {}) => gateway.executeTestCall({ companyId: org.id, connectionId: connection.connectionId, agentId, userId: actor.actorId, toolName, parameters });
+    expect((await call(denied.id, "execute")).decision).toBe("off");
+    expect((await call(allowed.id, "off")).decision).toBe("off");
+    expect(calls).toHaveLength(0);
+    const asked = await call(allowed.id, "execute");
+    expect(asked.decision).toBe("ask_first");
+    expect(calls).toHaveLength(0);
+    if (!("actionRequestId" in asked)) throw new Error("Missing approval request");
+    await gateway.approveActionRequest({ companyId: org.id, actionRequestId: asked.actionRequestId!, actor: { userId: actor.actorId } });
+    const status = await gateway.getTestCallStatus({ companyId: org.id, connectionId: connection.connectionId, actionRequestId: asked.actionRequestId! });
+    expect(status.phase).toBe("done");
+    expect(status.upstreamPending).toMatchObject({ executionId: "fixture-execution", resumeTool: "resume" });
+    const [invocation] = await db.select().from(toolInvocations).where(eq(toolInvocations.id, asked.invocationId));
+    expect(invocation.resultSummary?.summary).toContain("fixture-execution");
+    const resumed = await call(allowed.id, "resume", { executionId: status.upstreamPending!.executionId, action: "accept", content: "{}" });
+    expect(resumed.decision).toBe("allowed");
+    expect(calls.map((call) => call.name)).toEqual(["execute", "resume"]);
+    expect(calls.every((call) => call.session === "execution-session")).toBe(true);
+    const [issue] = await db.insert(issues).values({ companyId: org.id, title: "Provider approval", status: "in_progress", assigneeAgentId: allowed.id }).returning();
+    const [run] = await db.insert(heartbeatRuns).values({ companyId: org.id, agentId: allowed.id, invocationSource: "assignment", status: "running", contextSnapshot: { issueId: issue.id } }).returning();
+    const session = await gateway.createSession({ companyId: org.id, agentId: allowed.id, runId: run.id });
+    const execute = (await gateway.listToolsForSession(session.token)).find((entry) => entry.upstreamToolName === "execute")!;
+    expect(execute).toBeDefined();
+    let approvalId = "";
+    try { await gateway.executeTool({ sessionToken: session.token, tool: execute.name, parameters: {} }); }
+    catch (error) {
+      expect(error).toMatchObject({ reasonCode: "approval_required" });
+      approvalId = (error as { details: { actionRequestId: string } }).details.actionRequestId;
+    }
+    expect(approvalId).not.toBe("");
+    await gateway.approveActionRequest({ companyId: org.id, actionRequestId: approvalId, actor: { userId: actor.actorId } });
+    await expect(gateway.executeTool({ sessionToken: session.token, tool: execute.name, parameters: {} })).rejects.toMatchObject({
+      reasonCode: "provider_interaction_required", details: { upstreamPending: { executionId: "fixture-execution" } },
+    });
+    expect(calls).toHaveLength(3); // Retrying the approved action did not start another execution.
+    await service.finishGalleryAppConnection(org.id, connection.connectionId, { enabledCatalogEntryIds: [ids.execute, ids.resume], askFirstCatalogEntryIds: [ids.execute], access: { agentIds: [] } }, actor);
+    expect(await db.select().from(toolConnectionInstalls).where(eq(toolConnectionInstalls.connectionId, connection.connectionId))).toHaveLength(0);
+    expect((await call(allowed.id, "execute")).decision).toBe("off");
+    expect((await gateway.listToolsForSession(session.token)).filter((entry) => entry.connectionId === connection.connectionId)).toHaveLength(0);
+    await service.archiveConnection(connection.connectionId, org.id, actor);
+    await expect(call(allowed.id, "resume")).rejects.toThrow("not found");
+    expect(calls).toHaveLength(3);
+  });
+
+});

server/src/__tests__/remote-mcp-pending.test.ts

@@ -0,0 +1,59 @@
+import { describe, expect, it } from "vitest";
+import { extractRemoteMcpPending } from "../services/remote-mcp-pending.js";
+import { classifyRisk } from "../services/tool-access.js";
+
+describe("remote MCP provider handoffs", () => {
+  it("recognizes Arcade's JSON text authorization response before redaction", () => {
+    const result = { result: { isError: true, content: [{ type: "text", text: JSON.stringify({ authorization_url: "https://github.com/login/oauth/authorize?state=test", message: "Not executed; authorize first" }) }] } };
+    expect(extractRemoteMcpPending(result, "arcade")).toMatchObject({ kind: "authorization", links: [{ host: "github.com", url: "https://github.com/login/oauth/authorize?state=test" }] });
+  });
+  it("recognizes nested Composio connection links without treating ordinary results as handoffs", () => {
+    expect(extractRemoteMcpPending({ data: { results: [{ redirect_url: "https://connect.composio.dev/link/test" }] } }, "composio", "COMPOSIO_MANAGE_CONNECTIONS")?.links).toHaveLength(1);
+    expect(extractRemoteMcpPending({ url: "https://example.com/article", execution_id: "completed" }, "executor")).toBeNull();
+    expect(extractRemoteMcpPending({ redirect_url: "https://example.com/article" }, "unrelated")).toBeNull();
+  });
+  it("preserves URL elicitation identities and drops unsafe navigation targets", () => {
+    const elicitations = [
+      { mode: "url", elicitationId: "consent-1", url: "https://provider.example/approve" },
+      { mode: "url", elicitationId: "consent-2", url: "javascript:alert(1)" },
+      { mode: "url", elicitationId: "consent-3", url: "https://user:password@example.com" },
+    ];
+    const pending = extractRemoteMcpPending({ error: { code: -32042, data: { elicitations } } });
+    expect(pending?.links).toEqual([{ url: "https://provider.example/approve", host: "provider.example", elicitationId: "consent-1" }]);
+    expect(pending?.elicitationId).toBe("consent-1");
+  });
+  it("keeps an Executor execution identity for manual resume, without inventing a retry", () => {
+    expect(extractRemoteMcpPending({ status: "waiting_for_interaction", executionId: "run-42", interaction: { kind: "form", message: "Approve read?", requestedSchema: { type: "object", properties: {} } } }, "executor")).toMatchObject({ kind: "approval", executionId: "run-42", resumeTool: "resume" });
+  });
+  it("does not turn ordinary successful app data into an approval or authorization handoff", () => {
+    for (const provider of ["arcade", "composio", "executor"]) {
+      for (const status of ["suspended", "pending_approval", "awaiting_approval", "waiting_for_interaction"]) {
+        expect(extractRemoteMcpPending({ result: { structuredContent: { records: [{ status, executionId: "app-job", interaction: { kind: "form" }, authorization_url: "https://example.com/auth", redirect_url: "https://connect.composio.dev/link/test" }] } } }, provider, "get_records")).toBeNull();
+        expect(extractRemoteMcpPending({ result: { structuredContent: { status } } }, provider, "get_record")).toBeNull();
+      }
+      expect(extractRemoteMcpPending({ result: { structuredContent: { mode: "url", elicitationId: "app-field", url: "https://example.com" } } }, provider)).toBeNull();
+    }
+  });
+  it("classifies broad execution and resume as writes even without annotations", () => {
+    for (const name of ["execute", "resume", "edit-artifact"]) expect(classifyRisk({ name }, "executor")).toBe("write");
+    expect(classifyRisk({ name: "skills", annotations: { readOnlyHint: true } }, "executor")).toBe("read");
+    expect(classifyRisk({ name: "COMPOSIO_MULTI_EXECUTE_TOOL", annotations: { readOnlyHint: true } }, "composio")).toBe("write");
+  });
+  it("defaults unfamiliar and namespaced aggregator capabilities to writes despite read-only hints", () => {
+    for (const provider of ["executor", "composio", "arcade", "zapier"]) {
+      for (const name of ["vendor.execute", "custom_resume", "code", "workbench", "new_capability", "get_and_run_action"]) {
+        for (const annotations of [undefined, { readOnlyHint: true }, { readOnlyHint: false }]) {
+          expect(classifyRisk({ name, annotations }, provider)).toBe("write");
+        }
+      }
+      expect(classifyRisk({ name: "delete_everything", annotations: { readOnlyHint: true } }, provider)).toBe("destructive");
+    }
+    for (const [provider, name] of [["executor", "skills"], ["composio", "COMPOSIO_SEARCH_TOOLS"], ["arcade", "Github.GetRepository"]]) {
+      expect(classifyRisk({ name }, provider)).toBe("read");
+      expect(classifyRisk({ name, annotations: { readOnlyHint: false } }, provider)).toBe("write");
+      expect(classifyRisk({ name, annotations: { destructiveHint: true } }, provider)).toBe("destructive");
+      expect(classifyRisk({ name: `custom.${name}`, annotations: { readOnlyHint: true } }, provider)).toBe("write");
+    }
+    expect(classifyRisk({ name: "GITHUB_LIST_REPOSITORIES", annotations: { readOnlyHint: true } })).toBe("read");
+  });
+});

server/src/__tests__/remote-mcp-protocol.test.ts

@@ -0,0 +1,79 @@
+import { describe, expect, it, vi } from "vitest";
+import { forgetMcpHttpSessions, getMcpHttpSession, initializeMcpHttpSession, McpHttpInitializationError, readMcpHttpResponse } from "../services/mcp-http.js";
+
+const encoder = new TextEncoder();
+function stream(chunks: string[], close = true) {
+  const cancel = vi.fn();
+  const response = new Response(new ReadableStream({ start(controller) { for (const chunk of chunks) controller.enqueue(encoder.encode(chunk)); if (close) controller.close(); }, cancel }), { headers: { "content-type": "text/event-stream" } });
+  return { response, cancel };
+}
+function event(message: unknown) { return `data: ${JSON.stringify(message)}\r\n\r\n`; }
+
+describe("remote connector Streamable HTTP", () => {
+  it("matches the requested ID after progress and unrelated messages without waiting for stream closure", async () => {
+    const fixture = stream([event({ jsonrpc: "2.0", method: "notifications/progress", params: { progress: 1 } }), event({ jsonrpc: "2.0", id: "other", result: {} }), event({ jsonrpc: "2.0", id: "call", result: { content: [] } })], false);
+    expect(await readMcpHttpResponse(fixture.response, "call")).toEqual({ jsonrpc: "2.0", id: "call", result: { content: [] } });
+    expect(fixture.cancel).toHaveBeenCalledOnce();
+  });
+  it("decodes split CRLF and UTF-8 chunks", async () => {
+    const data = event({ id: 7, result: { text: "café" } });
+    expect(await readMcpHttpResponse(stream([...data]).response, 7)).toMatchObject({ result: { text: "café" } });
+  });
+  it("rejects an unrelated JSON result and bounds streamed responses", async () => {
+    await expect(readMcpHttpResponse(Response.json({ id: "wrong", result: {} }), "call")).rejects.toThrow("message ID");
+    await expect(readMcpHttpResponse(stream([event({ id: 1, result: "too much" })]).response, 1, { maxBytes: 8 })).rejects.toThrow("size limit");
+  });
+  it("applies matching, parse errors, and size limits to buffered HTTP transports", async () => {
+    const buffered = (body: string) => ({ headers: new Headers(), text: async () => body }) as Response;
+    expect(await readMcpHttpResponse(buffered('{"id":"call","result":{}}'), "call")).toMatchObject({ id: "call" });
+    await expect(readMcpHttpResponse(buffered('{"id":"other","result":{}}'), "call")).rejects.toMatchObject({ reason: "malformed_response" });
+    await expect(readMcpHttpResponse(buffered("not json"), "call")).rejects.toMatchObject({ reason: "invalid_json" });
+    await expect(readMcpHttpResponse(buffered("too large"), "call", { maxBytes: 2 })).rejects.toMatchObject({ reason: "too_large" });
+  });
+  it("delivers server requests before the matching response", async () => {
+    const onRequest = vi.fn(async () => {});
+    const request = { jsonrpc: "2.0", id: "auth", method: "elicitation/create", params: { mode: "url", url: "https://example.com/auth", elicitationId: "consent" } };
+    await readMcpHttpResponse(stream([event(request), event({ id: "call", result: {} })]).response, "call", { onRequest });
+    expect(onRequest).toHaveBeenCalledWith(request);
+  });
+  it("preserves an initialization authentication challenge for OAuth discovery", async () => {
+    const response = new Response(null, { status: 401, headers: { "www-authenticate": 'Bearer resource_metadata="https://example.com/.well-known/oauth-protected-resource"' } });
+    try { await initializeMcpHttpSession({ requestId: "a", send: async () => response }); throw new Error("Expected challenge"); }
+    catch (error) { expect(error).toBeInstanceOf(McpHttpInitializationError); expect((error as McpHttpInitializationError).response).toBe(response); }
+  });
+  it("initializes once per connection and identity and resets after revocation", async () => {
+    let count = 0;
+    const send = vi.fn(async (init: RequestInit) => {
+      const body = JSON.parse(String(init.body));
+      if (body.method === "initialize") { count++; return Response.json({ jsonrpc: "2.0", id: body.id, result: { protocolVersion: "2025-06-18", capabilities: {} } }, { headers: { "Mcp-Session-Id": `session-${count}` } }); }
+      expect(body.method).toBe("notifications/initialized");
+      return new Response(null, { status: 202 });
+    });
+    const input = { send, requestId: "a", scope: "connection-a:grant-a", headers: { Authorization: "Bearer secret-a" } };
+    const [a, repeated] = await Promise.all([getMcpHttpSession(input), getMcpHttpSession(input)]);
+    expect(a).toEqual(repeated);
+    expect((await getMcpHttpSession(input))["Mcp-Session-Id"]).toBe("session-1");
+    expect((await getMcpHttpSession({ ...input, scope: "connection-b:grant-a" }))["Mcp-Session-Id"]).toBe("session-2");
+    expect((await getMcpHttpSession({ ...input, scope: "connection-a:grant-b" }))["Mcp-Session-Id"]).toBe("session-3");
+    expect((await getMcpHttpSession({ ...input, headers: { Authorization: "Bearer rotated" } }))["Mcp-Session-Id"]).toBe("session-4");
+    forgetMcpHttpSessions("connection-a");
+    expect((await getMcpHttpSession(input))["Mcp-Session-Id"]).toBe("session-5");
+    expect(count).toBe(5);
+  });
+  it("does not resurrect an in-flight session after disconnect", async () => {
+    let finish: (() => void) | undefined;
+    const gate = new Promise<void>((resolve) => { finish = resolve; });
+    const send = async (init: RequestInit) => {
+      const body = JSON.parse(String(init.body));
+      if (body.method === "initialize") {
+        await gate;
+        return Response.json({ id: body.id, result: { protocolVersion: "2025-06-18" } }, { headers: { "Mcp-Session-Id": "revoked" } });
+      }
+      return new Response(null, { status: 202 });
+    };
+    const pending = getMcpHttpSession({ send, requestId: "revoking", scope: "revoking-connection:agent" });
+    forgetMcpHttpSessions("revoking-connection");
+    finish!();
+    await expect(pending).rejects.toThrow("connection changed");
+  });
+});

server/src/__tests__/remote-mcp-redaction.test.ts

@@ -0,0 +1,18 @@
+import { describe, expect, it } from "vitest";
+import { redactEventPayload, redactSensitiveText } from "../redaction.js";
+
+describe("public Executor selectors and secret redaction", () => {
+  it("keeps only exact known public helper addresses readable in MCP results", () => {
+    const path = "executor.coreTools.integrations.list";
+    expect(redactEventPayload({ path, text: `Call tools.${path}({})` })).toEqual({ path, text: `Call tools.${path}({})` });
+    expect(redactSensitiveText(`${path}.privateSecretSuffix`)).toContain("REDACTED");
+    expect(redactEventPayload({ path: "arbitrary.provider.path" })?.path).toContain("REDACTED");
+  });
+  it("still redacts secret fields, bearer values, and JWT-shaped strings", () => {
+    const value = "executor.coreTools.integrations.list";
+    expect(redactEventPayload({ token: value })?.token).toContain("REDACTED");
+    expect(redactSensitiveText(`Authorization: Bearer ${value}`)).not.toContain(value);
+    expect(redactSensitiveText(`TOKEN=${value}`)).not.toContain(value);
+    expect(redactSensitiveText("eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0.signature12345678")).toContain("REDACTED");
+  });
+});

server/src/__tests__/sentry.test.ts

@@ -546,6 +546,7 @@ describe("buildSentryInitOptions serverName", () => {
 
 describe("buildSentryInitOptions release", () => {
   const commit = "0123456789abcdef0123456789abcdef01234567";
+  const readBuildCommit = vi.fn<() => string | null>();
   const integrations = {
     httpIntegration: () => ({ name: "Http" }),
     onUnhandledRejectionIntegration: () => ({ name: "OnUnhandledRejection" }),
@@ -553,12 +554,14 @@ describe("buildSentryInitOptions release", () => {
 
   beforeEach(() => {
     vi.stubEnv("SENTRY_RELEASE", "");
-    vi.doMock("../build-commit.js", () => ({ readBuildCommit: () => commit }));
+    readBuildCommit.mockReturnValue(commit);
+    vi.doMock("../build-commit.js", () => ({ readBuildCommit }));
   });
 
   afterEach(() => {
     vi.unstubAllEnvs();
     vi.doUnmock("../build-commit.js");
+    readBuildCommit.mockReset();
   });
 
   it("uses the server build commit", async () => {
@@ -573,9 +576,12 @@ describe("buildSentryInitOptions release", () => {
   });
 
   it("leaves an unknown build unattributed", async () => {
-    vi.doMock("../build-commit.js", () => ({ readBuildCommit: () => null }));
+    // Keep one module factory and change its return value explicitly for this
+    // case, rather than depending on a second factory replacing the first.
+    readBuildCommit.mockReturnValue(null);
     const { buildSentryInitOptions } = await importFreshSentry();
     expect(buildSentryInitOptions("test-dsn", integrations).release).toBeUndefined();
+    expect(readBuildCommit).toHaveBeenCalled();
   });
 });
 

server/src/__tests__/tool-access-service.test.ts

@@ -72,6 +72,7 @@ import {
   toolAccessService,
 } from "../services/tool-access.js";
 import { accessService } from "../services/access.js";
+import { instanceSettingsService } from "../services/instance-settings.js";
 import { toolAccessPolicyService } from "../services/tool-access-policy.js";
 import { secretService } from "../services/secrets.js";
 import {
@@ -361,15 +362,12 @@ function mcpSseResponse(payload: unknown): Response {
 }
 
 function mockToolsList(tools: unknown[]) {
-  return vi
-    .spyOn(globalThis, "fetch")
-    .mockResolvedValue(
-      mcpHttpResponse({
-        jsonrpc: "2.0",
-        id: "paperclip-catalog-refresh",
-        result: { tools },
-      }),
-    );
+  return vi.spyOn(globalThis, "fetch").mockImplementation(async (_url, init) => {
+    const body = JSON.parse(String(init?.body));
+    if (body.method === "notifications/initialized") return new Response(null, { status: 202 });
+    return mcpHttpResponse({ jsonrpc: "2.0", id: body.id,
+      result: body.method === "initialize" ? { protocolVersion: "2025-06-18" } : { tools } });
+  });
 }
 
 const PUBLIC_MCP_FIXTURE_URL = "https://8.8.8.8/api/mcp";
@@ -852,12 +850,14 @@ describeEmbeddedPostgres("tool access service", () => {
       process.env.PAPERCLIP_TOOL_ACCESS_TEST_DATABASE_URL?.trim();
     if (externalDatabaseUrl) {
       db = createDb(externalDatabaseUrl);
+      await instanceSettingsService(db).updateExperimental({ enableMcpAggregators: true });
       return;
     }
     tempDb = await startEmbeddedPostgresTestDatabase(
       "paperclip-tool-access-service-",
     );
     db = createDb(tempDb.connectionString);
+    await instanceSettingsService(db).updateExperimental({ enableMcpAggregators: true });
   }, 20_000);
 
   afterEach(async () => {
@@ -3382,10 +3382,10 @@ describeEmbeddedPostgres("tool access service", () => {
       priority: 100,
       selectors: { connectionId: connection.id },
     });
-    const fetchMock = vi.spyOn(globalThis, "fetch").mockResolvedValue(
+    const fetchMock = vi.spyOn(globalThis, "fetch").mockImplementation(async (_url, init) =>
       mcpHttpResponse({
         jsonrpc: "2.0",
-        id: "paperclip-tool-test",
+        id: JSON.parse(String(init?.body)).id,
         result: { content: [{ type: "text", text: "sent" }] },
       }),
     );
@@ -3674,10 +3674,10 @@ describeEmbeddedPostgres("tool access service", () => {
     expect(waiting.body.result).toBeUndefined();
 
     // 3. Approving from the review queue is what runs the parked test call.
-    const fetchMock = vi.spyOn(globalThis, "fetch").mockResolvedValue(
+    const fetchMock = vi.spyOn(globalThis, "fetch").mockImplementation(async (_url, init) =>
       mcpHttpResponse({
         jsonrpc: "2.0",
-        id: "paperclip-tool-test",
+        id: JSON.parse(String(init?.body)).id,
         result: { content: [{ type: "text", text: "sent" }] },
       }),
     );
@@ -3741,10 +3741,10 @@ describeEmbeddedPostgres("tool access service", () => {
       .send(body)
       .expect(200);
 
-    vi.spyOn(globalThis, "fetch").mockResolvedValue(
+    vi.spyOn(globalThis, "fetch").mockImplementation(async (_url, init) =>
       mcpHttpResponse({
         jsonrpc: "2.0",
-        id: "paperclip-tool-test",
+        id: JSON.parse(String(init?.body)).id,
         result: { content: [{ type: "text", text: "sent" }] },
       }),
     );
@@ -5101,7 +5101,7 @@ describeEmbeddedPostgres("tool access service", () => {
         "youcom",
       ]),
     );
-    expect(res.body.apps).toHaveLength(48);
+    expect(res.body.apps).toHaveLength(51);
     expect(
       res.body.apps.find((app: { slug: string }) => app.slug === "gmail")
         .ownershipAvailability,

server/src/__tests__/tool-gateway-service.test.ts

@@ -1177,9 +1177,9 @@ describeEmbeddedPostgres("tool gateway service", () => {
       selectors: { riskLevel: "read" },
     });
     const originalFetch = globalThis.fetch;
-    globalThis.fetch = async () => new Response(JSON.stringify({
+    globalThis.fetch = async (_url, init) => new Response(JSON.stringify({
       jsonrpc: "2.0",
-      id: "paperclip-tool-test",
+      id: JSON.parse(String(init?.body)).id,
       result: {
         _meta: {
           elicitation: {
@@ -1765,9 +1765,9 @@ describeEmbeddedPostgres("tool gateway service", () => {
       selectors: { riskLevel: "read" },
     });
     const originalFetch = globalThis.fetch;
-    globalThis.fetch = async () => new Response(JSON.stringify({
+    globalThis.fetch = async (_url, init) => new Response(JSON.stringify({
       jsonrpc: "2.0",
-      id: "paperclip-tool-test",
+      id: JSON.parse(String(init?.body)).id,
       result: { elicitation: { message: "Need input" }, content: [] },
     }), { status: 200, headers: { "content-type": "application/json" } });
     try {

server/src/__tests__/tool-gateway.test.ts

@@ -585,10 +585,10 @@ describeEmbeddedPostgres("tool gateway acceptance", () => {
 
   it("exposes a named gateway with scoped bearer-token auth and revocation", async () => {
     const company = await createCompany(db);
-    const remote = await startFakeRemoteMcpServer(async () => ({
+    const remote = await startFakeRemoteMcpServer(async ({ body }) => ({
       body: {
         jsonrpc: "2.0",
-        id: "test",
+        id: body?.id,
         result: { content: [{ type: "text", text: "read ok" }], structuredContent: { ok: true } },
       },
     }));
@@ -1127,10 +1127,10 @@ describeEmbeddedPostgres("tool gateway acceptance", () => {
 
   it("rate limits public named gateway session setup, discovery, and calls with redacted audits", async () => {
     const company = await createCompany(db);
-    const remote = await startFakeRemoteMcpServer(async () => ({
+    const remote = await startFakeRemoteMcpServer(async ({ body }) => ({
       body: {
         jsonrpc: "2.0",
-        id: "test",
+        id: body?.id,
         result: { content: [{ type: "text", text: "read ok" }], structuredContent: { ok: true } },
       },
     }));
@@ -1561,7 +1561,7 @@ describeEmbeddedPostgres("tool gateway acceptance", () => {
         if (upstreamCalls === 1) return new Response("unauthorized", { status: 401 });
         return new Response(JSON.stringify({
           jsonrpc: "2.0",
-          id: "fixture",
+          id: JSON.parse(String(init.body)).id,
           result: { content: [{ type: "text", text: "repo-a" }], structuredContent: { repositories: ["repo-a"] } },
         }), { status: 200, headers: { "content-type": "application/json" } });
       },

server/src/redaction.ts

@@ -1,4 +1,5 @@
 import { redactCommandText } from "@paperclipai/adapter-utils";
+import { isPublicExecutorToolSelector } from "@paperclipai/adapter-utils/command-redaction";
 
 const SECRET_FIELD_NAME_PATTERN = String.raw`[A-Za-z0-9_-]*(?:api[-_]?key|access[-_]?token|auth(?:_?token)?|token|authorization|bearer|secret|passwd|password|credential|jwt|private[-_]?key|cookie|connectionstring|browser[-_]?code|login[-_]?url)[A-Za-z0-9_-]*`;
 
@@ -735,7 +736,7 @@ function sanitizeValue(value: unknown): unknown {
   // string leaf after validated protocol discriminators have had a chance to
   // opt in above in sanitizeRecord.
   if (typeof value === "string") {
-    return JWT_VALUE_RE.test(value)
+    return JWT_VALUE_RE.test(value) && !isPublicExecutorToolSelector(value)
       ? REDACTED_EVENT_VALUE
       : redactSensitiveText(value);
   }
@@ -926,6 +927,7 @@ export function sanitizeRecord(
     if (
       typeof value === "string" &&
       JWT_VALUE_RE.test(value) &&
+      !isPublicExecutorToolSelector(value) &&
       !isPaperclipSchemaDiscriminator(key, value)
     ) {
       redacted[key] = REDACTED_EVENT_VALUE;

server/src/routes/tool-access.ts

@@ -4,6 +4,7 @@ import { agents, companies, connectionGrants, issueThreadInteractions, toolConne
 import { and, eq, or } from "drizzle-orm";
 import {
   APP_STORE_DEFINITIONS,
+  isRemoteMcpConnectorId,
   GITHUB_CONNECTOR_PROFILES,
   GOOGLE_WORKSPACE_CONNECTOR_PROFILES,
   isAgentStatusAssignableToWork,
@@ -67,6 +68,7 @@ import {
   paperclipCloudConnectorCapabilitiesFromEnv,
 } from "../services/paperclip-cloud-connector.js";
 import { runtimeCanonicalOrigin } from "../services/cloud-runtime-identity.js";
+import { instanceSettingsService } from "../services/instance-settings.js";
 import {
   completePaperclipCloudConnectorEnrollment,
   loadPaperclipCloudConnectorIdentity,
@@ -815,6 +817,7 @@ function connectorEnrollmentPrincipal(req: Request): string {
         ? await options.paperclipCloudConnector.getCapabilities()
         : [];
     const vercelConnect = vercelConnectIntegrationStatus();
+    const { enableMcpAggregators } = await instanceSettingsService(db).getExperimental();
     res.json({
       capabilities: await describeConnectionCreateCapabilities(req, companyId),
       credentialSources: {
@@ -830,7 +833,7 @@ function connectorEnrollmentPrincipal(req: Request): string {
             : "Vercel Connect setup is disabled on this Paperclip instance.",
         },
       },
-      apps: APP_STORE_DEFINITIONS.map((app) =>
+      apps: APP_STORE_DEFINITIONS.filter((app) => enableMcpAggregators || !isRemoteMcpConnectorId(app.slug)).map((app) =>
         appWithPaperclipCloudConnectorAvailability(app, advertisedProfiles)
       ),
     });

server/src/services/instance-settings.ts

@@ -234,6 +234,7 @@ export function normalizeExperimentalSettings(raw: unknown): InstanceExperimenta
       // continuing to accept the compatibility key in stored settings.
       enableApps: true,
       enableChatConnectors: parsed.data.enableChatConnectors ?? false,
+      enableMcpAggregators: parsed.data.enableMcpAggregators ?? false,
       enablePipelines: parsed.data.enablePipelines ?? false,
       enableCases: parsed.data.enableCases ?? false,
       enableAgentChat: parsed.data.enableAgentChat ?? false,
@@ -275,6 +276,7 @@ export function normalizeExperimentalSettings(raw: unknown): InstanceExperimenta
     enableStreamlinedUi: true,
     enableApps: true,
     enableChatConnectors: false,
+    enableMcpAggregators: false,
     enablePipelines: false,
     enableCases: false,
     enableAgentChat: false,

server/src/services/mcp-http.ts

@@ -1,3 +1,4 @@
+import { createHash } from "node:crypto";
 // Helpers for talking to remote MCP servers over the Streamable HTTP transport.
 //
 // The MCP Streamable HTTP spec requires the client to advertise that it accepts
@@ -15,6 +16,13 @@
 export const MCP_HTTP_ACCEPT = "application/json, text/event-stream";
 export const MCP_PROTOCOL_VERSION = "2025-06-18";
 
+export class McpHttpResponseError extends Error {
+  constructor(readonly reason: "invalid_json" | "malformed_response" | "too_large", message: string) {
+    super(message);
+    this.name = "McpHttpResponseError";
+  }
+}
+
 /**
  * Default headers for an MCP Streamable HTTP JSON-RPC POST. Caller-supplied
  * headers (e.g. resolved credentials) are preserved, while the required
@@ -33,6 +41,7 @@ export class McpHttpInitializationError extends Error {
     message: string,
     readonly stage: "initialize" | "initialized_notification",
     readonly status: number | null,
+    readonly response?: Response,
   ) {
     super(message);
     this.name = "McpHttpInitializationError";
@@ -40,9 +49,8 @@ export class McpHttpInitializationError extends Error {
 }
 
 /**
- * Establish the short-lived Streamable HTTP session needed by stateful MCP
- * servers. The returned headers belong only to the caller's next request; no
- * session id is persisted with the connection or shared across operations.
+ * Establish a Streamable HTTP session. Callers may retain protocol headers in
+ * the in-memory cache scoped to the connection and effective credential identity.
  */
 export async function initializeMcpHttpSession(input: {
   send: (init: RequestInit) => Promise<Response>;
@@ -68,14 +76,12 @@ export async function initializeMcpHttpSession(input: {
       `Remote MCP initialization returned HTTP ${initializeResponse.status}`,
       "initialize",
       initializeResponse.status,
+      initializeResponse,
     );
   }
   let payload: unknown;
   try {
-    payload = parseMcpHttpResponseBody(
-      await initializeResponse.text(),
-      initializeResponse.headers.get("content-type"),
-    );
+    payload = await readMcpHttpResponse(initializeResponse, `${input.requestId}-initialize`);
   } catch {
     throw new McpHttpInitializationError("Remote MCP initialization returned an invalid response", "initialize", null);
   }
@@ -83,6 +89,7 @@ export async function initializeMcpHttpSession(input: {
     ? (payload as { result?: unknown }).result
     : null;
   const resultRecord = result && typeof result === "object" ? result as Record<string, unknown> : null;
+  if (!resultRecord) throw new McpHttpInitializationError("Remote MCP initialization failed", "initialize", null);
   const protocolVersion = typeof resultRecord?.protocolVersion === "string" && resultRecord.protocolVersion
     ? resultRecord.protocolVersion
     : MCP_PROTOCOL_VERSION;
@@ -166,3 +173,97 @@ export function parseMcpHttpResponseBody(bodyText: string, contentType: string |
   if (lastError) throw lastError;
   throw new SyntaxError("MCP SSE response contained no data events");
 }
+
+/** Read until the response for this request arrives, without waiting for an SSE
+ * connection to close. Notifications and responses for other IDs are ignored. */
+export async function readMcpHttpResponse(
+  response: Response,
+  requestId: string | number,
+  options: { maxBytes?: number; onRequest?: (message: Record<string, unknown>) => Promise<void> } = {},
+): Promise<unknown> {
+  const maxBytes = options.maxBytes ?? 8 * 1024 * 1024;
+  const isStream = response.headers.get("content-type")?.toLowerCase().includes("text/event-stream");
+  const reader = response.body?.getReader();
+  // Injected HTTP transports can expose a buffered text response rather than a
+  // Web ReadableStream. Keep the same size and message-ID checks for both forms.
+  if (!reader) {
+    const body = await response.text();
+    if (Buffer.byteLength(body, "utf8") > maxBytes) throw new McpHttpResponseError("too_large", "MCP response exceeded the size limit");
+    return readMcpHttpResponse(new Response(body, {
+      headers: { "content-type": response.headers.get("content-type") ?? "application/json" },
+    }), requestId, options);
+  }
+  const decoder = new TextDecoder();
+  let buffer = "";
+  let bytes = 0;
+  const parse = (text: string): unknown => {
+    try { return JSON.parse(text); }
+    catch { throw new McpHttpResponseError("invalid_json", "MCP response contained invalid JSON"); }
+  };
+  const inspect = async (message: unknown): Promise<unknown | undefined> => {
+    if (!message || typeof message !== "object") return undefined;
+    const record = message as Record<string, unknown>;
+    if (record.id === requestId && ("result" in record || "error" in record)) return record;
+    if ("method" in record && "id" in record) await options.onRequest?.(record);
+    return undefined;
+  };
+  const event = async (value: string) => {
+    const data = value.split("\n").filter((line) => line.startsWith("data:")).map((line) => line.slice(5).replace(/^ /, "")).join("\n");
+    if (!data) return undefined;
+    return inspect(parse(data));
+  };
+  try {
+    while (true) {
+      const { value, done } = await reader.read();
+      bytes += value?.byteLength ?? 0;
+      if (bytes > maxBytes) throw new McpHttpResponseError("too_large", "MCP response exceeded the size limit");
+      buffer += decoder.decode(value, { stream: !done });
+      if (isStream) {
+        // Normalize CRLF after concatenating chunks, including split CR/LF pairs.
+        buffer = buffer.replace(/\r\n/g, "\n");
+        let boundary: number;
+        while ((boundary = buffer.indexOf("\n\n")) >= 0) {
+          const result = await event(buffer.slice(0, boundary));
+          buffer = buffer.slice(boundary + 2);
+          if (result !== undefined) return result;
+        }
+      }
+      if (done) break;
+    }
+    const result = isStream ? await event(buffer) : await inspect(parse(buffer));
+    if (result !== undefined) return result;
+    throw new McpHttpResponseError("malformed_response", "MCP response did not contain the requested message ID");
+  } finally {
+    await reader.cancel().catch(() => undefined);
+    reader.releaseLock();
+  }
+}
+
+const sessions = new Map<string, { headers: Record<string, string>; expiresAt: number }>();
+const initializing = new Map<string, Promise<Record<string, string>>>();
+const SESSION_TTL_MS = 30 * 60_000;
+
+/** Cache only protocol headers; credential hashes and scope separate every
+ * connection and effective identity. Never cache a tool call or replay a write. */
+export async function getMcpHttpSession(input: Parameters<typeof initializeMcpHttpSession>[0] & { scope: string }) {
+  const key = `${input.scope}:${createHash("sha256").update(JSON.stringify(Object.entries(input.headers ?? {}).sort())).digest("hex")}`;
+  const cached = sessions.get(key);
+  if (cached && cached.expiresAt > Date.now()) return { ...input.headers, ...cached.headers };
+  const pending = initializing.get(key);
+  if (pending) return pending;
+  const promise = initializeMcpHttpSession(input).then((headers) => {
+    if (initializing.get(key) !== promise) throw new Error("MCP connection changed while initializing; reconnect before calling tools");
+    for (const [id, value] of sessions) if (value.expiresAt <= Date.now()) sessions.delete(id);
+    if (sessions.size >= 1000) sessions.delete(sessions.keys().next().value!);
+    const protocolHeaders = Object.fromEntries(Object.entries(headers).filter(([name]) => ["mcp-session-id", "mcp-protocol-version"].includes(name.toLowerCase())));
+    sessions.set(key, { headers: protocolHeaders, expiresAt: Date.now() + SESSION_TTL_MS });
+    return headers;
+  }).finally(() => { if (initializing.get(key) === promise) initializing.delete(key); });
+  initializing.set(key, promise);
+  return promise;
+}
+
+export function forgetMcpHttpSessions(connectionId: string) {
+  for (const key of sessions.keys()) if (key.startsWith(`${connectionId}:`)) sessions.delete(key);
+  for (const key of initializing.keys()) if (key.startsWith(`${connectionId}:`)) initializing.delete(key);
+}

server/src/services/remote-mcp-pending.ts

@@ -0,0 +1,78 @@
+import { redactEventPayload, redactSensitiveText } from "../redaction.js";
+import { checkOAuthEndpointUrl, type ToolUpstreamPending } from "@paperclipai/shared";
+
+function record(value: unknown): Record<string, unknown> | null {
+  return value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : null;
+}
+
+/** Recognize protocol/provider envelopes, never generic app-data statuses.
+ * Links are navigation targets kept outside durable result/audit storage. */
+export function extractRemoteMcpPending(value: unknown, provider?: string | null, toolName?: string): ToolUpstreamPending | null {
+  const root = record(value);
+  if (!root) return null;
+  const links = new Map<string, { url: string; host: string; elicitationId?: string }>();
+  let pending: ToolUpstreamPending | null = null;
+  const addLink = (url: unknown, id?: string) => {
+    const checked = checkOAuthEndpointUrl(url);
+    if (checked.ok && links.size < 8) links.set(checked.url, { url: checked.url, host: checked.host, ...(id ? { elicitationId: id } : {}) });
+  };
+  const urlElicitation = (value: unknown) => {
+    const item = record(value);
+    if (item?.mode !== "url" || typeof item.elicitationId !== "string") return;
+    pending ??= { kind: "authorization", links: [] };
+    const id = item.elicitationId.slice(0, 512);
+    addLink(item.url, id);
+    if (links.size && !pending.elicitationId) pending.elicitationId = id;
+  };
+  if (root.method === "elicitation/create") urlElicitation(root.params);
+  const error = record(root.error);
+  const elicitations = record(error?.data)?.elicitations;
+  if (error?.code === -32042 && Array.isArray(elicitations)) elicitations.slice(0, 8).forEach(urlElicitation);
+
+  const result = record(root.result) ?? root;
+  const payloads: Record<string, unknown>[] = [result];
+  const structured = record(result.structuredContent);
+  if (structured) payloads.push(structured);
+  if (Array.isArray(result.content)) {
+    for (const item of result.content.slice(0, 100)) {
+      const content = record(item);
+      if (content?.type !== "text" || typeof content.text !== "string" || content.text.length > 512_000) continue;
+      try { const parsed = record(JSON.parse(content.text)); if (parsed) payloads.push(parsed); } catch { /* Ordinary text. */ }
+    }
+  }
+  for (const payload of payloads) {
+    if (provider === "executor" && payload.status === "waiting_for_interaction"
+      && typeof payload.executionId === "string" && payload.executionId) {
+      const interaction = record(payload.interaction);
+      if (interaction?.kind !== "form" && interaction?.kind !== "url") continue;
+      pending = { kind: "approval", links: [], executionId: payload.executionId.slice(0, 512), resumeTool: "resume" };
+      if (typeof payload.expiresAt === "string" && Number.isFinite(Date.parse(payload.expiresAt))) pending.expiresAt = payload.expiresAt;
+      if (typeof interaction.message === "string") pending.message = redactSensitiveText(interaction.message).slice(0, 4000);
+      const schema = record(interaction.requestedSchema);
+      if (schema) pending.requestedSchema = redactEventPayload(schema) ?? undefined;
+      if (interaction.kind === "url") addLink(interaction.url);
+    }
+    if (provider === "arcade" && (result.isError === true || /(?:^|[._])ManageAuthorization$/.test(toolName ?? ""))
+      && typeof payload.authorization_url === "string") {
+      addLink(payload.authorization_url);
+      pending ??= { kind: "authorization", links: [] };
+    }
+    // This tool returns connection handoffs keyed by app. Other Composio tools
+    // may return arbitrary application data with redirect_url/status fields.
+    if (provider === "composio" && toolName === "COMPOSIO_MANAGE_CONNECTIONS") {
+      let visited = 0;
+      const visit = (item: unknown, depth: number) => {
+        if (++visited > 500 || depth > 10 || !item || typeof item !== "object") return;
+        const entry = record(item);
+        const checked = checkOAuthEndpointUrl(entry?.redirect_url);
+        if (checked.ok && checked.host === "connect.composio.dev" && new URL(checked.url).pathname.startsWith("/link/")) {
+          addLink(checked.url);
+          pending ??= { kind: "authorization", links: [] };
+        }
+        for (const child of Object.values(item).slice(0, 100)) visit(child, depth + 1);
+      };
+      visit(payload, 0);
+    }
+  }
+  return pending ? { ...pending, links: [...links.values()] } : null;
+}

server/src/services/tool-access-policy.ts

@@ -1183,6 +1183,11 @@ export function toolAccessPolicyService(db: Db) {
     const { ctx, redaction } = loaded;
     const profileState = await effectiveProfiles(ctx);
     const effectiveProfileIds = profileState.profiles.map((profile) => profile.id);
+    const permittedByProfile = profileState.profiles.some((profile) => {
+      const matchingEntries = profileState.entries.filter((entry) => entry.profileId === profile.id && profileEntryMatches(entry, ctx));
+      return !matchingEntries.some((entry) => entry.effect === "exclude")
+        && (profile.defaultAction === "allow" || matchingEntries.some((entry) => entry.effect === "include"));
+    });
     const policies = await db.select().from(toolPolicies).where(and(eq(toolPolicies.companyId, ctx.companyId), eq(toolPolicies.enabled, true))).orderBy(asc(toolPolicies.priority), asc(toolPolicies.createdAt));
     for (const policy of policies) {
       const conditions = policyConditions(policy);
@@ -1276,6 +1281,9 @@ export function toolAccessPolicyService(db: Db) {
         return decision("allow", "allow_trust_rule", policy.description ?? "Tool access allowed by trust rule.", effectiveProfileIds, [policy.id], { redactionPlan: redaction.redactionPlan, policyExplanation });
       }
       if (policy.policyType === "require_approval") {
+        // The connection's Ask first control restricts an existing action grant;
+        // it must never grant access to agents outside that connection's profile.
+        if (policy.config?.source === "app_gallery_finish" && !permittedByProfile) continue;
         return decision("require_approval", "requires_approval_policy", policy.description ?? "Tool access requires approval.", effectiveProfileIds, [policy.id], { redactionPlan: redaction.redactionPlan, policyExplanation });
       }
       if (policy.policyType === "allow") {

server/src/services/tool-access.ts

@@ -1,4 +1,5 @@
-import { connectionPurposeTransportSchema } from "@paperclipai/shared";
+import { isRemoteMcpConnectorMethod, connectionPurposeTransportSchema } from "@paperclipai/shared";
+import { instanceSettingsService } from "./instance-settings.js";
 import { syncConnectionCredentialBindings } from "./connection-credential-bindings.js";
 import { canBrowseProjectRepositoryGrant, mergeProjectRepository } from "./project-repositories.js";
 import { captureRunIdentity } from "./run-identity.js";
@@ -186,6 +187,10 @@ import { logger } from "../middleware/logger.js";
 import { logActivity } from "./activity-log.js";
 import {
   initializeMcpHttpSession,
+  McpHttpInitializationError,
+  getMcpHttpSession,
+  forgetMcpHttpSessions,
+  readMcpHttpResponse,
   mcpHttpRequestHeaders,
   parseMcpHttpResponseBody,
 } from "./mcp-http.js";
@@ -2316,6 +2321,17 @@ function normalizedProviderToolName(toolName: string): string {
     .replace(/[:._-]+/g, "-");
 }
 
+// Aggregators can add arbitrarily powerful tools without changing their MCP
+// endpoint. Only these reviewed, exact capabilities are read-only. Unknown or
+// renamed actions remain enabled by the usual access rules, but have write risk.
+// Legacy Composio child connections do not use these gallery template keys.
+const AGGREGATOR_READ_TOOLS = new Map<string, ReadonlySet<string>>([
+  ["executor", new Set(["skills"])],
+  ["composio", new Set(["COMPOSIO_SEARCH_TOOLS", "COMPOSIO_GET_TOOL_SCHEMAS", "COMPOSIO_SEARCH_SKILLS", "COMPOSIO_USE_SKILL"])],
+  ["arcade", new Set(["Github.GetRepository"])],
+  ["zapier", new Set()],
+]);
+
 export function classifyRisk(
   tool: McpToolDescriptor,
   sourceTemplateKey?: string | null,
@@ -2324,6 +2340,12 @@ export function classifyRisk(
   if (annotations.destructiveHint === true || annotations.destructive === true)
     return "destructive";
   const normalizedToolName = normalizedProviderToolName(tool.name);
+  const reviewedReads = AGGREGATOR_READ_TOOLS.get(sourceTemplateKey ?? "");
+  if (reviewedReads) {
+    if (verbMatches(tool.name, "delete|remove|destroy|unpublish")) return "destructive";
+    if (annotations.readOnlyHint === false || annotations.writeHint === true) return "write";
+    return reviewedReads.has(tool.name) ? "read" : "write";
+  }
   if (sourceTemplateKey === "railway") {
     const reviewed = railwayRisk(normalizedToolName);
     return reviewed === "read" && (annotations.readOnlyHint === false || annotations.writeHint === true) ? "write" : reviewed;
@@ -6068,6 +6090,7 @@ export function toolAccessService(
     removalOptions: { confirmComposioChildren?: boolean } = {},
   ): Promise<ToolConnectionRemovalResult> {
     const connection = await getConnectionRow(connectionId, companyId);
+    forgetMcpHttpSessions(connection.id);
     const now = new Date();
     const binding = actorBinding(actor);
 
@@ -6765,31 +6788,26 @@ export function toolAccessService(
     // Pinned to the address the guard approved: `config.url` is operator-supplied,
     // so a second DNS resolution here would reopen the rebinding window that
     // PAP-17098 closed for the OAuth endpoints.
-    const listRequestBody = JSON.stringify({
-      jsonrpc: "2.0",
-      id: "paperclip-catalog-refresh",
-      method: "tools/list",
-      params: {},
-    });
-    const sendRemote = (init: RequestInit) =>
-      requestRemoteHttpEndpoint(new URL(endpoint), init);
-    const sendToolsList = (requestHeaders: Record<string, string>) =>
-      sendRemote({
-        method: "POST",
-        // MCP Streamable HTTP requires advertising that we accept both a JSON body
-        // and an SSE stream; spec-compliant servers 406 without it (see mcp-http.ts).
-        headers: mcpHttpRequestHeaders(requestHeaders),
-        body: listRequestBody,
-      });
+    let listRequestId = "paperclip-catalog-refresh";
+    let sessionHeaders = headers;
+    const sendRemote = (init: RequestInit) => requestRemoteHttpEndpoint(new URL(endpoint), init);
+    const sendToolsList = (requestHeaders: Record<string, string>, cursor?: string) => {
+      sessionHeaders = requestHeaders;
+      return sendRemote({ method: "POST", headers: mcpHttpRequestHeaders(requestHeaders),
+        body: JSON.stringify({ jsonrpc: "2.0", id: listRequestId, method: "tools/list", params: cursor ? { cursor } : {} }) });
+    };
     let usedInitializedSession = connection.config.mcpSessionRequired === true;
     let response: Response;
     if (usedInitializedSession) {
-      const sessionHeaders = await initializeMcpHttpSession({
-        send: sendRemote,
-        headers,
-        requestId: "paperclip-catalog-refresh",
-      });
-      response = await sendToolsList(sessionHeaders);
+      try {
+        sessionHeaders = await getMcpHttpSession({ send: sendRemote, headers,
+          scope: `${connection.id}:catalog:${actor?.actorType}:${actor?.actorId}:${endpoint}`,
+          requestId: listRequestId });
+        response = await sendToolsList(sessionHeaders);
+      } catch (error) {
+        if (!(error instanceof McpHttpInitializationError) || !error.response) throw error;
+        response = error.response;
+      }
     } else {
       response = await sendToolsList(headers);
       // `tools/list` is read-only, so a 400 can safely be retried after the MCP
@@ -6810,6 +6828,17 @@ export function toolAccessService(
         }
       }
     }
+    if (response.status === 404 && new Headers(sessionHeaders).has("mcp-session-id")) {
+      // MCP uses 404 for an expired server session. Discovery is read-only, so
+      // discard the stale session and retry it once with a new handshake.
+      forgetMcpHttpSessions(connection.id);
+      await response.body?.cancel().catch(() => undefined);
+      sessionHeaders = await getMcpHttpSession({ send: sendRemote, headers,
+        scope: `${connection.id}:catalog:${actor?.actorType}:${actor?.actorId}:${endpoint}`,
+        requestId: listRequestId });
+      response = await sendToolsList(sessionHeaders);
+      if (response.status === 404) forgetMcpHttpSessions(connection.id);
+    }
     if (
       usedInitializedSession &&
       connection.config.mcpSessionRequired !== true
@@ -6833,6 +6862,8 @@ export function toolAccessService(
       const refreshed = await composioSessions.ensureSession(connection.id, {
         force: true,
       });
+      listRequestId = "paperclip-catalog-refresh-retry";
+      sessionHeaders = refreshed.headers;
       response = await requestRemoteHttpEndpoint(new URL(refreshed.url), {
         method: "POST",
         headers: mcpHttpRequestHeaders(refreshed.headers),
@@ -6954,20 +6985,23 @@ export function toolAccessService(
         status: response.status,
       });
     }
-    const payload = parseMcpHttpResponseBody(
-      await response.text(),
-      response.headers.get("content-type"),
-    );
-    const result = asRecord(asRecord(payload).result);
-    const payloadTools = asRecord(payload).tools;
-    const tools: unknown[] = Array.isArray(result.tools)
-      ? result.tools
-      : Array.isArray(payloadTools)
-        ? payloadTools
-        : [];
-    const descriptors = tools
-      .map((tool) => normalizeToolDescriptor(tool))
-      .filter((tool): tool is McpToolDescriptor => Boolean(tool));
+    const descriptors: McpToolDescriptor[] = [];
+    const seenCursors = new Set<string>();
+    for (let page = 0; ; page += 1) {
+      const payload = await readMcpHttpResponse(response, listRequestId);
+      const record = asRecord(payload);
+      if (record.error) throw new HttpError(502, "Remote MCP tool discovery failed", { code: "mcp_catalog_error" });
+      const result = asRecord(record.result);
+      if (!Array.isArray(result.tools)) throw new HttpError(502, "Remote MCP returned an invalid tool catalog", { code: "mcp_catalog_invalid" });
+      descriptors.push(...result.tools.map((tool) => normalizeToolDescriptor(tool)).filter((tool): tool is McpToolDescriptor => Boolean(tool)));
+      const cursor = typeof result.nextCursor === "string" ? result.nextCursor : null;
+      if (!cursor) break;
+      if (seenCursors.has(cursor) || page >= 99 || descriptors.length > 20_000) throw new HttpError(502, "Remote MCP tool catalog pagination did not finish", { code: "mcp_catalog_pagination" });
+      seenCursors.add(cursor);
+      listRequestId = `paperclip-catalog-refresh-${page + 1}`;
+      response = await sendToolsList(sessionHeaders, cursor);
+      if (!response.ok) throw new HttpError(502, "Remote MCP catalog page could not be read", { status: response.status });
+    }
     if (!isRailwayConnection(connection)) return descriptors;
     if (descriptors.some((tool) => normalizeRailwayToolName(tool.name).startsWith(RAILWAY_TOOL_PREFIX))) {
       throw unprocessable("Railway advertised a reserved Paperclip action name. Refresh is blocked pending review.", { code: "railway_tool_name_collision" });
@@ -7015,7 +7049,7 @@ export function toolAccessService(
     connection: typeof toolConnections.$inferSelect,
   ): boolean {
     return (
-      asRecord(connection.config).sourceTemplateKey === COMPOSIO_GALLERY_KEY
+      asRecord(connection.config).sourceTemplateKey === COMPOSIO_GALLERY_KEY && connection.transport === "rest_api"
     );
   }
 
@@ -7772,6 +7806,13 @@ export function toolAccessService(
     const existingByName = new Map(
       existingRows.map((entry) => [entry.toolName, entry]),
     );
+    if (isRemoteMcpConnectorMethod(connection.config.sourceTemplateKey, connection.config.connectionMethodKey)) {
+      const discoveredNames = new Set(descriptors.map((descriptor) => descriptor.name));
+      const removedIds = existingRows.filter((entry) => !discoveredNames.has(entry.toolName) && entry.status !== "disabled").map((entry) => entry.id);
+      if (removedIds.length) await db.update(toolCatalogEntries)
+        .set({ status: "disabled", quarantineReason: "mcp_tool_removed", updatedAt: refreshedAt })
+        .where(and(eq(toolCatalogEntries.companyId, connection.companyId), eq(toolCatalogEntries.connectionId, connection.id), inArray(toolCatalogEntries.id, removedIds)));
+    }
     // Retired native actions are absent from discovery, but old catalog rows
     // still need to show as disabled. Gateway denial also applies before refresh.
     const blockedRailwayEntryIds = isRailwayEndpoint(connection.config.url)
@@ -7835,7 +7876,7 @@ export function toolAccessService(
         ? "disabled"
         : shouldQuarantine
           ? "quarantined"
-          : existing?.status === "disabled"
+          : existing?.status === "disabled" && existing.quarantineReason !== "mcp_tool_removed"
             ? "disabled"
             : quarantineOnRefresh && existing?.status === "quarantined"
               ? "quarantined"
@@ -7950,10 +7991,12 @@ export function toolAccessService(
     const activeEntries = updatedEntries.filter(
       (entry) => entry.status === "active",
     );
-    if (!refreshOptions.skipDefaultProfileSync) {
+    const preserveMcpAccess = connection.config.mcpPreserveAccess === true
+      && isRemoteMcpConnectorMethod(connection.config.sourceTemplateKey, connection.config.connectionMethodKey);
+    if (!refreshOptions.skipDefaultProfileSync || preserveMcpAccess) {
       await enableCatalogEntriesByDefault({
         connection: updatedConnection,
-        newCatalogEntryIds: refreshOptions.enableAllByDefault
+        newCatalogEntryIds: refreshOptions.enableAllByDefault && !isRemoteMcpConnectorMethod(connection.config.sourceTemplateKey, connection.config.connectionMethodKey)
           ? activeEntries.map((entry) => entry.id)
           : activeEntries
               .filter((entry) => {
@@ -7962,7 +8005,7 @@ export function toolAccessService(
               })
               .map((entry) => entry.id),
         activeCatalogEntryIds: activeEntries.map((entry) => entry.id),
-        restoreDraftDefaults: refreshOptions.restoreDraftDefaults,
+        restoreDraftDefaults: refreshOptions.restoreDraftDefaults || preserveMcpAccess,
         actor,
       });
     }
@@ -12231,6 +12274,15 @@ export function toolAccessService(
     return `${base.slice(0, 151).trimEnd()} (${randomUUID().slice(0, 6)})`;
   }
 
+  async function assertMcpAggregatorSetupEnabled(provider: unknown, method: unknown) {
+    if (isRemoteMcpConnectorMethod(provider, method)
+      && !(await instanceSettingsService(db).getExperimental()).enableMcpAggregators) {
+      throw forbidden("Enable MCP aggregators in Settings → Experimental to set up this connection", {
+        code: "mcp_aggregators_disabled",
+      });
+    }
+  }
+
   async function connectGalleryApp(
     companyId: string,
     input: ConnectToolApp,
@@ -12377,18 +12429,20 @@ export function toolAccessService(
     const method = galleryEntry
       ? connectionMethodFor(galleryEntry, inferredMethodKey)
       : null;
+    const remoteMcpConnector = isRemoteMcpConnectorMethod(galleryEntry?.slug, method?.key);
+    await assertMcpAggregatorSetupEnabled(galleryEntry?.slug, method?.key);
     if (galleryEntry && input.link) {
       const acceptsProviderGeneratedUrl =
         method?.transport === "mcp_remote" &&
         method.auth === "none" &&
         !method.defaults?.serverUrl &&
         !method.defaults?.serverUrlTemplate;
-      if (!acceptsProviderGeneratedUrl) {
+      if (!acceptsProviderGeneratedUrl && !remoteMcpConnector) {
         throw badRequest(
           `${galleryEntry.name} does not accept a provider-generated connection URL`,
         );
       }
-      if (!getAppDefinitionForUrl(input.link, [galleryEntry])) {
+      if (!(remoteMcpConnector && galleryEntry.slug === "executor") && !getAppDefinitionForUrl(input.link, [galleryEntry])) {
         throw badRequest(
           `That connection URL does not belong to ${galleryEntry.name}`,
         );
@@ -12699,6 +12753,7 @@ export function toolAccessService(
       transport === "mcp_remote"
         ? {
             url:
+              (remoteMcpConnector ? remoteUrlCredential?.publicUrl : undefined) ??
               normalizedMethodConfig?.url ??
               method?.defaults?.serverUrl ??
               remoteUrlCredential?.publicUrl ??
@@ -12716,6 +12771,11 @@ export function toolAccessService(
           // the wizard projects the app's action defaults into policies at
           // finish time instead of using catalog quarantine as access state.
           quarantineNewEntries: galleryEntry.slug === "railway",
+          ...(remoteMcpConnector ? {
+            mcpSessionRequired: true,
+            mcpAuthMode: input.authMode ?? "auto",
+            mcpPreserveAccess: Boolean(retainedConnection && (retainedConnection.status === "active" || asRecord(retainedConnection.config).mcpPreserveAccess === true)),
+          } : {}),
           ...(galleryEntry.slug === "posthog" ? { safeDefault: true } : {}),
         }
       : { ...baseConfig, quarantineNewEntries: false, unverifiedServer: true };
@@ -12734,7 +12794,7 @@ export function toolAccessService(
       config.quarantineNewEntries = true;
     }
     const acceptsCustomerOAuthClient =
-      method?.auth === "oauth" && method.ownershipModes.includes("customer");
+      remoteMcpConnector || (method?.auth === "oauth" && method.ownershipModes.includes("customer"));
     if (galleryEntry && input.oauthClient && !acceptsCustomerOAuthClient) {
       throw badRequest(
         `${galleryEntry.name} does not accept customer-owned OAuth client credentials`,
@@ -12784,7 +12844,7 @@ export function toolAccessService(
     // operator supplied and is upgraded to `oauth` when discovery proves the
     // endpoint needs sign-in (see `remoteTools` and `startOAuth`).
     const genericAuthKind: ToolConnectionAuthKind =
-      method?.auth ??
+      (remoteMcpConnector ? undefined : method?.auth) ??
       (input.authMode === "oauth" || input.oauthClient
         ? "oauth"
         : input.authMode === "bearer" || input.authMode === "custom_headers"
@@ -12834,7 +12894,8 @@ export function toolAccessService(
       previousGrantKind === requestedGrantKind &&
       galleryEntry &&
       retainedSource === galleryEntry.slug &&
-      retainedMethodKey === method?.key,
+      retainedMethodKey === method?.key &&
+      (!remoteMcpConnector || (baseConfig.url === retainedConfig.url && genericAuthKind === retainedConnection.authKind)),
     );
     const retainedCredentialSecretRefs = canRetainCredentialMaterial
       ? (retainedPersonalIdentity?.grant?.credentialSecretRefs ??
@@ -12866,9 +12927,9 @@ export function toolAccessService(
       const credentialFields =
         credentialSource === "vercel_connect"
           ? []
-          : galleryEntry
+          : galleryEntry && !remoteMcpConnector
             ? credentialFieldsFor(galleryEntry, method?.key)
-            : linkCredentialFields(credentialValues);
+            : linkCredentialFields({ ...Object.fromEntries(retainedCredentialSecretRefs.filter((ref) => ref.configPath.startsWith("headers.") || ref.configPath === "credentials.authorization").map((ref) => [ref.configPath, "retained"])), ...credentialValues });
       for (const field of credentialFields) {
         const value = credentialValues[field.configPath];
         const retainedSecretRef = retainedCredentialSecretRefs.find(
@@ -12923,6 +12984,13 @@ export function toolAccessService(
         }
       }
 
+      if (!remoteUrlCredential?.secretUrl && canRetainCredentialMaterial && baseConfig.url === retainedConfig.url) {
+        const retainedUrl = retainedCredentialSecretRefs.find((ref) => ref.configPath === REMOTE_URL_SECRET_CONFIG_PATH);
+        if (retainedUrl) {
+          credentialSecretRefs.push(retainedUrl);
+          credentialRefs.push({ name: REMOTE_URL_SECRET_CONFIG_PATH, secretId: retainedUrl.secretId, version: retainedUrl.versionSelector ?? "latest", placement: "url", key: "url", prefix: null });
+        }
+      }
       if (remoteUrlCredential?.secretUrl) {
         const secret = await secrets.create(
           companyId,
@@ -13072,6 +13140,7 @@ export function toolAccessService(
       const connectionCredentialSecretRefs =
         personalIdentityUserId || dedicatedAgentId ? [] : credentialSecretRefs;
       if (revivedConnectionPrevious) {
+        forgetMcpHttpSessions(revivedConnectionPrevious.id);
         [connectionRow] = await db
           .update(toolConnections)
           .set({
@@ -13292,6 +13361,10 @@ export function toolAccessService(
       );
       await ensureRuntimeSlot(connectionRow);
 
+      if (input.saveDraft && remoteMcpConnector) {
+        return { connectionId: connectionRow.id, application: toApplication(applicationRow), connection: toConnection(connectionRow),
+          catalog: [], actions: { readOnly: [], canMakeChanges: [] }, suggestedDefaults: { access: "all_agents", askFirstRiskLevels: [] } };
+      }
       if (galleryEntry && method?.auth === "oauth") {
         const suggestedDefaults = recommendedDefaultsForApp(
           galleryEntry,
@@ -13317,7 +13390,7 @@ export function toolAccessService(
       // empty personal grant below so later catalog refreshes use the same
       // identity policy.
       const unauthenticatedPersonalProbe = Boolean(
-        !galleryEntry &&
+        (!galleryEntry || remoteMcpConnector) &&
           genericAuthKind === "none" &&
           credentialSecretRefs.length === 0 &&
           personalIdentityUserId &&
@@ -13330,7 +13403,10 @@ export function toolAccessService(
         });
       } catch (error) {
         if (
-          !galleryEntry &&
+          (!galleryEntry || remoteMcpConnector) &&
+          input.authMode !== "none" &&
+          input.authMode !== "bearer" &&
+          input.authMode !== "custom_headers" &&
           error instanceof HttpError &&
           asRecord(error.details).code === "oauth_challenge"
         ) {
@@ -13421,7 +13497,7 @@ export function toolAccessService(
         // later fails: another retry may already be using the committed grant.
         personalPublicSetupEstablished = true;
       }
-      if (galleryEntry?.slug === COMPOSIO_GALLERY_KEY) {
+      if (galleryEntry?.slug === COMPOSIO_GALLERY_KEY && transport === "rest_api") {
         const [application] = await db
           .select()
           .from(toolApplications)
@@ -13440,7 +13516,7 @@ export function toolAccessService(
       }
       const restoreDraftDefaults = Boolean(revivedConnectionPrevious);
       const refreshOptions = {
-        enableAllByDefault: restoreDraftDefaults,
+        enableAllByDefault: restoreDraftDefaults && !remoteMcpConnector,
         restoreDraftDefaults,
       };
       const catalogConnectionId = connectionRow.id;
@@ -13802,6 +13878,23 @@ export function toolAccessService(
     const connection = await getConnectionRow(connectionId, companyId);
     if (connection.status === "archived")
       throw conflict("Archived app connections cannot be finished");
+    if (connection.config.mcpPreserveAccess === true && isRemoteMcpConnectorMethod(connection.config.sourceTemplateKey, connection.config.connectionMethodKey)) {
+      const [profile] = await db.select().from(toolProfiles).where(and(
+        eq(toolProfiles.companyId, companyId), eq(toolProfiles.profileKey, `app:${connection.id}`),
+      )).limit(1);
+      if (!profile) throw conflict("The saved connection permissions could not be found");
+      const config = { ...connection.config };
+      delete config.mcpPreserveAccess;
+      const updated = await db.transaction(async (tx) => {
+        const [row] = await tx.update(toolConnections).set({ config, transportConfig: config, status: "active", enabled: true, updatedAt: new Date() })
+          .where(and(eq(toolConnections.id, connection.id), eq(toolConnections.companyId, companyId))).returning();
+        await tx.update(toolApplications).set({ status: "active", updatedAt: new Date() }).where(and(eq(toolApplications.id, connection.applicationId), eq(toolApplications.companyId, companyId)));
+        return row;
+      });
+      const details = await profileDetails(profile.id, companyId);
+      const policies = (await db.select().from(toolPolicies).where(eq(toolPolicies.companyId, companyId))).filter((policy) => asRecord(policy.config).connectionId === connection.id);
+      return { connection: toConnection(updated), profile: toProfile(profile), profileEntries: details.entries, profileBindings: details.bindings, policies: policies.map(toPolicy) };
+    }
     const enabledIds = [
       ...new Set([
         ...input.enabledCatalogEntryIds,
@@ -13825,7 +13918,7 @@ export function toolAccessService(
       connection.id,
       input.askFirstCatalogEntryIds,
     );
-    if (enabledRows.some((entry) => entry.status === "disabled")) {
+    if (enabledRows.some((entry) => entry.status === "disabled" && !(entry.quarantineReason === "mcp_tool_removed" && isRemoteMcpConnectorMethod(connection.config.sourceTemplateKey, connection.config.connectionMethodKey)))) {
       throw badRequest("Disabled actions cannot be enabled");
     }
     if (reviewedIds.length > 0) {
@@ -14032,6 +14125,28 @@ export function toolAccessService(
       }
 
       const profileBindings: ToolProfileBinding[] = [];
+      // These connectors expose one agent-access choice. Commit installation
+      // reach and permission bindings together, including an empty selection.
+      if (isRemoteMcpConnectorMethod(connection.config.sourceTemplateKey, connection.config.connectionMethodKey)) {
+        const existingInstalls = await tx.select().from(toolConnectionInstalls).where(and(
+          eq(toolConnectionInstalls.companyId, companyId), eq(toolConnectionInstalls.connectionId, connection.id),
+        ));
+        const desired = new Map(bindingInputs.flatMap((binding) => binding.targetType === "company" || binding.targetType === "agent"
+          ? [[`${binding.targetType}:${binding.targetId}`, { targetType: binding.targetType, targetId: binding.targetId }] as const] : []));
+        const removed = existingInstalls.filter((install) => !desired.has(`${install.targetType}:${install.targetId}`));
+        const added = [...desired.values()].filter((binding) => !existingInstalls.some((install) => install.targetType === binding.targetType && install.targetId === binding.targetId));
+        if (removed.length) await tx.delete(toolConnectionInstalls).where(inArray(toolConnectionInstalls.id, removed.map((install) => install.id)));
+        if (added.length) await tx.insert(toolConnectionInstalls).values(added.map((binding) => ({
+          companyId, connectionId: connection.id, targetType: binding.targetType, targetId: binding.targetId,
+          createdByAgentId: actor?.actorType === "agent" ? (actor.actorId ?? null) : null,
+          createdByUserId: actor?.actorType === "user" ? (actor.actorId ?? null) : null,
+        })));
+        if (added.length || removed.length) await tx.insert(toolAccessAuditEvents).values({
+          companyId, connectionId: connection.id, actorType: actor?.actorType ?? "system", actorId: actor?.actorId ?? null,
+          action: "connection_installs.changed", outcome: "success", reasonCode: "installs_changed",
+          details: { added: added.map(({ targetType, targetId }) => ({ targetType, targetId })), removed: removed.map(({ targetType, targetId }) => ({ targetType, targetId })) },
+        });
+      }
       for (const bindingInput of bindingInputs) {
         const [binding] = await tx
           .insert(toolProfileBindings)
@@ -14094,6 +14209,7 @@ export function toolAccessService(
               eq(toolCatalogEntries.companyId, companyId),
               inArray(toolCatalogEntries.id, enabledIds),
               ne(toolCatalogEntries.status, "quarantined"),
+              ne(toolCatalogEntries.status, "disabled"),
             ),
           );
       }
@@ -14228,6 +14344,7 @@ export function toolAccessService(
     actor?: ActorInfo,
   ): Promise<ToolConnectionHealthCheckResult> {
     const connection = await getConnectionRow(connectionId, companyId);
+    await assertMcpAggregatorSetupEnabled(connection.config.sourceTemplateKey, connection.config.connectionMethodKey);
     if (connection.status === "archived")
       throw conflict("Archived app connections cannot be reconnected");
     if (connection.credentialSource === "vercel_connect") {
@@ -14409,6 +14526,7 @@ export function toolAccessService(
     },
   ): Promise<ToolOAuthStartResult> {
     let connection = await getConnectionRow(connectionId, companyId);
+    await assertMcpAggregatorSetupEnabled(connection.config.sourceTemplateKey, connection.config.connectionMethodKey);
     if (connection.status === "archived")
       throw conflict("Archived app connections cannot start sign in");
     const sourceTemplateKey =
@@ -15143,10 +15261,11 @@ export function toolAccessService(
         : suggestedAgentIds.length > 0
           ? { agentIds: suggestedAgentIds }
           : "all_agents";
+    const remoteMcpAccess = isRemoteMcpConnectorMethod(input.connection.config.sourceTemplateKey, input.connection.config.connectionMethodKey);
     const access: FinishToolApp["access"] = deferTaskAccess
       ? { agentIds: [] }
       : installs.length === 0
-        ? normalizedSuggestedAccess
+        ? remoteMcpAccess ? { agentIds: [] } : normalizedSuggestedAccess
         : companyInstall
           ? "all_agents"
           : { agentIds };
@@ -15181,7 +15300,7 @@ export function toolAccessService(
       },
       input.actor,
     );
-    if (!deferTaskAccess && installs.length === 0) {
+    if (!deferTaskAccess && !remoteMcpAccess && installs.length === 0) {
       const installTargets =
         access === "all_agents"
           ? [
@@ -16856,6 +16975,7 @@ export function toolAccessService(
     if (!app || app.availability?.available === false)
       throw notFound("App not found");
     const method = connectionMethodFor(app, methodKey);
+    await assertMcpAggregatorSetupEnabled(app.slug, method.key);
     if (method.transport !== "mcp_remote" || !method.defaults?.serverUrl) {
       throw unprocessable(
         "This app method does not use a hosted remote MCP endpoint",

server/src/services/tool-gateway.ts

@@ -1,6 +1,7 @@
 import { runIdentityContexts } from "@paperclipai/db";
 import { captureRunIdentity } from "./run-identity.js";
 import { resolveManagedGitHubIdentitySelection } from "./git-credentials.js";
+import { extractRemoteMcpPending } from "./remote-mcp-pending.js";
 import { logger } from "../middleware/logger.js";
 import { spawn } from "node:child_process";
 import { createHash, randomBytes, randomUUID } from "node:crypto";
@@ -63,6 +64,7 @@ import type {
   ToolAccessDecision,
   ToolAccessDecisionInput,
   ToolConnectionTestCallStatus,
+  ToolUpstreamPending,
   ToolConnectionTestCallStatusPhase,
   ToolCredentialSecretRef,
   ToolMcpGateway,
@@ -89,6 +91,10 @@ import { railwayCommandBudgetMs, createRailwayClient, isRailwayConnection, isRai
 import { RAILWAY_SSH_SECRET_PATH, runRailwaySshCommand } from "./railway-ssh.js";
 import {
   initializeMcpHttpSession,
+  getMcpHttpSession,
+  forgetMcpHttpSessions,
+  readMcpHttpResponse,
+  McpHttpResponseError,
   mcpHttpRequestHeaders,
   parseMcpHttpResponseBody,
 } from "./mcp-http.js";
@@ -1049,6 +1055,28 @@ export function createToolGatewayService(
     now?: () => number;
   } = {},
 ) {
+  // Authorization links can contain one-time codes. Keep them briefly in memory;
+  // persist only the redacted request and execution identifiers for recovery.
+  const upstreamHandoffs = new Map<string, { pending: ToolUpstreamPending; expires: number }>();
+  async function retainUpstreamHandoff(invocationId: string, pending: ToolUpstreamPending) {
+    for (const [id, value] of upstreamHandoffs) if (value.expires <= Date.now()) upstreamHandoffs.delete(id);
+    while (upstreamHandoffs.size >= 256) upstreamHandoffs.delete(upstreamHandoffs.keys().next().value!);
+    upstreamHandoffs.set(invocationId, { pending, expires: Date.now() + 15 * 60_000 });
+    const summary = validateToolContent({
+      value: { upstreamPending: { ...pending, links: [] } }, direction: "result", sensitiveMode: "redact", promptInjectionMode: "ignore",
+    }).summary;
+    await db.update(toolInvocations).set({ resultSummary: summary }).where(eq(toolInvocations.id, invocationId));
+  }
+  function recoverUpstreamHandoff(invocation: typeof toolInvocations.$inferSelect): ToolUpstreamPending | undefined {
+    if (invocation.errorCode !== "provider_interaction_required") return undefined;
+    const cached = upstreamHandoffs.get(invocation.id);
+    if (cached && cached.expires > Date.now()) return cached.pending;
+    upstreamHandoffs.delete(invocation.id);
+    const stored = asRecord(storedInvocationResult(invocation));
+    const pending = asRecord(stored?.upstreamPending);
+    if (!pending || !["approval", "authorization"].includes(String(pending.kind))) return undefined;
+    return { ...pending, links: [] } as unknown as ToolUpstreamPending;
+  }
   const runtimeSupervisor = createToolRuntimeSupervisor(db, {
     deploymentMode: options.deploymentMode,
     deploymentExposure: options.deploymentExposure,
@@ -5843,7 +5871,8 @@ export function createToolGatewayService(
       }
       let requestHeaders = headers;
       if (connection.config.mcpSessionRequired === true) {
-        requestHeaders = await initializeMcpHttpSession({
+        requestHeaders = await getMcpHttpSession({
+          scope: `${connection.id}:grant:${grant.id}:actor:${session.agentId}:${endpoint}`,
           send: (init) =>
             dispatchRemote(endpoint, {
               ...init,
@@ -6004,7 +6033,32 @@ export function createToolGatewayService(
           headers: mcpHttpRequestHeaders(headers),
         });
       }
-      const body = await readBoundedRemoteResponse(response);
+      const sessionExpired = response.status === 404 && new Headers(requestHeaders).has("mcp-session-id");
+      if (sessionExpired) {
+        // The next explicit call initializes again. Never replay a tools/call
+        // automatically: the failed call may have changed app data.
+        forgetMcpHttpSessions(connection.id);
+      }
+      const body = response.ok
+        ? JSON.stringify(await readMcpHttpResponse(response, requestId, {
+            maxBytes: MAX_REMOTE_MCP_RESPONSE_BYTES,
+            onRequest: async (message) => {
+              const pending = extractRemoteMcpPending(message);
+              if (pending) {
+                await retainUpstreamHandoff(invocationId, pending);
+                // Defer this interaction to the user. Do not claim that consent
+                // was granted or repeat a potentially state-changing tool call.
+                await dispatchRemote(endpoint, {
+                  method: "POST", headers: mcpHttpRequestHeaders(requestHeaders),
+                  body: JSON.stringify({ jsonrpc: "2.0", id: message.id, result: { action: "cancel" } }),
+                });
+                throw new ToolGatewayHttpError(409, "This tool needs authorization in the provider.", "provider_interaction_required", { upstreamPending: pending, invocationId });
+              }
+              const request = extractMcpElicitationRequest(message);
+              if (request) await requestElicitationForRecordedToolCall({ session, tool, invocationId, request });
+            },
+          }))
+        : await readBoundedRemoteResponse(response);
       execution.response = {
         httpStatus: response.status,
         contentType: response.headers.get("content-type"),
@@ -6015,17 +6069,18 @@ export function createToolGatewayService(
           response.headers.get("traceparent"),
       };
       if (!response.ok) {
-        await markRemoteConnectionHealth(
-          connection,
-          "error",
-          "Remote MCP server returned an HTTP error.",
-        );
+        // Session expiration is recoverable on an explicit retry. Marking the
+        // connection unhealthy here would hide every tool and prevent it.
+        if (!sessionExpired) {
+          await markRemoteConnectionHealth(connection, "error", "Remote MCP server returned an HTTP error.");
+        }
         throw new ToolGatewayHttpError(
           502,
-          "Remote MCP server returned an HTTP error",
+          sessionExpired ? "Remote MCP session expired. Retry the action explicitly to start a new session." : "Remote MCP server returned an HTTP error",
           "mcp_remote_status",
           {
             status: response.status,
+            ...(sessionExpired ? { sessionExpired: true } : {}),
             connectionId: connection.id,
             catalogEntryId: entry.id,
             execution,
@@ -6034,10 +6089,7 @@ export function createToolGatewayService(
       }
       let payload: unknown;
       try {
-        payload = parseMcpHttpResponseBody(
-          body,
-          response.headers.get("content-type"),
-        );
+        payload = JSON.parse(body);
       } catch {
         await markRemoteConnectionHealth(
           connection,
@@ -6057,6 +6109,11 @@ export function createToolGatewayService(
       }
       const payloadRecord = asRecord(payload);
       if (!payloadRecord) throw malformedRemoteMcpResponse();
+      const upstreamPending = extractRemoteMcpPending(payloadRecord, String(connection.config.sourceTemplateKey ?? ""), entry.toolName);
+      if (upstreamPending) {
+        await retainUpstreamHandoff(invocationId, upstreamPending);
+        throw new ToolGatewayHttpError(409, "Complete the provider's authorization or approval before continuing. The original call has not been replayed.", "provider_interaction_required", { upstreamPending, invocationId });
+      }
       const topLevelElicitation = extractMcpElicitationRequest(payloadRecord);
       if (topLevelElicitation) {
         await requestElicitationForRecordedToolCall({
@@ -6117,6 +6174,15 @@ export function createToolGatewayService(
       );
       return { result, headerSummary, execution };
     } catch (error) {
+      if (error instanceof McpHttpResponseError) {
+        const failure = error.reason === "too_large" ? responseTooLargeError()
+          : error.reason === "malformed_response" ? malformedRemoteMcpResponse()
+          : new ToolGatewayHttpError(502, "Remote MCP server returned invalid JSON", "mcp_remote_invalid_json");
+        await markRemoteConnectionHealth(connection, "error", failure.message);
+        throw new ToolGatewayHttpError(failure.status, failure.message, failure.reasonCode, {
+          connectionId: connection.id, catalogEntryId: entry.id, execution,
+        });
+      }
       if (error instanceof RailwayError) {
         throw new ToolGatewayHttpError(error.status, error.message, error.code, { connectionId: connection.id, catalogEntryId: entry.id, execution });
       }
@@ -7181,6 +7247,9 @@ export function createToolGatewayService(
         decision: "allowed" as const,
         invocationId: args.invocationId,
         error: { message, reasonCode },
+        ...(err instanceof ToolGatewayHttpError && err.reasonCode === "provider_interaction_required"
+          ? { upstreamPending: err.details.upstreamPending as ToolUpstreamPending }
+          : {}),
       };
     }
   }
@@ -7977,12 +8046,14 @@ export function createToolGatewayService(
             "executing",
             "rejected",
             "executed",
+            "failed",
           ]),
         ),
       )
       .orderBy(desc(toolActionRequests.createdAt))
       .limit(1);
     if (!match) return null;
+    if (match.actionRequest.status === "failed" && match.invocation.errorCode !== "provider_interaction_required") return null;
     // The gateway builds an ask-first request in two steps inside one call: it
     // inserts the row with a null signature and a null expiry, then signs the
     // row and sets the expiry. A concurrent matching call can observe the row in
@@ -8041,6 +8112,14 @@ export function createToolGatewayService(
     const match = await matchingAgentActionRequest(input);
     if (!match) return null;
     const { actionRequest, invocation } = match;
+    if (actionRequest.status === "failed") {
+      throw new ToolGatewayHttpError(409,
+        "The provider is waiting for authorization or approval. Continue the existing execution; do not repeat the original call.",
+        "provider_interaction_required", {
+          invocationId: invocation.id, actionRequestId: actionRequest.id,
+          upstreamPending: recoverUpstreamHandoff(invocation),
+        });
+    }
     if (actionRequest.status === "pending") {
       await throwApprovalRequired({
         invocationId: invocation.id,
@@ -8126,7 +8205,7 @@ export function createToolGatewayService(
       phase = "expired";
     } else if (
       actionRequest.status === "approved" ||
-      actionRequest.status === "executed"
+      actionRequest.status === "executed" || actionRequest.status === "failed"
     ) {
       phase = invocationDone ? "done" : "running";
     } else {
@@ -8200,6 +8279,7 @@ export function createToolGatewayService(
       parameters,
       ...(result !== undefined ? { result } : {}),
       ...(error ? { error } : {}),
+      ...(recoverUpstreamHandoff(invocation) ? { upstreamPending: recoverUpstreamHandoff(invocation) } : {}),
       durationMs,
       requestedAt: actionRequest.createdAt.toISOString(),
       resolvedAt: actionRequest.resolvedAt