Carry account isolation and generation receipts into saved setup flows

Co-Authored-By: Paperclip <noreply@paperclip.ing>

* codex/agent-connection-api-parity:
  Enforce connection scopes and human ownership in account discovery

server/src/__tests__/connection-management.integration.test.ts

@@ -4,11 +4,12 @@ import { join } from "node:path";
 import { createServer } from "node:http";
 import { and, eq } from "drizzle-orm";
 import { beforeAll, afterAll, describe, expect, it, vi } from "vitest";
-import { activityLog, agentApiKeys, companies, chatEndpoints, chatExternalPrincipals, chatIdentityLinks, toolPolicies, connectionGrants, connectionGrantDelegations, toolAccessAuditEvents, toolOauthStates, issueThreadInteractions, connectionIntentDeliveries, agents, companyMemberships, heartbeatRuns, issues, principalPermissionGrants, projects, toolConnections, toolMcpGateways, toolMcpGatewayTokens, toolCatalogEntries, toolInvocations, toolProfileBindings, plugins, adapterAuthSessions, environments, routines, chatGitHubReviews } from "@paperclipai/db";
+import { activityLog, agentApiKeys, companies, companySecrets, chatEndpoints, chatExternalPrincipals, chatIdentityLinks, toolPolicies, connectionGrants, connectionGrantDelegations, toolAccessAuditEvents, toolOauthStates, issueThreadInteractions, connectionIntentDeliveries, agents, companyMemberships, heartbeatRuns, issues, principalPermissionGrants, projects, toolConnections, toolMcpGateways, toolMcpGatewayTokens, toolCatalogEntries, toolInvocations, toolProfileBindings, plugins, adapterAuthSessions, environments, routines, chatGitHubReviews } from "@paperclipai/db";
 import { findSatisfiedToolConnection, satisfiedConnectionIntentService } from "../services/satisfied-connection-intents.js";
 import { connectionManagementService, continuingConnectionManagement } from "../services/connection-management.js";
 import { redactRunnerApiPayloads } from "../services/native-runtime/runner-api-log-redaction.js";
 import * as remoteEndpointGuard from "../services/remote-http-endpoint-guard.js";
+import * as remoteHttpFetch from "../services/remote-http-fetch.js";
 import * as runIdentity from "../services/run-identity.js";
 import { createLocalAgentJwt } from "../agent-auth-jwt.js";
 import { startRunnerApiTestServer } from "./helpers/runner-api-server.js";
@@ -714,11 +715,13 @@ describe("connection management with production default agents", () => {
       expect((await server.db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, rotationIntentId)))[0].status).toBe("pending");
       const humanRotation = await humanRequest(`/api/chat-endpoints/${created.body.id}/setup-secret`, "POST", {});
       expect(humanRotation.status).toBe(201);
-      expect(await humanRotation.json()).toEqual({ webhookSecret: expect.stringMatching(/^[a-f0-9]{64}$/) });
+      const generatedRotation = await humanRotation.json();
+      expect(generatedRotation).toMatchObject({ webhookSecret: expect.stringMatching(/^[a-f0-9]{64}$/), credentialRotationId: expect.any(String) });
       const [humanRotated] = await server.db.select().from(toolConnections).where(eq(toolConnections.id, created.body.connectionId));
       expect(humanRotated.credentialSecretRefs).not.toEqual(beforeRotation.credentialSecretRefs);
       const rotationState = await f.request("GET", `/chat-endpoints/${endpoint.id}`);
       expect(rotationState.body.setup).toMatchObject({ credentialRotationId: expect.any(String), webhookVerifiedAt: null });
+      expect(rotationState.body.setup.credentialRotationId).toBe(generatedRotation.credentialRotationId);
       const nextRotation = await f.request("POST", `/chat-endpoints/${endpoint.id}/setup-secret`, {});
       expect(nextRotation.body.requiredUserAction.setupPath).toContain(`afterRotationId=${rotationState.body.setup.credentialRotationId}`);
       expect(await findSatisfiedToolConnection(server.db, task, rotationIntent)).toBeNull();
