aggregator
aggregator
¶
Multi-trail aggregation for governing multi-agent systems.
HandoffEdge(from_trail, from_record_id, to_trail, to_record_id, run_id='')
dataclass
¶
A tracked context handoff between two agents' trails.
Attributes:
| Name | Type | Description |
|---|---|---|
from_trail |
str
|
Label of the source trail. |
from_record_id |
int
|
Record ID in the source trail. |
to_trail |
str
|
Label of the destination trail. |
to_record_id |
int
|
Record ID in the destination trail. |
run_id |
str
|
Optional workflow/task run identifier grouping this handoff. |
TrailVerdict(label, verdict)
dataclass
¶
Per-trail chain verification result within an aggregation.
Attributes:
| Name | Type | Description |
|---|---|---|
label |
str
|
The trail's label in the aggregator. |
verdict |
ChainVerdict
|
The ChainVerdict from verifying that trail. |
AggregateVerdict(all_intact, trail_verdicts, total_records)
dataclass
¶
Aggregate chain verification across all trails.
Attributes:
| Name | Type | Description |
|---|---|---|
all_intact |
bool
|
True only if every trail's chain is intact. |
trail_verdicts |
tuple[TrailVerdict, ...]
|
Per-trail verification results. |
total_records |
int
|
Sum of records across all trails. |
EvidenceGap(trail, gap_type, details, record_id=None)
dataclass
¶
A detected governance gap in the multi-agent pipeline.
Attributes:
| Name | Type | Description |
|---|---|---|
trail |
str
|
Label of the trail where the gap was found. |
gap_type |
str
|
Category of gap (broken_chain, stale_context, missing_provenance, unlinked_handoff). |
details |
str
|
Human-readable description. |
record_id |
int | None
|
Optional record ID related to this gap. |
TrailAggregator()
¶
Aggregates governance data across multiple ContextTrail instances.
Each trail maintains its own independent hash chain. The aggregator provides unified querying, summary, verification, handoff tracking, and evidence gap detection across all trails.
labels
property
¶
Return the registered trail labels.
trail_count
property
¶
Number of registered trails.
handoffs
property
¶
Return all recorded handoff edges.
add(label, trail)
¶
Register a named trail.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
label
|
str
|
Unique label identifying this trail (e.g. agent name). |
required |
trail
|
Any
|
The ContextTrail instance. |
required |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the label is already registered. |
remove(label)
¶
Remove a trail by label.
get_trail(label)
¶
Return the trail for a given label, or None.
record_handoff(from_trail, from_record_id, to_trail, to_record_id, run_id='')
¶
Record a context handoff between two agents' trails.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
from_trail
|
str
|
Label of the source agent's trail. |
required |
from_record_id
|
int
|
Record ID of the output in the source trail. |
required |
to_trail
|
str
|
Label of the destination agent's trail. |
required |
to_record_id
|
int
|
Record ID of the input in the destination trail. |
required |
run_id
|
str
|
Optional workflow/task identifier. |
''
|
Returns:
| Type | Description |
|---|---|
HandoffEdge
|
The created HandoffEdge. |
summary()
¶
Aggregate summary across all trails.
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
A dictionary with per-trail and aggregate counts for |
dict[str, Any]
|
provenance, freshness, and source breakdowns. |
query(*, trail_label=None, source=None, start=None, end=None, provenance_status=None, freshness_status=None, run_id=None, limit=100)
¶
Query across all trails with optional filters.
Each returned record includes a _trail field with the trail label.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
trail_label
|
str | None
|
Restrict to a single trail by label. |
None
|
source
|
ContextSource | str | None
|
Filter by context source type. |
None
|
start
|
datetime | None
|
Include only records at or after this timestamp. |
None
|
end
|
datetime | None
|
Include only records at or before this timestamp. |
None
|
provenance_status
|
str | None
|
Filter by provenance validation status. |
None
|
freshness_status
|
str | None
|
Filter by freshness check status. |
None
|
run_id
|
str | None
|
Filter by run_id (matches records involved in handoffs tagged with this run_id). |
None
|
limit
|
int
|
Maximum total records to return. |
100
|
Returns:
| Type | Description |
|---|---|
list[dict[str, Any]]
|
A list of record dicts, each tagged with |
verify_chain()
¶
Verify chain integrity of every registered trail.
Returns:
| Type | Description |
|---|---|
AggregateVerdict
|
An AggregateVerdict with per-trail results. |
timeline(*, start=None, end=None, limit=200)
¶
Merged chronological view across all trails.
Returns records sorted by timestamp, each tagged with _trail.
Handoff edges are interleaved as synthetic _handoff entries.
detect_gaps()
¶
Detect governance gaps across all trails.
Checks for: - Broken hash chains - Stale context entries - Missing provenance - Unlinked handoffs (referencing non-existent records/trails)
Returns:
| Type | Description |
|---|---|
list[EvidenceGap]
|
A list of EvidenceGap instances. |
handoffs_for_run(run_id)
¶
Return all handoff edges for a specific run/task ID.
close()
¶
Close all registered trails.
Closing continues even if a trail raises, so one failing trail cannot leak the backends of the trails registered after it. If any trail failed to close, the first exception is raised after every trail has had a chance to close; any additional exceptions are logged rather than dropped silently.