Tracing

indusagi.tracing is the framework's homegrown, dependency-light observability subsystem — an OpenTelemetry-free way to trace agent runs, model calls, and tool actions. Imported as import indusagi.tracing, or via the lazy tracing submodule on the top-level indusagi package.

A trace is modeled as immutable Segment values driven by a functional SegmentHandle (note/child/fail/close). Each state transition emits exactly one TraceSignal onto an async-iterable SignalChannel, and terminal Sinks drain that channel to a console, an NDJSON file, or an arbitrary stream. Admission control (SampleGate), a named-recorder registry (TelemetryHub), credential redaction (SecretScrubber), and an optional RunEvent adapter round out the surface. The vocabulary is deliberately fresh — there is no span/exporter/tracer-provider naming.

Table of Contents

• Public exports
• Sub-directories
• The segment lifecycle
• Recorders and sampling
• Draining to sinks
• Redaction
• The telemetry hub
• Bridging the runtime
• Relationship to neighbors

Public exports

The package barrel re-exports its layers in a bottom-up order (signal, channel, redaction, sinks, recorder, registry, adapter):

Sub-directories

The segment lifecycle

A Segment is a frozen dataclass — the OTel-free analogue of a span. It is never mutated in place: the lifecycle open → note → close is modeled by producing new frozen values through open_segment, with_attributes, and close_segment. Because every value is immutable, a closed TraceRecord is trivially serializable.

Callers rarely touch those functions directly. The behavioural face is SegmentHandle, a Protocol with note, child, fail, and close. There is intentionally no class hierarchy:

• A sampled-in handle is the plain _LiveHandle returned by open_handle. It holds the segment's current frozen value, emits an OpenSignal on construction, and on each note/fail/close derives the next frozen value and emits an UpdateSignal/CloseSignal through an injected SignalSink callable.
• A sampled-out handle is the single shared NOOP_HANDLE — its methods do nothing and its child() returns itself, so an entire sampled-out subtree costs nothing.

note, child, and fail are no-ops once a handle is closed, and close is idempotent (the first call wins), so a double-close in a finally block is safe. fail(error) records a SegmentError and sets the closing status to error without closing — the actual close carries it. close_segment preserves a pre-existing error even on an explicit close("ok"): only terminal fields are replaced.

Each handle transition emits exactly one TraceSignal. These flow onto a SignalChannel: a single-consumer async iterable backed by a deque plus an asyncio future waiter. emit() appends and wakes a parked consumer; async for pulls; close() ends iteration after draining. The channel is lossy under backpressure — past bound it drops the oldest unread signal (newest wins); a None/non-positive bound is unbounded. emit/close are synchronous and must run on the consumer's event-loop thread; off-loop producers must marshal via loop.call_soon_threadsafe.

import asyncio
from indusagi.tracing import Recorder, RecorderOptions, ConsoleSink


async def main() -> None:
    rec = Recorder(RecorderOptions(service_name="my-agent"))
    sink = ConsoleSink()
    drain_task = asyncio.create_task(sink.drain(rec.channel()))

    run = rec.open("run", "answer-question")
    inf = run.child("inference", "chat.completion")
    inf.note({"model": "opus", "tokens.in": 1200})
    inf.close("ok")
    act = run.child("action", "write_file")
    act.fail("EACCES: permission denied")  # sets the closing status to error
    act.close()
    run.close("ok")

    rec.channel().close()   # end the stream so drain() returns
    await drain_task


asyncio.run(main())

Recorders and sampling

A Recorder is the flat tracer instance — it replaces the OTel tracer-provider / tracer pair with one object. It holds a service_name, a SampleGate, and a SignalChannel. open(kind, name, opts) resolves a trace id (a supplied opts.trace_id joins an existing trace, otherwise a fresh one is minted), consults the gate, and either returns NOOP_HANDLE (sampled out — nothing is emitted) or calls open_handle bound to the channel's emit. A root open folds service.name into the segment's attributes. channel() exposes the stream sinks drain.

SampleGate.decide(trace_id) -> bool is admission control. "always" and "never" are constant verdicts; RatioStrategy(ratio=...) is deterministic ratio sampling, not a random draw. The decision is _unit_hash(trace_id) < ratio, where _unit_hash is a dependency-free FNV-1a folded to [0, 1). The hash masks with & 0xFFFFFFFF after every multiply to reproduce a uint32 wrap-around exactly, so the same trace id always yields the same verdict — across processes and replays — and children opened on an admitted trace id are themselves admitted. A gate holds no mutable state, so one instance is safely shared across recorders and tasks.

from indusagi.tracing import Recorder, RecorderOptions, RatioStrategy, SampleGate

# Either pass a strategy directly...
rec = Recorder(RecorderOptions(service_name="svc", sampling=RatioStrategy(ratio=0.25)))

# ...or a prebuilt gate (shareable, stateless):
gate = SampleGate(RatioStrategy(ratio=0.25))
admitted = gate.decide("a1b2c3d4e5f60718a1b2c3d4e5f60718")  # deterministic per trace id

h = rec.open("run", "task")
print(h.active)  # False when sampled out -> this is the shared NOOP_HANDLE

Draining to sinks

A Sink is the terminal end of the transport — the OTel-free analogue of an exporter, but a plain ABC with one mandatory async drain(channel) pull-loop plus optional flush()/close() lifecycle hooks (both default to no-ops). Each sink owns its own async for … in channel loop and acts only on CloseSignal, ignoring open/update chatter. Three concrete sinks ship:

