Audit

zeroth.governance.audit

Audit evidence and emission domain.

This package provides everything you need to record, store, query, and review audit trails: data models, a SQLite-backed repository, payload sanitization (to strip secrets), a timeline assembler for viewing events in order, and the governed audit emitters consolidated from the vendored governai bundle (see docs/backend-import-migration.md).

AuditEmitter

Bases: ABC

emit abstractmethod async

emit(event: AuditEvent) -> None

Emit.

ApprovalActionRecord

Bases: BaseModel

A record of an approval-related action (e.g. requested, approved, denied).

Used when a node requires human or system approval before proceeding. Tracks who took the action and when it happened.

AuditContinuityReport

Bases: BaseModel

Verification result for a run or deployment audit chain.

AuditQuery

Bases: BaseModel

Filters for searching audit records.

Set one or more fields to narrow down which audit records you want. Leave a field as None to not filter on it. For example, set run_id to retrieve all audit records from a specific run.

AuditRedactionConfig

Bases: BaseModel

Rules that control which parts of audit payloads get hidden or removed.

Use this to protect sensitive data (like API keys or passwords) from appearing in audit logs. You can redact specific dictionary keys or omit entire nested paths.

AuditTimeline

Bases: BaseModel

A time-ordered list of audit records for a single run or scope.

Think of this as a "replay log" -- it shows you exactly what happened and in what order, making it easy to trace through a run step by step.

MemoryAccessRecord

Bases: BaseModel

A record of a single memory read or write during a node execution.

Tracks which memory store was accessed, what operation was performed (e.g. read, write, delete), and the key/value involved.

NodeAuditRecord

Bases: BaseModel

The main audit record for a single node execution.

This is the core audit object. It captures everything that happened when a node ran: inputs, outputs, tool calls, memory accesses, approval actions, validation results, timing, and any errors.

ToolCallRecord

Bases: BaseModel

A record of a single tool call made during a node execution.

Captures which tool was called, what arguments were passed in, what the tool returned, and whether it produced an error.

The operation_* fields carry the durable receipt outcome for guarded executable tools. MCP calls populate only the support and residual-risk marker because they bypass the operation boundary entirely.

AuditRepository

AuditRepository(
    database: AsyncDatabase,
    scope_context: ScopeContext
    | NullWorkspaceScopeContext
    | TenantWideScopeContext,
    signer: SigningKeyProvider | None = None,
)

Saves and retrieves audit records from an async database.

Use this class to store audit records when nodes run and to look them up later for debugging, compliance, or building timelines.

scoped classmethod

scoped(
    database: AsyncDatabase,
    scope_context: ScopeContext
    | NullWorkspaceScopeContext
    | TenantWideScopeContext,
    signer: SigningKeyProvider | None = None,
) -> AuditRepository

Construct an audit repository bound to one trusted tenant/workspace.

for_default_compatibility classmethod

for_default_compatibility(
    database: AsyncDatabase,
    *,
    signer: SigningKeyProvider | None = None,
) -> AuditRepository

Bind legacy tests and migration tools to the reserved default scope.

configure_capture

configure_capture(classifier: CaptureClassifier) -> None

Install the deployment's capture classifier, once, at wiring time.

The classifier is the only replaceable part of the capture boundary: it picks between two fixed outcomes and cannot author either, so a deployment can opt into retaining content without supplying the transform that decides what "retained" means.

Parameters:

Name	Type	Description	Default
classifier	CaptureClassifier	Decides per record whether content may be retained.	required

Raises:

Type	Description
ValueError	If a classifier was already installed. Capture posture is wiring, not a runtime switch: a repository whose policy can be swapped mid-flight has no posture at all.

write async

write(record: NodeAuditRecord) -> NodeAuditRecord

Save an audit record to the database.

Writes are append-only. Duplicate audit IDs are rejected so history cannot be silently rewritten. The record is classified and redacted first -- always, with no way for a caller to signal otherwise -- so what is digested and inserted is what the capture policy allows.

Raises:

Type	Description
DuplicateAuditIdError	If record.audit_id is already stored -- and only then. It subclasses ValueError so callers catching ValueError are unaffected, while one that treats "already stored" as a successful delivery can narrow to the exact type instead of reading every pre-commit failure as a durable record.

