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    );
94    Ok(())
95}
96
97/// Emit `event` over the `rig_tap` tracing target. Serialization failures
98/// are logged at `warn` level under the same target and otherwise swallowed
99/// so that telemetry never panics the agent loop.
100pub fn emit(event: &ObservabilityEvent) {
101    if let Err(err) = try_emit(event) {
102        tracing::warn!(
103            target: EVENT_TARGET,
104            error = %err,
105            error_kind = err.kind(),
106            "rig-tap: failed to emit event",
107        );
108    }
109}
110
111/// Convenience: build + emit in one call.
112pub fn emit_kind(conversation_id: impl Into<String>, kind: EventKind) {
113    let event = build_event(conversation_id, kind);
114    emit(&event);
115}
116
117#[cfg(test)]
118#[allow(
119    clippy::unwrap_used,
120    clippy::panic,
121    clippy::indexing_slicing,
122    clippy::expect_used
123)]
124mod tests {
125    use super::*;
126
127    #[test]
128    fn tick_is_monotonic() {
129        let a = next_tick();
130        let b = next_tick();
131        assert!(b > a);
132    }
133
134    #[test]
135    fn build_event_stamps_envelope() {
136        let evt = build_event(
137            "c",
138            EventKind::PromptStarted {
139                model: "m".into(),
140                messages_in: 0,
141            },
142        );
143        assert_eq!(evt.version, SCHEMA_VERSION);
144        assert_eq!(evt.conversation_id, "c");
145        // occurred_at_millis is 0 only if the system clock is broken; in tests
146        // it should always be a positive value.
147        assert!(evt.occurred_at_millis > 0);
148    }
149}
rig_tap/emit.rs

rig_tap/
emit.rs
