leviath_core/telemetry.rs
1//! The telemetry seam: pure-data lifecycle events and the sink they flow into.
2//!
3//! The runtime's observability system translates ECS state changes into
4//! [`TelemetryEvent`] values and hands them to whatever [`TelemetrySink`] the
5//! host installed. The events carry plain data only - no SDK types - so the
6//! runtime never depends on an exporter, and tests can assert on the exact
7//! event stream with [`MemorySink`]. The OpenTelemetry-backed sink lives in
8//! `leviath-telemetry`; a host that installs nothing gets [`NoopSink`].
9
10/// What kind of per-run log line a [`TelemetryEvent::Log`] carries.
11///
12/// Mirrors the two per-stage files the persistence layer writes: `output.log`
13/// (the model's own text) and `logs.log` (tool results, token counts, errors).
14#[derive(Debug, Clone, Copy, PartialEq, Eq)]
15pub enum LogKind {
16 /// A line of assistant output (`output.log`).
17 Output,
18 /// A runtime log line - tool results, token counts, errors (`logs.log`).
19 Runtime,
20}
21
22/// One observable moment in an agent run's life.
23///
24/// Timestamps are milliseconds since the Unix epoch (`at_ms`) so a sink can
25/// reconstruct span boundaries without sub-second drift; durations are
26/// measured wall-clock milliseconds at the point the work actually ran.
27#[derive(Debug, Clone, PartialEq)]
28pub enum TelemetryEvent {
29 /// An agent run became visible to the observer.
30 RunStarted {
31 /// The run this is about; the correlation key for every later event.
32 run_id: String,
33 /// The blueprint's `[agent] name`.
34 agent_name: String,
35 /// The run-level model hint from spawn metadata, if one was recorded.
36 model: Option<String>,
37 /// Present when this run is a sub-agent of another run.
38 parent_run_id: Option<String>,
39 /// True when the run was reloaded from disk rather than freshly
40 /// spawned - its earlier life was traced (if at all) by a previous
41 /// daemon process, so this trace starts mid-run.
42 recovered: bool,
43 /// When it happened, in milliseconds since the Unix epoch.
44 at_ms: i64,
45 },
46 /// The run entered a stage (including the first).
47 StageEntered {
48 /// The run this is about.
49 run_id: String,
50 /// Zero-based position of the stage in the blueprint's stage list.
51 stage_index: usize,
52 /// The stage's name, matching its key under `[stages]`.
53 stage_name: String,
54 /// When it happened, in milliseconds since the Unix epoch.
55 at_ms: i64,
56 },
57 /// The run left a stage; token counts are the stage's own totals.
58 StageExited {
59 /// The run this is about.
60 run_id: String,
61 /// Zero-based position of the stage in the blueprint's stage list.
62 stage_index: usize,
63 /// The stage's name, matching its key under `[stages]`.
64 stage_name: String,
65 /// Input tokens billed across this stage's whole life, so a revisited
66 /// stage reports the accumulated figure rather than the last visit's.
67 prompt_tokens: usize,
68 /// Output tokens billed across this stage's whole life.
69 completion_tokens: usize,
70 /// When it happened, in milliseconds since the Unix epoch.
71 at_ms: i64,
72 },
73 /// One inference call finished (successfully or not).
74 InferenceCompleted {
75 /// The run this is about.
76 run_id: String,
77 /// The stage the call was made from.
78 stage_name: String,
79 /// Which provider served it, after fallback resolution - so this is the
80 /// one that answered, not the one the blueprint listed first.
81 provider: String,
82 /// The model identifier sent on the wire.
83 model: String,
84 /// Wall-clock time of the provider call, including retries.
85 latency_ms: u64,
86 /// Input tokens this single call billed.
87 prompt_tokens: usize,
88 /// Output tokens this single call billed.
89 completion_tokens: usize,
90 /// Input tokens served from the provider's prompt cache, already counted
91 /// within `prompt_tokens` rather than in addition to it.
92 cached_tokens: usize,
93 /// Whether a response came back. A refusal or a tool call is still a
94 /// success; only a failed call is not.
95 success: bool,
96 },
97 /// One tool call finished.
98 ToolCallCompleted {
99 /// The run this is about.
100 run_id: String,
101 /// The stage the call was made from.
102 stage_name: String,
103 /// The tool as the model named it, before alias resolution.
104 tool_name: String,
105 /// Wall-clock time of the batch the call ran in. Tool calls execute
106 /// in batches and the executor reports one duration per batch, so
107 /// every call in a batch carries the same figure.
108 batch_latency_ms: u64,
109 /// Derived from the `[error] ` result-text convention every executor
110 /// uses; a heuristic, not a structured status.
111 success: bool,
112 },
113 /// A context compaction finished.
114 CompactionCompleted {
115 /// The run this is about.
116 run_id: String,
117 /// The stage whose context was compacted.
118 stage_name: String,
119 /// Whether the compaction produced a usable summary. A failure leaves
120 /// the region as it was rather than emptying it.
121 success: bool,
122 },
123 /// The run reached a terminal status; totals are run-wide.
124 RunCompleted {
125 /// The run this is about.
126 run_id: String,
127 /// The terminal status label: `complete`, `error`, or `cancelled`.
128 status: String,
129 /// Input tokens billed across the whole run.
130 prompt_tokens: usize,
131 /// Output tokens billed across the whole run.
132 completion_tokens: usize,
133 /// How many tool calls the run made, across every stage.
134 tool_calls: usize,
135 /// Whether the run stopped having modified nothing, when its blueprint
136 /// gave it a way to. `complete` says the pipeline reached the end, not
137 /// that it achieved anything; this is the difference.
138 empty_output: bool,
139 /// When it happened, in milliseconds since the Unix epoch.
140 at_ms: i64,
141 },
142 /// One per-run log line, as also written to the stage's log files.
143 Log {
144 /// The run this is about. An OTLP sink uses it to look up the run's open
145 /// span and stamp the record with its trace context.
146 run_id: String,
147 /// Which stage's log files the line also went to.
148 stage_index: usize,
149 /// Which log the line belongs in.
150 kind: LogKind,
151 /// The line itself, without a trailing newline.
152 line: String,
153 },
154}
155
156impl TelemetryEvent {
157 /// A stable short name for this event's variant.
158 ///
159 /// Exists so a test can `assert_eq!(event.kind(), "run_started")` rather
160 /// than `assert!(matches!(event, ...))` - the `matches!` non-matching arm
161 /// is a region only a *failing* assertion ever reaches, which reads as
162 /// uncovered under the workspace's 100% gate. Useful in its own right for
163 /// structured logging, where the kind is the field worth indexing on.
164 #[must_use]
165 pub fn kind(&self) -> &'static str {
166 match self {
167 Self::RunStarted { .. } => "run_started",
168 Self::StageEntered { .. } => "stage_entered",
169 Self::StageExited { .. } => "stage_exited",
170 Self::InferenceCompleted { .. } => "inference_completed",
171 Self::ToolCallCompleted { .. } => "tool_call_completed",
172 Self::CompactionCompleted { .. } => "compaction_completed",
173 Self::RunCompleted { .. } => "run_completed",
174 Self::Log { .. } => "log",
175 }
176 }
177
178 /// The run this event belongs to.
179 pub fn run_id(&self) -> &str {
180 match self {
181 Self::RunStarted { run_id, .. }
182 | Self::StageEntered { run_id, .. }
183 | Self::StageExited { run_id, .. }
184 | Self::InferenceCompleted { run_id, .. }
185 | Self::ToolCallCompleted { run_id, .. }
186 | Self::CompactionCompleted { run_id, .. }
187 | Self::RunCompleted { run_id, .. }
188 | Self::Log { run_id, .. } => run_id,
189 }
190 }
191}
192
193/// How the daemon as a whole is doing, sampled once per safety re-drive.
194///
195/// Deliberately not a [`TelemetryEvent`]: every variant of that enum belongs to
196/// one run, and this belongs to none of them. The distinction is the point. A
197/// daemon whose lanes are full and whose runs have all stopped moving emits no
198/// per-run telemetry at all, precisely because nothing is happening, so the
199/// silence that issue #191 reported was indistinguishable from an idle night.
200#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
201pub struct LaneHealth {
202 /// Agents doing work, or ready to.
203 pub agents_active: usize,
204 /// Agents blocked on input, a child, or a prompt.
205 pub agents_waiting: usize,
206 /// Tool batches holding lane capacity and running.
207 pub tools_busy: usize,
208 /// Tool batches waiting for lane capacity.
209 pub tools_queued: usize,
210 /// Tool batches parked on an unbounded wait, holding no capacity.
211 pub tools_parked: usize,
212 /// The tool lane's concurrency cap, including any relief granted.
213 pub tools_workers: usize,
214 /// Consecutive re-drives that found a lane at capacity and no run moving.
215 pub dead_cycles: u32,
216 /// Extra tool-lane capacity handed out on this sample, if any.
217 pub relief_granted: usize,
218}
219
220/// One provider the daemon has stopped sending work to, sampled alongside
221/// [`LaneHealth`].
222///
223/// Daemon-wide for the same reason as `LaneHealth`: a provider out of credits
224/// belongs to no single run, and the runs it kills emit nothing useful because
225/// they die before doing anything (issue #201).
226///
227/// `reason` is the label rather than the runtime's own enum: this crate sits
228/// below `leviath-providers`, so the type that names it is not in scope here.
229#[derive(Debug, Clone, PartialEq, Eq)]
230pub struct ProviderHealth {
231 /// The provider taken out of service.
232 pub provider: String,
233 /// Why, as a stable lowercase label (`credits-exhausted`, `auth-failed`).
234 pub reason: String,
235 /// Consecutive failures accumulated against it.
236 pub consecutive_failures: u32,
237 /// Seconds until it is probed again.
238 pub retry_in_secs: u64,
239}
240
241/// Where telemetry events go.
242///
243/// Implementations must tolerate being called from the engine's tick loop:
244/// `emit` should hand off or record cheaply, never block on network I/O.
245pub trait TelemetrySink: Send + Sync {
246 /// Record one event.
247 fn emit(&self, event: TelemetryEvent);
248
249 /// Record one daemon-wide health sample. Default: ignore it, so a sink that
250 /// only cares about runs needs no changes.
251 fn observe_lanes(&self, _health: LaneHealth) {}
252
253 /// Record which providers are currently out of service, sampled on the same
254 /// re-drive tick as [`TelemetrySink::observe_lanes`]. Default: ignore it,
255 /// so an existing sink keeps compiling unchanged.
256 fn observe_providers(&self, _down: &[ProviderHealth]) {}
257
258 /// Flush any buffered export before shutdown. Default: nothing buffered.
259 fn force_flush(&self) {}
260}
261
262/// The sink used when no telemetry backend is installed: drops everything.
263pub struct NoopSink;
264
265impl TelemetrySink for NoopSink {
266 fn emit(&self, _event: TelemetryEvent) {}
267}
268
269/// A sink that records every event in memory, for tests to assert on.
270#[derive(Default)]
271pub struct MemorySink {
272 events: std::sync::Mutex<Vec<TelemetryEvent>>,
273 lanes: std::sync::Mutex<Vec<LaneHealth>>,
274 providers: std::sync::Mutex<Vec<Vec<ProviderHealth>>>,
275 flushes: std::sync::atomic::AtomicUsize,
276}
277
278impl MemorySink {
279 /// A snapshot of everything emitted so far, in order.
280 pub fn events(&self) -> Vec<TelemetryEvent> {
281 crate::sync::lock(&self.events).clone()
282 }
283
284 /// Every lane-health sample recorded so far, in order.
285 pub fn lane_samples(&self) -> Vec<LaneHealth> {
286 crate::sync::lock(&self.lanes).clone()
287 }
288
289 /// Every provider-health sample recorded so far, in order.
290 pub fn provider_samples(&self) -> Vec<Vec<ProviderHealth>> {
291 crate::sync::lock(&self.providers).clone()
292 }
293
294 /// How many times `force_flush` was called.
295 pub fn flush_count(&self) -> usize {
296 self.flushes.load(std::sync::atomic::Ordering::SeqCst)
297 }
298}
299
300impl TelemetrySink for MemorySink {
301 fn emit(&self, event: TelemetryEvent) {
302 crate::sync::lock(&self.events).push(event);
303 }
304
305 fn observe_lanes(&self, health: LaneHealth) {
306 crate::sync::lock(&self.lanes).push(health);
307 }
308
309 fn observe_providers(&self, down: &[ProviderHealth]) {
310 crate::sync::lock(&self.providers).push(down.to_vec());
311 }
312
313 fn force_flush(&self) {
314 self.flushes
315 .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
316 }
317}
318
319#[cfg(test)]
320mod tests {
321 use super::*;
322
323 fn run_started(run_id: &str) -> TelemetryEvent {
324 TelemetryEvent::RunStarted {
325 run_id: run_id.to_string(),
326 agent_name: "coder".to_string(),
327 model: Some("claude-sonnet-5".to_string()),
328 parent_run_id: None,
329 recovered: false,
330 at_ms: 1_000,
331 }
332 }
333
334 #[test]
335 fn memory_sink_records_events_in_order() {
336 let sink = MemorySink::default();
337 sink.emit(run_started("r1"));
338 sink.emit(TelemetryEvent::StageEntered {
339 run_id: "r1".to_string(),
340 stage_index: 0,
341 stage_name: "plan".to_string(),
342 at_ms: 1_001,
343 });
344 let events = sink.events();
345 assert_eq!(events.len(), 2);
346 assert_eq!(events[0].kind(), "run_started");
347 assert_eq!(events[1].kind(), "stage_entered");
348 }
349
350 #[test]
351 fn memory_sink_records_lane_samples_and_the_noop_ignores_them() {
352 let sink = MemorySink::default();
353 assert!(sink.lane_samples().is_empty());
354 sink.observe_lanes(LaneHealth {
355 dead_cycles: 3,
356 ..Default::default()
357 });
358 assert_eq!(sink.lane_samples().len(), 1);
359 assert_eq!(sink.lane_samples()[0].dead_cycles, 3);
360
361 // The default trait body: a sink that only cares about runs drops it.
362 NoopSink.observe_lanes(LaneHealth::default());
363 }
364
365 #[test]
366 fn memory_sink_records_provider_samples_and_the_noop_ignores_them() {
367 let sink = MemorySink::default();
368 assert!(sink.provider_samples().is_empty());
369 let down = ProviderHealth {
370 provider: "openrouter".to_string(),
371 reason: "credits-exhausted".to_string(),
372 consecutive_failures: 3,
373 retry_in_secs: 240,
374 };
375 sink.observe_providers(std::slice::from_ref(&down));
376 // The empty sample matters too: it is how a collector sees a provider
377 // come back, not just go away.
378 sink.observe_providers(&[]);
379 assert_eq!(sink.provider_samples().len(), 2);
380 assert_eq!(sink.provider_samples()[0], vec![down]);
381 assert!(sink.provider_samples()[1].is_empty());
382
383 NoopSink.observe_providers(&[]);
384 }
385
386 #[test]
387 fn memory_sink_counts_flushes() {
388 let sink = MemorySink::default();
389 assert_eq!(sink.flush_count(), 0);
390 sink.force_flush();
391 sink.force_flush();
392 assert_eq!(sink.flush_count(), 2);
393 }
394
395 #[test]
396 fn noop_sink_accepts_events_and_default_flush() {
397 let sink = NoopSink;
398 sink.emit(run_started("r1"));
399 // The trait's default force_flush is a no-op; exercise it through the
400 // trait object the runtime actually holds.
401 let boxed: Box<dyn TelemetrySink> = Box::new(NoopSink);
402 boxed.force_flush();
403 }
404
405 #[test]
406 fn run_id_reaches_every_variant() {
407 let events = [
408 run_started("r1"),
409 TelemetryEvent::StageEntered {
410 run_id: "r1".to_string(),
411 stage_index: 0,
412 stage_name: "plan".to_string(),
413 at_ms: 0,
414 },
415 TelemetryEvent::StageExited {
416 run_id: "r1".to_string(),
417 stage_index: 0,
418 stage_name: "plan".to_string(),
419 prompt_tokens: 10,
420 completion_tokens: 5,
421 at_ms: 0,
422 },
423 TelemetryEvent::InferenceCompleted {
424 run_id: "r1".to_string(),
425 stage_name: "plan".to_string(),
426 provider: "anthropic".to_string(),
427 model: "claude-sonnet-5".to_string(),
428 latency_ms: 120,
429 prompt_tokens: 10,
430 completion_tokens: 5,
431 cached_tokens: 0,
432 success: true,
433 },
434 TelemetryEvent::ToolCallCompleted {
435 run_id: "r1".to_string(),
436 stage_name: "build".to_string(),
437 tool_name: "read_file".to_string(),
438 batch_latency_ms: 8,
439 success: true,
440 },
441 TelemetryEvent::CompactionCompleted {
442 run_id: "r1".to_string(),
443 stage_name: "build".to_string(),
444 success: true,
445 },
446 TelemetryEvent::RunCompleted {
447 run_id: "r1".to_string(),
448 status: "complete".to_string(),
449 prompt_tokens: 10,
450 completion_tokens: 5,
451 tool_calls: 1,
452 empty_output: false,
453 at_ms: 0,
454 },
455 TelemetryEvent::Log {
456 run_id: "r1".to_string(),
457 stage_index: 0,
458 kind: LogKind::Runtime,
459 line: "[Tokens: 10 in, 5 out]".to_string(),
460 },
461 ];
462 let kinds: Vec<&str> = events.iter().map(TelemetryEvent::kind).collect();
463 assert_eq!(
464 kinds,
465 [
466 "run_started",
467 "stage_entered",
468 "stage_exited",
469 "inference_completed",
470 "tool_call_completed",
471 "compaction_completed",
472 "run_completed",
473 "log",
474 ]
475 );
476 for event in &events {
477 assert_eq!(event.run_id(), "r1");
478 }
479 }
480
481 #[test]
482 fn event_clone_debug_and_eq() {
483 let event = run_started("r1");
484 let cloned = event.clone();
485 assert_eq!(event, cloned);
486 assert!(format!("{event:?}").contains("RunStarted"));
487 assert_ne!(LogKind::Output, LogKind::Runtime);
488 assert!(format!("{:?}", LogKind::Output).contains("Output"));
489 }
490}