write_in_transaction async

write_in_transaction(
    connection: AsyncConnection | BoundStructuredTable,
    record: NodeAuditRecord,
) -> NodeAuditRecord

Append record using a caller-owned database transaction.

This is the atomic composition boundary for service state transitions: callers can mutate their scoped table and append the signed audit record under one commit. The supplied connection must belong to this repository's database; ScopedTable.in_transaction enforces that binding instead of opening a second transaction.

get async

get(
    audit_id: str, *, tenant_id: str | None = None
) -> NodeAuditRecord | None

Look up one audit record, optionally constrained by tenant in SQL.

list async

list(
    query: AuditQuery | None = None,
    *,
    limit: int | None = None,
) -> list[NodeAuditRecord]

Return audit records matching the given filters, ordered by time.

Pass an AuditQuery to filter by run, thread, node, etc. If no query is given, all records are returned.

limit bounds the read to the most recent limit records, still returned oldest-first. Without it a caller reads the deployment's entire audit history, which is what the econ-analytics and rightsizing routes used to do before filtering in Python.

list_by_run async

list_by_run(
    run_id: str,
    *,
    tenant_id: str | None = None,
    workspace_id: str | None = None,
    workspace_scoped: bool = False,
    deployment_ref: str | None = None,
) -> list[NodeAuditRecord]

Return all audit records for a specific run.

list_by_run_in_transaction async

list_by_run_in_transaction(
    connection: AsyncConnection, run_id: str
) -> list[NodeAuditRecord]

Return a run's records using the caller's database transaction.

list_by_thread async

list_by_thread(thread_id: str) -> list[NodeAuditRecord]

Return all audit records for a specific thread.

list_by_node async

list_by_node(node_id: str) -> list[NodeAuditRecord]

Return all audit records for a specific node.

list_by_graph_version async

list_by_graph_version(
    graph_version_ref: str,
) -> list[NodeAuditRecord]

Return all audit records for a specific graph version.

list_by_deployment async

list_by_deployment(
    deployment_ref: str,
    *,
    tenant_id: str | None = None,
    workspace_id: str | None = None,
    workspace_scoped: bool = False,
) -> list[NodeAuditRecord]

Return all audit records for a specific deployment.

write_many async

write_many(
    records: Sequence[NodeAuditRecord],
) -> list[NodeAuditRecord]

Save multiple audit records at once. Returns all saved records.

crypto_erase async

crypto_erase(
    audit_id: str, *, reason: str
) -> NodeAuditRecord | None

Crypto-erase a single record's PII while keeping the chain verifiable.

A SANCTIONED, append-only-preserving single-row UPDATE: it nulls the PII payload fields (input_snapshot, output_snapshot, stdout, tool calls, memory interactions, …), keeps pii_commitments and the digest, and stamps erased/erased_at/erasure_reason. Because a v2 digest is computed over the commitments (not the plaintext), the record digest, its signature, and the whole hash-chain still verify afterwards.

created_at, audit_id, previous_record_digest and record_digest are NEVER touched — re-chaining history is itself a tamper event. digest_version=1 (legacy) records are un-erasable and raise; an already-erased or missing record is a no-op (idempotent).

crypto_erase_in_transaction async

crypto_erase_in_transaction(
    connection: AsyncConnection | BoundStructuredTable,
    audit_id: str,
    *,
    reason: str,
    record: NodeAuditRecord | None = None,
) -> NodeAuditRecord | None

Crypto-erase one audit through an existing transaction.

list_erasable async

list_erasable(
    tenant_id: str,
    older_than: datetime,
    *,
    exclude_run_ids: Sequence[str] | None = None,
) -> list[NodeAuditRecord]

Return erasable commitment-digest records older than a cutoff.

Scoped to one tenant and to records created before older_than (compared as UTC isoformat, matching the write path). Legacy v1 records are excluded (un-erasable), as are records for any run in exclude_run_ids (legal-hold protected).

list_erasable_in_transaction async

list_erasable_in_transaction(
    connection: AsyncConnection | BoundStructuredTable,
    tenant_id: str,
    older_than: datetime,
    *,
    exclude_run_ids: Sequence[str] | None = None,
) -> list[NodeAuditRecord]

