trail

trail

Core ContextTrail class for logging and auditing context inputs.

ContextTrail(*, storage_path='provena.db', backend='sqlite', required_fields=None, max_age_days=90, temporal_detection=True, max_content_bytes=65536, signing_key=None, otel_enabled=False, otel_service_name='provena', strict_mode=False, on_error=None, policies=None, buffered=False, buffer_size=500, flush_interval=1.0, config=None)

Tamper-evident audit trail for AI agent context inputs.

Logs every context input with provenance validation, freshness checking, and SHA-256 hash chaining. Supports both programmatic logging via log() and automatic tracking via the @trail.track() decorator.

Can be used as a context manager to ensure the storage backend is closed.

Initialize a ContextTrail.

Parameters:

Name	Type	Description	Default
storage_path	str	File path for the SQLite database.	'provena.db'
backend	str	Storage backend type, "sqlite", "memory", or "postgresql".	'sqlite'
required_fields	list[str] | tuple[str, ...] | None	Provenance fields to require for VALID status.	None
max_age_days	int	Content older than this is marked STALE.	90
temporal_detection	bool	Enable regex-based date detection in content text.	True
max_content_bytes	int	Truncate content beyond this size.	65536
signing_key	str | bytes | None	HMAC key for signed hash chains. Also read from PROVENA_SIGNING_KEY env var.	None
otel_enabled	bool	Emit OpenTelemetry spans for each logged entry.	False
otel_service_name	str	Service name for OTel spans.	'provena'
strict_mode	bool	If True, governance errors propagate as exceptions.	False
on_error	Callable[[Exception], None] | None	Optional callback invoked on governance errors.	None
policies	list[Policy] | None	List of Policy objects to enforce on every logged entry.	None
buffered	bool	If True, batch writes via a background thread.	False
buffer_size	int	Flush when buffer reaches this many records.	500
flush_interval	float	Seconds between periodic flushes.	1.0
config	dict[str, Any] | str | Path | None	Configuration dict, or path to a .toml/.yaml file.	None

Raises:

Type	Description
ValueError	If max_age_days or max_content_bytes is less than 1.

error_count property

Number of governance errors encountered during this trail's lifetime.

is_signed property

Whether this trail uses HMAC-signed hash chains.

record_count property

Total number of records in the audit trail, including unflushed buffered records.

last_record property

The most recently logged record, or None if the trail is empty.

log(content, source, source_name='', *, provenance=None, metadata=None)

Log a context input to the audit trail.

Computes a content hash, validates provenance, checks freshness, and appends a hash-chained record to storage.

Parameters:

Name	Type	Description	Default
content	str | bytes	The raw context content (string or bytes).	required
source	ContextSource | str	A ContextSource enum value or string like "tool:api".	required
source_name	str	Optional explicit source name.	''
provenance	ProvenanceMetadata | None	Optional origin metadata for validation.	None
metadata	dict[str, Any] | None	Arbitrary key-value pairs to attach to the record.	None

Returns:

Type	Description
TrailRecord | None	The created TrailRecord, or None if a non-strict error occurred.
TrailRecord | None	In buffered mode the record has not reached the backend yet, so
TrailRecord | None	its id is the placeholder -1 rather than a persisted row
TrailRecord | None	id. Call flush() before using it with annotate().

track(source, source_name='', *, content_extractor=None, policies=None)

Decorator that automatically logs function return values to the trail.

Supports both sync and async functions. The decorated function's return value is passed through unchanged.

Parameters:

Name	Type	Description	Default
source	ContextSource | str	A ContextSource enum value or string identifying the source.	required
source_name	str	Optional explicit source name.	''
content_extractor	Callable[..., str | bytes | list[Any]] | None	Optional callable to extract loggable content from the return value.	None
policies	list[Policy] | None	Optional per-decorator policy override. When provided, these policies are used instead of the trail-level policies for calls through this decorator.	None

Returns:

Type	Description
Callable[[F], F]	A decorator that wraps the target function with trail logging.

verify_chain()

Verify the integrity of the entire hash chain.

Flushes pending buffered records before verification so the verdict always covers the complete trail.

Recomputes every chain hash from the genesis hash forward and checks each against the stored value.

Returns:

Type	Description
ChainVerdict	A ChainVerdict indicating whether the chain is intact.

query(*, source=None, start=None, end=None, provenance_status=None, freshness_status=None, limit=100, offset=0, run_id=None)

Return trail rows matching the filters, oldest first.

When buffering is enabled, unflushed records are included. Those rows have not been assigned a backend id yet and are returned with id=-1. Filters match InMemoryBackend.query (string timestamp comparison, metadata_json.run_id).

annotate(record_id, note, reviewer='')

Add a human oversight annotation to a trail record.

Parameters:

Name	Type	Description	Default
record_id	int	The ID of the record to annotate. Must be >= 1.	required
note	str	The annotation text.	required
reviewer	str	Optional name of the reviewer.	''

Returns:

Type	Description
int	The ID of the created annotation.

Raises:

Type	Description
ValueError	If record_id is less than 1. In buffered mode log() returns TrailRecord(id=-1) until the record reaches the backend, and annotating that placeholder is rejected by every backend with a different low-level error.

get_annotations(record_id)

Return all annotations for record_id in insertion order.

Returns an empty list if the record does not exist or has no annotations (does not raise).

summary()

Generate an aggregate summary of the audit trail.

Returns:

Type	Description
dict[str, Any]	A dictionary with total count, provenance/freshness/source
dict[str, Any]	breakdowns, and signing status.

export(format='json')

Export all trail records in the specified format.

Parameters:

Name	Type	Description	Default
format	str	Output format, either "json" or "csv" or "json_with_annotations".	'json'

Returns:

Type	Description
str	The serialized trail data as a string.

health()

Return a health-check dictionary for the trail.

Returns:

Type	Description
dict[str, Any]	A dictionary with status, record count, backend type,
dict[str, Any]	signing state, and error count.

flush()

Flush buffered records to the storage backend.

Returns:

Type	Description
int	Number of records flushed, or 0 if buffering is not enabled.

close()

Close the storage backend and release resources.

__enter__()

Enter the context manager, returning this trail.

__exit__(*args)

Exit the context manager, closing the storage backend.