@@ -747,9 +750,54 @@ describe("connection management with production default agents", () => {
       expect((await f.request("GET", `/companies/${f.companyId}/email/connections`)).body).toEqual(expect.arrayContaining([expect.objectContaining({ id: created.body.id })]));
       const [saved] = await server.db.select().from(toolConnections).where(eq(toolConnections.id, created.body.id));
       expect(saved.createdByUserId).toBe(f.userId);
+      const other = await f.request("POST", `/companies/${f.companyId}/email/connections`, { ...body, apiKey: "other-agentmail-secret", grantKind: "organization", idempotencyKey: randomUUID() });
+      expect(other).toMatchObject({ status: 201 });
+      await server.db.update(principalPermissionGrants).set({ scope: { connectionIds: [created.body.id] } }).where(and(eq(principalPermissionGrants.principalId, f.agentId), eq(principalPermissionGrants.permissionKey, "tools:manage_connections")));
+      for (const token of [f.key.token, f.jwt]) {
+        const listed = await f.request("GET", `/companies/${f.companyId}/email/connections?connectionId=${created.body.id}`, undefined, token);
+        expect(listed, JSON.stringify(listed)).toMatchObject({ status: 200 });
+        expect(listed.body.map((account: { id: string }) => account.id)).toEqual([created.body.id]);
+      }
     } finally { provider = null; }
   });
 
