# rig-tap
**Emits Uniform Telemetry** for [Rig](https://crates.io/crates/rig-core)
agents and companion crates. `rig-tap` defines one stable, versioned
`ObservabilityEvent` stream for prompts, tools, memory, context, and dispatch,
then emits it through `tracing` so any consumer (OpenTelemetry collectors,
Phoenix, Langfuse, a custom dashboard, or a log shipper) can subscribe without
crate-specific glue.
This crate is the telemetry contract, not a UI. It gives every producer in the
Rig ecosystem the same event vocabulary, whether the event came from a Rig
agent hook, `rig-compose` dispatch, `rig-memvid` memory behavior, or a host
application.
## Emits Uniform Telemetry
`rig-tap` exists to make ecosystem telemetry look the same at the boundary:
- **One schema** — every event is an `ObservabilityEvent` envelope with a stable
`kind`, schema `version`, timestamp, monotonic `tick`, and `conversation_id`.
- **One transport** — events are emitted under the dedicated `rig_tap` tracing
target, so existing `tracing`, JSON log, and OpenTelemetry pipelines can keep
their normal setup.
- **One lifecycle vocabulary** — prompt, tool, context, memory, and
`rig-compose` kernel-loop events share names and correlation fields across
crates.
- **One collector shape** — each event includes the full JSON envelope plus
scalar `rig_tap.*` attributes for collector-side routing and indexing.
Use `rig-tap` when you want memory crates, model metadata hooks, orchestration
kernels, and application code to speak the same telemetry language.
## What it provides
- **`ObservabilityEvent` + `EventKind`** — the wire schema. Event kinds cover
prompt / tool / context / memory plus `rig-compose` kernel and loop
lifecycle events, all emitted with the same envelope shape.
- **`TelemetryHook<M>`** — implements `rig::agent::PromptHook<M>` and emits
`prompt.*` and `tool.*` events from the five `PromptHook` lifecycle methods.
- **`DispatchObserveHook`** (feature `compose`) — implements
`rig_compose::ToolDispatchHook` and emits `tool.invoked` /
`tool.completed` / `tool.skipped` / `tool.terminated` from the kernel-direct
dispatch path. It also implements `rig_compose::AgentLifecycleHook` and emits
`compose.*` events around `GenericAgent` step and skill execution.
- **`ObservedMemory<M>`** — decorator that wraps any `rig::memory::ConversationMemory`
and emits `context.sampled` on every `load`.
- **`EventQuery` + `EventFilter`** — in-process query helpers for captured
`ObservabilityEvent` snapshots. Useful for tests, demos, and small local
dashboards without adding a service runtime.
- **`ChainedHook<A, B>`** — compose two `PromptHook`s on a single agent (e.g.
pair `MemvidPersistHook` with `TelemetryHook`).
## Quick start
Wire a `tracing` subscriber that keeps the dedicated `rig_tap` target, then
attach the hooks at the lifecycle boundary you want to observe:
```rust,no_run
use rig_tap::{ObservedMemory, TelemetryHook};
use tracing_subscriber::{EnvFilter, prelude::*};
fn install_observe_sink() {
tracing_subscriber::registry()
.with(EnvFilter::new("rig_tap=info"))
.with(tracing_subscriber::fmt::layer().json())
.init();
}
# fn build<M: rig::completion::CompletionModel>() -> TelemetryHook<M> {
let hook = TelemetryHook::<M>::with_defaults("qwen3.5:9b", "thread-1");
let memory = ObservedMemory::new(rig::memory::InMemoryConversationMemory::new());
// Attach `hook` to a Rig agent and use `memory` anywhere a
// `ConversationMemory` implementation is accepted.
# let _ = memory;
# hook }
```
For kernel-direct tool dispatch, enable the `compose` feature and register
`DispatchObserveHook` with `dispatch_tool_invocations_with_hooks`. The same hook
can be passed to `GenericAgentBuilder::with_lifecycle_hook` to observe the
agent step and skill loop. For deterministic tests or examples, enable
`subscriber` and use `CapturingLayer` to collect typed `ObservabilityEvent`
values in-process, then call `capture.query().filter(&EventFilter::new().kind("tool.completed"))`
to inspect a bounded snapshot.
## Architecture
`rig-tap` acts as a tap, listening to various hooks in the Rig lifecycle and
writing uniform telemetry into the `tracing` ecosystem under a dedicated
`rig_tap` target. Producer crates do not need to agree on storage backends,
model providers, or orchestration strategy; they only need to emit the shared
event vocabulary.
```text
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ │ │ │ │ │
│ Host Agent │──────►│ TelemetryHook ├──────►│ │
│ (rig::pipeline) │ │ │ │ │
└─────────────────┘ └─────────────────┘ │ │
│ │
┌─────────────────┐ ┌─────────────────┐ │ tracing::info! │
│ │ │ │ │ (target: │
│ Host Runtime │──────►│DispatchObserve..├──────►│ "rig_tap") │
│ (rig_compose) │ │ │ │ │
└─────────────────┘ └─────────────────┘ │ │
│ │
┌─────────────────┐ ┌─────────────────┐ │ │
│ │ │ │ │ │
│ ConversationMem ├──────►│ ObservedMemory ├──────►│ │
│ (rig::memory) │ │ │ └────────┬────────┘
└─────────────────┘ └─────────────────┘ │
▼
┌─────────────────┐
│ │
│ Telemetry Sink │
│ (OTEL/Langfuse/ │
│ Phoenix/etc.) │
└─────────────────┘
```
## Uniform Wire Format
All events are flat JSON serialized via `tracing::info!(target: "rig_tap", event = %json, ...)`.
The `event` field carries the complete envelope, and scalar `rig_tap.*`
attributes expose the fields collectors most often need for routing:
```json
{
"version": 1,
"occurred_at_millis": 1715000000000,
"tick": 42,
"conversation_id": "thread-1",
"kind": "context.compacted",
"evicted_count": 8,
"evicted_bytes": 4096,
"carry_over": true,
"summary_bytes": 512
}
```
`tick` is a monotonic per-process counter so consumers can order events
without clock skew. `version` is the schema version (currently `1`).
When exported through `tracing-opentelemetry`, an OpenTelemetry collector can
filter on `rig_tap.kind = "tool.skipped"`, group by
`rig_tap.conversation_id`, or route all `rig_tap` target events without parsing
the JSON body.
## Event kinds (v1)
| `prompt.started` | `TelemetryHook::on_completion_call` |
| `prompt.completed` | `TelemetryHook::on_completion_response` |
| `tool.invoked` | `TelemetryHook::on_tool_call` / `DispatchObserveHook` |
| `tool.completed` | `TelemetryHook::on_tool_result` / `DispatchObserveHook` |
| `tool.skipped` | Producer crate (kernel hook with `Skip` semantics) |
| `tool.terminated` | `DispatchObserveHook` (kernel gate / runtime error) |
| `tool.hosted_invoked` | Producer crate (Responses streaming/WebSocket tap), `responses_extract::emit_hosted_tools`, or `ObservedResponsesSession` |
| `tool.hosted_completed` | Producer crate (Responses streaming/WebSocket tap), `responses_extract::emit_hosted_tools`, or `ObservedResponsesSession` |
| `context.sampled` | `ObservedMemory::load` |
| `context.compacted` | Producer crate (e.g. `rig-memvid`) |
| `memory.demoted` | Producer crate |
| `memory.frame_written` | Producer crate |
| `compose.kernel_start` | Producer crate (`rig-compose` kernel lifecycle) |
| `compose.kernel_shutdown` | Producer crate (`rig-compose` kernel lifecycle) |
| `compose.loop_iteration` | Producer crate (`rig-compose` agent loop) |
| `compose.skill_resolved` | Producer crate (`rig-compose` skill resolution) |
| `compose.retry_attempt` | Producer crate (`rig-compose` retry path) |
| `compose.recovery` | Producer crate (`rig-compose` recovery path) |
| `response.session_started` | `ObservedResponsesSession` (`openai-responses-websocket`) |
| `response.turn_started` | `ObservedResponsesSession` (`openai-responses-websocket`) |
| `response.turn_completed` | `ObservedResponsesSession` (`openai-responses-websocket`) |
| `response.session_ended` | `ObservedResponsesSession` (`openai-responses-websocket`) |
| `eval.report` | Producer crate (`rig-retrieval-evals` `MultiReport` / `ReportDiff`) |
`prompt.*`, `tool.*` (via `TelemetryHook`/`DispatchObserveHook`), and
`context.sampled` are emitted by this crate. The remaining `tool.*` and
`memory.*` / `context.compacted` / `compose.*` events are emitted by producer
crates (e.g. `rig-memvid` and `rig-compose`) using the same schema — construct
an `EventKind` variant and pass it through
`ObservabilityEvent::new(conversation_id, kind)` or the `emit_kind` helper.
## Consumer example
```rust,no_run
use tracing_subscriber::{EnvFilter, prelude::*};
fn main() {
tracing_subscriber::registry()
.with(EnvFilter::new("rig_tap=info"))
.with(tracing_subscriber::fmt::layer().json())
.init();
// ... build agent with `TelemetryHook` and `ObservedMemory<...>` ...
}
```
A consumer wanting typed events can attach a custom `tracing_subscriber::Layer`
that parses the `event` field via `serde_json::from_str::<ObservabilityEvent>`.
For in-process tests, demos, or local dashboards, the optional `subscriber`
feature exposes `CapturingLayer::query()` and the default-build
`EventQuery`/`EventFilter` helpers for filtering by conversation, kind, tick
range, and scalar correlators such as tool name, call ID, skill ID, kernel ID,
or model.
## OpenAI Responses WebSocket sessions
The `openai-responses-websocket` feature (forwards `rig/websocket`, non-WASM
only) wires `rig-core`'s
`rig::providers::openai::responses_api::websocket::ResponsesWebSocketSession`
into the schema:
```rust,no_run
use rig::client::CompletionClient;
use rig::providers::openai;
use rig_tap::ObservedResponsesSession;
# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let client = openai::Client::new("YOUR_API_KEY")?;
let session = client.responses_websocket(openai::GPT_5_2).await?;
let mut observed = ObservedResponsesSession::new(
session,
"conversation-1", // conversation_id stamped on every envelope
openai::GPT_5_2, // model id, recorded once on session_started
"ws-session-abc", // stable session correlator
);
// observed.send(request).await?;
// while let Ok(event) = observed.next_event().await { /* ... */ }
// observed.close().await?;
# Ok(()) }
```
The decorator emits, in order, one `response.session_started`, an
alternating `response.turn_started` / `response.turn_completed` pair per
turn, paired `tool.hosted_invoked` / `tool.hosted_completed` events for
every hosted tool call extracted from the raw `Done.response` payload
(`web_search`, `file_search`, `computer_use`, `code_interpreter`, and any
future `*_call` kind), and exactly one `response.session_ended` on close.
Turn finalization is lazy: when callers stop reading after the terminal
`Response` chunk (as `ResponsesWebSocketSession::completion` does
upstream) the open turn is closed on the next `send`, on `close`, or on
`into_inner` — so the envelope stays well-formed even when the caller
short-circuits before `Done`. For raw HTTP / streaming hosted-tool
extraction without the WebSocket session, see
`responses_extract::{extract_hosted_tools, emit_hosted_tools}` under the
lighter `openai-responses` feature.
## Sampling controls
`TelemetryHook` accepts a [`SamplingPolicy`] so high-volume `tool.*`
traffic can be downsampled without losing low-volume lifecycle events
such as `prompt.*` or `memory.*`. The default policy is `AlwaysSample`
(keep everything); the bundled `RatePolicy` is a deterministic per-kind
rate sampler:
```rust,no_run
use std::sync::Arc;
use rig_tap::{RatePolicy, TelemetryHook, TelemetryHookConfig};
# fn make_hook<M: rig::completion::CompletionModel>() -> TelemetryHook<M> {
TelemetryHook::new(TelemetryHookConfig::new("gpt-4o", "thread-1"))
.with_sampling_policy(Arc::new(
RatePolicy::new()
.with_rate("tool.invoked", 0.1)
.with_rate("tool.completed", 0.1),
// `prompt.*`, `memory.*`, `compose.*` keep their default
// rate of 1.0 and are emitted unchanged.
))
# }
```
Sampling decisions are deterministic: the policy hashes a per-event
correlator with a fixed seed. The hook passes the resolved conversation
id on `prompt.*` events and the internal call id on `tool.*` events, so
a `tool.invoked` and its matching `tool.completed` either both ship or
are both dropped — pairs stay coherent.
Custom policies (allowlists, error-only, tail-based) can implement
`SamplingPolicy::should_sample(kind, correlator)` and be plugged in via
`with_sampling_policy(Arc::new(...))`.
## Coexistence with `rig-core::telemetry`
This crate is additive to Rig's existing GenAI span conventions
(`gen_ai.input.messages`, `gen_ai.usage.input_tokens`, etc.). Consumers using
`tracing-opentelemetry` for Phoenix / Langfuse keep their existing setup;
`rig_tap` events live under a separate target and can be filtered
independently. OpenTelemetry collectors receive the full JSON envelope in the
`event` attribute plus stable scalar attributes (`rig_tap.kind`,
`rig_tap.conversation_id`, `rig_tap.version`, `rig_tap.tick`, and
`rig_tap.occurred_at_millis`) for routing, filtering, and indexing without a
collector-side JSON transform.
## OpenTelemetry exporter recipe
`rig-tap` events are emitted as `tracing::info!` records on the `rig_tap`
target with structured fields whose names are already valid OpenTelemetry
attribute keys — **no rename or JSON transform is required to ship them
through an OTel pipeline**. Wire `tracing-opentelemetry` into your existing
subscriber stack and the attributes flow through verbatim.
### Stable attribute keys
Every event carries the envelope scalars:
| `rig_tap.version` | `ObservabilityEvent::version` | schema version (currently `1`) |
| `rig_tap.kind` | `ObservabilityEvent::kind.discriminant()` | e.g. `prompt.completed`, `eval.report` |
| `rig_tap.conversation_id` | envelope | join key for multi-event traces |
| `rig_tap.tick` | envelope | monotonic per-process sequence |
| `rig_tap.occurred_at_millis` | envelope | UNIX epoch milliseconds |
| `rig_tap.span_id` | envelope | numeric id of the `tracing::Span` that was current when the event was emitted (`0` = absent); also serialized into the JSON envelope as `span_id` when present |
Plus per-variant correlators, populated when present and emitted as empty
strings otherwise (filter `field != ""` collector-side):
| `rig_tap.kernel_id` | `compose.*` |
| `rig_tap.tool_name` | `tool.*` |
| `rig_tap.call_id` | `tool.*` |
| `rig_tap.skill_id` | `compose.skill_resolved` / `compose.loop_iteration` |
| `rig_tap.model` | `prompt.*`, `response.session_started` |
| `rig_tap.response_id` | `prompt.completed`, `response.turn_*` |
| `rig_tap.previous_response_id` | stateful Responses-API turns |
| `rig_tap.dataset` | `eval.report` |
| `rig_tap.metric` | `eval.report` |
| `rig_tap.verdict` | `eval.report` |
The full JSON envelope (including non-scalar fields like `args_json`,
`tokens_in`, `ci_low`/`ci_high`, etc.) ships as the `event` attribute.
### Minimum-viable collector config
If you forward `tracing` records to an out-of-process OpenTelemetry
Collector, the only processor you need is `filter` (to scope to the
`rig_tap` target) plus optional `attributes` (to rename keys to your
backend's preferred taxonomy). Example fragment:
```yaml
processors:
filter/rig_tap:
logs:
include:
match_type: strict
record_attributes:
- key: target
value: rig_tap
attributes/rig_tap:
actions:
# Optional: align with OTel GenAI semconv where it overlaps.
- key: gen_ai.response.model
from_attribute: rig_tap.model
action: insert
- key: gen_ai.response.id
from_attribute: rig_tap.response_id
action: insert
```
### Runnable preview
The [`otel_exporter_recipe`](examples/otel_exporter_recipe.rs) example
emits one event per major family and prints the exact attribute set an
OTel pipeline would receive:
```bash
cargo run --example otel_exporter_recipe --features subscriber
```
### In-process tracing-opentelemetry
For in-process exporters, drop the OTel layer into the same subscriber
stack as any other `tracing-opentelemetry` user:
```rust,no_run
# #[cfg(false)]
# {
use tracing_subscriber::{EnvFilter, prelude::*};
let tracer = /* your `opentelemetry_otlp::new_pipeline()...install_simple()?` */;
let otel = tracing_opentelemetry::layer().with_tracer(tracer);
tracing_subscriber::registry()
.with(EnvFilter::new("rig_tap=info,rig=info"))
.with(otel)
.init();
# }
```
No `rig-tap`-specific configuration is required — the `rig_tap.*` fields
are propagated as span attributes automatically.
## Status
Crate version: `0.2.1`. Rust edition: 2024. MSRV: 1.89. The library is
runtime-agnostic and emits through `tracing`; production consumers should use a
non-blocking tracing sink when exporting events off-host. The optional
`subscriber` feature is for tests/examples, while the optional `compose`
feature adds the `rig-compose` dispatch tap.
## License
Licensed under either of [Apache-2.0](LICENSE-APACHE) or [MIT](LICENSE-MIT)
at your option.