leviath_runtime/telemetry.rs
1//! Observability as an ECS system: watch the components every other system
2//! already writes and narrate them into the installed [`TelemetrySink`].
3//!
4//! The pipeline's collect systems leave pure-data [`ActivityRecord`]s on the
5//! agent (an inference landed, a tool batch ran, a compaction finished);
6//! [`observe_lifecycle`] runs once per schedule pass near the end of the tick,
7//! turns those plus the agent's own state into [`TelemetryEvent`]s, and emits
8//! them. Ordering in the tick chain is load-bearing twice over: the system
9//! must run *before* `sync_tool_stages` (which consumes the transient
10//! `StageJustEntered` marker) and *before* `dispatch_persistence` (which
11//! drains `StageIoBuffer` - this system only reads the buffer, so running
12//! first is what makes each log line observed exactly once).
13
14use std::sync::Arc;
15
16use bevy_ecs::prelude::*;
17use leviath_core::telemetry::{LogKind, TelemetryEvent, TelemetrySink};
18
19use crate::components::{AgentState, AgentStatus};
20use crate::persistence::{RunMetadata, TokenTotals};
21use crate::pipeline::{StageCursor, StageIoBuffer, StageJustEntered, StageLedger};
22
23/// The installed telemetry sink. [`crate::world::PipelineWorld::new`] installs
24/// [`leviath_core::telemetry::NoopSink`]; a host that wants export replaces
25/// the resource (the same way it installs `PolicyGate` or `TitleSettings`).
26#[derive(Resource, Clone)]
27pub struct Telemetry(pub Arc<dyn TelemetrySink>);
28
29/// One completed piece of stage work, recorded by the collect system that
30/// applied it and drained into events by [`observe_lifecycle`]. Carries only
31/// what the collect site knows; run/stage identity is added at drain time.
32#[derive(Debug, Clone, PartialEq)]
33pub enum ActivityRecord {
34 /// An inference call finished (either way).
35 Inference {
36 provider: String,
37 model: String,
38 latency_ms: u64,
39 prompt_tokens: usize,
40 completion_tokens: usize,
41 cached_tokens: usize,
42 success: bool,
43 },
44 /// One tool call out of a finished batch.
45 ToolCall {
46 tool_name: String,
47 batch_latency_ms: u64,
48 success: bool,
49 },
50 /// A compaction pass finished.
51 Compaction { success: bool },
52}
53
54/// Buffered [`ActivityRecord`]s awaiting the observer. Inserted alongside
55/// [`TelemetryState`] the first time the observer sees an agent, so the
56/// collect systems treat it as optional and skip recording until then (an
57/// agent's first inference cannot land before the observer has run once).
58#[derive(Component, Debug, Default)]
59pub struct StageActivity(pub Vec<ActivityRecord>);
60
61/// The observer's per-agent memory: what it has already narrated.
62#[derive(Component, Debug, Clone, Default)]
63pub struct TelemetryState {
64 /// A `RunStarted` was emitted and no `RunCompleted` yet.
65 run_open: bool,
66 /// The stage the observer last reported as entered.
67 last_stage: Option<(usize, String)>,
68}
69
70/// The terminal status label for [`TelemetryEvent::RunCompleted`], or `None`
71/// while the run is still going.
72fn terminal_label(status: &AgentStatus) -> Option<&'static str> {
73 match status {
74 AgentStatus::Complete => Some("complete"),
75 AgentStatus::Error { .. } => Some("error"),
76 AgentStatus::Cancelled => Some("cancelled"),
77 AgentStatus::Idle | AgentStatus::Active | AgentStatus::Waiting => None,
78 }
79}
80
81/// The (prompt, completion) token totals a stage accrued, from its ledger
82/// record; zeros when the ledger has no record for it.
83fn stage_tokens(ledger: Option<&StageLedger>, index: usize) -> (usize, usize) {
84 ledger
85 .and_then(|l| l.0.get(index))
86 .map_or((0, 0), |rec| (rec.prompt_tokens, rec.completion_tokens))
87}
88
89/// Emit lifecycle, activity, and log events for every agent run.
90///
91/// Stage boundaries come from the `StageJustEntered` marker (with the
92/// agent's first sighting standing in for the marker-less initial stage);
93/// a re-entry into the same stage index keeps the stage open rather than
94/// closing and reopening it, matching how the stage ledger accrues.
95#[allow(clippy::type_complexity)]
96pub fn observe_lifecycle(
97 telemetry: Res<Telemetry>,
98 mut agents: Query<(
99 Entity,
100 &RunMetadata,
101 &AgentState,
102 Option<&StageCursor>,
103 Option<&TokenTotals>,
104 Option<&StageLedger>,
105 Option<&StageJustEntered>,
106 Option<&mut TelemetryState>,
107 Option<&mut StageActivity>,
108 Option<&StageIoBuffer>,
109 )>,
110 mut commands: Commands,
111) {
112 crate::tick_scope::clear();
113 for (entity, md, state, cursor, totals, ledger, entered, ts, activity, buffer) in
114 agents.iter_mut()
115 {
116 crate::tick_scope::enter(entity);
117 let now_ms = chrono::Utc::now().timestamp_millis();
118 let sink = telemetry.0.as_ref();
119 let mut ts = ts;
120 let (mut st, is_new) = match ts.as_deref() {
121 Some(existing) => (existing.clone(), false),
122 None => (TelemetryState::default(), true),
123 };
124
125 if is_new {
126 // First sighting. A run restored from disk is already mid-flight:
127 // its earlier spans (if any) belong to a previous daemon process,
128 // so the trace it gets here starts now and says so.
129 let recovered = state.iteration > 0 || cursor.is_some_and(|c| c.index > 0);
130 sink.emit(TelemetryEvent::RunStarted {
131 run_id: md.run_id.clone(),
132 agent_name: md.agent_name.clone(),
133 model: md.model.clone(),
134 parent_run_id: md.parent_run_id.clone(),
135 recovered,
136 at_ms: now_ms,
137 });
138 st.run_open = true;
139 }
140
141 if st.run_open {
142 // Stage boundary: the transition marker, or - for the marker-less
143 // first sighting - the agent's current stage.
144 let boundary = match entered {
145 Some(marker) => Some((marker.index, marker.name.clone())),
146 None if st.last_stage.is_none() => {
147 Some((cursor.map_or(0, |c| c.index), state.current_stage.clone()))
148 }
149 None => None,
150 };
151 if let Some((index, name)) = boundary {
152 let same_stage = st.last_stage.as_ref().is_some_and(|(i, _)| *i == index);
153 if !same_stage {
154 if let Some((prev_index, prev_name)) = st.last_stage.take() {
155 let (prompt, completion) = stage_tokens(ledger, prev_index);
156 sink.emit(TelemetryEvent::StageExited {
157 run_id: md.run_id.clone(),
158 stage_index: prev_index,
159 stage_name: prev_name,
160 prompt_tokens: prompt,
161 completion_tokens: completion,
162 at_ms: now_ms,
163 });
164 }
165 sink.emit(TelemetryEvent::StageEntered {
166 run_id: md.run_id.clone(),
167 stage_index: index,
168 stage_name: name.clone(),
169 at_ms: now_ms,
170 });
171 st.last_stage = Some((index, name));
172 }
173 }
174
175 // Completed work the collect systems recorded since the last pass.
176 if let Some(mut activity) = activity {
177 // An open run always has an entered stage: the first sighting
178 // above set one before this point.
179 let (_, ref stage_name) = *st.last_stage.as_ref().expect("stage set at sighting");
180 let stage_name = stage_name.clone();
181 for record in activity.0.drain(..) {
182 sink.emit(match record {
183 ActivityRecord::Inference {
184 provider,
185 model,
186 latency_ms,
187 prompt_tokens,
188 completion_tokens,
189 cached_tokens,
190 success,
191 } => TelemetryEvent::InferenceCompleted {
192 run_id: md.run_id.clone(),
193 stage_name: stage_name.clone(),
194 provider,
195 model,
196 latency_ms,
197 prompt_tokens,
198 completion_tokens,
199 cached_tokens,
200 success,
201 },
202 ActivityRecord::ToolCall {
203 tool_name,
204 batch_latency_ms,
205 success,
206 } => TelemetryEvent::ToolCallCompleted {
207 run_id: md.run_id.clone(),
208 stage_name: stage_name.clone(),
209 tool_name,
210 batch_latency_ms,
211 success,
212 },
213 ActivityRecord::Compaction { success } => {
214 TelemetryEvent::CompactionCompleted {
215 run_id: md.run_id.clone(),
216 stage_name: stage_name.clone(),
217 success,
218 }
219 }
220 });
221 }
222 }
223
224 // Log lines: read, never drain - `dispatch_persistence` (which
225 // runs after this system in the same pass) owns the drain, so
226 // each line passes through here exactly once.
227 if let Some(buffer) = buffer {
228 for ((idx, line), kind) in buffer
229 .output
230 .iter()
231 .map(|l| (l, LogKind::Output))
232 .chain(buffer.logs.iter().map(|l| (l, LogKind::Runtime)))
233 {
234 sink.emit(TelemetryEvent::Log {
235 run_id: md.run_id.clone(),
236 stage_index: *idx,
237 kind,
238 line: line.clone(),
239 });
240 }
241 }
242
243 if let Some(status) = terminal_label(&state.status) {
244 // Same invariant as the drain above: an open run always has an
245 // entered stage to close.
246 let (prev_index, prev_name) = st.last_stage.take().expect("stage set at sighting");
247 let (prompt, completion) = stage_tokens(ledger, prev_index);
248 sink.emit(TelemetryEvent::StageExited {
249 run_id: md.run_id.clone(),
250 stage_index: prev_index,
251 stage_name: prev_name,
252 prompt_tokens: prompt,
253 completion_tokens: completion,
254 at_ms: now_ms,
255 });
256 let totals = totals.copied().unwrap_or_default();
257 sink.emit(TelemetryEvent::RunCompleted {
258 run_id: md.run_id.clone(),
259 status: status.to_string(),
260 prompt_tokens: totals.prompt_tokens,
261 completion_tokens: totals.completion_tokens,
262 tool_calls: totals.tool_calls,
263 at_ms: now_ms,
264 });
265 st.run_open = false;
266 }
267 }
268
269 if is_new {
270 commands
271 .entity(entity)
272 .insert((st, StageActivity::default()));
273 } else {
274 *ts.as_deref_mut().expect("state exists when not new") = st;
275 }
276 }
277}
278
279#[cfg(test)]
280mod tests;