feat(audit): attribute workflow audits to their run and merge a run's repeated row audits

apps/docs/content/docs/platform/enterprise/audit-logs.mdx

@@ -105,6 +105,8 @@ 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.
+
 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.
 
 The interface and API provide no audit-entry editing or deletion. Database administrators can alter stored records, and entries are not cryptographically sealed. Use a drain to retain your own archive.

apps/sim/lib/auth/principal.test.ts

@@ -575,6 +575,7 @@ describe('principal actors', () => {
         kind: 'delegated',
         serviceId: 'executor',
         delegationId: 'delegation-1',
+        workflowId: 'workflow-1',
       },
       attributedUserId: 'billing-owner-1',
     })

apps/sim/lib/core/application/authorized-workspace-use-case.ts

@@ -1,12 +1,16 @@
 import {
   type AuditActionType,
+  type AuditLogParams,
   type AuditResourceTypeValue,
+  type AuditRollupSummary,
   recordAudit,
   recordAuditOnce,
+  recordAuditRollup,
 } from '@sim/audit'
-import type { Principal, PrincipalAuditAttribution } from '@sim/auth/principal'
+import type { Principal, PrincipalActor, PrincipalAuditAttribution } from '@sim/auth/principal'
 import { resolvePrincipalAuditAttribution } from '@sim/auth/principal'
 import { createLogger } from '@sim/logger'
