Skip to main content

harn_vm/
tracing.rs

1//! Pipeline Observability: structured tracing spans with parent/child relationships.
2//!
3//! When tracing is enabled (`vm.enable_tracing()`), the VM automatically emits
4//! spans for pipeline execution, function calls, LLM calls, tool invocations,
5//! imports, and async operations. Spans form a tree via parent_span_id.
6//!
7//! Access via builtins: `trace_spans()` returns all completed spans,
8//! `trace_summary()` returns a formatted summary.
9
10use std::cell::RefCell;
11use std::collections::BTreeMap;
12use std::future::Future;
13use std::pin::Pin;
14use std::task::{Context, Poll};
15use std::time::{Instant, SystemTime, UNIX_EPOCH};
16
17use crate::value::VmValue;
18
19/// The kind of operation a span represents.
20#[derive(Debug, Clone, Copy, PartialEq, Eq)]
21pub enum SpanKind {
22    Pipeline,
23    FnCall,
24    LlmCall,
25    ToolCall,
26    Import,
27    Parallel,
28    Spawn,
29    /// A `@step`-annotated function while its frame is on the call stack.
30    Step,
31    /// Host-side VM setup before user bytecode starts executing.
32    VmSetup,
33    /// Cooperative worker suspension while the durable checkpoint is written.
34    Suspension,
35    /// Worker resumption after a cooperative suspension.
36    Resume,
37    /// Pipeline drain / settlement phase.
38    Drain,
39    /// One drain settlement decision.
40    DrainDecision,
41    /// `pool.submit()` boundary — accepted, rejected, or queued (PL-06).
42    PoolSubmit,
43    /// Pool worker picks the task out of the queue (PL-06). Links back to
44    /// the originating `PoolSubmit` span across the async boundary so
45    /// queue dwell time can be reconstructed from a single trace.
46    PoolDequeue,
47    /// `emit_channel(...)` boundary — opened at `emit_channel`, closed
48    /// after the durable append + trigger fan-out finishes (CH-06 / #1877).
49    ChannelEmit,
50    /// Channel-source trigger match boundary — opened at trigger fan-out
51    /// just before the handler is invoked, closed once dispatch finishes.
52    /// Links back to the originating `ChannelEmit` span (multi-link for
53    /// batched / aggregated triggers).
54    ChannelMatch,
55    /// Script-opened user timing span via `std/timing`. Modeled as an
56    /// OTel INTERNAL span — distinct from `FnCall` so OTel exporters and
57    /// `harn run --profile-json` do not confuse them with LLM/tool work.
58    UserTiming,
59    /// A model routing / escalation decision — the agent switched the
60    /// serving model mid-run. Metadata carries `from_model`, `to_model`,
61    /// and `reason` (see [`meta`]). Emitted as a zero-duration marker at
62    /// the decision point so viewers can annotate the flame graph with the
63    /// switch instead of inferring it from adjacent `llm_call` models.
64    ModelRoute,
65    /// A batch of tools promoted into the active surface (MCP bootstrap,
66    /// skill activation, or a search-driven mount). Metadata carries
67    /// `tool_names`, `tool_count`, `source`, and an optional `detail`.
68    ToolMount,
69    /// A single deferred tool schema promoted via `tool_search`. Metadata
70    /// carries `tool_name`, `query`, and the match `score`.
71    DeferredToolLoad,
72}
73
74impl SpanKind {
75    pub fn as_str(self) -> &'static str {
76        match self {
77            Self::Pipeline => "pipeline",
78            Self::FnCall => "fn_call",
79            Self::LlmCall => "llm_call",
80            Self::ToolCall => "tool_call",
81            Self::Import => "import",
82            Self::Parallel => "parallel",
83            Self::Spawn => "spawn",
84            Self::Step => "step",
85            Self::VmSetup => "vm_setup",
86            Self::Suspension => "suspension",
87            Self::Resume => "resume",
88            Self::Drain => "drain",
89            Self::DrainDecision => "drain_decision",
90            Self::PoolSubmit => "pool_submit",
91            Self::PoolDequeue => "pool_dequeue",
92            Self::ChannelEmit => "channel_emit",
93            Self::ChannelMatch => "channel_match",
94            Self::UserTiming => "user_timing",
95            Self::ModelRoute => "model_route",
96            Self::ToolMount => "tool_mount",
97            Self::DeferredToolLoad => "deferred_tool_load",
98        }
99    }
100}
101
102/// Canonical metadata keys for VM trace spans. Downstream viewers (Burin
103/// portal, harn-cloud dashboard) key off these exact strings to render
104/// token flame graphs and tool-selection events, so they are defined once
105/// here rather than retyped at each emission site.
106pub mod meta {
107    // llm_call token + cost attribution.
108    pub const MODEL: &str = "model";
109    pub const PROVIDER: &str = "provider";
110    pub const INPUT_TOKENS: &str = "input_tokens";
111    pub const OUTPUT_TOKENS: &str = "output_tokens";
112    pub const CACHE_READ_TOKENS: &str = "cache_read_tokens";
113    pub const CACHE_WRITE_TOKENS: &str = "cache_write_tokens";
114    pub const COST_USD: &str = "cost_usd";
115
116    // model_route.
117    pub const FROM_MODEL: &str = "from_model";
118    pub const TO_MODEL: &str = "to_model";
119    pub const REASON: &str = "reason";
120
121    // tool_mount.
122    pub const TOOL_NAMES: &str = "tool_names";
123    pub const TOOL_COUNT: &str = "tool_count";
124    pub const SOURCE: &str = "source";
125    pub const DETAIL: &str = "detail";
126
127    // deferred_tool_load.
128    pub const TOOL_NAME: &str = "tool_name";
129    pub const QUERY: &str = "query";
130    pub const SCORE: &str = "score";
131}
132
133/// Structured per-LLM-call token and cost attribution for an `llm_call`
134/// span. Built once at the call site and lowered to metadata pairs via
135/// [`LlmCallUsage::metadata_pairs`] so every emission uses the canonical
136/// [`meta`] keys instead of ad-hoc strings. `cost_usd` is `None` when the
137/// (provider, model) pair has no catalog pricing.
138#[derive(Debug, Clone, Default, PartialEq)]
139pub struct LlmCallUsage {
140    pub model: String,
141    pub provider: String,
142    pub input_tokens: i64,
143    pub output_tokens: i64,
144    pub cache_read_tokens: i64,
145    pub cache_write_tokens: i64,
146    pub cost_usd: Option<f64>,
147}
148
149impl LlmCallUsage {
150    /// Lower to `(key, value)` pairs keyed by the canonical [`meta`]
151    /// constants, suitable for `annotate_current_span` / `span_set_metadata`.
152    pub fn metadata_pairs(&self) -> Vec<(&'static str, serde_json::Value)> {
153        let mut pairs = vec![
154            (meta::MODEL, serde_json::json!(self.model)),
155            (meta::PROVIDER, serde_json::json!(self.provider)),
156            (meta::INPUT_TOKENS, serde_json::json!(self.input_tokens)),
157            (meta::OUTPUT_TOKENS, serde_json::json!(self.output_tokens)),
158            (
159                meta::CACHE_READ_TOKENS,
160                serde_json::json!(self.cache_read_tokens),
161            ),
162            (
163                meta::CACHE_WRITE_TOKENS,
164                serde_json::json!(self.cache_write_tokens),
165            ),
166        ];
167        if let Some(cost) = self.cost_usd {
168            pairs.push((meta::COST_USD, serde_json::json!(cost)));
169        }
170        pairs
171    }
172}
173
174/// Emit a zero-duration marker span of `kind` carrying `metadata`. Marker
175/// spans model point-in-time telemetry events (model routing, tool mounts,
176/// deferred-tool promotions) that have no meaningful duration but need to
177/// appear in the trace tree at their causal position under the current
178/// active span. No-op when tracing is disabled.
179pub fn emit_marker_span(
180    kind: SpanKind,
181    name: impl Into<String>,
182    metadata: Vec<(&str, serde_json::Value)>,
183) {
184    let span_id = span_start(kind, name.into());
185    if span_id == 0 {
186        return;
187    }
188    for (key, value) in metadata {
189        span_set_metadata(span_id, key, value);
190    }
191    span_end(span_id);
192}
193
194/// Emit a [`SpanKind::ModelRoute`] marker for a model switch / escalation.
195pub fn emit_model_route(from_model: &str, to_model: &str, reason: &str) {
196    emit_marker_span(
197        SpanKind::ModelRoute,
198        "model_route",
199        vec![
200            (meta::FROM_MODEL, serde_json::json!(from_model)),
201            (meta::TO_MODEL, serde_json::json!(to_model)),
202            (meta::REASON, serde_json::json!(reason)),
203        ],
204    );
205}
206
207/// Emit a [`SpanKind::ToolMount`] marker for a batch of tools promoted into
208/// the active surface. `source` is the promotion origin (`"mcp"`,
209/// `"skill"`, `"search"`); `detail` optionally names the concrete source
210/// (e.g. an MCP server name).
211pub fn emit_tool_mount(tool_names: &[String], source: &str, detail: Option<&str>) {
212    if tool_names.is_empty() {
213        return;
214    }
215    let mut metadata = vec![
216        (meta::TOOL_NAMES, serde_json::json!(tool_names)),
217        (meta::TOOL_COUNT, serde_json::json!(tool_names.len())),
218        (meta::SOURCE, serde_json::json!(source)),
219    ];
220    if let Some(detail) = detail {
221        metadata.push((meta::DETAIL, serde_json::json!(detail)));
222    }
223    emit_marker_span(SpanKind::ToolMount, "tool_mount", metadata);
224}
225
226/// Emit a [`SpanKind::DeferredToolLoad`] marker for a single deferred tool
227/// schema promoted via `tool_search`.
228pub fn emit_deferred_tool_load(tool_name: &str, query: &str, score: Option<f64>) {
229    let mut metadata = vec![
230        (meta::TOOL_NAME, serde_json::json!(tool_name)),
231        (meta::QUERY, serde_json::json!(query)),
232    ];
233    if let Some(score) = score {
234        metadata.push((meta::SCORE, serde_json::json!(score)));
235    }
236    emit_marker_span(SpanKind::DeferredToolLoad, "deferred_tool_load", metadata);
237}
238
239/// Link to a span that is causally related but not the parent.
240#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
241#[serde(default)]
242pub struct SpanLink {
243    pub trace_id: String,
244    pub span_id: String,
245    #[serde(skip_serializing_if = "BTreeMap::is_empty")]
246    pub attributes: BTreeMap<String, String>,
247}
248
249impl SpanLink {
250    pub fn new(trace_id: impl Into<String>, span_id: impl Into<String>) -> Self {
251        Self {
252            trace_id: trace_id.into(),
253            span_id: span_id.into(),
254            attributes: BTreeMap::new(),
255        }
256    }
257
258    pub fn with_attributes(mut self, attributes: BTreeMap<String, String>) -> Self {
259        self.attributes = attributes;
260        self
261    }
262}
263
264/// One sub-phase annotation attached to a span. Modeled after OTel span
265/// events: a named checkpoint with optional structured attributes that
266/// piggy-backs on the enclosing span rather than allocating a new one.
267#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
268#[serde(default)]
269pub struct SpanEvent {
270    pub name: String,
271    /// Wall-clock time of the event in milliseconds since the UNIX epoch.
272    pub time_unix_ms: u64,
273    /// Monotonic offset from the parent span's start, in milliseconds.
274    pub offset_ms: u64,
275    #[serde(skip_serializing_if = "BTreeMap::is_empty")]
276    pub attributes: BTreeMap<String, serde_json::Value>,
277}
278
279/// A completed tracing span.
280#[derive(Debug, Clone)]
281pub struct Span {
282    pub trace_id: String,
283    pub span_id: u64,
284    pub parent_id: Option<u64>,
285    pub kind: SpanKind,
286    pub name: String,
287    /// Monotonic offset from the collector's epoch, in milliseconds.
288    pub start_ms: u64,
289    /// Wall-clock start in milliseconds since the UNIX epoch. Recorded
290    /// once at `start` for external correlation; duration is always
291    /// derived from the monotonic clock, not from wall-clock end - start.
292    pub start_unix_ms: u64,
293    pub duration_ms: u64,
294    pub metadata: BTreeMap<String, serde_json::Value>,
295    pub links: Vec<SpanLink>,
296    pub events: Vec<SpanEvent>,
297}
298
299/// An in-flight span (not yet completed).
300struct OpenSpan {
301    trace_id: String,
302    span_id: u64,
303    parent_id: Option<u64>,
304    kind: SpanKind,
305    name: String,
306    started_at: Instant,
307    /// Mock-monotonic snapshot at start, captured only when a
308    /// `clock_mock` override was active. Pairs with the closing snapshot
309    /// to compute deterministic durations under `mock_time(...)`.
310    started_at_mock_mono_ms: Option<u64>,
311    start_unix_ms: u64,
312    metadata: BTreeMap<String, serde_json::Value>,
313    links: Vec<SpanLink>,
314    events: Vec<SpanEvent>,
315}
316
317/// Thread-local span collector. Accumulates completed spans and tracks the
318/// active span stack for automatic parent assignment.
319pub struct SpanCollector {
320    trace_id: String,
321    next_id: u64,
322    /// Stack of open span IDs — the top is the current active span.
323    active_stack: Vec<u64>,
324    /// Open (in-flight) spans keyed by ID.
325    open: BTreeMap<u64, OpenSpan>,
326    /// Completed spans in chronological order.
327    completed: Vec<Span>,
328    /// Epoch for relative timing.
329    epoch: Instant,
330}
331
332impl Default for SpanCollector {
333    fn default() -> Self {
334        Self::new()
335    }
336}
337
338impl SpanCollector {
339    pub fn new() -> Self {
340        Self {
341            next_id: 1,
342            trace_id: format!("trace_{}", uuid::Uuid::now_v7()),
343            active_stack: Vec::new(),
344            open: BTreeMap::new(),
345            completed: Vec::new(),
346            epoch: Instant::now(),
347        }
348    }
349
350    /// Start a new span. Returns the span ID.
351    pub fn start(&mut self, kind: SpanKind, name: String) -> u64 {
352        let parent_id = self.active_stack.last().copied();
353        self.start_with_parent(kind, name, Vec::new(), parent_id)
354    }
355
356    /// Start a new span with non-parent causal links. Returns the span ID.
357    pub fn start_with_links(&mut self, kind: SpanKind, name: String, links: Vec<SpanLink>) -> u64 {
358        let parent_id = self.active_stack.last().copied();
359        self.start_with_parent(kind, name, links, parent_id)
360    }
361
362    /// Start a root span with non-parent causal links. Returns the span ID.
363    pub fn start_detached_with_links(
364        &mut self,
365        kind: SpanKind,
366        name: String,
367        links: Vec<SpanLink>,
368    ) -> u64 {
369        self.start_with_parent(kind, name, links, None)
370    }
371
372    fn start_with_parent(
373        &mut self,
374        kind: SpanKind,
375        name: String,
376        links: Vec<SpanLink>,
377        parent_id: Option<u64>,
378    ) -> u64 {
379        let id = self.next_id;
380        self.next_id += 1;
381        let now = Instant::now();
382        let started_at_mock_mono_ms = mock_monotonic_ms();
383        let start_unix_ms = wall_clock_ms();
384
385        let mut event_metadata = BTreeMap::new();
386        if !links.is_empty() {
387            event_metadata.insert("links".to_string(), serde_json::json!(links));
388        }
389        crate::events::emit_span_start(id, parent_id, &name, kind.as_str(), event_metadata);
390
391        self.open.insert(
392            id,
393            OpenSpan {
394                trace_id: self.trace_id.clone(),
395                span_id: id,
396                parent_id,
397                kind,
398                name,
399                started_at: now,
400                started_at_mock_mono_ms,
401                start_unix_ms,
402                metadata: BTreeMap::new(),
403                links,
404                events: Vec::new(),
405            },
406        );
407        self.active_stack.push(id);
408        id
409    }
410
411    /// Attach metadata to an open span.
412    pub fn set_metadata(&mut self, span_id: u64, key: &str, value: serde_json::Value) {
413        if let Some(span) = self.open.get_mut(&span_id) {
414            span.metadata.insert(key.to_string(), value);
415        }
416    }
417
418    /// Attach metadata to an open or completed span unless the key exists.
419    pub fn attach_metadata_if_absent(&mut self, span_id: u64, key: &str, value: serde_json::Value) {
420        if let Some(span) = self.open.get_mut(&span_id) {
421            span.metadata.entry(key.to_string()).or_insert(value);
422            return;
423        }
424        if let Some(span) = self
425            .completed
426            .iter_mut()
427            .rev()
428            .find(|span| span.span_id == span_id)
429        {
430            span.metadata.entry(key.to_string()).or_insert(value);
431        }
432    }
433
434    /// Append a sub-phase annotation to an open span. Returns `true` if
435    /// the event was attached; `false` if `span_id` does not match any
436    /// open span (already closed or never opened).
437    pub fn record_event(
438        &mut self,
439        span_id: u64,
440        name: String,
441        attributes: BTreeMap<String, serde_json::Value>,
442    ) -> bool {
443        let Some(span) = self.open.get_mut(&span_id) else {
444            return false;
445        };
446        let offset_ms = match (span.started_at_mock_mono_ms, mock_monotonic_ms()) {
447            (Some(start), Some(now)) => now.saturating_sub(start),
448            _ => span.started_at.elapsed().as_millis() as u64,
449        };
450        span.events.push(SpanEvent {
451            name,
452            time_unix_ms: wall_clock_ms(),
453            offset_ms,
454            attributes,
455        });
456        true
457    }
458
459    /// Read the wall-clock start of an open span.
460    pub fn open_start_unix_ms(&self, span_id: u64) -> Option<u64> {
461        self.open.get(&span_id).map(|span| span.start_unix_ms)
462    }
463
464    /// End a span. Moves it from open to completed and returns the
465    /// finalized span so callers (e.g. `std/timing`) can read its
466    /// `duration_ms` directly without re-scanning `take_spans()`.
467    pub fn end(&mut self, span_id: u64) -> Option<Span> {
468        let span = self.open.remove(&span_id)?;
469        let start_ms = span.started_at.duration_since(self.epoch).as_millis() as u64;
470        let duration_ms = match (span.started_at_mock_mono_ms, mock_monotonic_ms()) {
471            (Some(start), Some(end)) => end.saturating_sub(start),
472            _ => span.started_at.elapsed().as_millis() as u64,
473        };
474
475        let mut end_meta = span.metadata.clone();
476        end_meta.insert(
477            "duration_ms".to_string(),
478            serde_json::Value::Number(serde_json::Number::from(duration_ms)),
479        );
480        crate::events::emit_span_end(span_id, end_meta);
481
482        let completed = Span {
483            trace_id: span.trace_id,
484            span_id: span.span_id,
485            parent_id: span.parent_id,
486            kind: span.kind,
487            name: span.name,
488            start_ms,
489            start_unix_ms: span.start_unix_ms,
490            duration_ms,
491            metadata: span.metadata,
492            links: span.links,
493            events: span.events,
494        };
495        self.completed.push(completed.clone());
496
497        if let Some(pos) = self.active_stack.iter().rposition(|&id| id == span_id) {
498            self.active_stack.remove(pos);
499        }
500        Some(completed)
501    }
502
503    /// Get the current active span ID (if any).
504    pub fn current_span_id(&self) -> Option<u64> {
505        self.active_stack.last().copied()
506    }
507
508    /// Build a serializable link for an open span.
509    pub fn span_link(&self, span_id: u64) -> Option<SpanLink> {
510        self.open
511            .get(&span_id)
512            .map(|span| SpanLink::new(span.trace_id.clone(), span.span_id.to_string()))
513    }
514
515    /// Build a serializable link for the current active span.
516    pub fn current_span_link(&self) -> Option<SpanLink> {
517        self.current_span_id()
518            .and_then(|span_id| self.span_link(span_id))
519    }
520
521    /// Take all completed spans (drains the collector).
522    pub fn take_spans(&mut self) -> Vec<Span> {
523        std::mem::take(&mut self.completed)
524    }
525
526    /// Peek at all completed spans (non-destructive).
527    pub fn spans(&self) -> &[Span] {
528        &self.completed
529    }
530
531    /// Reset the collector.
532    pub fn reset(&mut self) {
533        self.active_stack.clear();
534        self.open.clear();
535        self.completed.clear();
536        self.next_id = 1;
537        self.trace_id = format!("trace_{}", uuid::Uuid::now_v7());
538        self.epoch = Instant::now();
539    }
540}
541
542thread_local! {
543    static COLLECTOR: RefCell<SpanCollector> = RefCell::new(SpanCollector::new());
544    static TRACING_ENABLED: RefCell<bool> = const { RefCell::new(false) };
545}
546
547/// Close only spans opened after this checkpoint if its owning future is
548/// cancelled. Existing caller spans remain active and completed history is
549/// retained.
550pub(crate) fn checkpoint() -> TracingCheckpoint {
551    let open_at_checkpoint =
552        COLLECTOR.with(|collector| collector.borrow().open.keys().copied().collect::<Vec<_>>());
553    TracingCheckpoint {
554        open_at_checkpoint: Some(open_at_checkpoint),
555    }
556}
557
558pub(crate) struct TracingCheckpoint {
559    open_at_checkpoint: Option<Vec<u64>>,
560}
561
562impl TracingCheckpoint {
563    pub(crate) fn complete(mut self) {
564        self.open_at_checkpoint = None;
565    }
566}
567
568impl Drop for TracingCheckpoint {
569    fn drop(&mut self) {
570        let Some(baseline) = self.open_at_checkpoint.take() else {
571            return;
572        };
573        COLLECTOR.with(|collector| {
574            let mut collector = collector.borrow_mut();
575            let mut abandoned = collector
576                .active_stack
577                .iter()
578                .rev()
579                .copied()
580                .filter(|span_id| !baseline.contains(span_id))
581                .collect::<Vec<_>>();
582            for span_id in collector.open.keys().rev().copied() {
583                if !baseline.contains(&span_id) && !abandoned.contains(&span_id) {
584                    abandoned.push(span_id);
585                }
586            }
587            for span_id in abandoned {
588                collector.set_metadata(span_id, "status", serde_json::json!("abandoned"));
589                collector.end(span_id);
590            }
591        });
592    }
593}
594
595pin_project_lite::pin_project! {
596    pub(crate) struct Checkpointed<F> {
597        // Field order matters on cancellation: abandon open spans before
598        // dropping `inner`, whose ordinary RAII span guards would otherwise
599        // close them as if execution completed normally.
600        checkpoint: Option<TracingCheckpoint>,
601        #[pin]
602        inner: F,
603    }
604}
605
606pub(crate) fn checkpoint_future<F: Future>(inner: F) -> Checkpointed<F> {
607    Checkpointed {
608        checkpoint: Some(checkpoint()),
609        inner,
610    }
611}
612
613impl<F: Future> Future for Checkpointed<F> {
614    type Output = F::Output;
615
616    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
617        let this = self.project();
618        let result = this.inner.poll(cx);
619        if result.is_ready() {
620            this.checkpoint
621                .take()
622                .expect("tracing checkpoint")
623                .complete();
624        }
625        result
626    }
627}
628
629/// Best-effort wall-clock millis since the UNIX epoch. Honors an active
630/// `clock_mock` override so spans recorded inside `mock_time(...)` blocks
631/// align with the rest of the runtime's clock reads; returns 0 only if
632/// the host clock is behind the epoch (e.g. unusual sandbox shims).
633fn wall_clock_ms() -> u64 {
634    if let Some(clock) = crate::clock_mock::active_clock() {
635        return harn_clock::now_wall_ms(clock.as_ref()).max(0) as u64;
636    }
637    SystemTime::now()
638        .duration_since(UNIX_EPOCH)
639        .map(|d| d.as_millis() as u64)
640        .unwrap_or(0)
641}
642
643/// Mock-aware monotonic snapshot. Returns `Some(ms)` when a
644/// `clock_mock` override is active, `None` otherwise. Span lifecycle
645/// pairs the start/end snapshots so durations recorded under
646/// `mock_time(...)` reflect `advance_time(...)` instead of real
647/// wall-clock progress; spans without an active mock at start fall
648/// through to the standard `Instant::elapsed` path on close.
649fn mock_monotonic_ms() -> Option<u64> {
650    crate::clock_mock::active_clock().map(|clock| clock.monotonic_ms().max(0) as u64)
651}
652
653/// Enable or disable VM tracing for the current thread.
654pub fn set_tracing_enabled(enabled: bool) {
655    TRACING_ENABLED.with(|e| *e.borrow_mut() = enabled);
656    if enabled {
657        COLLECTOR.with(|c| c.borrow_mut().reset());
658    }
659}
660
661/// Enable VM tracing for the current thread WITHOUT clobbering an
662/// in-flight trace. Unlike [`set_tracing_enabled(true)`], this only
663/// resets the span collector when it is idle (no open spans); when a
664/// caller is mid-trace — e.g. holding an enclosing `timed(...)` /
665/// `start_timing` span across a nested `workflow_execute` run — the
666/// collector is left intact so the caller's open span and its completed
667/// siblings survive. Note that user-timing spans populate the collector
668/// regardless of the enabled flag (see [`span_start_user_timing`]), so
669/// the open-span check (not [`is_tracing_enabled`]) is what distinguishes
670/// "someone is mid-trace" from "clean slate". When idle, behavior is
671/// identical to `set_tracing_enabled(true)`.
672pub fn enable_tracing_preserving_open_spans() {
673    let has_open_spans = COLLECTOR.with(|c| !c.borrow().open.is_empty());
674    TRACING_ENABLED.with(|e| *e.borrow_mut() = true);
675    if !has_open_spans {
676        COLLECTOR.with(|c| c.borrow_mut().reset());
677    }
678}
679
680/// Check if tracing is enabled.
681pub fn is_tracing_enabled() -> bool {
682    TRACING_ENABLED.with(|e| *e.borrow())
683}
684
685/// Start a span (no-op if tracing disabled). Returns span ID or 0.
686pub fn span_start(kind: SpanKind, name: String) -> u64 {
687    if !is_tracing_enabled() {
688        return 0;
689    }
690    COLLECTOR.with(|c| c.borrow_mut().start(kind, name))
691}
692
693/// Start a span with non-parent causal links (no-op if tracing disabled).
694pub fn span_start_with_links(kind: SpanKind, name: String, links: Vec<SpanLink>) -> u64 {
695    if !is_tracing_enabled() {
696        return 0;
697    }
698    COLLECTOR.with(|c| c.borrow_mut().start_with_links(kind, name, links))
699}
700
701/// Start a root span with non-parent causal links (no-op if tracing disabled).
702pub fn span_start_detached_with_links(kind: SpanKind, name: String, links: Vec<SpanLink>) -> u64 {
703    if !is_tracing_enabled() {
704        return 0;
705    }
706    COLLECTOR.with(|c| c.borrow_mut().start_detached_with_links(kind, name, links))
707}
708
709/// Attach metadata to an open span (no-op if span_id is 0).
710pub fn span_set_metadata(span_id: u64, key: &str, value: serde_json::Value) {
711    if span_id == 0 {
712        return;
713    }
714    COLLECTOR.with(|c| c.borrow_mut().set_metadata(span_id, key, value));
715}
716
717/// End a span (no-op if span_id is 0). Returns the finalized span when
718/// the id was a live open span.
719pub fn span_end(span_id: u64) -> Option<Span> {
720    if span_id == 0 {
721        return None;
722    }
723    COLLECTOR.with(|c| c.borrow_mut().end(span_id))
724}
725
726/// Start a user-timing span. Unlike [`span_start`], this always records
727/// regardless of [`is_tracing_enabled`] — `std/timing` callers depend on
728/// the returned `duration_ms` to function as a primitive replacement for
729/// hand-rolled `now_ms()` subtraction.
730pub fn span_start_user_timing(
731    name: String,
732    attrs: BTreeMap<String, serde_json::Value>,
733) -> (u64, String, Option<u64>, u64) {
734    COLLECTOR.with(|c| {
735        let mut c = c.borrow_mut();
736        let id = c.start(SpanKind::UserTiming, name);
737        for (key, value) in attrs {
738            c.set_metadata(id, &key, value);
739        }
740        let parent = c.open.get(&id).and_then(|span| span.parent_id);
741        let trace_id = c
742            .open
743            .get(&id)
744            .map(|span| span.trace_id.clone())
745            .unwrap_or_default();
746        let start_unix_ms = c.open_start_unix_ms(id).unwrap_or(0);
747        (id, trace_id, parent, start_unix_ms)
748    })
749}
750
751/// Record a sub-phase event on an open span. No-op when `span_id` is 0
752/// or already closed; returns whether the event was attached so callers
753/// can surface no-op feedback.
754pub fn span_record_event(
755    span_id: u64,
756    name: String,
757    attributes: BTreeMap<String, serde_json::Value>,
758) -> bool {
759    if span_id == 0 {
760        return false;
761    }
762    COLLECTOR.with(|c| c.borrow_mut().record_event(span_id, name, attributes))
763}
764
765/// Attach metadata to an open span. No-op when `span_id` is 0 or already
766/// closed.
767pub fn span_attach_metadata(span_id: u64, key: &str, value: serde_json::Value) {
768    if span_id == 0 {
769        return;
770    }
771    COLLECTOR.with(|c| c.borrow_mut().set_metadata(span_id, key, value));
772}
773
774/// Attach metadata to an open or completed span unless the key already exists.
775pub fn span_attach_metadata_if_absent(span_id: u64, key: &str, value: serde_json::Value) {
776    if span_id == 0 {
777        return;
778    }
779    COLLECTOR.with(|c| {
780        c.borrow_mut()
781            .attach_metadata_if_absent(span_id, key, value);
782    });
783}
784
785/// Get the currently active span id, if tracing is enabled and a span is open.
786pub fn current_span_id() -> Option<u64> {
787    if !is_tracing_enabled() {
788        return None;
789    }
790    COLLECTOR.with(|c| c.borrow().current_span_id())
791}
792
793/// Return a link reference for an open span.
794pub fn span_link(span_id: u64) -> Option<SpanLink> {
795    if span_id == 0 || !is_tracing_enabled() {
796        return None;
797    }
798    COLLECTOR.with(|c| c.borrow().span_link(span_id))
799}
800
801/// Return a link reference for the current active span.
802pub fn current_span_link() -> Option<SpanLink> {
803    if !is_tracing_enabled() {
804        return None;
805    }
806    COLLECTOR.with(|c| c.borrow().current_span_link())
807}
808
809/// Take all completed spans.
810pub fn take_spans() -> Vec<Span> {
811    COLLECTOR.with(|c| c.borrow_mut().take_spans())
812}
813
814/// Peek at completed spans (cloned).
815pub fn peek_spans() -> Vec<Span> {
816    COLLECTOR.with(|c| c.borrow().spans().to_vec())
817}
818
819/// Reset the tracing collector.
820pub fn reset_tracing() {
821    COLLECTOR.with(|c| c.borrow_mut().reset());
822}
823
824/// Convert a span to a VmValue dict for user access.
825pub fn span_to_vm_value(span: &Span) -> VmValue {
826    let mut d: BTreeMap<String, VmValue> = BTreeMap::new();
827    d.insert(
828        "trace_id".into(),
829        VmValue::String(arcstr::ArcStr::from(span.trace_id.as_str())),
830    );
831    d.insert("span_id".into(), VmValue::Int(span.span_id as i64));
832    d.insert(
833        "parent_id".into(),
834        span.parent_id
835            .map(|id| VmValue::Int(id as i64))
836            .unwrap_or(VmValue::Nil),
837    );
838    d.insert(
839        "kind".into(),
840        VmValue::String(arcstr::ArcStr::from(span.kind.as_str())),
841    );
842    d.insert(
843        "name".into(),
844        VmValue::String(arcstr::ArcStr::from(span.name.as_str())),
845    );
846    d.insert("start_ms".into(), VmValue::Int(span.start_ms as i64));
847    d.insert(
848        "start_unix_ms".into(),
849        VmValue::Int(span.start_unix_ms as i64),
850    );
851    d.insert("duration_ms".into(), VmValue::Int(span.duration_ms as i64));
852
853    if !span.metadata.is_empty() {
854        let meta: crate::value::DictMap = span
855            .metadata
856            .iter()
857            .map(|(k, v)| {
858                (
859                    crate::value::intern_key(k),
860                    crate::stdlib::json_to_vm_value(v),
861                )
862            })
863            .collect();
864        d.insert("metadata".into(), VmValue::dict(meta));
865    }
866    if !span.links.is_empty() {
867        d.insert(
868            "links".into(),
869            crate::stdlib::json_to_vm_value(&serde_json::json!(span.links)),
870        );
871    }
872    if !span.events.is_empty() {
873        d.insert(
874            "events".into(),
875            crate::stdlib::json_to_vm_value(&serde_json::json!(span.events)),
876        );
877    }
878
879    VmValue::dict(d)
880}
881
882/// Generate a formatted summary of all spans.
883pub fn format_summary() -> String {
884    let spans = peek_spans();
885    if spans.is_empty() {
886        return "No spans recorded.".into();
887    }
888
889    let mut lines = Vec::new();
890    let total_ms: u64 = spans
891        .iter()
892        .filter(|s| s.parent_id.is_none())
893        .map(|s| s.duration_ms)
894        .sum();
895
896    lines.push(format!("Trace: {} spans, {total_ms}ms total", spans.len()));
897    lines.push(String::new());
898
899    fn print_tree(spans: &[Span], parent_id: Option<u64>, depth: usize, lines: &mut Vec<String>) {
900        let children: Vec<&Span> = spans.iter().filter(|s| s.parent_id == parent_id).collect();
901        for span in children {
902            let indent = "  ".repeat(depth);
903            let meta_str = if span.metadata.is_empty() {
904                String::new()
905            } else {
906                let parts: Vec<String> = span
907                    .metadata
908                    .iter()
909                    .map(|(k, v)| format!("{k}={v}"))
910                    .collect();
911                format!(" ({})", parts.join(", "))
912            };
913            lines.push(format!(
914                "{indent}{} {} {}ms{meta_str}",
915                span.kind.as_str(),
916                span.name,
917                span.duration_ms,
918            ));
919            print_tree(spans, Some(span.span_id), depth + 1, lines);
920        }
921    }
922
923    print_tree(&spans, None, 0, &mut lines);
924    lines.join("\n")
925}
926
927#[cfg(test)]
928mod tests {
929    use super::*;
930
931    #[test]
932    fn test_span_collector_basic() {
933        let mut c = SpanCollector::new();
934        let id = c.start(SpanKind::Pipeline, "main".into());
935        assert_eq!(id, 1);
936        assert_eq!(c.current_span_id(), Some(1));
937        assert!(c.span_link(id).is_some());
938        c.end(id);
939        assert_eq!(c.current_span_id(), None);
940        assert_eq!(c.spans().len(), 1);
941        assert_eq!(c.spans()[0].name, "main");
942        assert_eq!(c.spans()[0].parent_id, None);
943    }
944
945    #[test]
946    fn test_span_parent_child() {
947        let mut c = SpanCollector::new();
948        let parent = c.start(SpanKind::Pipeline, "main".into());
949        let child = c.start(SpanKind::FnCall, "helper".into());
950        c.end(child);
951        c.end(parent);
952        assert_eq!(c.spans().len(), 2);
953        assert_eq!(c.spans()[0].parent_id, Some(parent));
954        assert_eq!(c.spans()[1].parent_id, None);
955    }
956
957    #[test]
958    fn test_span_metadata() {
959        let mut c = SpanCollector::new();
960        let id = c.start(SpanKind::LlmCall, "gpt-4".into());
961        c.set_metadata(id, "tokens", serde_json::json!(100));
962        c.end(id);
963        assert_eq!(c.spans()[0].metadata["tokens"], serde_json::json!(100));
964    }
965
966    #[test]
967    fn test_completed_span_metadata_can_be_attached_late() {
968        let mut c = SpanCollector::new();
969        let id = c.start(SpanKind::LlmCall, "gpt-4".into());
970        c.end(id);
971
972        c.attach_metadata_if_absent(id, "first_token_ms", serde_json::json!(125));
973        c.attach_metadata_if_absent(id, "first_token_ms", serde_json::json!(250));
974
975        assert_eq!(
976            c.spans()[0].metadata["first_token_ms"],
977            serde_json::json!(125)
978        );
979    }
980
981    #[test]
982    fn test_span_links_are_preserved() {
983        let mut c = SpanCollector::new();
984        let parent = c.start(SpanKind::Suspension, "suspend worker".into());
985        let link = c.span_link(parent).expect("link for open span");
986        c.end(parent);
987
988        let child = c.start_with_links(SpanKind::Resume, "resume worker".into(), vec![link]);
989        c.end(child);
990
991        assert_eq!(c.spans().len(), 2);
992        assert_eq!(c.spans()[1].parent_id, None);
993        assert_eq!(c.spans()[1].links.len(), 1);
994        assert_eq!(c.spans()[1].links[0].span_id, parent.to_string());
995    }
996
997    #[test]
998    fn test_detached_span_links_do_not_inherit_active_parent() {
999        let mut c = SpanCollector::new();
1000        let pipeline = c.start(SpanKind::Pipeline, "pipeline".into());
1001        let link = c.span_link(pipeline).expect("pipeline link");
1002        let drain = c.start_detached_with_links(SpanKind::Drain, "drain".into(), vec![link]);
1003        c.end(drain);
1004        c.end(pipeline);
1005
1006        let drain = c
1007            .spans()
1008            .iter()
1009            .find(|span| span.kind == SpanKind::Drain)
1010            .expect("drain span");
1011        assert_eq!(drain.parent_id, None);
1012        assert_eq!(drain.links.len(), 1);
1013        assert_eq!(drain.links[0].span_id, pipeline.to_string());
1014    }
1015
1016    #[test]
1017    fn test_noop_when_disabled() {
1018        set_tracing_enabled(false);
1019        let id = span_start(SpanKind::Pipeline, "test".into());
1020        assert_eq!(id, 0);
1021        assert!(span_end(id).is_none());
1022    }
1023
1024    #[test]
1025    fn test_user_timing_records_when_tracing_disabled() {
1026        // UserTiming is the substrate behind `std/timing`. Script
1027        // callers depend on a real `duration_ms` even when global VM
1028        // tracing is off, so the collector must always record this
1029        // kind.
1030        set_tracing_enabled(false);
1031        reset_tracing();
1032        let mut attrs = BTreeMap::new();
1033        attrs.insert("phase".into(), serde_json::json!("warmup"));
1034        let (id, trace_id, parent, start_unix_ms) =
1035            span_start_user_timing("script.work".into(), attrs);
1036        assert!(id != 0);
1037        assert!(!trace_id.is_empty());
1038        assert_eq!(parent, None);
1039        assert!(start_unix_ms > 0);
1040
1041        assert!(span_record_event(id, "checkpoint".into(), BTreeMap::new()));
1042
1043        let closed = span_end(id).expect("user timing always records");
1044        assert_eq!(closed.kind, SpanKind::UserTiming);
1045        assert_eq!(closed.events.len(), 1);
1046        assert_eq!(closed.events[0].name, "checkpoint");
1047        assert_eq!(closed.metadata["phase"], serde_json::json!("warmup"));
1048
1049        // The recorded user_timing span survives in the collector
1050        // snapshot so `trace_spans()` / `harn run --profile-json`
1051        // surface it alongside the other VM-emitted spans.
1052        let snapshot = peek_spans();
1053        assert!(snapshot
1054            .iter()
1055            .any(|span| span.kind == SpanKind::UserTiming && span.name == "script.work"));
1056    }
1057
1058    #[test]
1059    fn test_new_span_kinds_stringify() {
1060        assert_eq!(SpanKind::ModelRoute.as_str(), "model_route");
1061        assert_eq!(SpanKind::ToolMount.as_str(), "tool_mount");
1062        assert_eq!(SpanKind::DeferredToolLoad.as_str(), "deferred_tool_load");
1063    }
1064
1065    #[test]
1066    fn test_llm_call_usage_metadata_pairs_carry_cache_tokens() {
1067        let usage = LlmCallUsage {
1068            model: "claude-sonnet-4".into(),
1069            provider: "anthropic".into(),
1070            input_tokens: 100,
1071            output_tokens: 20,
1072            cache_read_tokens: 40,
1073            cache_write_tokens: 8,
1074            cost_usd: Some(0.0123),
1075        };
1076        let pairs: BTreeMap<&str, serde_json::Value> = usage.metadata_pairs().into_iter().collect();
1077        assert_eq!(pairs[meta::MODEL], serde_json::json!("claude-sonnet-4"));
1078        assert_eq!(pairs[meta::PROVIDER], serde_json::json!("anthropic"));
1079        assert_eq!(pairs[meta::INPUT_TOKENS], serde_json::json!(100));
1080        assert_eq!(pairs[meta::OUTPUT_TOKENS], serde_json::json!(20));
1081        assert_eq!(pairs[meta::CACHE_READ_TOKENS], serde_json::json!(40));
1082        assert_eq!(pairs[meta::CACHE_WRITE_TOKENS], serde_json::json!(8));
1083        assert_eq!(pairs[meta::COST_USD], serde_json::json!(0.0123));
1084    }
1085
1086    #[test]
1087    fn test_llm_call_usage_omits_cost_when_unpriced() {
1088        let usage = LlmCallUsage {
1089            model: "local-model".into(),
1090            provider: "local".into(),
1091            input_tokens: 5,
1092            output_tokens: 1,
1093            cost_usd: None,
1094            ..LlmCallUsage::default()
1095        };
1096        let pairs: BTreeMap<&str, serde_json::Value> = usage.metadata_pairs().into_iter().collect();
1097        assert!(!pairs.contains_key(meta::COST_USD));
1098        // Token attribution is still present even when unpriced.
1099        assert_eq!(pairs[meta::INPUT_TOKENS], serde_json::json!(5));
1100    }
1101
1102    #[test]
1103    fn test_marker_spans_nest_under_active_span_and_carry_metadata() {
1104        set_tracing_enabled(true);
1105        reset_tracing();
1106        let parent = span_start(SpanKind::Pipeline, "agent_loop".into());
1107        emit_model_route("cheap-model", "smart-model", "no_progress");
1108        emit_tool_mount(
1109            &["read".to_string(), "write".to_string()],
1110            "mcp",
1111            Some("filesystem"),
1112        );
1113        emit_deferred_tool_load("grep", "search files", Some(4.5));
1114        span_end(parent);
1115
1116        let spans = peek_spans();
1117        let route = spans
1118            .iter()
1119            .find(|s| s.kind == SpanKind::ModelRoute)
1120            .expect("model_route span");
1121        assert_eq!(route.parent_id, Some(parent));
1122        assert_eq!(
1123            route.metadata[meta::TO_MODEL],
1124            serde_json::json!("smart-model")
1125        );
1126        assert_eq!(
1127            route.metadata[meta::FROM_MODEL],
1128            serde_json::json!("cheap-model")
1129        );
1130
1131        let mount = spans
1132            .iter()
1133            .find(|s| s.kind == SpanKind::ToolMount)
1134            .expect("tool_mount span");
1135        assert_eq!(mount.metadata[meta::TOOL_COUNT], serde_json::json!(2));
1136        assert_eq!(mount.metadata[meta::SOURCE], serde_json::json!("mcp"));
1137        assert_eq!(
1138            mount.metadata[meta::DETAIL],
1139            serde_json::json!("filesystem")
1140        );
1141
1142        let deferred = spans
1143            .iter()
1144            .find(|s| s.kind == SpanKind::DeferredToolLoad)
1145            .expect("deferred_tool_load span");
1146        assert_eq!(
1147            deferred.metadata[meta::TOOL_NAME],
1148            serde_json::json!("grep")
1149        );
1150        assert_eq!(deferred.metadata[meta::SCORE], serde_json::json!(4.5));
1151        set_tracing_enabled(false);
1152    }
1153
1154    #[test]
1155    fn test_empty_tool_mount_is_a_noop() {
1156        set_tracing_enabled(true);
1157        reset_tracing();
1158        let parent = span_start(SpanKind::Pipeline, "loop".into());
1159        emit_tool_mount(&[], "mcp", Some("empty-server"));
1160        span_end(parent);
1161        let spans = peek_spans();
1162        assert!(!spans.iter().any(|s| s.kind == SpanKind::ToolMount));
1163        set_tracing_enabled(false);
1164    }
1165
1166    #[test]
1167    fn test_span_event_offset_is_monotonic() {
1168        // Use a mock clock so the test is deterministic under any load and
1169        // requires zero wall-clock time. The mock is advanced by 10 ms
1170        // between the two events, guaranteeing a strictly-increasing offset
1171        // rather than relying on the OS scheduler to deliver >= 1 ms of
1172        // real elapsed time between the two `record_event` calls.
1173        let clock = crate::clock_mock::MockClock::at_wall_ms(1_000_000_000_000);
1174        let _guard = crate::clock_mock::install_override(clock.clone());
1175        let mut c = SpanCollector::new();
1176        let id = c.start(SpanKind::UserTiming, "outer".into());
1177        assert!(c.record_event(id, "before".into(), BTreeMap::new()));
1178        clock.advance_std_sync(std::time::Duration::from_millis(10));
1179        assert!(c.record_event(id, "after".into(), BTreeMap::new()));
1180        let closed = c.end(id).expect("open span");
1181        assert_eq!(closed.events.len(), 2);
1182        assert!(
1183            closed.events[1].offset_ms > closed.events[0].offset_ms,
1184            "second event should have a strictly greater offset after a 10ms advance; \
1185             before={} after={}",
1186            closed.events[0].offset_ms,
1187            closed.events[1].offset_ms
1188        );
1189    }
1190}