Skip to main content

SDK observability

Replay one governed run, compare related calls, export records, or follow live traffic from the same Python surface.

Status: Beta Companion docs: Logs API · Metrics, health, and logging · Receipts Source of truth: the five REST methods and optional live stream implemented by tr.observability.

Replay a governed run​

from trinitite import Trinitite

tr = Trinitite(env="prod")
run = tr.observability.replay("plog_abc123")

print(run.log_id)
print(run.subsystem)
print(run.verdict)
print(run.correlation_id)
print(run.governance)

The replay endpoint accepts governed log ids. Prefix routing is handled by the service for plog_, log_, and icap_ records. The canonical replay verdict is read from governance.verdict, with a flat field fallback for older records.

Compare correlated calls​

diff = tr.observability.diff("corr_checkout_4821")
if diff.ok:
    print(diff.body)

# A replay handle reuses its own correlation id when available.
same_diff = run.diff()

Use a correlation id to compare changes across calls that belong to one application flow.

Query recent logs​

recent = tr.observability.filter(
    subsystem="proxy",
    status="completed",
    has_violations=True,
).last(50)

for row in recent.body["logs"]:
    print(row["log_id"])

last(n) requests JSON export, parses the NDJSON lines, and returns:

{
    "logs": [...],
    "count": 2,
    "_raw": "...",
    "row_count_header": "2",
}

The filter handle is mutable. Build and consume one chain at a time instead of sharing it between concurrent tasks.

Export records​

cef = tr.observability.export(
    format="siem",
    subsystem="proxy",
    guardian_id="refund-guardian",
    start_timestamp="2026-08-01T00:00:00Z",
    end_timestamp="2026-08-23T23:59:59Z",
    limit=500,
)

print(cef.body["_raw"])
print(cef.headers.get("x-export-row-count"))

Formats are siem, cef, leef, and json. The SDK maps siem to cef. CEF and LEEF are returned as text under body["_raw"]; JSON export is NDJSON.

Audit chain integrity​

audit = tr.observability.chain_audit(
    since="2026-08-01T00:00:00Z",
    to_timestamp="2026-08-23T23:59:59Z",
    max_entries=1000,
)

assert audit.body["valid"] is True
print(audit.body["entries_verified"])
print(audit.body["coverage"])

The response uses valid, total_entries, entries_verified, optional first_invalid_index, optional invalid_entries, coverage, and verified_at.

shredding = tr.observability.shredding_status()
print(shredding.body)

Stream live traffic​

Install the optional stream dependency:

pip install 'trinitite[stream]'

Then consume matching events:

from trinitite import ObservabilityStreamUnavailableError

try:
    for event in tr.observability.stream(
        scope="refund-flow",
        verdict="blocked",
        subsystem="proxy",
        timeout_seconds=30,
    ):
        print(event)
except ObservabilityStreamUnavailableError as exc:
    recent = tr.observability.filter(
        subsystem=exc.subsystem,
    ).last(50)

The stream is a synchronous generator. timeout_seconds is a silence timeout. The generator completes when no event arrives within that interval.

Next steps​

  • Traces and graph: move from a replay envelope into connected policy and decision nodes.
  • Receipts: verify the signed receipt on a governed result.
  • Authentication: use the right key and permissions for exports and streams.
Last updated on