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 /// The provider that answered, after fallback resolution.
37 provider: String,
38 /// The model identifier sent on the wire.
39 model: String,
40 /// Wall-clock time of the call, including retries.
41 latency_ms: u64,
42 /// Input tokens this call billed.
43 prompt_tokens: usize,
44 /// Output tokens this call billed.
45 completion_tokens: usize,
46 /// Input tokens served from the provider's prompt cache, counted within
47 /// `prompt_tokens` rather than on top of it.
48 cached_tokens: usize,
49 /// Whether a response came back at all.
50 success: bool,
51 },
52 /// One tool call out of a finished batch.
53 ToolCall {
54 /// The tool as the model named it.
55 tool_name: String,
56 /// Wall-clock time of the whole batch. The executor reports one figure
57 /// per batch, so every call in it carries the same number.
58 batch_latency_ms: u64,
59 /// Derived from the `[error] ` result-text convention, so a heuristic
60 /// rather than a structured status.
61 success: bool,
62 },
63 /// A compaction pass finished.
64 Compaction {
65 /// Whether it produced a usable summary. A failure leaves the region
66 /// alone rather than emptying it.
67 success: bool,
68 },
69}
70
71/// Buffered [`ActivityRecord`]s awaiting the observer. Inserted alongside
72/// [`TelemetryState`] the first time the observer sees an agent, so the
73/// collect systems treat it as optional and skip recording until then (an
74/// agent's first inference cannot land before the observer has run once).
75#[derive(Component, Debug, Default)]
76pub struct StageActivity(pub Vec<ActivityRecord>);
77
78/// The observer's per-agent memory: what it has already narrated.
79#[derive(Component, Debug, Clone, Default)]
80pub struct TelemetryState {
81 /// A `RunStarted` was emitted and no `RunCompleted` yet.
82 run_open: bool,
83 /// The stage the observer last reported as entered.
84 last_stage: Option<(usize, String)>,
85}
86
87/// The terminal status label for [`TelemetryEvent::RunCompleted`], or `None`
88/// while the run is still going.
89fn terminal_label(status: &AgentStatus) -> Option<&'static str> {
90 match status {
91 AgentStatus::Complete => Some("complete"),
92 AgentStatus::Error { .. } => Some("error"),
93 AgentStatus::Cancelled => Some("cancelled"),
94 AgentStatus::Idle | AgentStatus::Active | AgentStatus::Waiting | AgentStatus::Paused => {
95 None
96 }
97 }
98}
99
100/// The (prompt, completion) token totals a stage accrued, from its ledger
101/// record; zeros when the ledger has no record for it.
102fn stage_tokens(ledger: Option<&StageLedger>, index: usize) -> (usize, usize) {
103 ledger
104 .and_then(|l| l.0.get(index))
105 .map_or((0, 0), |rec| (rec.prompt_tokens, rec.completion_tokens))
106}
107
108/// What `observe_lifecycle` selects.
109///
110/// `&'static` is bevy's `WorldQuery` convention, not a claim about
111/// lifetimes: the borrow is bound when the query is fetched.
112type LifecycleQuery = (
113 Entity,
114 &'static RunMetadata,
115 &'static AgentState,
116 Option<&'static StageCursor>,
117 Option<&'static TokenTotals>,
118 Option<&'static StageLedger>,
119 Option<&'static StageJustEntered>,
120 Option<&'static mut TelemetryState>,
121 Option<&'static mut StageActivity>,
122 Option<&'static StageIoBuffer>,
123 Option<&'static crate::persistence::RunOutcomeFlags>,
124);
125
126/// Emit lifecycle, activity, and log events for every agent run.
127///
128/// Stage boundaries come from the `StageJustEntered` marker (with the
129/// agent's first sighting standing in for the marker-less initial stage);
130/// a re-entry into the same stage index keeps the stage open rather than
131/// closing and reopening it, matching how the stage ledger accrues.
132pub fn observe_lifecycle(
133 telemetry: Res<Telemetry>,
134 mut agents: Query<LifecycleQuery>,
135 mut commands: Commands,
136) {
137 crate::tick_scope::clear();
138 for (entity, md, state, cursor, totals, ledger, entered, ts, activity, buffer, flags) in
139 agents.iter_mut()
140 {
141 crate::tick_scope::enter(entity);
142 let now_ms = chrono::Utc::now().timestamp_millis();
143 let sink = telemetry.0.as_ref();
144 let mut ts = ts;
145 let (mut st, is_new) = match ts.as_deref() {
146 Some(existing) => (existing.clone(), false),
147 None => (TelemetryState::default(), true),
148 };
149
150 if is_new {
151 // First sighting. A run restored from disk is already mid-flight:
152 // its earlier spans (if any) belong to a previous daemon process,
153 // so the trace it gets here starts now and says so.
154 let recovered = state.iteration > 0 || cursor.is_some_and(|c| c.index > 0);
155 sink.emit(TelemetryEvent::RunStarted {
156 run_id: md.run_id.clone(),
157 agent_name: md.agent_name.clone(),
158 model: md.model.clone(),
159 parent_run_id: md.parent_run_id.clone(),
160 recovered,
161 at_ms: now_ms,
162 });
163 st.run_open = true;
164 }
165
166 if st.run_open {
167 // Stage boundary: the transition marker, or - for the marker-less
168 // first sighting - the agent's current stage.
169 let boundary = match entered {
170 Some(marker) => Some((marker.index, marker.name.clone())),
171 None if st.last_stage.is_none() => {
172 Some((cursor.map_or(0, |c| c.index), state.current_stage.clone()))
173 }
174 None => None,
175 };
176 if let Some((index, name)) = boundary {
177 let same_stage = st.last_stage.as_ref().is_some_and(|(i, _)| *i == index);
178 if !same_stage {
179 if let Some((prev_index, prev_name)) = st.last_stage.take() {
180 let (prompt, completion) = stage_tokens(ledger, prev_index);
181 sink.emit(TelemetryEvent::StageExited {
182 run_id: md.run_id.clone(),
183 stage_index: prev_index,
184 stage_name: prev_name,
185 prompt_tokens: prompt,
186 completion_tokens: completion,
187 at_ms: now_ms,
188 });
189 }
190 sink.emit(TelemetryEvent::StageEntered {
191 run_id: md.run_id.clone(),
192 stage_index: index,
193 stage_name: name.clone(),
194 at_ms: now_ms,
195 });
196 st.last_stage = Some((index, name));
197 }
198 }
199
200 // Completed work the collect systems recorded since the last pass.
201 if let Some(mut activity) = activity {
202 // An open run always has an entered stage: the first sighting
203 // above set one before this point.
204 let (_, ref stage_name) = *st.last_stage.as_ref().expect("stage set at sighting");
205 let stage_name = stage_name.clone();
206 for record in activity.0.drain(..) {
207 sink.emit(match record {
208 ActivityRecord::Inference {
209 provider,
210 model,
211 latency_ms,
212 prompt_tokens,
213 completion_tokens,
214 cached_tokens,
215 success,
216 } => TelemetryEvent::InferenceCompleted {
217 run_id: md.run_id.clone(),
218 stage_name: stage_name.clone(),
219 provider,
220 model,
221 latency_ms,
222 prompt_tokens,
223 completion_tokens,
224 cached_tokens,
225 success,
226 },
227 ActivityRecord::ToolCall {
228 tool_name,
229 batch_latency_ms,
230 success,
231 } => TelemetryEvent::ToolCallCompleted {
232 run_id: md.run_id.clone(),
233 stage_name: stage_name.clone(),
234 tool_name,
235 batch_latency_ms,
236 success,
237 },
238 ActivityRecord::Compaction { success } => {
239 TelemetryEvent::CompactionCompleted {
240 run_id: md.run_id.clone(),
241 stage_name: stage_name.clone(),
242 success,
243 }
244 }
245 });
246 }
247 }
248
249 // Log lines: read, never drain - `dispatch_persistence` (which
250 // runs after this system in the same pass) owns the drain, so
251 // each line passes through here exactly once.
252 if let Some(buffer) = buffer {
253 for ((idx, line), kind) in buffer
254 .output
255 .iter()
256 .map(|l| (l, LogKind::Output))
257 .chain(buffer.logs.iter().map(|l| (l, LogKind::Runtime)))
258 {
259 sink.emit(TelemetryEvent::Log {
260 run_id: md.run_id.clone(),
261 stage_index: *idx,
262 kind,
263 line: line.clone(),
264 });
265 }
266 }
267
268 if let Some(status) = terminal_label(&state.status) {
269 // Same invariant as the drain above: an open run always has an
270 // entered stage to close.
271 let (prev_index, prev_name) = st.last_stage.take().expect("stage set at sighting");
272 let (prompt, completion) = stage_tokens(ledger, prev_index);
273 sink.emit(TelemetryEvent::StageExited {
274 run_id: md.run_id.clone(),
275 stage_index: prev_index,
276 stage_name: prev_name,
277 prompt_tokens: prompt,
278 completion_tokens: completion,
279 at_ms: now_ms,
280 });
281 let totals = totals.copied().unwrap_or_default();
282 sink.emit(TelemetryEvent::RunCompleted {
283 run_id: md.run_id.clone(),
284 status: status.to_string(),
285 prompt_tokens: totals.prompt_tokens,
286 completion_tokens: totals.completion_tokens,
287 tool_calls: totals.tool_calls,
288 empty_output: flags
289 .is_some_and(|f| crate::persistence::is_empty_output(&state.status, &f.0)),
290 at_ms: now_ms,
291 });
292 st.run_open = false;
293 }
294 }
295
296 if is_new {
297 commands
298 .entity(entity)
299 .insert((st, StageActivity::default()));
300 } else {
301 *ts.as_deref_mut().expect("state exists when not new") = st;
302 }
303 }
304}
305
306#[cfg(test)]
307mod tests;