rig_tap/
emit.rs

1//! Tracing transport for [`ObservabilityEvent`].
2//!
3//! All events are emitted as a single `tracing::info!` call under the
4//! `rig_tap` target. The legacy `event` field carries the JSON-encoded
5//! envelope, while stable scalar `rig_tap.*` fields make OpenTelemetry
6//! collector routing and indexing possible without JSON parsing.
7
8use std::sync::atomic::{AtomicU64, Ordering};
9use std::time::{SystemTime, UNIX_EPOCH};
10
11use crate::error::Error;
12use crate::event::{EventKind, ObservabilityEvent, SCHEMA_VERSION};
13
14/// Target string used on every `rig_tap` event.
15pub const EVENT_TARGET: &str = "rig_tap";
16
17static TICK: AtomicU64 = AtomicU64::new(0);
18
19/// Return the next monotonic per-process tick.
20pub fn next_tick() -> u64 {
21    TICK.fetch_add(1, Ordering::Relaxed)
22}
23
24/// Return the current wall-clock time in milliseconds since the Unix epoch.
25/// Returns `0` if the clock is set before the epoch.
26pub fn now_millis() -> u64 {
27    SystemTime::now()
28        .duration_since(UNIX_EPOCH)
29        .map(|d| u64::try_from(d.as_millis()).unwrap_or(u64::MAX))
30        .unwrap_or(0)
31}
32
33/// Return the numeric id of the currently-active `tracing::Span`, if any.
34///
35/// Mirrors [`tracing::span::Id::into_u64`]. Subscribers that respect
36/// span context (e.g. `tracing-opentelemetry`) attach this id to every
37/// event automatically; surfacing it on the envelope lets consumers
38/// reading only structured fields stitch events into the same
39/// waterfall.
40pub fn current_span_id() -> Option<u64> {
41    tracing::Span::current().id().map(|id| id.into_u64())
42}
43
44/// Build a fully-formed [`ObservabilityEvent`] for `kind` belonging to
45/// `conversation_id`, stamped with the next tick and current wall time.
46pub fn build_event(conversation_id: impl Into<String>, kind: EventKind) -> ObservabilityEvent {
47    ObservabilityEvent {
48        version: SCHEMA_VERSION,
49        occurred_at_millis: now_millis(),
50        tick: next_tick(),
51        conversation_id: conversation_id.into(),
52        span_id: current_span_id(),
53        kind,
54    }
55}
56
57/// Emit `event` over the `rig_tap` tracing target as a single
58/// `info!`-level event carrying a JSON-encoded `event` field.
59///
60/// Returns an [`Error`] if the event fails to serialize. Callers in library
61/// code typically discard the result via [`emit`] which logs serialization
62/// failures rather than propagating them.
63pub fn try_emit(event: &ObservabilityEvent) -> Result<(), Error> {
64    let json = serde_json::to_string(event)?;
65    let fields = event.kind.scalar_fields();
66    tracing::info!(
67        target: EVENT_TARGET,
68        event = %json,
69        rig_tap.version = event.version,
70        rig_tap.kind = event.kind.discriminant(),
71        rig_tap.conversation_id = %event.conversation_id,
72        rig_tap.tick = event.tick,
73        rig_tap.occurred_at_millis = event.occurred_at_millis,
74        // Numeric `tracing::Span` id captured at emit time. `0` =
75        // absent (no span was active). Consumers correlating via
76        // `tracing-opentelemetry` already get the span via subscriber
77        // context; this field is for collectors that read only the
78        // structured `rig_tap.*` attributes.
79        rig_tap.span_id = event.span_id.unwrap_or(0),
80        // Per-variant scalar correlators. Absent values are emitted as
81        // empty strings (see `ScalarFields` rustdoc) — collectors should
82        // filter `rig_tap.<field> != ""` to detect presence.
83        rig_tap.kernel_id = fields.kernel_id,
84        rig_tap.tool_name = fields.tool_name,
85        rig_tap.call_id = fields.call_id,
86        rig_tap.skill_id = fields.skill_id,
87        rig_tap.model = fields.model,
88        rig_tap.response_id = fields.response_id,
89        rig_tap.previous_response_id = fields.previous_response_id,
90        rig_tap.dataset = fields.dataset,
91        rig_tap.metric = fields.metric,
92        rig_tap.verdict = fields.verdict,
93        rig_tap.error_class = fields.error_class,
94    );
95    Ok(())
96}
97
98/// Emit `event` over the `rig_tap` tracing target. Serialization failures
99/// are logged at `warn` level under the same target and otherwise swallowed
100/// so that telemetry never panics the agent loop.
101pub fn emit(event: &ObservabilityEvent) {
102    if let Err(err) = try_emit(event) {
103        tracing::warn!(
104            target: EVENT_TARGET,
105            error = %err,
106            error_kind = err.kind(),
107            "rig-tap: failed to emit event",
108        );
109    }
110}
111
112/// Convenience: build + emit in one call.
113pub fn emit_kind(conversation_id: impl Into<String>, kind: EventKind) {
114    let event = build_event(conversation_id, kind);
115    emit(&event);
116}
117
118#[cfg(test)]
119#[allow(
120    clippy::unwrap_used,
121    clippy::panic,
122    clippy::indexing_slicing,
123    clippy::expect_used
124)]
125mod tests {
126    use super::*;
127
128    #[test]
129    fn tick_is_monotonic() {
130        let a = next_tick();
131        let b = next_tick();
132        assert!(b > a);
133    }
134
135    #[test]
136    fn build_event_stamps_envelope() {
137        let evt = build_event(
138            "c",
139            EventKind::PromptStarted {
140                model: "m".into(),
141                messages_in: 0,
142            },
143        );
144        assert_eq!(evt.version, SCHEMA_VERSION);
145        assert_eq!(evt.conversation_id, "c");
146        // occurred_at_millis is 0 only if the system clock is broken; in tests
147        // it should always be a positive value.
148        assert!(evt.occurred_at_millis > 0);
149    }
150}
rig_tap/emit.rs

rig_tap/
emit.rs
