diff --git a/server/src/__tests__/connection-management.integration.test.ts b/server/src/__tests__/connection-management.integration.test.ts index d6333d362e5..328b973e019 100644 --- a/server/src/__tests__/connection-management.integration.test.ts +++ b/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; + 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: [] }); diff --git a/server/src/__tests__/openapi-routes.test.ts b/server/src/__tests__/openapi-routes.test.ts index 03bb0a040e7..0cda5bd1579 100644 --- a/server/src/__tests__/openapi-routes.test.ts +++ b/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(); diff --git a/server/src/routes/openapi.ts b/server/src/routes/openapi.ts index 4dd89f38ad9..dab6135db41 100644 --- a/server/src/routes/openapi.ts +++ b/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({ diff --git a/server/src/services/chat-channels/github/setup-secret.ts b/server/src/services/chat-channels/github/setup-secret.ts index 2a9ff16c797..5844c69da16 100644 --- a/server/src/services/chat-channels/github/setup-secret.ts +++ b/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, diff --git a/server/src/services/email-connections.ts b/server/src/services/email-connections.ts index 72f5498f4ff..2e9676d618a 100644 --- a/server/src/services/email-connections.ts +++ b/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 { + 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) { diff --git a/server/src/services/tool-access/catalog/aggregator-accounts.ts b/server/src/services/tool-access/catalog/aggregator-accounts.ts index 6abf2ba68d2..7c89458bfcd 100644 --- a/server/src/services/tool-access/catalog/aggregator-accounts.ts +++ b/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 { if (value && typeof value === "object" && !Array.isArray(value)) return value as Record; @@ -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) };