Transaction-scoped :meth:list_erasable for coordinated sweeps.

One cutoff-bounded projection query; only aged rows are hydrated — never the tenant's full audit history.

PayloadSanitizer

PayloadSanitizer(
    config: AuditRedactionConfig | None = None,
)

Cleans audit payloads by redacting or removing sensitive data.

Given a redaction config, this class walks through a payload (dicts, lists, etc.) and replaces sensitive keys with "REDACTED" or drops entire paths that should not appear in audit logs.

sanitize

sanitize(payload: Any) -> Any

Clean a payload by applying all configured redaction rules.

Pass in any data structure (dict, list, or primitive) and get back a copy with sensitive values masked or removed.

AuditTimelineAssembler

Builds a time-ordered timeline from a collection of audit records.

Use this when you have a bunch of audit records and want to see them in the order they actually happened, like a replay of the run.

assemble

assemble(
    records: Sequence[NodeAuditRecord],
) -> AuditTimeline

Sort the given records by time and return them as an AuditTimeline.

Records are ordered by their start time, with ties broken by audit_id to keep the ordering stable and predictable.

AuditContinuityVerifier

AuditContinuityVerifier(
    repository, signer: SigningKeyProvider | None = None
)

Verify digest continuity and (WS-D) signatures for audit history.

signer is optional: without it the digest axis still verifies and the signature axis reports None (unsigned-legacy). The API threads the process signer in so signed records are checked against the real key.

verify_run async

verify_run(
    run_id: str,
    *,
    tenant_id: str | None = None,
    workspace_id: str | None = None,
    workspace_scoped: bool = False,
    deployment_ref: str | None = None,
) -> AuditContinuityReport

Verify digest continuity and signatures for one run's audit chain.

Records are strictly ordered first; an ordering violation yields a failed report (verified=False) rather than an exception.

verify_deployment async

verify_deployment(
    deployment_ref: str,
    *,
    tenant_id: str | None = None,
    workspace_id: str | None = None,
    workspace_scoped: bool = False,
) -> AuditContinuityReport

Verify every run recorded under deployment_ref.

Chains are per-run: each run is ordered and verified independently, with per-run signature states aggregated into one three-state result. Verification loads the complete tenant/workspace-scoped run chain because legacy service-request identities were shared across deployments. The response count remains the number of records attributed to this deployment; no records from another deployment are returned. The first failing run short-circuits into a failed deployment report.

emit_event async

emit_event(
    emitter: AuditEmitter,
    *,
    run_id: str,
    thread_id: str | None = None,
    workflow_name: str,
    event_type: EventType,
    step_name: str | None = None,
    payload: dict[str, Any] | None = None,
    extensions: list[AuditExtension] | None = None,
) -> AuditEvent

Emit event.

build_summary

build_summary(
    audits: list[NodeAuditRecord],
    approvals: list[object],
    *,
    resolve_artifacts: bool = False,
    artifact_store: Any | None = None,
) -> dict[str, int | float | str | bool]

Summarize the key governance signals in a bundle.

When resolve_artifacts is True and an artifact_store is provided, the summary includes an artifacts_resolved flag to indicate that artifact payloads have been resolved in the evidence export.

collect_policy_events

collect_policy_events(
    audits: list[NodeAuditRecord],
) -> list[str]

Extract policy and authorization failures into a review-friendly list.

Parameters:

Name	Type	Description	Default
audits	list[NodeAuditRecord]	The bundle's audit records, already capture-transformed.	required

Returns:

Type	Description
list[str]	One line per denied attempt, naming the node, the decision and the
list[str]	normalized reason code. Every part comes from an allowlisted metadata
list[str]	key or a typed column, so the list survives a metadata-only capture and
list[str]	carries no producer text.

compute_chained_record

compute_chained_record(
    record: NodeAuditRecord,
    previous_digest: str | None,
    signer: SigningKeyProvider | None = None,
) -> NodeAuditRecord

Fill chain + (optional) signature fields deterministically.

Computes record_digest over the predecessor-linked record, then signs signable_bytes(record_digest, key_id, alg). When signer is None (or a :class:NullSigner) the signature fields stay None — unsigned-legacy, never signed-invalid.