#2112 Serve Audit: Extension sink API — design and implementation

Summary

Let extension authors register custom audit sinks (Kafka, Elasticsearch, Azure Event Hubs, PagerDuty, etc.) via the existing extension system. An extension exports an auditSink factory, operators reference it in serve config, and the serve instance loads it at startup alongside built-in sinks.

This was deferred from Phase 5 (#2103) because it needs design work. This issue contains the full design.

Design doc: https://claude.ai/code/artifact/40e51500-334b-40ef-89a9-dbd9185a109e

Background

The audit log shipped across five phases (#2004, #2028, #2049, #2074, #2103). All 106 handlers emit events. Events flow through a ring buffer to pluggable sinks. Today there are four built-in sinks: StoreSink (durable JSONL), WalSink (WAL wrapper), WebSocketSink (live streaming), WebhookSink (JSON/CEF to HTTP), and SyslogSink (RFC 5424). The emitter fans out to all registered sinks independently — a failing sink does not block others.

Existing infrastructure to build on

AuditSink interface (src/domain/serve_audit/audit_sink.ts:22-28)

interface AuditSink { readonly name: string; readonly durable: boolean; write(events: readonly AuditEvent[]): Promise; flush(): Promise; close(): Promise; }

The durable flag controls chain-state rollback in the emitter: if no durable sink succeeds, the chain rolls back. Extension sinks are durable: false unless they write to an append-only store.

AuditEmitter (src/domain/serve_audit/audit_emitter.ts)

• Accepts AuditSink[] in constructor
• replaceSinks(newSinks) for atomic swap (preserves cursors by name, used by hot-reload)
• Per-sink timeout (default 30s) wraps sink.write() in Promise.race with abort
• Sink errors are caught and logged per-sink, never propagate (line 279-287)

AuditSinkHotReloader (src/domain/serve_audit/audit_sink_hot_reloader.ts:25-73)

async reload(currentSinks, buildNewSinks): flush old, close old, return new. Already handles the rebuild-from-config pattern — extension sinks slot in naturally.

Sink config (src/serve/serve_config.ts)

AuditSinkConfigEntry has type: webhook | syslog. Validation at line 1498-1538 switches on type, applies per-type validation. The sink wiring at serve startup (src/cli/commands/serve.ts:2985-3084) iterates config entries and constructs WebhookSink or SyslogSink instances.

Extension system pattern (KindAdapter)

Each extension kind (model, vault, datastore, report) follows the same pattern:

1. A KindAdapter (src/domain/extensions/kind_adapter.ts) defines how to validate, register, and load
2. A TypeRegistry singleton — register(), get(), has(), lazy loading
3. A conformance test suite in packages/testing/
4. An export convention — the extension module exports a named const (e.g. export const datastore = ...)

DatastoreKindAdapter (src/domain/extensions/datastore_kind_adapter.ts:45-133) is the closest prior art:

• kind: datastore, primaryExportKey: datastore, exportRegex for bundle detection
• validatePrimaryExport() validates against a Zod schema: { type, name, description, configSchema?, createProvider }
• register() calls typeRegistry.register()
• registerLazy() / promoteFromLazy() for on-demand loading

ExtensionKind type (src/infrastructure/persistence/extension_catalog_store.ts:82-87) is the union: model | extension | vault | datastore | report.

Design

Extension author's export contract

Extension authors export a named const from their module:

export const auditSink = { type: string, // e.g. @acme/kafka-audit-sink name: string, // human-readable: Kafka Audit Sink description: string, // what it does configSchema: ZodType, // validates operator config from swamp-serve.yaml createSink(config, options): AuditSink, };

The createSink factory receives the validated config object and an options bag with: signal (AbortSignal for shutdown), logger (LogTape logger), instanceId (serve instance ID).

The returned AuditSink must implement write/flush/close. The extension controls its own batching, retry, and connection management.

Operator config

audit: sinks: - type: extension extension: @acme/kafka-audit-sink config: brokers: [kafka-1:9092, kafka-2:9092] topic: swamp-audit-events filter: categories: [auth, access, secrets] tier: management

The config block is validated against the extension's configSchema at serve startup. Filter works identically to webhook/syslog sinks — applied before events reach the sink.

Trust model

Extension sinks receive audit events post-HMAC and post-chain-hashing — sensitive fields are already hashed. This is operator-trusted code: the operator explicitly declares the extension in serve config, same trust model as datastore and vault extensions which already run in-process with full access to data.

No additional sandbox is needed. The existing per-sink protections are sufficient:

• Per-sink timeout (30s default, configurable) prevents a hung sink from blocking the drain loop
• Error isolation: emitter catches sink.write() errors per-sink (line 279-287), logs them, continues to other sinks
• A failing extension sink does not block store writes, WebSocket streaming, or other sinks
• AbortSignal propagation for graceful shutdown

Lifecycle

• Created at serve startup from config (src/cli/commands/serve.ts)
• Extension loaded via KindAdapter + TypeRegistry — same as datastores
• Hot-reloaded via AuditSinkHotReloader when serve.reload fires — old sink flushed and closed, new one built from reloaded config
• Closed on graceful shutdown

Implementation

Files to create

src/domain/extensions/audit_sink_kind_adapter.ts + test

• KindAdapter for audit sinks
• kind: audit-sink
• primaryExportKey: auditSink
• exportRegex: /export\s+const\s+auditSink\s*[=:]/
• validatePrimaryExport() validates against AuditSinkExportSchema (Zod): { type, name, description, configSchema, createSink }
• register() calls auditSinkTypeRegistry.register()

src/domain/serve_audit/audit_sink_type_registry.ts + test

• TypeRegistry singleton for audit sink extensions
• register(entry), get(type), has(type), list()
• Mirrors DatastoreTypeRegistry pattern

packages/testing/audit_sink_conformance.ts

• assertAuditSinkExportConformance(auditSinkExport, options)
• Validates: type matches naming pattern, name/description non-empty, configSchema accepts valid config and rejects invalid, createSink returns an object with write/flush/close methods
• Extension authors import this and call it in their test suite

packages/testing/audit_sink_types.ts

• Re-exports AuditSink and AuditEvent types so extension authors can import them from the testing package
• IMPORTANT: do not require extension authors to import a new package — these types come from the existing @swamp-club/swamp-testing package

Files to modify

src/infrastructure/persistence/extension_catalog_store.ts:82

• Add audit-sink to ExtensionKind union

src/serve/serve_config.ts

• Add extension to AuditSinkConfigEntry.type union (line 184)
• Add optional extension field (string, the extension name)
• Add validation branch for type === extension: requires extension field, loads extension's configSchema, validates config against it

src/cli/commands/serve.ts

• Sink construction loop (line 3002-3084): add else if (sinkEntry.type === extension) branch that loads extension via auditSinkTypeRegistry, validates config against extension's schema, calls createSink(config, { signal, logger, instanceId })
• Hot-reload lambda (line 3195-3291): same branch for rebuilding extension sinks from new config

src/cli/mod.ts

• Wire auditSinkKindAdapter into extension loader

src/cli/auto_resolver_adapters.ts

• Wire audit-sink adapter for auto-resolution

src/domain/serve_audit/mod.ts

• Re-export types needed by the testing package

packages/testing/mod.ts

• Re-export [REDACTED-SECRET-1] and audit sink types

design/enablers/serve-audit.md

• Update Phase 6 / future phases section to reflect this is in progress
• Add extension sink to the domain model table

What not to build

• No new npm/deno package for extension authors — types come from @swamp-club/swamp-testing (existing)
• No sandbox or isolation runtime — operator-trusted code, same as other extension kinds
• No cross-instance sink coordination — each instance loads and manages its own sink instances
• No built-in Kafka/Elasticsearch sinks — this is the API; community builds the sinks

Verification

1. Unit tests for AuditSinkKindAdapter: validate export shape, register/get/has, reject invalid exports
2. Unit tests for TypeRegistry: register, lazy load, promote
3. Conformance test: create a mock audit sink extension, run [REDACTED-SECRET-1] against it
4. Integration test: configure an extension sink in serve config, start serve, make requests, verify events reach the mock sink
5. Hot-reload test: start serve with extension sink, reload config, verify old sink closed and new one receives events
6. Error isolation test: extension sink that throws on write(), verify other sinks still receive events
7. Timeout test: extension sink that hangs on write(), verify drain loop continues after timeout

Prior art in this codebase

• DatastoreKindAdapter (src/domain/extensions/datastore_kind_adapter.ts) — closest pattern
• VaultKindAdapter (src/domain/extensions/vault_kind_adapter.ts) — same shape
• DatastoreTypeRegistry — map-backed, same lifecycle
• [REDACTED-SECRET-2] (packages/testing/) — same testing contract