+  it("saves delegated Arcade discovery for the responsible user and shares its inventory with that user", async () => {
+    const f = await fixture();
+    const transport = vi.spyOn(remoteHttpFetch, "guardedRemoteHttpFetch").mockImplementation(async (url, init) => provider!(String(url), init));
+    provider = async (url, init) => {
+      const path = new URL(url).pathname;
+      const rpc = init?.body ? JSON.parse(String(init.body)) : null;
+      if (rpc?.method === "initialize") return Response.json({ jsonrpc: "2.0", id: rpc.id, result: { protocolVersion: "2025-03-26", capabilities: { tools: {} }, serverInfo: { name: "fixture", version: "1" } } });
+      if (rpc?.method === "notifications/initialized") return new Response(null, { status: 202 });
+      if (rpc?.method === "tools/list") return Response.json({ jsonrpc: "2.0", id: rpc.id, result: { tools: [{ name: "Notion.ListPages", inputSchema: { type: "object" } }] } });
+      expect(new Headers(init?.headers).get("Authorization")).toBe("Bearer fixture-arcade-discovery");
+      if (path === "/v1/tools") return Response.json({ items: [{ qualified_name: "Notion.ListPages", toolkit: { name: "Notion" }, requirements: { met: true, authorization: { provider_id: "notion", token_status: "completed" } } }] });
+      if (path === "/v1/admin/user_connections") return Response.json({ items: [{ id: "arcade-work", provider_id: "notion", user_id: "arcade-user", connection_status: "active" }] });
+      throw new Error(`Unexpected Arcade fixture operation: ${path}`);
+    };
+    try {
+      const created = await f.request("POST", `/companies/${f.companyId}/tools/connections`, { name: "Arcade fixture", transport: "mcp_remote", authKind: "none", credentialPolicy: "shared", status: "active", enabled: true, config: { sourceTemplateKey: "arcade", url: "https://api.arcade.dev/mcp/test" } });
+      expect(created, JSON.stringify(created)).toMatchObject({ status: 201 });
+      const configured = await f.request("PUT", `/tool-connections/${created.body.id}/aggregator/discovery`, { apiKey: "fixture-arcade-discovery", userId: "arcade-user" }, f.jwt);
+      expect(configured, JSON.stringify(configured)).toMatchObject({ status: 200 });
+      const [saved] = await server.db.select().from(toolConnections).where(eq(toolConnections.id, created.body.id));
+      const discovery = saved.config.aggregatorDiscovery as Record<string, { secretId: string }>;
+      expect(discovery[f.agentId]).toBeUndefined();
+      expect(discovery[f.userId]).toMatchObject({ userId: "arcade-user", secretId: expect.any(String) });
+      const [secret] = await server.db.select().from(companySecrets).where(eq(companySecrets.id, discovery[f.userId].secretId));
+      expect(secret.ownerUserId).toBe(f.userId);
+      const events = await server.db.select().from(activityLog).where(eq(activityLog.entityId, saved.id));
+      expect(events).toEqual(expect.arrayContaining([expect.objectContaining({ action: "tool_connection.account_sync_configured", actorType: "agent", actorId: f.agentId, responsibleUserId: f.userId })]));
+      const cookie = await server.humanSession(f.userId);
+      await vi.waitFor(async () => {
+        const response = await fetch(`${server.apiUrl}/api/tool-connections/${saved.id}/aggregator/apps`, { headers: { cookie } });
+        const result = await response.json();
+        expect(result.discovery).toMatchObject({ availability: "available" });
+        expect(result.apps).toEqual(expect.arrayContaining([expect.objectContaining({ provider: "arcade", appSlug: "notion" })]));
+      }, { timeout: 10_000 });
+    } finally { transport.mockRestore(); provider = null; }
+  });
+
   it("keeps durable gateway client tokens bound to the issuer and responsible user", async () => {
     const f = await fixture();
     const profile = await f.request("POST", `/companies/${f.companyId}/tools/profiles`, { profileKey: "gateway-profile", name: "Gateway profile", entries: [] });

server/src/__tests__/openapi-routes.test.ts

@@ -797,7 +797,7 @@ describe("openapi routes", () => {
     ).toMatchObject({
       type: "object",
       additionalProperties: false,
-      required: ["webhookSecret"],
+      required: ["webhookSecret", "credentialRotationId"],
     });
     expect(setupSecret.responses["409"]).toBeDefined();
     expect(setupSecret.responses["422"]).toBeDefined();

server/src/routes/openapi.ts

@@ -859,7 +859,7 @@ const chatEndpointSetupResponseSchema = z
   .strict();
 
 const chatEndpointSetupSecretResponseSchema = z
-  .object({ webhookSecret: z.string().length(64) })
+  .object({ webhookSecret: z.string().length(64), credentialRotationId: z.uuid() })
   .strict();
 
 const chatEndpointCredentialHandoffSchema = z.object({

server/src/services/chat-channels/github/setup-secret.ts

@@ -305,7 +305,7 @@ export function createGitHubSetupSecret(db: Db, options: GitHubSetupSecretOption
               "GitHub setup secret rotated but completion audit failed",
             );
           });
-        return { webhookSecret };
+        return { webhookSecret, credentialRotationId: rotationId };
       },
       {
         beforeFinalOwnershipCheck: options.setupSecretFinalOwnershipBarrier,

server/src/services/email-connections.ts

@@ -15,6 +15,7 @@ import { lockToolConnectionRuntimeInputs, toolAccessService } from "./tool-acces
 import { agentmailApi } from "./agentmail-api.js";
 import { logActivity } from "./activity-log.js";
 import type { EmailActor } from "./email-channels.js";
+import { connectionManagementService } from "./connection-management.js";
 
 export function emailConnectionService(
   db: Db,
@@ -76,6 +77,8 @@ export function emailConnectionService(
     return connection;
   }
   async function listCredentials(companyId: string, actor: EmailActor): Promise<EmailCredentialOption[]> {
+    const management = actor.managementContext;
+    if (actor.agentId && !management) throw forbidden("Connection management authority is required");
     const connections = await db.select().from(toolConnections).where(and(
       eq(toolConnections.companyId, companyId), eq(toolConnections.status, "active"),
       eq(toolConnections.enabled, true),
@@ -92,6 +95,11 @@ export function emailConnectionService(
     await Promise.all(Array.from({ length: Math.min(4, candidates.length) }, async () => {
       while (next < candidates.length) {
         const connection = candidates[next++];
+        if (actor.agentId && management) {
+          const access = await connectionManagementService(db).capability(management.actor, "connections:read",
+            { ...management.resource, connectionId: connection.id }, { mutation: false, readContext: management });
+          if (!access.allowed) continue;
+        }
         try {
           await get(companyId, connection.id, actor);
         } catch (error) {

server/src/services/tool-access/catalog/aggregator-accounts.ts

@@ -23,6 +23,12 @@ import type { ConnectionCredentialAccess } from "../credential-access.js";
 import type { ToolRemoteHttp } from "../remote-http.js";
 import type { ToolCatalogDiscovery } from "./discovery.js";
 
+function aggregatorUserId(actor: ActorInfo): string {
+  const userId = connectionActorUserId(actor);
+  if (!userId) throw forbidden("An authorized responsible user is required");
+  return userId;
+}
+
 function asRecord(value: unknown): Record<string, unknown> {
   if (value && typeof value === "object" && !Array.isArray(value))
     return value as Record<string, unknown>;
@@ -51,7 +57,7 @@ export function createAggregatorAccounts(db: Db, { options, secrets, audit, cred
     const usesConnectionRefs = ["arcade", "executor"].includes(String(connection.config.sourceTemplateKey)) && connection.authKind !== "oauth" && connection.credentialPolicy === "shared";
     const grant = usesConnectionRefs ? null : await vaultGrantForConnection(connection, actor);
     const refs = grant?.credentialSecretRefs ?? connection.credentialSecretRefs;
-    const discovery = asRecord(asRecord(connection.config.aggregatorDiscovery)[actor.actorId!]);
+    const discovery = asRecord(asRecord(connection.config.aggregatorDiscovery)[aggregatorUserId(actor)]);
     const ids = [...new Set([...refs.map(ref => ref.secretId), ...(typeof discovery.secretId === "string" ? [discovery.secretId] : [])])].sort();
     const versions = ids.length ? await db.select({ id: companySecrets.id, version: companySecrets.latestVersion }).from(companySecrets)
       .where(and(eq(companySecrets.companyId, connection.companyId), inArray(companySecrets.id, ids))).orderBy(asc(companySecrets.id)) : [];
@@ -65,7 +71,7 @@ export function createAggregatorAccounts(db: Db, { options, secrets, audit, cred
     const credentialKey = await composioCredentialKey(connection, actor);
     const rows = await db.select().from(toolConnectionAppSnapshots).where(and(
       eq(toolConnectionAppSnapshots.companyId, connection.companyId), eq(toolConnectionAppSnapshots.connectionId, connection.id),
-      eq(toolConnectionAppSnapshots.userId, actor.actorId!), eq(toolConnectionAppSnapshots.credentialKey, credentialKey),
+      eq(toolConnectionAppSnapshots.userId, aggregatorUserId(actor)), eq(toolConnectionAppSnapshots.credentialKey, credentialKey),
     ));
     return rows.map(row => ({ connectionId: row.connectionId, toolkit: row.toolkit, status: row.status, accounts: row.accounts,
       checkedAt: row.checkedAt.toISOString(), errorAt: row.errorAt?.toISOString() ?? null }));
@@ -75,7 +81,7 @@ export function createAggregatorAccounts(db: Db, { options, secrets, audit, cred
     const credentialKey = await composioCredentialKey(connection, actor);
     const [sync] = await db.select().from(toolConnectionAppSyncs).where(and(
       eq(toolConnectionAppSyncs.companyId, connection.companyId), eq(toolConnectionAppSyncs.connectionId, connection.id),
-      eq(toolConnectionAppSyncs.userId, actor.actorId!), eq(toolConnectionAppSyncs.credentialKey, credentialKey),
+      eq(toolConnectionAppSyncs.userId, aggregatorUserId(actor)), eq(toolConnectionAppSyncs.credentialKey, credentialKey),
     ));
     const interrupted = sync?.status === "syncing" && sync.updatedAt.getTime() < Date.now() - composioSyncLeaseMs;
     return { apps: await cachedComposioApps(connection, actor), sync: {
@@ -115,7 +121,7 @@ export function createAggregatorAccounts(db: Db, { options, secrets, audit, cred
         const names = toolkits.filter(tool => tool.action === "list").map(tool => tool.name);
         if (names.length) await db.update(toolConnectionAppSnapshots).set({ errorAt: checkedAt }).where(and(
           eq(toolConnectionAppSnapshots.companyId, connection.companyId), eq(toolConnectionAppSnapshots.connectionId, connectionId),
-          eq(toolConnectionAppSnapshots.userId, actor.actorId!), eq(toolConnectionAppSnapshots.credentialKey, credentialKey),
+          eq(toolConnectionAppSnapshots.userId, aggregatorUserId(actor)), eq(toolConnectionAppSnapshots.credentialKey, credentialKey),
           inArray(toolConnectionAppSnapshots.toolkit, names), lte(toolConnectionAppSnapshots.checkedAt, checkedAt),
         ));
       };
@@ -145,7 +151,7 @@ export function createAggregatorAccounts(db: Db, { options, secrets, audit, cred
         } catch (error) {
           await db.update(toolConnectionAppSnapshots).set({ errorAt: checkedAt }).where(and(
             eq(toolConnectionAppSnapshots.companyId, connection.companyId), eq(toolConnectionAppSnapshots.connectionId, connectionId),
-            eq(toolConnectionAppSnapshots.userId, actor.actorId!), eq(toolConnectionAppSnapshots.credentialKey, credentialKey),
+            eq(toolConnectionAppSnapshots.userId, aggregatorUserId(actor)), eq(toolConnectionAppSnapshots.credentialKey, credentialKey),
             eq(toolConnectionAppSnapshots.toolkit, tool.name), lte(toolConnectionAppSnapshots.checkedAt, checkedAt),
           ));
           if (!tolerateToolkitErrors) throw error;
@@ -153,7 +159,7 @@ export function createAggregatorAccounts(db: Db, { options, secrets, audit, cred
           continue;
         }
         const status = accounts.some(account => account.status === "ACTIVE") ? "connected" as const : "not_connected" as const;
-        await db.insert(toolConnectionAppSnapshots).values({ companyId: connection.companyId, connectionId, userId: actor.actorId!,
+        await db.insert(toolConnectionAppSnapshots).values({ companyId: connection.companyId, connectionId, userId: aggregatorUserId(actor),
           credentialKey, toolkit: tool.name, accounts, status, checkedAt, errorAt: null,
         }).onConflictDoUpdate({ target: [toolConnectionAppSnapshots.companyId, toolConnectionAppSnapshots.connectionId, toolConnectionAppSnapshots.userId, toolConnectionAppSnapshots.toolkit],
           set: { accounts, status, checkedAt, credentialKey, errorAt: null }, setWhere: lte(toolConnectionAppSnapshots.checkedAt, checkedAt) });
@@ -173,7 +179,7 @@ export function createAggregatorAccounts(db: Db, { options, secrets, audit, cred
     const leaseId = randomUUID();
     const startedAt = new Date();
     const [lease] = await db.insert(toolConnectionAppSyncs).values({ companyId: connection.companyId, connectionId,
-      userId: actor.actorId!, credentialKey, leaseId, status: "syncing", total: toolkits.length, updatedAt: startedAt,
+      userId: aggregatorUserId(actor), credentialKey, leaseId, status: "syncing", total: toolkits.length, updatedAt: startedAt,
     }).onConflictDoUpdate({ target: [toolConnectionAppSyncs.companyId, toolConnectionAppSyncs.connectionId, toolConnectionAppSyncs.userId, toolConnectionAppSyncs.credentialKey],
       set: { leaseId, status: "syncing", checked: 0, total: toolkits.length, failed: 0, updatedAt: startedAt },
       setWhere: and(
@@ -209,7 +215,7 @@ export function createAggregatorAccounts(db: Db, { options, secrets, audit, cred
           // Preserve observations, but never present a failed check as current authorization.
           await db.update(toolConnectionAppSnapshots).set({ errorAt: new Date() }).where(and(
             eq(toolConnectionAppSnapshots.companyId, connection.companyId), eq(toolConnectionAppSnapshots.connectionId, connectionId),
-            eq(toolConnectionAppSnapshots.userId, actor.actorId!), eq(toolConnectionAppSnapshots.credentialKey, credentialKey),
+            eq(toolConnectionAppSnapshots.userId, aggregatorUserId(actor)), eq(toolConnectionAppSnapshots.credentialKey, credentialKey),
             lte(toolConnectionAppSnapshots.checkedAt, startedAt),
           ));
           await updateLease({ status: "error", failed }).catch(() => undefined);
@@ -236,7 +242,7 @@ export function createAggregatorAccounts(db: Db, { options, secrets, audit, cred
     const provider = aggregatorProvider(connection);
     if (connection.status !== "active" || !connection.enabled) return { availability: "disabled", message: "Restore this gateway to check its accounts." };
     if (provider === "arcade") {
-      const discovery = asRecord(asRecord(connection.config.aggregatorDiscovery)[actor.actorId!]);
+      const discovery = asRecord(asRecord(connection.config.aggregatorDiscovery)[aggregatorUserId(actor)]);
       const headers = projectedConnectionHeaders(connection);
       const grant = connection.authKind === "oauth" || connection.credentialPolicy !== "shared" ? await vaultGrantForConnection(connection, actor) : null;
       const refs = grant?.credentialSecretRefs ?? connection.credentialSecretRefs;
@@ -257,7 +263,7 @@ export function createAggregatorAccounts(db: Db, { options, secrets, audit, cred
       const credentialKey = await composioCredentialKey(connection, actor);
       const [sync] = await db.select({ status: toolConnectionAppSyncs.status }).from(toolConnectionAppSyncs).where(and(
         eq(toolConnectionAppSyncs.companyId, connection.companyId), eq(toolConnectionAppSyncs.connectionId, connection.id),
-        eq(toolConnectionAppSyncs.userId, actor.actorId!), eq(toolConnectionAppSyncs.credentialKey, credentialKey),
+        eq(toolConnectionAppSyncs.userId, aggregatorUserId(actor)), eq(toolConnectionAppSyncs.credentialKey, credentialKey),
       ));
       if (sync?.status === "unsupported") return { availability: "unsupported", message: "Executor account discovery is unavailable on this server. You can still use the gateway’s actions." };
     }
@@ -286,14 +292,14 @@ export function createAggregatorAccounts(db: Db, { options, secrets, audit, cred
     if (provider === "arcade") {
       // The extra project credential never joins invocation headers or tool grants.
       if (endpoint.hostname !== "api.arcade.dev" || endpoint.protocol !== "https:") throw unprocessable("Account sync requires an Arcade Cloud gateway");
-      const discovery = asRecord(asRecord(connection.config.aggregatorDiscovery)[actor.actorId!]);
+      const discovery = asRecord(asRecord(connection.config.aggregatorDiscovery)[aggregatorUserId(actor)]);
       let key: string | undefined;
       let userId: string | undefined;
       if (typeof discovery.secretId === "string" && typeof discovery.userId === "string") {
-        const [secret] = await db.select().from(companySecrets).where(and(eq(companySecrets.id, discovery.secretId), eq(companySecrets.companyId, connection.companyId), eq(companySecrets.ownerUserId, actor.actorId!)));
+        const [secret] = await db.select().from(companySecrets).where(and(eq(companySecrets.id, discovery.secretId), eq(companySecrets.companyId, connection.companyId), eq(companySecrets.ownerUserId, aggregatorUserId(actor))));
         if (!secret?.userSecretDefinitionId) throw forbidden("Your Arcade discovery credential is unavailable");
-        const resolved = await secrets.resolveUserSecretValue(connection.companyId, { definitionId: secret.userSecretDefinitionId, responsibleUserId: actor.actorId!, required: true }, {
-          actorType: "user", actorId: actor.actorId!, consumerType: "tool_connection", consumerId: connection.id, responsibleUserId: actor.actorId!,
+        const resolved = await secrets.resolveUserSecretValue(connection.companyId, { definitionId: secret.userSecretDefinitionId, responsibleUserId: aggregatorUserId(actor), required: true }, {
+          actorType: actor.actorType === "agent" ? "agent" : "user", actorId: actor.actorId!, heartbeatRunId: actor.managementContext?.workContext.runId, consumerType: "tool_connection", consumerId: connection.id, responsibleUserId: aggregatorUserId(actor),
         });
         key = resolved?.value;
         userId = discovery.userId;
@@ -340,7 +346,7 @@ export function createAggregatorAccounts(db: Db, { options, secrets, audit, cred
     const credentialKey = await composioCredentialKey(connection, actor);
     const leaseId = randomUUID();
     const startedAt = new Date();
-    const [lease] = await db.insert(toolConnectionAppSyncs).values({ companyId: connection.companyId, connectionId, userId: actor.actorId!, credentialKey, leaseId, status: "syncing", updatedAt: startedAt })
+    const [lease] = await db.insert(toolConnectionAppSyncs).values({ companyId: connection.companyId, connectionId, userId: aggregatorUserId(actor), credentialKey, leaseId, status: "syncing", updatedAt: startedAt })
       .onConflictDoUpdate({ target: [toolConnectionAppSyncs.companyId, toolConnectionAppSyncs.connectionId, toolConnectionAppSyncs.userId, toolConnectionAppSyncs.credentialKey],
         set: { leaseId, status: "syncing", checked: 0, total: 0, failed: 0, updatedAt: startedAt },
         setWhere: and(or(ne(toolConnectionAppSyncs.status, "syncing"), lt(toolConnectionAppSyncs.updatedAt, new Date(Date.now() - composioSyncLeaseMs))),
@@ -358,9 +364,9 @@ export function createAggregatorAccounts(db: Db, { options, secrets, audit, cred
             const checkedAt = new Date();
             // Commit the entire enumeration at once; a failed later page cannot delete accounts.
             await tx.update(toolConnectionAppSnapshots).set({ accounts: [], status: "not_connected", checkedAt, errorAt: null }).where(and(
-              eq(toolConnectionAppSnapshots.companyId, connection.companyId), eq(toolConnectionAppSnapshots.connectionId, connectionId), eq(toolConnectionAppSnapshots.userId, actor.actorId!), eq(toolConnectionAppSnapshots.credentialKey, credentialKey),
+              eq(toolConnectionAppSnapshots.companyId, connection.companyId), eq(toolConnectionAppSnapshots.connectionId, connectionId), eq(toolConnectionAppSnapshots.userId, aggregatorUserId(actor)), eq(toolConnectionAppSnapshots.credentialKey, credentialKey),
             ));
-            for (const app of apps) await tx.insert(toolConnectionAppSnapshots).values({ companyId: connection.companyId, connectionId, userId: actor.actorId!, credentialKey,
+            for (const app of apps) await tx.insert(toolConnectionAppSnapshots).values({ companyId: connection.companyId, connectionId, userId: aggregatorUserId(actor), credentialKey,
               toolkit: app.toolkit, accounts: app.accounts, status: app.accounts.some(account => account.status === "ACTIVE") ? "connected" : "not_connected", checkedAt,
             }).onConflictDoUpdate({ target: [toolConnectionAppSnapshots.companyId, toolConnectionAppSnapshots.connectionId, toolConnectionAppSnapshots.userId, toolConnectionAppSnapshots.toolkit],
               set: { accounts: app.accounts, status: app.accounts.some(account => account.status === "ACTIVE") ? "connected" : "not_connected", checkedAt, errorAt: null, credentialKey },
@@ -371,7 +377,7 @@ export function createAggregatorAccounts(db: Db, { options, secrets, audit, cred
           await db.transaction(async tx => {
             const [active] = await tx.select().from(toolConnectionAppSyncs).where(and(eq(toolConnectionAppSyncs.id, lease.id), eq(toolConnectionAppSyncs.leaseId, leaseId))).for("update");
             if (!active) return;
-            await tx.update(toolConnectionAppSnapshots).set({ errorAt: new Date() }).where(and(eq(toolConnectionAppSnapshots.companyId, connection.companyId), eq(toolConnectionAppSnapshots.connectionId, connectionId), eq(toolConnectionAppSnapshots.userId, actor.actorId!), eq(toolConnectionAppSnapshots.credentialKey, credentialKey)));
+            await tx.update(toolConnectionAppSnapshots).set({ errorAt: new Date() }).where(and(eq(toolConnectionAppSnapshots.companyId, connection.companyId), eq(toolConnectionAppSnapshots.connectionId, connectionId), eq(toolConnectionAppSnapshots.userId, aggregatorUserId(actor)), eq(toolConnectionAppSnapshots.credentialKey, credentialKey)));
             await tx.update(toolConnectionAppSyncs).set({ status: error instanceof AggregatorDiscoveryUnavailableError ? "unsupported" : "error", failed: 1, updatedAt: new Date() }).where(eq(toolConnectionAppSyncs.id, lease.id));
           });
         }
@@ -390,17 +396,17 @@ export function createAggregatorAccounts(db: Db, { options, secrets, audit, cred
       const [connection] = await tx.select().from(toolConnections).where(eq(toolConnections.id, connectionId)).for("update");
       if (!connection || aggregatorProvider(connection) !== "arcade" || connection.status === "archived") throw notFound("Arcade gateway not found");
       const all = asRecord(connection.config.aggregatorDiscovery);
-      const before = asRecord(all[actor.actorId!]);
+      const before = asRecord(all[aggregatorUserId(actor)]);
       const { secret } = await writeConnectionCredential(tx, {
         companyId: connection.companyId, connectionName: connection.name, configPath: "aggregatorDiscovery.apiKey", label: "Arcade account sync API key",
-        value: input.apiKey, ownerUserId: actor.actorId!, actor: actorForSecret(actor),
+        value: input.apiKey, ownerUserId: aggregatorUserId(actor), actor: actorForSecret(actor),
         existingRef: typeof before.secretId === "string" ? { secretId: before.secretId, configPath: "aggregatorDiscovery.apiKey", versionSelector: "latest" } : undefined,
         definitionKey: `arcade_discovery.${connection.id}`,
       });
-      const config = { ...connection.config, aggregatorDiscovery: { ...all, [actor.actorId!]: { userId: input.userId, secretId: secret.id } } };
+      const config = { ...connection.config, aggregatorDiscovery: { ...all, [aggregatorUserId(actor)]: { userId: input.userId, secretId: secret.id } } };
       await tx.update(toolConnections).set({ config, transportConfig: config, updatedAt: new Date() }).where(eq(toolConnections.id, connectionId));
     });
-    await logActivity(db, { companyId, actorType: "user", actorId: actor.actorId!, action: "tool_connection.account_sync_configured", entityType: "tool_connection", entityId: connectionId, details: { provider: "arcade" } });
+    await logActivity(db, { companyId, ...(actor.managementContext?.audit ?? { actorType: "user" as const, actorId: actor.actorId! }), action: "tool_connection.account_sync_configured", entityType: "tool_connection", entityId: connectionId, details: { provider: "arcade" } });
     return syncAggregatorApps(connectionId, true, actor);
   }
   return {
@@ -470,7 +476,7 @@ export function createAggregatorAccounts(db: Db, { options, secrets, audit, cred
         if (!before?.some(account => account.id === input.accountId)) throw notFound("This account is no longer available in the selected Composio app");
       }
       const details = { toolkit, ...(input.action !== "add" ? { accountId: input.accountId } : {}) };
-      await logActivity(db, { companyId: connection.companyId, actorType: "user", actorId: actor.actorId!,
+      await logActivity(db, { companyId: connection.companyId, ...(actor.managementContext?.audit ?? { actorType: "user" as const, actorId: actor.actorId! }),
         action: `tool_connection.upstream_account_${input.action}_requested`, entityType: "tool_connection", entityId: connectionId, details });
       let payload: unknown;
       try {
@@ -490,7 +496,7 @@ export function createAggregatorAccounts(db: Db, { options, secrets, audit, cred
           throw conflict("Composio has not confirmed this change. Refresh the accounts before trying again.");
         }
       }
-      await logActivity(db, { companyId: connection.companyId, actorType: "user", actorId: actor.actorId!,
+      await logActivity(db, { companyId: connection.companyId, ...(actor.managementContext?.audit ?? { actorType: "user" as const, actorId: actor.actorId! }),
         action: `tool_connection.upstream_account_${input.action}`, entityType: "tool_connection", entityId: connectionId,
         details: { toolkit, ...(input.action !== "add" ? { accountId: input.accountId } : {}) } });
       return { ...result, apps: await cachedComposioApps(connection, actor) };