theway_core/
observability.rs

1//! Transport-neutral runtime observations.
2//!
3//! Product events (`LoopEvent`, `SessionEvent`, `SubagentJobEvent`, and `DagEvent`) retain
4//! their UI, persistence, and wire semantics. This module is a separate, content-safe port
5//! for embedders that need traces, metrics, or structured operational logs.
6
7use std::sync::Arc;
8use std::sync::atomic::{AtomicU64, Ordering};
9use std::time::{Duration, Instant};
10
11static NEXT_OPERATION_ID: AtomicU64 = AtomicU64::new(1);
12
13/// Process-local identity for one observed operation.
14#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
15pub struct OperationId(u64);
16
17impl OperationId {
18    fn next() -> Self {
19        Self(NEXT_OPERATION_ID.fetch_add(1, Ordering::Relaxed))
20    }
21
22    pub fn get(self) -> u64 {
23        self.0
24    }
25}
26
27/// Safe correlation values shared by runtime operations.
28///
29/// Identifiers are useful trace/log attributes but MUST NOT be used as metric labels.
30#[derive(Clone, Debug, Default, PartialEq, Eq)]
31pub struct ObservationContext {
32    pub session_id: Option<String>,
33    pub run_id: Option<String>,
34    pub turn_id: Option<u32>,
35    pub job_id: Option<String>,
36    pub node_id: Option<String>,
37}
38
39impl ObservationContext {
40    pub fn with_turn(&self, turn_id: u32) -> Self {
41        Self {
42            turn_id: Some(turn_id),
43            ..self.clone()
44        }
45    }
46
47    pub fn with_graph(
48        &self,
49        run_id: Option<String>,
50        node_id: Option<String>,
51        job_id: Option<String>,
52    ) -> Self {
53        Self {
54            run_id,
55            node_id,
56            job_id,
57            ..self.clone()
58        }
59    }
60}
61
62#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
63#[non_exhaustive]
64pub enum OperationKind {
65    AgentRun,
66    Turn,
67    LlmRequest,
68    ToolExecution,
69    Compaction,
70    SubagentJob,
71    DagRun,
72    DagNode,
73}
74
75impl OperationKind {
76    pub fn as_str(self) -> &'static str {
77        match self {
78            Self::AgentRun => "agent.run",
79            Self::Turn => "agent.turn",
80            Self::LlmRequest => "llm.request",
81            Self::ToolExecution => "tool.execute",
82            Self::Compaction => "session.compaction",
83            Self::SubagentJob => "multiagent.job",
84            Self::DagRun => "dag.run",
85            Self::DagNode => "dag.node",
86        }
87    }
88}
89
90#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
91#[non_exhaustive]
92pub enum OperationOutcome {
93    Succeeded,
94    Failed,
95    Cancelled,
96    Interrupted,
97    TimedOut,
98    Skipped,
99    Abandoned,
100}
101
102impl OperationOutcome {
103    pub fn as_str(self) -> &'static str {
104        match self {
105            Self::Succeeded => "succeeded",
106            Self::Failed => "failed",
107            Self::Cancelled => "cancelled",
108            Self::Interrupted => "interrupted",
109            Self::TimedOut => "timed_out",
110            Self::Skipped => "skipped",
111            Self::Abandoned => "abandoned",
112        }
113    }
114}
115
116#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
117#[non_exhaustive]
118pub enum ErrorCategory {
119    Provider,
120    Tool,
121    Permission,
122    Persistence,
123    Validation,
124    Timeout,
125    Cancellation,
126    Runtime,
127}
128
129impl ErrorCategory {
130    pub fn as_str(self) -> &'static str {
131        match self {
132            Self::Provider => "provider",
133            Self::Tool => "tool",
134            Self::Permission => "permission",
135            Self::Persistence => "persistence",
136            Self::Validation => "validation",
137            Self::Timeout => "timeout",
138            Self::Cancellation => "cancellation",
139            Self::Runtime => "runtime",
140        }
141    }
142}
143
144/// Neutral measurements. USD pricing and application budget policy are intentionally absent.
145#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
146pub struct RuntimeMeasurements {
147    pub input_tokens: u64,
148    pub output_tokens: u64,
149    pub cache_read_tokens: u64,
150    pub cache_write_tokens: u64,
151    pub characters: u64,
152    pub turns: u64,
153    pub tool_calls: u64,
154}
155
156#[derive(Clone, Debug, PartialEq, Eq)]
157#[non_exhaustive]
158pub enum OperationDetail {
159    AgentRun,
160    Turn {
161        index: u32,
162    },
163    LlmRequest {
164        provider: String,
165        model: String,
166    },
167    ToolExecution {
168        tool_name: String,
169    },
170    Compaction {
171        algorithm: String,
172        provider: String,
173        model: String,
174    },
175    SubagentJob {
176        agent: String,
177        source: String,
178    },
179    DagRun,
180    DagNode,
181}
182
183impl OperationDetail {
184    pub fn kind(&self) -> OperationKind {
185        match self {
186            Self::AgentRun => OperationKind::AgentRun,
187            Self::Turn { .. } => OperationKind::Turn,
188            Self::LlmRequest { .. } => OperationKind::LlmRequest,
189            Self::ToolExecution { .. } => OperationKind::ToolExecution,
190            Self::Compaction { .. } => OperationKind::Compaction,
191            Self::SubagentJob { .. } => OperationKind::SubagentJob,
192            Self::DagRun => OperationKind::DagRun,
193            Self::DagNode => OperationKind::DagNode,
194        }
195    }
196}
197
198#[derive(Clone, Debug, PartialEq, Eq)]
199pub struct OperationStarted {
200    pub id: OperationId,
201    pub parent_id: Option<OperationId>,
202    pub context: ObservationContext,
203    pub detail: OperationDetail,
204}
205
206/// Opt-in full input/output payload attached to a finished operation.
207///
208/// JSON-shaped so exporters can serialize it directly (e.g. OTLP span
209/// attributes for Langfuse). Producers only build this when the embedder's
210/// [`RuntimeObserver::include_content`] returns `true`; the default keeps
211/// runtime observations content-safe.
212#[derive(Clone, Debug, Default, PartialEq)]
213pub struct ObservationContent {
214    pub input: Option<serde_json::Value>,
215    pub output: Option<serde_json::Value>,
216}
217
218#[derive(Clone, Debug, PartialEq)]
219pub struct OperationFinished {
220    pub id: OperationId,
221    pub kind: OperationKind,
222    pub context: ObservationContext,
223    pub outcome: OperationOutcome,
224    pub error_category: Option<ErrorCategory>,
225    pub duration: Duration,
226    pub measurements: RuntimeMeasurements,
227    /// Full input/output content. `None` unless the embedder opted in via
228    /// [`RuntimeObserver::include_content`].
229    pub content: Option<ObservationContent>,
230}
231
232#[derive(Clone, Debug, PartialEq)]
233#[non_exhaustive]
234pub enum RuntimeObservation {
235    OperationStarted(OperationStarted),
236    OperationFinished(OperationFinished),
237}
238
239/// Embedder-owned, non-blocking observation port.
240pub trait RuntimeObserver: Send + Sync {
241    fn observe(&self, observation: RuntimeObservation);
242
243    /// Whether operation producers should attach full input/output content to
244    /// finished observations. Defaults to `false` — observations stay
245    /// content-safe unless an embedder explicitly opts in.
246    fn include_content(&self) -> bool {
247        false
248    }
249}
250
251#[derive(Debug, Default)]
252pub struct NoopRuntimeObserver;
253
254impl RuntimeObserver for NoopRuntimeObserver {
255    fn observe(&self, _observation: RuntimeObservation) {}
256}
257
258pub fn noop_runtime_observer() -> Arc<dyn RuntimeObserver> {
259    Arc::new(NoopRuntimeObserver)
260}
261
262/// Invoke an observer without allowing its panic to enter runtime control flow.
263pub fn dispatch(observer: &Arc<dyn RuntimeObserver>, observation: RuntimeObservation) {
264    let observer = Arc::clone(observer);
265    let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(move || {
266        observer.observe(observation);
267    }));
268}
269
270/// RAII lifecycle helper. Dropping an unfinished scope emits `Abandoned` exactly once.
271pub struct OperationScope {
272    id: OperationId,
273    kind: OperationKind,
274    context: ObservationContext,
275    observer: Arc<dyn RuntimeObserver>,
276    started_at: Instant,
277    finished: bool,
278    content: Option<ObservationContent>,
279}
280
281impl OperationScope {
282    pub fn start(
283        observer: Arc<dyn RuntimeObserver>,
284        parent_id: Option<OperationId>,
285        context: ObservationContext,
286        detail: OperationDetail,
287    ) -> Self {
288        let id = OperationId::next();
289        let kind = detail.kind();
290        dispatch(
291            &observer,
292            RuntimeObservation::OperationStarted(OperationStarted {
293                id,
294                parent_id,
295                context: context.clone(),
296                detail,
297            }),
298        );
299        Self {
300            id,
301            kind,
302            context,
303            observer,
304            started_at: Instant::now(),
305            finished: false,
306            content: None,
307        }
308    }
309
310    pub fn id(&self) -> OperationId {
311        self.id
312    }
313
314    /// Attach opt-in full input/output content carried into the finish
315    /// observation. Call before [`OperationScope::finish`].
316    pub fn attach_content(&mut self, content: ObservationContent) {
317        self.content = Some(content);
318    }
319
320    pub fn finish(
321        mut self,
322        outcome: OperationOutcome,
323        error_category: Option<ErrorCategory>,
324        measurements: RuntimeMeasurements,
325    ) {
326        self.emit_finish(outcome, error_category, measurements);
327    }
328
329    fn emit_finish(
330        &mut self,
331        outcome: OperationOutcome,
332        error_category: Option<ErrorCategory>,
333        measurements: RuntimeMeasurements,
334    ) {
335        if self.finished {
336            return;
337        }
338        self.finished = true;
339        dispatch(
340            &self.observer,
341            RuntimeObservation::OperationFinished(OperationFinished {
342                id: self.id,
343                kind: self.kind,
344                context: self.context.clone(),
345                outcome,
346                error_category,
347                duration: self.started_at.elapsed(),
348                measurements,
349                content: self.content.take(),
350            }),
351        );
352    }
353}
354
355impl Drop for OperationScope {
356    fn drop(&mut self) {
357        self.emit_finish(
358            OperationOutcome::Abandoned,
359            Some(ErrorCategory::Runtime),
360            RuntimeMeasurements::default(),
361        );
362    }
363}
364
365#[cfg(test)]
366tests_bridge_macro::tests_bridge!("observability");
theway_core/observability.rs

theway_core/
observability.rs
