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 | AgentStatus::Paused => {
78 None
79 }
80 }
81}
82
83/// The (prompt, completion) token totals a stage accrued, from its ledger
84/// record; zeros when the ledger has no record for it.
85fn stage_tokens(ledger: Option<&StageLedger>, index: usize) -> (usize, usize) {
86 ledger
87 .and_then(|l| l.0.get(index))
88 .map_or((0, 0), |rec| (rec.prompt_tokens, rec.completion_tokens))
89}
90
91/// Emit lifecycle, activity, and log events for every agent run.
92///
93/// Stage boundaries come from the `StageJustEntered` marker (with the
94/// agent's first sighting standing in for the marker-less initial stage);
95/// a re-entry into the same stage index keeps the stage open rather than
96/// closing and reopening it, matching how the stage ledger accrues.
97#[allow(clippy::type_complexity)]
98pub fn observe_lifecycle(
99 telemetry: Res<Telemetry>,
100 mut agents: Query<(
101 Entity,
102 &RunMetadata,
103 &AgentState,
104 Option<&StageCursor>,
105 Option<&TokenTotals>,
106 Option<&StageLedger>,
107 Option<&StageJustEntered>,
108 Option<&mut TelemetryState>,
109 Option<&mut StageActivity>,
110 Option<&StageIoBuffer>,
111 Option<&crate::persistence::RunOutcomeFlags>,
112 )>,
113 mut commands: Commands,
114) {
115 crate::tick_scope::clear();
116 for (entity, md, state, cursor, totals, ledger, entered, ts, activity, buffer, flags) in
117 agents.iter_mut()
118 {
119 crate::tick_scope::enter(entity);
120 let now_ms = chrono::Utc::now().timestamp_millis();
121 let sink = telemetry.0.as_ref();
122 let mut ts = ts;
123 let (mut st, is_new) = match ts.as_deref() {
124 Some(existing) => (existing.clone(), false),
125 None => (TelemetryState::default(), true),
126 };
127
128 if is_new {
129 // First sighting. A run restored from disk is already mid-flight:
130 // its earlier spans (if any) belong to a previous daemon process,
131 // so the trace it gets here starts now and says so.
132 let recovered = state.iteration > 0 || cursor.is_some_and(|c| c.index > 0);
133 sink.emit(TelemetryEvent::RunStarted {
134 run_id: md.run_id.clone(),
135 agent_name: md.agent_name.clone(),
136 model: md.model.clone(),
137 parent_run_id: md.parent_run_id.clone(),
138 recovered,
139 at_ms: now_ms,
140 });
141 st.run_open = true;
142 }
143
144 if st.run_open {
145 // Stage boundary: the transition marker, or - for the marker-less
146 // first sighting - the agent's current stage.
147 let boundary = match entered {
148 Some(marker) => Some((marker.index, marker.name.clone())),
149 None if st.last_stage.is_none() => {
150 Some((cursor.map_or(0, |c| c.index), state.current_stage.clone()))
151 }
152 None => None,
153 };
154 if let Some((index, name)) = boundary {
155 let same_stage = st.last_stage.as_ref().is_some_and(|(i, _)| *i == index);
156 if !same_stage {
157 if let Some((prev_index, prev_name)) = st.last_stage.take() {
158 let (prompt, completion) = stage_tokens(ledger, prev_index);
159 sink.emit(TelemetryEvent::StageExited {
160 run_id: md.run_id.clone(),
161 stage_index: prev_index,
162 stage_name: prev_name,
163 prompt_tokens: prompt,
164 completion_tokens: completion,
165 at_ms: now_ms,
166 });
167 }
168 sink.emit(TelemetryEvent::StageEntered {
169 run_id: md.run_id.clone(),
170 stage_index: index,
171 stage_name: name.clone(),
172 at_ms: now_ms,
173 });
174 st.last_stage = Some((index, name));
175 }
176 }
177
178 // Completed work the collect systems recorded since the last pass.
179 if let Some(mut activity) = activity {
180 // An open run always has an entered stage: the first sighting
181 // above set one before this point.
182 let (_, ref stage_name) = *st.last_stage.as_ref().expect("stage set at sighting");
183 let stage_name = stage_name.clone();
184 for record in activity.0.drain(..) {
185 sink.emit(match record {
186 ActivityRecord::Inference {
187 provider,
188 model,
189 latency_ms,
190 prompt_tokens,
191 completion_tokens,
192 cached_tokens,
193 success,
194 } => TelemetryEvent::InferenceCompleted {
195 run_id: md.run_id.clone(),
196 stage_name: stage_name.clone(),
197 provider,
198 model,
199 latency_ms,
200 prompt_tokens,
201 completion_tokens,
202 cached_tokens,
203 success,
204 },
205 ActivityRecord::ToolCall {
206 tool_name,
207 batch_latency_ms,
208 success,
209 } => TelemetryEvent::ToolCallCompleted {
210 run_id: md.run_id.clone(),
211 stage_name: stage_name.clone(),
212 tool_name,
213 batch_latency_ms,
214 success,
215 },
216 ActivityRecord::Compaction { success } => {
217 TelemetryEvent::CompactionCompleted {
218 run_id: md.run_id.clone(),
219 stage_name: stage_name.clone(),
220 success,
221 }
222 }
223 });
224 }
225 }
226
227 // Log lines: read, never drain - `dispatch_persistence` (which
228 // runs after this system in the same pass) owns the drain, so
229 // each line passes through here exactly once.
230 if let Some(buffer) = buffer {
231 for ((idx, line), kind) in buffer
232 .output
233 .iter()
234 .map(|l| (l, LogKind::Output))
235 .chain(buffer.logs.iter().map(|l| (l, LogKind::Runtime)))
236 {
237 sink.emit(TelemetryEvent::Log {
238 run_id: md.run_id.clone(),
239 stage_index: *idx,
240 kind,
241 line: line.clone(),
242 });
243 }
244 }
245
246 if let Some(status) = terminal_label(&state.status) {
247 // Same invariant as the drain above: an open run always has an
248 // entered stage to close.
249 let (prev_index, prev_name) = st.last_stage.take().expect("stage set at sighting");
250 let (prompt, completion) = stage_tokens(ledger, prev_index);
251 sink.emit(TelemetryEvent::StageExited {
252 run_id: md.run_id.clone(),
253 stage_index: prev_index,
254 stage_name: prev_name,
255 prompt_tokens: prompt,
256 completion_tokens: completion,
257 at_ms: now_ms,
258 });
259 let totals = totals.copied().unwrap_or_default();
260 sink.emit(TelemetryEvent::RunCompleted {
261 run_id: md.run_id.clone(),
262 status: status.to_string(),
263 prompt_tokens: totals.prompt_tokens,
264 completion_tokens: totals.completion_tokens,
265 tool_calls: totals.tool_calls,
266 empty_output: flags
267 .is_some_and(|f| crate::persistence::is_empty_output(&state.status, &f.0)),
268 at_ms: now_ms,
269 });
270 st.run_open = false;
271 }
272 }
273
274 if is_new {
275 commands
276 .entity(entity)
277 .insert((st, StageActivity::default()));
278 } else {
279 *ts.as_deref_mut().expect("state exists when not new") = st;
280 }
281 }
282}
283
284#[cfg(test)]
285mod tests;