• ConsoleSink prints one tidy line per closed segment — glyph, kind, name, duration, and a short trace/segment id tag — through an injected log callable (default print). Stateless and resource-free, so it needs neither flush nor close.
• FileSink(path, FileSinkOptions(flush_every=...)) appends each closed record as an NDJSON line, buffering and flushing via asyncio.to_thread on the flush_every threshold, at the end of drain, and on close.
• StreamSink is the most general: it writes each closed record as a JSON line to an asyncio.StreamWriter (or any text writer), honours backpressure via the writer's drain(), and optionally ends the stream on close (end_on_close).

FileSink and StreamSink serialize through the shared base._record_to_json, which emits camelCase keys (traceId, parentId, startedAt, endedAt) with compact separators and default=str so a non-serializable attribute value won't crash the sink.

Redaction

SecretScrubber is a stateless attribute processor. scrub(attributes) returns a deep copy with secret-looking keys and values replaced by REDACTION_TOKEN — the guillemet string '‹redacted›', deliberately not '[REDACTED]'. Rules come from DEFAULT_SECRET_PATTERNS (credential key names, auth headers, bearer/vendor-key/JWT/ PEM value shapes), each a SecretPattern with optional key/value regexes matched unanchored via re.search. The walk is iterative with an explicit stack (no recursion) and never mutates its input; when a key name looks secret it short-circuits descent and collapses the whole subtree to the token.

The scrubber is exported so sinks can sanitize attributes before persisting, but the bundled ConsoleSink/FileSink/StreamSink do not call it automatically — a caller must apply it explicitly.

import asyncio
from indusagi.tracing import (
    Recorder, RecorderOptions, FileSink, FileSinkOptions, SecretScrubber,
)

scrubber = SecretScrubber()  # uses DEFAULT_SECRET_PATTERNS + REDACTION_TOKEN
safe = scrubber.scrub({"authorization": "Bearer abcd1234efgh", "model": "opus"})
# -> {'authorization': '‹redacted›', 'model': 'opus'}


async def main() -> None:
    rec = Recorder(RecorderOptions(service_name="svc"))
    sink = FileSink("/tmp/trace.ndjson", FileSinkOptions(flush_every=10))
    task = asyncio.create_task(sink.drain(rec.channel()))

    step = rec.open("custom", "step")
    step.note(safe)
    step.close("ok")

    rec.channel().close()
    await task
    await sink.close()  # final flush


asyncio.run(main())

The telemetry hub

TelemetryHub is a named registry of Recorders. register(name, options) adds (or replaces) a recorder and returns it; recorder(name=None) looks one up, falling back to the default and raising LookupError if the requested or default recorder is missing (so dropped traces fail loudly). get_default() returns the default recorder or None, and has(name) checks membership. The first registration becomes the default unless a later register uses the reserved "default" name, which always re-designates it.

get_default_hub() lazily creates and returns a process-wide default hub, and set_default_hub(hub) replaces it — both strictly opt-in; nothing in the subsystem reads the process default implicitly.

from indusagi.tracing import TelemetryHub, RecorderOptions, get_default_hub

hub = TelemetryHub()
hub.register("agent-a", RecorderOptions(service_name="agent-a"))  # becomes default
hub.register("agent-b", RecorderOptions(service_name="agent-b"))

rec = hub.recorder("agent-b")
default_rec = hub.recorder()        # the default (agent-a)
print(hub.has("agent-b"))           # True

shared = get_default_hub()          # opt-in process-wide hub

Bridging the runtime

trace_agent_run is the optional, decoupled bridge that projects the runtime's flat RunEvent stream into nested run/inference/action segments on a Recorder. It subscribes to an event stream whose shape matches Agent.subscribe (captured as RunEventSubscribe) and maintains the run/inference/action handles in a closure: snapshot phases drive the inference (an invoking phase opens one; dispatching/compacting/idle close it), tool_started/tool_finished map to actions, and settled/faulted close the run. It also folds streamed text_delta/thinking_delta counts onto the open inference as stream.text_deltas/stream.thinking_deltas attributes. The call returns a disposer that detaches and closes anything still open. It is best-effort and never raises on malformed input — out-of-order, missing, or repeated events just close what's open and ignore the rest.

from indusagi.tracing import Recorder, RecorderOptions, trace_agent_run

rec = Recorder(RecorderOptions(service_name="agent"))

# `agent.subscribe(handler) -> unsubscribe` matches RunEventSubscribe's shape.
dispose = trace_agent_run(rec, agent.subscribe)
try:
    ...  # run the agent; run/inference/action segments flow onto rec.channel()
finally:
    dispose()  # detaches and closes any still-open segments

Relationship to neighbors

The tracing core (signal, channel, recorder, registry, redaction, sinks) is entirely runtime-agnostic and sits below or beside the rest of the framework. The only coupling to the Runtime is the optional adapter layer: adapter/runtime_trace.py imports RunEvent under TYPE_CHECKING only and reads its public shape (event.kind, snapshot.run_id/phase, event.id/name, event.outcome.is_error, event.error) via getattr/match — it never touches runtime internals, and nothing in the runtime depends on it.

Internally the layers form an acyclic stack: handle imports channel.signal and signal.segment; channel.signal references segment only under TYPE_CHECKING to stay acyclic; recorder composes channel, handle, and sampling; registry wraps recorder; sinks consume the channel and share base._record_to_json. The top-level package exposes tracing as a lazy submodule. For the model conversations being traced, see the LLM Gateway; for the broader picture of how subsystems fit together, see the Architecture.