Runtime

Sinks

techrevati.runtime.sinks

Sinks — pluggable observability outputs for the runtime.

The orchestrator emits two streams while running:

  • AgentEvent lifecycle events (state transitions, recovery, gate, permission, handoff, budget). These go to an EventSink.
  • Per-turn usage (model + UsageSnapshot + cost). These go to a UsageSink.

Both protocols are tiny and synchronous. Default implementations buffer in memory with a bounded ring so long-running sessions can't balloon. Durable and cross-process observability can be added through separate sink implementations.

EventSink

Bases: Protocol

Receives every AgentEvent the runtime produces.

UsageSink

Bases: Protocol

Receives per-turn usage tuples (model, snapshot, cost in USD).

NoopEventSink dataclass 

NoopEventSink()

Validate and discard every event.

This keeps no-op behavior observability-free while still enforcing the same protocol boundary as concrete sinks.

NoopUsageSink dataclass 

NoopUsageSink()

Validate and discard every usage record.

FanoutEventSink dataclass 

FanoutEventSink(sinks, suppress_errors=False)

Forward each event to multiple event sinks.

The fan-out attempts every configured sink even when one fails. By default it re-raises the first sink exception after fan-out so AgentSession can record its normal local diagnostic event while still keeping the session alive.

FanoutUsageSink dataclass 

FanoutUsageSink(sinks, suppress_errors=False)

Forward each usage record to multiple usage sinks.

RingBufferEventSink dataclass 

RingBufferEventSink(capacity=DEFAULT_RING_CAPACITY)

In-memory bounded ring of recent events.

Useful for tests, debug consoles, and short-lived processes. The buffer drops oldest entries silently once capacity is reached; if you need durability, plug in a persistent sink or write a custom one.

RingBufferUsageSink dataclass 

RingBufferUsageSink(capacity=DEFAULT_RING_CAPACITY)

In-memory bounded ring of recent usage records.