feat(audit): org-wide audit log with actor attribution and durable queue delivery

apps/alerting/src/worker.ts

@@ -6,6 +6,7 @@ import {
 	AlertRulesService,
 	AlertsService,
 	AnomalyDetectionService,
+	AuditLogService,
 	BucketCacheService,
 	CacheBackendLive,
 	CloudflareAnalyticsService,
@@ -144,7 +145,9 @@ export const buildLayer = (env: AlertingWorkerEnv) => {
 
 	const ErrorActorsServiceLive = ErrorActorsService.layer.pipe(Layer.provide(BaseLive))
 	const ErrorIssueWorkflowServiceLive = ErrorIssueWorkflowService.layer.pipe(
-		Layer.provide(Layer.mergeAll(BaseLive, ErrorActorsServiceLive)),
+		Layer.provide(
+			Layer.mergeAll(BaseLive, ErrorActorsServiceLive, AuditLogService.layer.pipe(Layer.provide(BaseLive))),
+		),
 	)
 	const ErrorPolicyServiceLive = ErrorPolicyService.layer.pipe(Layer.provide(BaseLive))
 	const ErrorIssueReadModelsServiceLive = ErrorIssueReadModelsService.layer.pipe(

apps/api/alchemy.run.ts

@@ -178,6 +178,8 @@ const apiConfiguredEnv = (stage: MapleStage) =>
 		// Agent LLM path. `MAPLE_LLM_PROVIDER` flips between OpenRouter (default) and
 		// Workers AI; both stay wired, so a switch is this one var plus a redeploy.
 		// See `@/platform/Llm` for the provider-scoped model overrides.
+		// Audit log retention horizon in days; the sweep defaults to 400 when unset.
+		optionalPlain("AUDIT_LOG_RETENTION_DAYS"),
 		optionalPlain("MAPLE_LLM_PROVIDER"),
 		optionalPlain("MAPLE_TRIAGE_MODEL_OPENROUTER"),
 		optionalPlain("MAPLE_TRIAGE_MODEL_WORKERS_AI"),
@@ -318,6 +320,15 @@ export const createMapleApi = ({ stage, domains, replayBlobs }: CreateMapleApiOp
 		const planetScaleWebhookQueue = yield* Cloudflare.Queues.Queue("planetscale-webhooks", {
 			name: planetScaleWebhookQueueName,
 		})
+		const auditEventsQueueName = resolveWorkerName("audit-events", stage)
+		const auditEventsQueue = yield* Cloudflare.Queues.Queue("audit-events", {
+			name: auditEventsQueueName,
+		})
+		// Parking lot for audit entries that exhausted their retries. Deliberately
+		// has no consumer: an entry landing here is a lost audit record, and the
+		// point is that it survives for inspection instead of being dropped.
+		const auditEventsDlqName = resolveWorkerName("audit-events-dlq", stage)
+		yield* Cloudflare.Queues.Queue("audit-events-dlq", { name: auditEventsDlqName })
 
 		const worker = (yield* Cloudflare.Worker("api", {
 			name: resolveWorkerName("api", stage),
@@ -369,6 +380,8 @@ export const createMapleApi = ({ stage, domains, replayBlobs }: CreateMapleApiOp
 				VCS_SYNC_QUEUE_NAME: vcsSyncQueueName,
 				PLANETSCALE_WEBHOOK_QUEUE: planetScaleWebhookQueue,
 				PLANETSCALE_WEBHOOK_QUEUE_NAME: planetScaleWebhookQueueName,
+				AUDIT_EVENTS_QUEUE: auditEventsQueue,
+				AUDIT_EVENTS_QUEUE_NAME: auditEventsQueueName,
 				CLICKHOUSE_SCHEMA_APPLY_WORKFLOW: schemaApplyWorkflow,
 				INVESTIGATION_FANOUT_WORKFLOW: investigationFanoutWorkflow,
 				API_V2_RATE_LIMITER: Cloudflare.RateLimit("API_V2_RATE_LIMITER", {
@@ -428,6 +441,21 @@ export const createMapleApi = ({ stage, domains, replayBlobs }: CreateMapleApiOp
 				maxWaitTimeMs: 5000,
 			},
 		})
+		// Audit entries tolerate a few seconds of delivery latency; batch wider and
+		// wait longer so one insert round-trip covers many entries.
+		yield* Cloudflare.Queues.Consumer("audit-events-consumer", {
+			queueId: auditEventsQueue.queueId,
+			scriptName: worker.workerName,
+			// `maxRetries` must stay in sync with AUDIT_EVENTS_MAX_RETRIES in
+			// audit-events-runtime.ts, which logs the drop on the final attempt.
+			deadLetterQueue: auditEventsDlqName,
+			settings: {
+				batchSize: 25,
+				maxConcurrency: 2,
+				maxRetries: 5,
+				maxWaitTimeMs: 5000,
+			},
+		})
 
 		// `db` is undefined on ref stages — alerting resolves the same ref itself.
 		return { worker, db: mapleDb }

apps/api/src/alerting.ts

@@ -4,6 +4,7 @@ export { AlertDestinationsService } from "./services/alerts/AlertDestinationsSer
 export { AlertReadModelsService } from "./services/alerts/AlertReadModelsService"
 export { AlertRulesService } from "./services/alerts/AlertRulesService"
 export { AnomalyDetectionService } from "./services/alerts/AnomalyDetectionService"
+export { AuditLogService } from "./services/audit/AuditLogService"
 export { BucketCacheService } from "@maple/query-engine/caching"
 export { CacheBackendLive } from "@/platform/CacheBackendLive"
 export { CloudflareAnalyticsService } from "./services/integrations/CloudflareAnalyticsService"

apps/api/src/audit-events-runtime.test.ts

@@ -0,0 +1,127 @@
+import { afterEach, describe, expect, it } from "@effect/vitest"
+import { OrgId } from "@maple/domain/primitives"
+import { Effect, Layer, Schema } from "effect"
+import { auditLogEntries } from "@maple/db"
+import { cleanupTestDbs, createTestDb, type TestDb } from "@/platform/test-pglite"
+import { Database, DatabaseError } from "@/platform/DatabaseLive"
+import { processAuditEventsBatch } from "./audit-events-runtime"
+import { AuditLogEvent, encodeAuditLogEventSync } from "./services/audit/audit-event"
+
+const asOrgId = Schema.decodeUnknownSync(OrgId)
+const ORG = asOrgId("org_audit_consumer_test")
+const createdDbs: TestDb[] = []
+
+afterEach(() => cleanupTestDbs(createdDbs))
+
+const event = (id: string) =>
+	encodeAuditLogEventSync(
+		new AuditLogEvent({
+			orgId: ORG,
+			id: Schema.decodeUnknownSync(AuditLogEvent.fields.id)(id),
+			actorType: "user",
+			source: "dashboard",
+			action: "dashboard.created",
+			outcome: "allowed",
+			occurredAtMs: 1_700_000_000_000,
+		}),
+	)
+
+/** One queue message, recording which terminal call the consumer made on it. */
+const message = (body: unknown, attempts: number) => {
+	const calls: string[] = []
+	return {
+		message: {
+			body,
+			attempts,
+			ack: () => calls.push("ack"),
+			retry: () => calls.push("retry"),
+		},
+		calls,
+	}
+}
+
+const run = <A>(effect: Effect.Effect<A, never, Database>) => {
+	const db = createTestDb(createdDbs)
+	return effect.pipe(Effect.provide(db.layer))
+}
+
+const batchOf = (...messages: ReadonlyArray<{ readonly message: unknown }>) =>
+	({ messages: messages.map((entry) => entry.message) }) as never
+
+/**
+ * A database whose every write fails, so the consumer's retry path is exercised
+ * without depending on a real Postgres fault.
+ */
+const failingDatabase = Layer.succeed(Database, {
+	execute: () =>
+		Effect.fail(new DatabaseError({ message: "insert failed", cause: new Error("insert failed") })),
+})
+
+describe("processAuditEventsBatch", () => {
+	it.effect("inserts a well-formed event and acks it", () =>
+		run(
+			Effect.gen(function* () {
+				const first = message(event("11111111-1111-4111-8111-111111111111"), 1)
+				yield* processAuditEventsBatch(batchOf(first))
+
+				expect(first.calls).toEqual(["ack"])
+				const database = yield* Database
+				const rows = yield* database.execute((db) => db.select().from(auditLogEntries))
+				expect(rows.map((row) => row.action)).toEqual(["dashboard.created"])
+			}),
+		),
+	)
+
+	// Redelivery is expected — the queue retries whole batches — so a second
+	// delivery of an already-inserted event must be a no-op, not a duplicate row.
+	it.effect("is idempotent across redelivery of the same event", () =>
+		run(
+			Effect.gen(function* () {
+				const body = event("22222222-2222-4222-8222-222222222222")
+				yield* processAuditEventsBatch(batchOf(message(body, 1)))
+				yield* processAuditEventsBatch(batchOf(message(body, 2)))
+
+				const database = yield* Database
+				const rows = yield* database.execute((db) => db.select().from(auditLogEntries))
+				expect(rows).toHaveLength(1)
+			}),
+		),
+	)
+
+	// Cloudflare routes a message to the DLQ only when the consumer retries it
+	// past `max_retries`. Acking on the final attempt would discard the entry
+	// instead, which is exactly the silent drop this branch exists to prevent.
+	it.effect("retries a failed insert on the final attempt so the message reaches the DLQ", () =>
+		Effect.gen(function* () {
+			const exhausted = message(event("33333333-3333-4333-8333-333333333333"), 6)
+			yield* processAuditEventsBatch(batchOf(exhausted))
+
+			expect(exhausted.calls).toEqual(["retry"])
+		}).pipe(Effect.provide(failingDatabase)),
+	)
+
+	it.effect("retries a failed insert while attempts remain", () =>
+		Effect.gen(function* () {
+			const failed = message(event("44444444-4444-4444-8444-444444444444"), 2)
+			yield* processAuditEventsBatch(batchOf(failed))
+
+			expect(failed.calls).toEqual(["retry"])
+		}).pipe(Effect.provide(failingDatabase)),
+	)
+
+	// A message that cannot decode will never decode. Retrying only burns the
+	// attempts that would otherwise carry a recoverable message to the DLQ.
+	it.effect("acks a malformed message instead of retrying it forever", () =>
+		run(
+			Effect.gen(function* () {
+				const malformed = message({ not: "an audit event" }, 1)
+				yield* processAuditEventsBatch(batchOf(malformed))
+
+				expect(malformed.calls).toEqual(["ack"])
+				const database = yield* Database
+				const rows = yield* database.execute((db) => db.select().from(auditLogEntries))
+				expect(rows).toEqual([])
+			}),
+		),
+	)
+})

apps/api/src/audit-events-runtime.ts

@@ -0,0 +1,120 @@
+import type { MessageBatch } from "@cloudflare/workers-types"
+import * as MapleCloudflareSDK from "@maple-dev/effect-sdk/cloudflare"
+import { ANTICIPATED_ERROR_IDENTIFIERS } from "@maple/domain/anticipated-errors"
+import { WorkerConfigProviderLayer, WorkerEnvironment } from "@maple/effect-cloudflare"
+import { auditLogEntries } from "@maple/db"
+import { Clock, Effect, Layer } from "effect"
+import { layerPg } from "@/platform/DatabasePgLive"
+import { Database } from "@/platform/DatabaseLive"
+import { auditEventToInsert, decodeAuditLogEvent } from "./services/audit/audit-event"
+
+const telemetry = MapleCloudflareSDK.make({
+	serviceName: "maple-api",
+	serviceNamespace: "core",
+	repositoryUrl: "https://github.com/MapleTechLabs/maple",
+	anticipatedErrorIdentifiers: [...ANTICIPATED_ERROR_IDENTIFIERS],
+})
+
+export const buildAuditEventsLayer = (_env: Record<string, unknown>) => {
+	const DatabaseLive = layerPg.pipe(Layer.provide(WorkerEnvironment.layer))
+	return DatabaseLive.pipe(
+		Layer.provideMerge(telemetry.layer),
+		Layer.provideMerge(WorkerEnvironment.layer),
+		Layer.provideMerge(WorkerConfigProviderLayer),
+	)
+}
+
+export const flushAuditEventsTelemetry = (env: Record<string, unknown>) => telemetry.flush(env)
+
+/**
+ * Must match `maxRetries` on the audit-events consumer in `alchemy.run.ts` and
+ * `wrangler.jsonc`. Cloudflare routes the message to the DLQ after this many
+ * retries without telling us; the check below is what makes the hand-off
+ * visible in logs at the moment it happens.
+ */
+const AUDIT_EVENTS_MAX_RETRIES = 5
+
+/**
+ * Best-effort identity for the exhaustion log. The body reached us as queue
+ * JSON and may be anything at all, so these read defensively rather than
+ * decoding — a drop must still be reported when the payload is the problem.
+ */
+const auditEventField = (body: unknown, field: string): string => {
+	if (typeof body !== "object" || body === null || !(field in body)) return "<unknown>"
+	// SAFETY: `field in body` established the key exists on this object.
+	const value = (body as Record<string, unknown>)[field]
+	return typeof value === "string" ? value : "<unknown>"
+}
+const auditEventOrgId = (body: unknown) => auditEventField(body, "orgId")
+const auditEventAction = (body: unknown) => auditEventField(body, "action")
+
+/**
+ * Audit events queue consumer: lowers each event to its `audit_log_entries`
+ * row. The `(org_id, id)` primary key plus `onConflictDoNothing` makes queue
+ * redelivery idempotent; insert failures retry through the queue's policy and,
+ * once exhausted, land in `audit-events-dlq` rather than disappearing.
+ */
+export const processAuditEventsBatch = (batch: MessageBatch<unknown>) =>
+	Effect.gen(function* () {
+		const database = yield* Database
+		yield* Effect.forEach(
+			batch.messages,
+			(message) =>
+				decodeAuditLogEvent(message.body).pipe(
+					Effect.matchEffect({
+						// Undecodable now means undecodable on every redelivery, so retrying
+						// only burns attempts. Acked, but at Error: an audit entry that
+						// never reaches a row is lost evidence, not routine noise.
+						onFailure: (error) =>
+							Effect.logError("Discarding malformed audit event queue message").pipe(
+								Effect.annotateLogs({ attempt: message.attempts, error: String(error) }),
+								Effect.flatMap(() => Effect.sync(() => message.ack())),
+							),
+						onSuccess: (event) =>
+							Effect.gen(function* () {
+								const now = yield* Clock.currentTimeMillis
+								yield* database.execute((db) =>
+									db
+										.insert(auditLogEntries)
+										.values(auditEventToInsert(event, now))
+										.onConflictDoNothing(),
+								)
+								yield* Effect.sync(() => message.ack())
+							}).pipe(
+								Effect.withSpan("auditEvents.processMessage"),
+								Effect.catchCause((cause) => {
+									// Retrying past the limit is what hands the message to the
+									// DLQ; acking here would silently discard it instead.
+									const isFinalAttempt = message.attempts > AUDIT_EVENTS_MAX_RETRIES
+									const outcome = isFinalAttempt ? "exhausted_dlq" : "retry"
+									return Effect.annotateCurrentSpan({
+										"audit.queue.message.outcome": outcome,
+									}).pipe(
+										Effect.flatMap(() =>
+											isFinalAttempt
+												? Effect.logError(
+														"Audit event exhausted retries; routed to dead letter queue",
+													).pipe(
+														Effect.annotateLogs({
+															attempt: message.attempts,
+															orgId: auditEventOrgId(message.body),
+															action: auditEventAction(message.body),
+															error: String(cause),
+														}),
+													)
+												: Effect.logWarning("Audit event insert failed; retrying").pipe(
+														Effect.annotateLogs({
+															attempt: message.attempts,
+															error: String(cause),
+														}),
+													),
+										),
+										Effect.flatMap(() => Effect.sync(() => message.retry())),
+									)
+								}),
+							),
+					}),
+				),
+			{ concurrency: 5, discard: true },
+		)
+	}).pipe(Effect.withSpan("auditEvents.processBatch"))

apps/api/src/mcp/app.ts

@@ -10,6 +10,8 @@ import { InstructionsResource } from "./resources/instructions"
 import { sessionStore } from "./lib/session-store"
 import type { McpToolExecutor } from "./dispatcher"
 import { CurrentMcpRequestTenant, CurrentMcpTenant, resolveHttpMcpTenant } from "./lib/query-warehouse"
+import { type AuditActorInfo, CurrentAuditActor } from "@/services/auth/audit-actor"
+import { INTERNAL_SERVICE_PREFIX } from "./lib/resolve-tenant"
 import { ApiKeysService } from "@/services/org/ApiKeysService"
 import { AuthService } from "@/services/auth/AuthService"
 import { Env } from "@/platform/Env"
@@ -93,6 +95,23 @@ const mcpUnavailable = () =>
 		),
 	)
 
+/**
+ * Which credential an MCP request presented, as far as the transport can tell.
+ * Mirrors the branches in `resolveMcpTenantContext`: an internal service token
+ * is Maple acting on its own behalf, any other bearer is an API key or OAuth
+ * token, and no bearer at all means a forwarded dashboard session.
+ */
+const mcpAuditActor = (headers: Record<string, string | undefined>): AuditActorInfo => {
+	const authorization = headers["authorization"] ?? headers["Authorization"]
+	if (authorization?.toLowerCase().startsWith("bearer ") !== true) {
+		return { type: "user", source: "mcp" }
+	}
+	const bearer = authorization.slice("bearer ".length).trim()
+	return bearer.startsWith(INTERNAL_SERVICE_PREFIX)
+		? { type: "system", source: "system" }
+		: { type: "api_key", source: "mcp" }
+}
+
 const McpAuthorizationMiddleware = HttpRouter.middleware<{ provides: CurrentMcpTenant }>()(
 	Effect.gen(function* () {
 		const apiKeys = yield* ApiKeysService
@@ -108,6 +127,12 @@ const McpAuthorizationMiddleware = HttpRouter.middleware<{ provides: CurrentMcpT
 					Effect.flatMap((tenant) =>
 						Effect.provideService(httpEffect, CurrentMcpTenant, tenant).pipe(
 							Effect.provideService(CurrentMcpRequestTenant, tenant),
+							// Without this an MCP mutation reads the reference's `undefined`
+							// default and is audited as a dashboard session. The credential
+							// kind is all this layer can see — `resolveMcpTenantContext`
+							// returns the tenant, not the key it resolved — so the key id is
+							// deliberately absent rather than guessed.
+							Effect.provideService(CurrentAuditActor, mcpAuditActor(request.headers)),
 						),
 					),
 					Effect.catchTags({

apps/api/src/mcp/lib/resolve-tenant.ts

@@ -14,7 +14,8 @@ import {
 import { recordExpectedMcpFailure } from "@/mcp/expected-failures"
 import { sessionStore } from "@/mcp/lib/session-store"
 
-const INTERNAL_SERVICE_PREFIX = "maple_svc_"
+/** Exported so the audit layer classifies the same token the same way. */
+export const INTERNAL_SERVICE_PREFIX = "maple_svc_"
 const decodeOrgId = Schema.decodeUnknownEffect(OrgId)
 const decodeUserId = Schema.decodeUnknownEffect(UserId)
 const decodeActorIdOption = Schema.decodeUnknownOption(ActorId)