diff --git a/apps/docs/content/docs/platform/enterprise/audit-logs.mdx b/apps/docs/content/docs/platform/enterprise/audit-logs.mdx index 84614751b4c..de42fc1c380 100644 --- a/apps/docs/content/docs/platform/enterprise/audit-logs.mdx +++ b/apps/docs/content/docs/platform/enterprise/audit-logs.mdx @@ -105,7 +105,7 @@ Events use identifiers such as `resource.action`. These categories are represent Authentication failures and password-reset requests use an anonymous actor and record the target account separately. Impersonation and administrative session revocations identify the operator and target organization. Table-row mutations record summaries and counts without copying row contents. -When a workflow run makes the same kind of change to a table many times in quick succession, such as inserting rows in a loop, the repeats are merged into one `table.updated` entry. The merged entry keeps every row ID in `metadata.rowIds`, sums the counts, and records in `metadata.rollup` how many changes it covers and when the first and last happened. The actor of any workflow-made entry names the workflow (`metadata.actor.workflowId`) and the run (`metadata.actor.executionId`). Credential use by a workflow run is merged the same way: one `credential.accessed` entry per credential, counting the repeats. +When a workflow run makes the same kind of change to a table many times in quick succession, such as inserting rows in a loop, the repeats are merged into one `table.updated` entry. The merged entry keeps every row ID in `metadata.rowIds`, sums the counts, and records in `metadata.rollup` how many changes it covers and when the first and last happened. The actor of any workflow-made entry names the workflow (`metadata.actor.workflowId`) and the run (`metadata.actor.executionId`). Credential use by a workflow run is merged the same way: one `credential.accessed` entry per credential, counting the repeats, that names the workflow (`metadata.workflowId`) and the run (`metadata.executionId`). Most audit writes are best effort: an outage or process loss can prevent persistence after an action succeeds. Deployment activation uses an awaited, idempotent insert before its outbox checkpoint. The feed is not a complete tool-call stream or a guaranteed transactional ledger for every operation. diff --git a/apps/sim/executor/handlers/agent/handler.ts b/apps/sim/executor/handlers/agent/handler.ts index 0de3b244c88..96a953799ee 100644 --- a/apps/sim/executor/handlers/agent/handler.ts +++ b/apps/sim/executor/handlers/agent/handler.ts @@ -2988,6 +2988,7 @@ export class AgentBlockHandler implements BlockHandler { actingUserId: ctx.userId, workspaceId: ctx.workspaceId, workflowId: ctx.workflowId, + executionId: ctx.executionId, callerLabel: 'vertex-agent', }) } diff --git a/apps/sim/executor/handlers/evaluator/handler.ts b/apps/sim/executor/handlers/evaluator/handler.ts index 71bf795b96d..68118cd0e31 100644 --- a/apps/sim/executor/handlers/evaluator/handler.ts +++ b/apps/sim/executor/handlers/evaluator/handler.ts @@ -185,6 +185,7 @@ export class EvaluatorBlockHandler implements BlockHandler { actingUserId: ctx.userId, workspaceId: ctx.workspaceId, workflowId: ctx.workflowId, + executionId: ctx.executionId, callerLabel: 'vertex-evaluator', }) } diff --git a/apps/sim/executor/handlers/router/handler.ts b/apps/sim/executor/handlers/router/handler.ts index 1b3ad1d9711..8abcf52792b 100644 --- a/apps/sim/executor/handlers/router/handler.ts +++ b/apps/sim/executor/handlers/router/handler.ts @@ -120,6 +120,7 @@ export class RouterBlockHandler implements BlockHandler { actingUserId: ctx.userId, workspaceId: ctx.workspaceId, workflowId: ctx.workflowId, + executionId: ctx.executionId, callerLabel: 'vertex-router', }) } @@ -289,6 +290,7 @@ export class RouterBlockHandler implements BlockHandler { actingUserId: ctx.userId, workspaceId: ctx.workspaceId, workflowId: ctx.workflowId, + executionId: ctx.executionId, callerLabel: 'vertex-router', }) } diff --git a/apps/sim/executor/utils/credential-token.ts b/apps/sim/executor/utils/credential-token.ts index a32bf7b090d..7199e1bc41a 100644 --- a/apps/sim/executor/utils/credential-token.ts +++ b/apps/sim/executor/utils/credential-token.ts @@ -22,6 +22,7 @@ export interface ResolveExecutorCredentialTokenParams { credentialId: string userId?: string workflowId?: string + executionId?: string /** Tool consuming the token; required by the managed-OAuth scope policy. */ toolId?: string /** Display label for the thrown failure ("Failed to obtain credential for X: ..."). */ @@ -112,6 +113,7 @@ export async function resolveExecutorCredentialToken( requestId, credentialId, workflowId, + executionId: params.executionId, toolId, scopes: params.scopes, impersonateEmail: params.impersonateEmail, diff --git a/apps/sim/executor/utils/vertex-credential.ts b/apps/sim/executor/utils/vertex-credential.ts index c05c9f943a8..bb9289d473a 100644 --- a/apps/sim/executor/utils/vertex-credential.ts +++ b/apps/sim/executor/utils/vertex-credential.ts @@ -1,6 +1,7 @@ import { createLogger } from '@sim/logger' import { canUseCredential, getCredentialActorContext } from '@/lib/credentials/access' import { getServiceAccountToken } from '@/lib/oauth/credential-service' +import { recordCredentialAccess } from '@/lib/oauth/token-resolution' import { resolveExecutorCredentialToken } from '@/executor/utils/credential-token' const logger = createLogger('VertexCredential') @@ -12,6 +13,7 @@ export interface ResolveVertexCredentialParams { workspaceId: string | null | undefined /** Pins the token request to this workflow's workspace. */ workflowId?: string + executionId?: string callerLabel?: string } @@ -27,6 +29,7 @@ export async function resolveVertexCredential({ actingUserId, workspaceId, workflowId, + executionId, callerLabel = 'vertex', }: ResolveVertexCredentialParams): Promise { const requestId = `${callerLabel}-${Date.now()}` @@ -57,6 +60,15 @@ export async function resolveVertexCredential({ const accessToken = await getServiceAccountToken(cred.id, [ 'https://www.googleapis.com/auth/cloud-platform', ]) + recordCredentialAccess({ + actorId: actingUserId, + workspaceId: cred.workspaceId, + resourceId: cred.id, + providerId: cred.providerId, + credentialType: 'service_account', + workflowId, + executionId, + }) logger.info(`[${requestId}] Successfully resolved Vertex AI service account credential`) return accessToken } @@ -70,6 +82,7 @@ export async function resolveVertexCredential({ credentialId, userId: actingUserId, workflowId, + executionId, toolLabel: 'Vertex AI', }) diff --git a/apps/sim/lib/oauth/token-resolution.ts b/apps/sim/lib/oauth/token-resolution.ts index 08b77d05d36..6bde91721b7 100644 --- a/apps/sim/lib/oauth/token-resolution.ts +++ b/apps/sim/lib/oauth/token-resolution.ts @@ -58,6 +58,8 @@ export interface ResolveCredentialTokenInput { requestId: string credentialId?: string workflowId?: string + /** Workflow run the token is for; recorded on the credential-access audit entry. */ + executionId?: string /** Canonical provider scopes, used only by service-account token minting. */ scopes?: string[] /** Google domain-wide-delegation subject for service-account credentials. */ @@ -112,9 +114,12 @@ export function recordCredentialAccess(params: { resourceId: string providerId: string | null | undefined credentialType: 'oauth' | 'service_account' + workflowId?: string + executionId?: string auditRequest?: CredentialAuditRequest }): void { - const { actorId, workspaceId, resourceId, providerId, credentialType } = params + const { actorId, workspaceId, resourceId, providerId, credentialType, workflowId, executionId } = + params const credentialLabel = `${credentialType === 'oauth' ? 'OAuth' : 'service account'} credential for provider ${providerId ?? 'unknown'}` const audit = { workspaceId, @@ -126,6 +131,8 @@ export function recordCredentialAccess(params: { metadata: { provider: providerId, credentialType, + ...(workflowId ? { workflowId } : {}), + ...(executionId ? { executionId } : {}), }, } if (params.auditRequest) { @@ -211,9 +218,20 @@ export async function completeOAuthCredentialToken(params: { resolvedCredentialId: string actorId?: string workspaceId: string | null + workflowId?: string + executionId?: string auditRequest?: CredentialAuditRequest }): Promise { - const { requestId, credential, resolvedCredentialId, actorId, workspaceId, auditRequest } = params + const { + requestId, + credential, + resolvedCredentialId, + actorId, + workspaceId, + workflowId, + executionId, + auditRequest, + } = params const contextValidation = validateOAuthCredentialContext(credential) if (!contextValidation.ok) { return { ok: false, status: 401, error: contextValidation.error } @@ -229,6 +247,8 @@ export async function completeOAuthCredentialToken(params: { resourceId: resolvedCredentialId, providerId: credential.providerId, credentialType: 'oauth', + workflowId, + executionId, auditRequest, }) } @@ -301,6 +321,7 @@ export async function resolveCredentialToken( requestId, credentialId, workflowId, + executionId, scopes, impersonateEmail, callerUserId, @@ -355,6 +376,8 @@ export async function resolveCredentialToken( resourceId: resolved.credentialId, providerId: resolved.providerId, credentialType: 'service_account', + workflowId, + executionId, auditRequest, }) } @@ -433,6 +456,8 @@ export async function resolveCredentialToken( resolvedCredentialId, actorId: authz.requesterUserId, workspaceId: authz.workspaceId ?? null, + workflowId, + executionId, auditRequest, }) } catch (error) { @@ -477,6 +502,7 @@ export async function resolveCredentialAccessToken( requestId, credentialId, workflowId: input.workflowId, + executionId: input.executionId, scopes: input.scopes, /** * In-process callers forward raw subblock state, where an untouched diff --git a/apps/sim/tools/index.ts b/apps/sim/tools/index.ts index 5fa80c13678..66c4ff4088c 100644 --- a/apps/sim/tools/index.ts +++ b/apps/sim/tools/index.ts @@ -1931,6 +1931,7 @@ async function executeToolImplementation( credentialId, userId, workflowId, + executionId: scope.executionId, toolId, toolLabel, scopes: providerScopes, diff --git a/packages/audit/src/rollup.ts b/packages/audit/src/rollup.ts index 56e669dad7f..d18b529950a 100644 --- a/packages/audit/src/rollup.ts +++ b/packages/audit/src/rollup.ts @@ -13,7 +13,10 @@ const ROLLUP_MAX_ENTRIES = 100 /** Pending rollups held by the process; reaching it writes every one of them. */ const ROLLUP_MAX_PENDING = 1_000 -/** Rows inserted at once by one write, so a large flush never floods the pool. */ +/** + * Rollup rows the process inserts at once, across every window timer, size + * flush, and scope end, so a burst of distinct rollups never floods the pool. + */ const ROLLUP_WRITE_CONCURRENCY = 10 export interface AuditRollupSummary { @@ -67,6 +70,8 @@ interface PendingRollup { } const pending = new Map() +const queuedWrites: Array<{ rollup: PendingRollup; written: () => void }> = [] +let activeWrites = 0 const scopeStorage = new AsyncLocalStorage() let nextScopeId = 0 @@ -196,11 +201,31 @@ function toAuditParams(rollup: PendingRollup): AuditLogParams { /** * One insert per row, so a row that cannot be written never takes others with - * it, at most {@link ROLLUP_WRITE_CONCURRENCY} at a time. + * it. Rows wait in one process-wide queue, at most + * {@link ROLLUP_WRITE_CONCURRENCY} in flight. */ -async function writePending(rollups: PendingRollup[]): Promise { - for (let index = 0; index < rollups.length; index += ROLLUP_WRITE_CONCURRENCY) { - await Promise.all(rollups.slice(index, index + ROLLUP_WRITE_CONCURRENCY).map(writeRollup)) +function writePending(rollups: PendingRollup[]): Promise { + return Promise.all( + rollups.map( + (rollup) => + new Promise((written) => { + queuedWrites.push({ rollup, written }) + drainWrites() + }) + ) + ).then(() => undefined) +} + +function drainWrites(): void { + while (activeWrites < ROLLUP_WRITE_CONCURRENCY) { + const next = queuedWrites.shift() + if (!next) return + activeWrites += 1 + void writeRollup(next.rollup).finally(() => { + activeWrites -= 1 + next.written() + drainWrites() + }) } }