+import { toNumberOrNull, toStringOrNull } from '@sim/utils/coerce'
 import { runDeclaredContentOperation } from '@/lib/core/application/content-delivery'
 import type { ApplicationOperation, OperationUseCase } from '@/lib/core/application/operation'
 import {
@@ -26,6 +30,11 @@ import type { ResourcePolicyBinding } from '@/lib/resource-policies/registry'
 
 const logger = createLogger('WorkspaceUseCaseAudit')
 
+/** The audit actor of a workflow-execution delegation that names its run. */
+type ExecutionDelegatedActor = Extract<PrincipalActor, { kind: 'delegated' }> & {
+  executionId: string
+}
+
 export interface WorkspaceUseCaseAuditEntry {
   /** Canonical affected workspace; null keeps an organization event outside the authorization workspace. */
   workspaceId?: string | null
@@ -40,6 +49,21 @@ export interface WorkspaceUseCaseAuditEntry {
    * retry after losing its response: each retry projects the same entry again.
    */
   idempotencyKey?: string
+  /**
+   * Merges this entry with its repeats from the same workflow execution into
+   * one row, keeping every count and identifier (see `recordAuditRollup`).
+   * Applies only to an executor delegation that names its execution; every
+   * other caller records the entry as usual.
+   */
+  executionRollup?: WorkspaceUseCaseAuditRollup
+}
+
+export interface WorkspaceUseCaseAuditRollup {
+  /** Metadata keys summed across merged entries. */
+  counterKeys: readonly string[]
+  /** Metadata keys collected into a `<key>s` array on a merged row. */
+  identifierKeys: readonly string[]
+  describe(summary: AuditRollupSummary): string
 }
 
 export interface AuthorizedWorkspaceUseCaseContext<
@@ -119,7 +143,12 @@ export function recordProjectedUseCaseAuditEntries(
   organizationId?: string
 ): void {
   const attribution: PrincipalAuditAttribution = resolvePrincipalAuditAttribution(principal)
-  for (const { idempotencyKey, ...entry } of entries) {
+  const { actor } = attribution
+  const rollupActor =
+    actor.kind === 'delegated' && actor.executionId && !request
+      ? { ...actor, executionId: actor.executionId }
+      : undefined
+  for (const { idempotencyKey, executionRollup, ...entry } of entries) {
     const params = {
       workspaceId: entry.workspaceId === undefined ? workspaceId : entry.workspaceId,
       actorId: attribution.actorId,
@@ -137,6 +166,10 @@ export function recordProjectedUseCaseAuditEntries(
       },
       request,
     }
+    if (executionRollup && rollupActor && !idempotencyKey) {
+      recordExecutionAuditRollup(params, rollupActor, executionRollup)
+      continue
+    }
     if (!idempotencyKey) {
       recordAudit(params)
       continue
@@ -147,6 +180,37 @@ export function recordProjectedUseCaseAuditEntries(
   }
 }
 
+/**
+ * Entries merge when everything but their counters, identifiers, and the
+ * per-call delegation matches, so a merged row never blends two operations.
+ */
+function recordExecutionAuditRollup(
+  params: AuditLogParams & { metadata: Record<string, unknown> },
+  { delegationId, ...sharedActor }: ExecutionDelegatedActor,
+  rollup: WorkspaceUseCaseAuditRollup
+): void {
+  const shared: Record<string, unknown> = { ...params.metadata, actor: sharedActor }
+  const counters: Record<string, number> = {}
+  const identifiers: Record<string, string | undefined> = { delegationId }
+  for (const key of rollup.counterKeys) {
+    const value = toNumberOrNull(shared[key])
+    if (value !== null) counters[key] = value
+    delete shared[key]
+  }
+  for (const key of rollup.identifierKeys) {
+    identifiers[key] = toStringOrNull(shared[key]) ?? undefined
+    delete shared[key]
+  }
+  recordAuditRollup({
+    key: JSON.stringify(shared),
+    params,
+    rollupMetadata: shared,
+    counters,
+    identifiers,
+    describe: rollup.describe,
+  })
+}
+
 /**
  * A use case that always answers `authorize`, so a caller that must run the
  * funnel without executing — a `HEAD` on a route declaring `headSafe: false`,

apps/sim/lib/core/application/index.ts

@@ -8,6 +8,7 @@ export {
   defineAuthorizedWorkspaceUseCase,
   recordProjectedUseCaseAuditEntries,
   type WorkspaceUseCaseAuditEntry,
+  type WorkspaceUseCaseAuditRollup,
 } from '@/lib/core/application/authorized-workspace-use-case'
 export {
   FORBIDDEN_DETAIL_CODES,

apps/sim/lib/oauth/token-resolution.test.ts

@@ -75,6 +75,7 @@ describe('resolveCredentialToken', () => {
     expect(mockGetCredential).not.toHaveBeenCalled()
     expect(mockRefreshTokenIfNeeded).not.toHaveBeenCalled()
     expect(mockRecordAudit).not.toHaveBeenCalled()
+    expect(auditMockFns.mockRecordAuditRollup).not.toHaveBeenCalled()
   })
 
   it('refreshes the token, records the access trail, and returns the payload', async () => {
@@ -104,12 +105,14 @@ describe('resolveCredentialToken', () => {
       token: { accessToken: 'fresh', credentialType: 'oauth', idToken: 'id-token' },
     })
     expect(mockGetCredential).toHaveBeenCalledWith('req-1', 'account-1', 'owner-1')
-    expect(mockRecordAudit).toHaveBeenCalledWith(
+    expect(auditMockFns.mockRecordAuditRollup).toHaveBeenCalledWith(
       expect.objectContaining({
-        actorId: 'user-1',
-        workspaceId: 'ws-1',
-        resourceId: 'account-1',
-        action: 'credential.accessed',
+        params: expect.objectContaining({
+          actorId: 'user-1',
+          workspaceId: 'ws-1',
+          resourceId: 'account-1',
+          action: 'credential.accessed',
+        }),
       })
     )
   })
@@ -164,6 +167,7 @@ describe('resolveCredentialToken', () => {
 
     expect(result).toEqual({ ok: false, status: 401, error: 'Failed to refresh access token' })
     expect(mockRecordAudit).not.toHaveBeenCalled()
+    expect(auditMockFns.mockRecordAuditRollup).not.toHaveBeenCalled()
   })
 
   it('authorizes service-account credentials before minting a token', async () => {
@@ -467,6 +471,7 @@ describe('ordinary tool credential binding', () => {
       expect(mockResolveServiceAccountToken).not.toHaveBeenCalled()
       expect(mockRefreshTokenIfNeeded).not.toHaveBeenCalled()
       expect(mockRecordAudit).not.toHaveBeenCalled()
+      expect(auditMockFns.mockRecordAuditRollup).not.toHaveBeenCalled()
     }
   )
 

apps/sim/lib/oauth/token-resolution.ts

@@ -1,4 +1,4 @@
-import { AuditAction, AuditResourceType, recordAudit } from '@sim/audit'
+import { AuditAction, AuditResourceType, recordAudit, recordAuditRollup } from '@sim/audit'
 import { type DelegatedPrincipal, resolvePrincipalSubject } from '@sim/auth/principal'
 import { createLogger } from '@sim/logger'
 import { getErrorMessage } from '@sim/utils/errors'
@@ -102,7 +102,9 @@ export function validateOAuthCredentialContext(
 
 /**
  * Emits the semantic "credential used" trail for one resolved credential.
- * Both the audit row and the analytics event are fire-and-forget.
+ * Both the audit row and the analytics event are fire-and-forget. A server-side
+ * access with no client request, such as a workflow run refreshing the token
+ * for every tool call, merges with its repeats into one row that counts them.
  */
 export function recordCredentialAccess(params: {
   actorId: string
@@ -113,19 +115,31 @@ export function recordCredentialAccess(params: {
   auditRequest?: CredentialAuditRequest
 }): void {
   const { actorId, workspaceId, resourceId, providerId, credentialType } = params
-  recordAudit({
+  const credentialLabel = `${credentialType === 'oauth' ? 'OAuth' : 'service account'} credential for provider ${providerId ?? 'unknown'}`
+  const audit = {
     workspaceId,
     actorId,
     action: AuditAction.CREDENTIAL_ACCESSED,
     resourceType: AuditResourceType.CREDENTIAL,
     resourceId,
-    description: `Accessed ${credentialType === 'oauth' ? 'OAuth' : 'service account'} credential for provider ${providerId ?? 'unknown'}`,
+    description: `Accessed ${credentialLabel}`,
     metadata: {
       provider: providerId,
       credentialType,
     },
-    request: params.auditRequest,
-  })
+  }
+  if (params.auditRequest) {
+    recordAudit({ ...audit, request: params.auditRequest })
+  } else {
+    recordAuditRollup({
+      key: JSON.stringify(audit.metadata),
+      params: audit,
+      rollupMetadata: audit.metadata,
+      counters: {},
+      identifiers: {},
+      describe: ({ entries }) => `Accessed ${credentialLabel} ${entries} times`,
+    })
+  }
   captureServerEvent(
     actorId,
     'credential_used',

apps/sim/lib/table/application/row-audit.integration.ts

@@ -1,3 +1,4 @@
+import { withAuditRollupScope } from '@sim/audit'
 import { db } from '@sim/db'
 import {
   auditLog,
@@ -8,14 +9,17 @@ import {
   workspace,
 } from '@sim/db/schema'
 import { deleteWorkspaceFixture, insertWorkspaceFixture } from '@sim/db/testing/workspace-fixtures'
-import { createSessionPrincipal } from '@sim/testing/factories/principal.factory'
+import {
+  createExecutorPrincipal,
+  createSessionPrincipal,
+} from '@sim/testing/factories/principal.factory'
 import { storageServiceMock } from '@sim/testing/mocks/storage-service.mock'
 import { tableBillingMock, tableBillingMockFns } from '@sim/testing/mocks/table-billing.mock'
 import { tableEventsMock } from '@sim/testing/mocks/table-events.mock'
 import { tableTriggerMock } from '@sim/testing/mocks/table-trigger.mock'
 import { tableWorkflowColumnsMock } from '@sim/testing/mocks/table-workflow-columns.mock'
 import { generateId } from '@sim/utils/id'
-import { eq } from 'drizzle-orm'
+import { asc, eq } from 'drizzle-orm'
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 
 vi.mock('@/lib/table/billing', () => tableBillingMock)
@@ -173,4 +177,100 @@ describe('table mutation audit history in PostgreSQL', () => {
     })
     expect(JSON.stringify(entry)).not.toContain(privateValue)
   })
+
+  function executorPrincipal(executionId: string) {
+    return createExecutorPrincipal({
+      subjectUserId: userId,
+      workspaceId,
+      delegationId: generateId(),
+      audience: 'sim:tables',
+      resourceScope: { tableId },
+      delegationContext: { kind: 'workflow_execution', workflowId: 'wf-audit', executionId },
+    })
+  }
+
+  async function insertAsExecution(executionId: string, value: string) {
+    const result = await createTableRows.execute({
+      principal: executorPrincipal(executionId),
+      input: {
+        tableId,
+        assertedWorkspaceId: workspaceId,
+        strictWrite: true,
+        dataKeying: 'ids',
+        kind: 'single',
+        data: { value },
+      },
+    })
+    if (result.kind !== 'single') throw new Error('Expected a single-row insert')
+    return result.row.id
+  }
+
+  async function auditRows() {
+    return db
+      .select()
+      .from(auditLog)
+      .where(eq(auditLog.workspaceId, workspaceId))
+      .orderBy(asc(auditLog.createdAt))
+  }
+
+  it("merges one execution's repeated row inserts into one row that keeps every row id and count", async () => {
+    const rowIds: string[] = []
+    await withAuditRollupScope(async () => {
+      for (const value of ['a', 'b', 'c']) rowIds.push(await insertAsExecution('exec-1', value))
+    })
+
+    const entries = await auditRows()
+    expect(entries).toHaveLength(1)
+    const [entry] = entries
+    expect(entry).toMatchObject({
+      actorId: userId,
+      action: 'table.updated',
+      resourceId: tableId,
+      metadata: {
+        op: 'insert',
+        operation: 'tables.rows.create',
+        rowsInserted: 3,
+        rowIds,
+        actor: { kind: 'delegated', workflowId: 'wf-audit', executionId: 'exec-1' },
+        rollup: { entries: 3 },
+      },
+    })
+    const metadata = entry.metadata as Record<string, unknown>
+    expect(metadata.delegationIds).toHaveLength(3)
+    expect(new Set(metadata.delegationIds as string[]).size).toBe(3)
+    expect(metadata).not.toHaveProperty('rowId')
+    expect(entry.description).toContain('3 inserted')
+  })
+
+  it('never merges row changes from two executions', async () => {
+    await withAuditRollupScope(async () => {
+      await insertAsExecution('exec-1', 'a')
+      await insertAsExecution('exec-2', 'b')
+      await insertAsExecution('exec-1', 'c')
+    })
+
+    const entries = await auditRows()
+    expect(entries).toHaveLength(2)
+    const byExecution = new Map(
+      entries.map((entry) => {
+        const metadata = entry.metadata as { actor: { executionId: string }; rowsInserted: number }
+        return [metadata.actor.executionId, metadata.rowsInserted]
+      })
+    )
+    expect(Object.fromEntries(byExecution)).toEqual({ 'exec-1': 2, 'exec-2': 1 })
+  })
+
+  it('writes a lone execution change exactly as an unmerged entry', async () => {
+    const insertedRowId = await withAuditRollupScope(() => insertAsExecution('exec-1', 'a'))
+
+    const [entry] = await auditRows()
+    expect(entry.metadata).toMatchObject({
+      op: 'insert',
+      rowId: insertedRowId,
+      rowsInserted: 1,
+      actor: { kind: 'delegated', workflowId: 'wf-audit', executionId: 'exec-1' },
+    })
+    expect(entry.metadata).not.toHaveProperty('rollup')
+    expect(entry.description).toBe('Inserted 1 row(s) into table "Audited table"')
+  })
 })