Skip to content

Commit a9d5dc8

Browse files
authored
feat(audit): attribute workflow audits to their run and merge a run's repeated row audits (#9404)
* feat(audit): attribute workflow audits to their run and merge a run's repeated row audits * fix(audit): scope rollups to the run that recorded them and bound flush concurrency * test(workflows): expect run attribution on executor deployment actors
1 parent 09976fd commit a9d5dc8

16 files changed

Lines changed: 528 additions & 28 deletions

File tree

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -105,6 +105,8 @@ Events use identifiers such as `resource.action`. These categories are represent
105105

106106
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.
107107

108+
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.
109+
108110
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.
109111

110112
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‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -575,6 +575,7 @@ describe('principal actors', () => {
575575
kind: 'delegated',
576576
serviceId: 'executor',
577577
delegationId: 'delegation-1',
578+
workflowId: 'workflow-1',
578579
},
579580
attributedUserId: 'billing-owner-1',
580581
})

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

Lines changed: 66 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,16 @@
11
import {
22
type AuditActionType,
3+
type AuditLogParams,
34
type AuditResourceTypeValue,
5+
type AuditRollupSummary,
46
recordAudit,
57
recordAuditOnce,
8+
recordAuditRollup,
69
} from '@sim/audit'
7-
import type { Principal, PrincipalAuditAttribution } from '@sim/auth/principal'
10+
import type { Principal, PrincipalActor, PrincipalAuditAttribution } from '@sim/auth/principal'
811
import { resolvePrincipalAuditAttribution } from '@sim/auth/principal'
912
import { createLogger } from '@sim/logger'
13+
import { toNumberOrNull, toStringOrNull } from '@sim/utils/coerce'
1014
import { runDeclaredContentOperation } from '@/lib/core/application/content-delivery'
1115
import type { ApplicationOperation, OperationUseCase } from '@/lib/core/application/operation'
1216
import {
@@ -26,6 +30,11 @@ import type { ResourcePolicyBinding } from '@/lib/resource-policies/registry'
2630

2731
const logger = createLogger('WorkspaceUseCaseAudit')
2832

33+
/** The audit actor of a workflow-execution delegation that names its run. */
34+
type ExecutionDelegatedActor = Extract<PrincipalActor, { kind: 'delegated' }> & {
35+
executionId: string
36+
}
37+
2938
export interface WorkspaceUseCaseAuditEntry {
3039
/** Canonical affected workspace; null keeps an organization event outside the authorization workspace. */
3140
workspaceId?: string | null
@@ -40,6 +49,21 @@ export interface WorkspaceUseCaseAuditEntry {
4049
* retry after losing its response: each retry projects the same entry again.
4150
*/
4251
idempotencyKey?: string
52+
/**
53+
* Merges this entry with its repeats from the same workflow execution into
54+
* one row, keeping every count and identifier (see `recordAuditRollup`).
55+
* Applies only to an executor delegation that names its execution; every
56+
* other caller records the entry as usual.
57+
*/
58+
executionRollup?: WorkspaceUseCaseAuditRollup
59+
}
60+
61+
export interface WorkspaceUseCaseAuditRollup {
62+
/** Metadata keys summed across merged entries. */
63+
counterKeys: readonly string[]
64+
/** Metadata keys collected into a `<key>s` array on a merged row. */
65+
identifierKeys: readonly string[]
66+
describe(summary: AuditRollupSummary): string
4367
}
4468

4569
export interface AuthorizedWorkspaceUseCaseContext<
@@ -119,7 +143,12 @@ export function recordProjectedUseCaseAuditEntries(
119143
organizationId?: string
120144
): void {
121145
const attribution: PrincipalAuditAttribution = resolvePrincipalAuditAttribution(principal)
122-
for (const { idempotencyKey, ...entry } of entries) {
146+
const { actor } = attribution
147+
const rollupActor =
148+
actor.kind === 'delegated' && actor.executionId && !request
149+
? { ...actor, executionId: actor.executionId }
150+
: undefined
151+
for (const { idempotencyKey, executionRollup, ...entry } of entries) {
123152
const params = {
124153
workspaceId: entry.workspaceId === undefined ? workspaceId : entry.workspaceId,
125154
actorId: attribution.actorId,
@@ -137,6 +166,10 @@ export function recordProjectedUseCaseAuditEntries(
137166
},
138167
request,
139168
}
169+
if (executionRollup && rollupActor && !idempotencyKey) {
170+
recordExecutionAuditRollup(params, rollupActor, executionRollup)
171+
continue
172+
}
140173
if (!idempotencyKey) {
141174
recordAudit(params)
142175
continue
@@ -147,6 +180,37 @@ export function recordProjectedUseCaseAuditEntries(
147180
}
148181
}
149182

183+
/**
184+
* Entries merge when everything but their counters, identifiers, and the
185+
* per-call delegation matches, so a merged row never blends two operations.
186+
*/
187+
function recordExecutionAuditRollup(
188+
params: AuditLogParams & { metadata: Record<string, unknown> },
189+
{ delegationId, ...sharedActor }: ExecutionDelegatedActor,
190+
rollup: WorkspaceUseCaseAuditRollup
191+
): void {
192+
const shared: Record<string, unknown> = { ...params.metadata, actor: sharedActor }
193+
const counters: Record<string, number> = {}
194+
const identifiers: Record<string, string | undefined> = { delegationId }
195+
for (const key of rollup.counterKeys) {
196+
const value = toNumberOrNull(shared[key])
197+
if (value !== null) counters[key] = value
198+
delete shared[key]
199+
}
200+
for (const key of rollup.identifierKeys) {
201+
identifiers[key] = toStringOrNull(shared[key]) ?? undefined
202+
delete shared[key]
203+
}
204+
recordAuditRollup({
205+
key: JSON.stringify(shared),
206+
params,
207+
rollupMetadata: shared,
208+
counters,
209+
identifiers,
210+
describe: rollup.describe,
211+
})
212+
}
213+
150214
/**
151215
* A use case that always answers `authorize`, so a caller that must run the
152216
* funnel without executing — a `HEAD` on a route declaring `headSafe: false`,

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ export {
88
defineAuthorizedWorkspaceUseCase,
99
recordProjectedUseCaseAuditEntries,
1010
type WorkspaceUseCaseAuditEntry,
11+
type WorkspaceUseCaseAuditRollup,
1112
} from '@/lib/core/application/authorized-workspace-use-case'
1213
export {
1314
FORBIDDEN_DETAIL_CODES,

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

Lines changed: 10 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -75,6 +75,7 @@ describe('resolveCredentialToken', () => {
7575
expect(mockGetCredential).not.toHaveBeenCalled()
7676
expect(mockRefreshTokenIfNeeded).not.toHaveBeenCalled()
7777
expect(mockRecordAudit).not.toHaveBeenCalled()
78+
expect(auditMockFns.mockRecordAuditRollup).not.toHaveBeenCalled()
7879
})
7980

8081
it('refreshes the token, records the access trail, and returns the payload', async () => {
@@ -104,12 +105,14 @@ describe('resolveCredentialToken', () => {
104105
token: { accessToken: 'fresh', credentialType: 'oauth', idToken: 'id-token' },
105106
})
106107
expect(mockGetCredential).toHaveBeenCalledWith('req-1', 'account-1', 'owner-1')
107-
expect(mockRecordAudit).toHaveBeenCalledWith(
108+
expect(auditMockFns.mockRecordAuditRollup).toHaveBeenCalledWith(
108109
expect.objectContaining({
109-
actorId: 'user-1',
110-
workspaceId: 'ws-1',
111-
resourceId: 'account-1',
112-
action: 'credential.accessed',
110+
params: expect.objectContaining({
111+
actorId: 'user-1',
112+
workspaceId: 'ws-1',
113+
resourceId: 'account-1',
114+
action: 'credential.accessed',
115+
}),
113116
})
114117
)
115118
})
@@ -164,6 +167,7 @@ describe('resolveCredentialToken', () => {
164167

165168
expect(result).toEqual({ ok: false, status: 401, error: 'Failed to refresh access token' })
166169
expect(mockRecordAudit).not.toHaveBeenCalled()
170+
expect(auditMockFns.mockRecordAuditRollup).not.toHaveBeenCalled()
167171
})
168172

169173
it('authorizes service-account credentials before minting a token', async () => {
@@ -467,6 +471,7 @@ describe('ordinary tool credential binding', () => {
467471
expect(mockResolveServiceAccountToken).not.toHaveBeenCalled()
468472
expect(mockRefreshTokenIfNeeded).not.toHaveBeenCalled()
469473
expect(mockRecordAudit).not.toHaveBeenCalled()
474+
expect(auditMockFns.mockRecordAuditRollup).not.toHaveBeenCalled()
470475
}
471476
)
472477

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

Lines changed: 20 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
import { AuditAction, AuditResourceType, recordAudit } from '@sim/audit'
1+
import { AuditAction, AuditResourceType, recordAudit, recordAuditRollup } from '@sim/audit'
22
import { type DelegatedPrincipal, resolvePrincipalSubject } from '@sim/auth/principal'
33
import { createLogger } from '@sim/logger'
44
import { getErrorMessage } from '@sim/utils/errors'
@@ -102,7 +102,9 @@ export function validateOAuthCredentialContext(
102102

103103
/**
104104
* Emits the semantic "credential used" trail for one resolved credential.
105-
* Both the audit row and the analytics event are fire-and-forget.
105+
* Both the audit row and the analytics event are fire-and-forget. A server-side
106+
* access with no client request, such as a workflow run refreshing the token
107+
* for every tool call, merges with its repeats into one row that counts them.
106108
*/
107109
export function recordCredentialAccess(params: {
108110
actorId: string
@@ -113,19 +115,31 @@ export function recordCredentialAccess(params: {
113115
auditRequest?: CredentialAuditRequest
114116
}): void {
115117
const { actorId, workspaceId, resourceId, providerId, credentialType } = params
116-
recordAudit({
118+
const credentialLabel = `${credentialType === 'oauth' ? 'OAuth' : 'service account'} credential for provider ${providerId ?? 'unknown'}`
119+
const audit = {
117120
workspaceId,
118121
actorId,
119122
action: AuditAction.CREDENTIAL_ACCESSED,
120123
resourceType: AuditResourceType.CREDENTIAL,
121124
resourceId,
122-
description: `Accessed ${credentialType === 'oauth' ? 'OAuth' : 'service account'} credential for provider ${providerId ?? 'unknown'}`,
125+
description: `Accessed ${credentialLabel}`,
123126
metadata: {
124127
provider: providerId,
125128
credentialType,
126129
},
127-
request: params.auditRequest,
128-
})
130+
}
131+
if (params.auditRequest) {
132+
recordAudit({ ...audit, request: params.auditRequest })
133+
} else {
134+
recordAuditRollup({
135+
key: JSON.stringify(audit.metadata),
136+
params: audit,
137+
rollupMetadata: audit.metadata,
138+
counters: {},
139+
identifiers: {},
140+
describe: ({ entries }) => `Accessed ${credentialLabel} ${entries} times`,
141+
})
142+
}
129143
captureServerEvent(
130144
actorId,
131145
'credential_used',

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

Lines changed: 102 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
import { withAuditRollupScope } from '@sim/audit'
12
import { db } from '@sim/db'
23
import {
34
auditLog,
@@ -8,14 +9,17 @@ import {
89
workspace,
910
} from '@sim/db/schema'
1011
import { deleteWorkspaceFixture, insertWorkspaceFixture } from '@sim/db/testing/workspace-fixtures'
11-
import { createSessionPrincipal } from '@sim/testing/factories/principal.factory'
12+
import {
13+
createExecutorPrincipal,
14+
createSessionPrincipal,
15+
} from '@sim/testing/factories/principal.factory'
1216
import { storageServiceMock } from '@sim/testing/mocks/storage-service.mock'
1317
import { tableBillingMock, tableBillingMockFns } from '@sim/testing/mocks/table-billing.mock'
1418
import { tableEventsMock } from '@sim/testing/mocks/table-events.mock'
1519
import { tableTriggerMock } from '@sim/testing/mocks/table-trigger.mock'
1620
import { tableWorkflowColumnsMock } from '@sim/testing/mocks/table-workflow-columns.mock'
1721
import { generateId } from '@sim/utils/id'
18-
import { eq } from 'drizzle-orm'
22+
import { asc, eq } from 'drizzle-orm'
1923
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
2024

2125
vi.mock('@/lib/table/billing', () => tableBillingMock)
@@ -173,4 +177,100 @@ describe('table mutation audit history in PostgreSQL', () => {
173177
})
174178
expect(JSON.stringify(entry)).not.toContain(privateValue)
175179
})
180+
181+
function executorPrincipal(executionId: string) {
182+
return createExecutorPrincipal({
183+
subjectUserId: userId,
184+
workspaceId,
185+
delegationId: generateId(),
186+
audience: 'sim:tables',
187+
resourceScope: { tableId },
188+
delegationContext: { kind: 'workflow_execution', workflowId: 'wf-audit', executionId },
189+
})
190+
}
191+
192+
async function insertAsExecution(executionId: string, value: string) {
193+
const result = await createTableRows.execute({
194+
principal: executorPrincipal(executionId),
195+
input: {
196+
tableId,
197+
assertedWorkspaceId: workspaceId,
198+
strictWrite: true,
199+
dataKeying: 'ids',
200+
kind: 'single',
201+
data: { value },
202+
},
203+
})
204+
if (result.kind !== 'single') throw new Error('Expected a single-row insert')
205+
return result.row.id
206+
}
207+
208+
async function auditRows() {
209+
return db
210+
.select()
211+
.from(auditLog)
212+
.where(eq(auditLog.workspaceId, workspaceId))
213+
.orderBy(asc(auditLog.createdAt))
214+
}
215+
216+
it("merges one execution's repeated row inserts into one row that keeps every row id and count", async () => {
217+
const rowIds: string[] = []
218+
await withAuditRollupScope(async () => {
219+
for (const value of ['a', 'b', 'c']) rowIds.push(await insertAsExecution('exec-1', value))
220+
})
221+
222+
const entries = await auditRows()
223+
expect(entries).toHaveLength(1)
224+
const [entry] = entries
225+
expect(entry).toMatchObject({
226+
actorId: userId,
227+
action: 'table.updated',
228+
resourceId: tableId,
229+
metadata: {
230+
op: 'insert',
231+
operation: 'tables.rows.create',
232+
rowsInserted: 3,
233+
rowIds,
234+
actor: { kind: 'delegated', workflowId: 'wf-audit', executionId: 'exec-1' },
235+
rollup: { entries: 3 },
236+
},
237+
})
238+
const metadata = entry.metadata as Record<string, unknown>
239+
expect(metadata.delegationIds).toHaveLength(3)
240+
expect(new Set(metadata.delegationIds as string[]).size).toBe(3)
241+
expect(metadata).not.toHaveProperty('rowId')
242+
expect(entry.description).toContain('3 inserted')
243+
})
244+
245+
it('never merges row changes from two executions', async () => {
246+
await withAuditRollupScope(async () => {
247+
await insertAsExecution('exec-1', 'a')
248+
await insertAsExecution('exec-2', 'b')
249+
await insertAsExecution('exec-1', 'c')
250+
})
251+
252+
const entries = await auditRows()
253+
expect(entries).toHaveLength(2)
254+
const byExecution = new Map(
255+
entries.map((entry) => {
256+
const metadata = entry.metadata as { actor: { executionId: string }; rowsInserted: number }
257+
return [metadata.actor.executionId, metadata.rowsInserted]
258+
})
259+
)
260+
expect(Object.fromEntries(byExecution)).toEqual({ 'exec-1': 2, 'exec-2': 1 })
261+
})
262+
263+
it('writes a lone execution change exactly as an unmerged entry', async () => {
264+
const insertedRowId = await withAuditRollupScope(() => insertAsExecution('exec-1', 'a'))
265+
266+
const [entry] = await auditRows()
267+
expect(entry.metadata).toMatchObject({
268+
op: 'insert',
269+
rowId: insertedRowId,
270+
rowsInserted: 1,
271+
actor: { kind: 'delegated', workflowId: 'wf-audit', executionId: 'exec-1' },
272+
})
273+
expect(entry.metadata).not.toHaveProperty('rollup')
274+
expect(entry.description).toBe('Inserted 1 row(s) into table "Audited table"')
275+
})
176276
})

0 commit comments

Comments
 (0)