greentic_runner_host/trace/
recorder.rs

1use std::collections::VecDeque;
2use std::env;
3use std::fs;
4use std::path::{Path, PathBuf};
5use std::time::Instant;
6
7use anyhow::{Context, Result};
8use chrono::Utc;
9use greentic_types::TenantCtx;
10use parking_lot::Mutex;
11use rand::{RngExt, rng};
12use serde_json::Value;
13
14use crate::runner::engine::{ExecutionObserver, NodeEvent};
15use crate::validate::ValidationIssue;
16
17use super::audit_event::{NodeAuditRecord, Outcome, audit_subject, build_audit_event};
18use super::audit_sink::AuditSink;
19use super::model::{TraceEnvelope, TraceError, TraceFlow, TraceHash, TracePack, TraceStep};
20
21const DEFAULT_TRACE_FILE: &str = "trace.json";
22const DEFAULT_BUFFER_SIZE: usize = 20;
23const HASH_ALGORITHM: &str = "blake3";
24
25#[derive(Clone, Debug)]
26pub struct TraceConfig {
27    pub mode: TraceMode,
28    pub out_path: PathBuf,
29    pub buffer_size: usize,
30    pub capture_inputs: bool,
31}
32
33impl TraceConfig {
34    pub fn from_env() -> Self {
35        let out_path = env::var_os("GREENTIC_TRACE_OUT")
36            .map(PathBuf::from)
37            .unwrap_or_else(|| PathBuf::from(DEFAULT_TRACE_FILE));
38        Self {
39            mode: TraceMode::On,
40            out_path,
41            buffer_size: DEFAULT_BUFFER_SIZE,
42            capture_inputs: env::var("GREENTIC_TRACE_CAPTURE_INPUTS").ok().as_deref() == Some("1"),
43        }
44    }
45
46    pub fn with_overrides(mut self, mode: TraceMode, out_path: Option<PathBuf>) -> Self {
47        self.mode = mode;
48        if let Some(path) = out_path {
49            self.out_path = path;
50        }
51        self
52    }
53
54    pub fn with_capture_inputs(mut self, capture: bool) -> Self {
55        self.capture_inputs = capture;
56        self
57    }
58}
59
60#[derive(Copy, Clone, Debug, PartialEq, Eq)]
61pub enum TraceMode {
62    Off,
63    On,
64    Always,
65}
66
67#[derive(Clone, Debug)]
68pub struct PackTraceInfo {
69    pub pack_ref: String,
70    pub resolved_digest: Option<String>,
71}
72
73#[derive(Clone, Debug)]
74pub struct TraceContext {
75    pub pack_ref: String,
76    pub resolved_digest: Option<String>,
77    pub flow_id: String,
78    pub flow_version: String,
79}
80
81pub struct TraceRecorder {
82    config: TraceConfig,
83    context: TraceContext,
84    state: Mutex<TraceState>,
85    /// Best-effort audit publisher; `None` when no NATS client is available
86    /// (the default, off path — see `docs/superpowers/specs/2026-07-03-runner-audit-emitter-design.md`).
87    audit_sink: Option<AuditSink>,
88    /// The flow's `TenantCtx`, captured at construction time. The per-node
89    /// `NodeEvent.context.tenant` seen in `on_node_end`/`on_node_error` is a
90    /// bare `&str` (no env), so the full context needed to build the audit
91    /// `EventEnvelope.tenant` field is threaded in here instead, from the
92    /// construction site where it is already in scope (`src/engine/runtime.rs`).
93    audit_tenant: Option<TenantCtx>,
94}
95
96struct TraceState {
97    buffer: VecDeque<TraceStep>,
98    in_flight: Option<InFlightStep>,
99    flushed: bool,
100}
101
102struct InFlightStep {
103    node_id: String,
104    component_id: String,
105    operation: String,
106    input_hash: TraceHash,
107    started_at: Instant,
108    validation_issues: Vec<ValidationIssue>,
109    invocation_json: Option<Value>,
110}
111
112impl TraceRecorder {
113    pub fn new(config: TraceConfig, context: TraceContext) -> Self {
114        Self::new_with_audit(config, context, None, None)
115    }
116
117    /// Constructs a recorder that additionally fans out a best-effort audit
118    /// `EventEnvelope` to `audit_sink` on `on_node_end`/`on_node_error`, in
119    /// addition to the existing file-buffering (which is unchanged). Pass
120    /// `None` for both `audit_sink` and `audit_tenant` to keep the exact
121    /// existing (file-only) behavior — this is what `new` delegates to.
122    pub fn new_with_audit(
123        config: TraceConfig,
124        context: TraceContext,
125        audit_sink: Option<AuditSink>,
126        audit_tenant: Option<TenantCtx>,
127    ) -> Self {
128        Self {
129            config,
130            context,
131            state: Mutex::new(TraceState {
132                buffer: VecDeque::new(),
133                in_flight: None,
134                flushed: false,
135            }),
136            audit_sink,
137            audit_tenant,
138        }
139    }
140
141    pub fn mode(&self) -> TraceMode {
142        self.config.mode
143    }
144
145    pub fn flush_success(&self) -> Result<()> {
146        if self.config.mode != TraceMode::Always {
147            return Ok(());
148        }
149        self.flush_with_steps(None)
150    }
151
152    pub fn flush_error(&self, err: &dyn std::error::Error) -> Result<()> {
153        if self.config.mode == TraceMode::Off {
154            return Ok(());
155        }
156        self.flush_with_steps(Some(err))
157    }
158
159    pub fn flush_buffer(&self) -> Result<()> {
160        if self.config.mode == TraceMode::Off {
161            return Ok(());
162        }
163        self.flush_with_steps(None)
164    }
165
166    fn flush_with_steps(&self, fallback_error: Option<&dyn std::error::Error>) -> Result<()> {
167        let mut state = self.state.lock();
168        if state.flushed {
169            return Ok(());
170        }
171        if let Some(err) = fallback_error {
172            let step = if let Some(in_flight) = state.in_flight.take() {
173                TraceStep {
174                    node_id: in_flight.node_id,
175                    component_id: in_flight.component_id,
176                    operation: in_flight.operation,
177                    input_hash: in_flight.input_hash,
178                    invocation_json: in_flight.invocation_json,
179                    invocation_path: None,
180                    output_hash: None,
181                    state_delta_hash: None,
182                    duration_ms: in_flight.started_at.elapsed().as_millis() as u64,
183                    validation_issues: in_flight.validation_issues,
184                    error: Some(TraceError {
185                        code: "node_error".to_string(),
186                        message: err.to_string(),
187                        details: Value::Null,
188                    }),
189                }
190            } else {
191                TraceStep {
192                    node_id: "unknown".to_string(),
193                    component_id: "unknown".to_string(),
194                    operation: "unknown".to_string(),
195                    input_hash: hash_value(&Value::Null),
196                    invocation_json: None,
197                    invocation_path: None,
198                    output_hash: None,
199                    state_delta_hash: None,
200                    duration_ms: 0,
201                    validation_issues: Vec::new(),
202                    error: Some(TraceError {
203                        code: "flow_error".to_string(),
204                        message: err.to_string(),
205                        details: Value::Null,
206                    }),
207                }
208            };
209            state.buffer.push_back(step);
210            while state.buffer.len() > self.config.buffer_size {
211                state.buffer.pop_front();
212            }
213        }
214        let steps = state.buffer.iter().cloned().collect::<Vec<_>>();
215        state.flushed = true;
216        drop(state);
217        let trace = self.build_trace(steps);
218        write_trace_atomic(&self.config.out_path, &trace)?;
219        Ok(())
220    }
221
222    fn build_trace(&self, steps: Vec<TraceStep>) -> TraceEnvelope {
223        TraceEnvelope {
224            trace_version: 1,
225            runner_version: Some(env!("CARGO_PKG_VERSION").to_string()),
226            git_sha: git_sha(),
227            pack: TracePack {
228                pack_ref: self.context.pack_ref.clone(),
229                resolved_digest: self.context.resolved_digest.clone(),
230            },
231            flow: TraceFlow {
232                id: self.context.flow_id.clone(),
233                version: self.context.flow_version.clone(),
234            },
235            steps,
236        }
237    }
238
239    /// Best-effort audit fan-out for one completed node execution. No-op
240    /// when audit is disabled (`audit_sink`/`audit_tenant` are `None`), and
241    /// never blocks or fails execution — `AuditSink::emit` itself never
242    /// errors or panics.
243    fn emit_audit(&self, event: &NodeEvent<'_>, step: &TraceStep, outcome: Outcome) {
244        let (Some(sink), Some(tenant)) = (&self.audit_sink, &self.audit_tenant) else {
245            return;
246        };
247        let event_name = match outcome {
248            Outcome::Ok => "node_end",
249            Outcome::Error => "node_error",
250        };
251        let rec = NodeAuditRecord {
252            tenant,
253            flow_id: event.context.flow_id,
254            node_id: event.node_id,
255            component_id: &step.component_id,
256            operation: &step.operation,
257            session_id: event.context.session_id.unwrap_or_default(),
258            duration_ms: step.duration_ms,
259            outcome,
260            error: step.error.as_ref().map(|e| e.message.as_str()),
261        };
262        let envelope = build_audit_event(&rec, Utc::now(), generate_audit_event_id());
263        sink.emit(audit_subject(tenant.tenant.as_str(), event_name), &envelope);
264    }
265}
266
267impl ExecutionObserver for TraceRecorder {
268    fn on_node_start(&self, event: &NodeEvent<'_>) {
269        if self.config.mode == TraceMode::Off {
270            return;
271        }
272        let operation = event
273            .node
274            .operation_name()
275            .or_else(|| event.node.operation_in_mapping())
276            .unwrap_or("unknown")
277            .to_string();
278        let input_hash = hash_value(event.payload);
279        let component_id = event.node.component_id().to_string();
280        let mut state = self.state.lock();
281        state.in_flight = Some(InFlightStep {
282            node_id: event.node_id.to_string(),
283            component_id: component_id.clone(),
284            operation,
285            input_hash,
286            started_at: Instant::now(),
287            validation_issues: Vec::new(),
288            invocation_json: if self.config.capture_inputs {
289                Some(build_invocation(event, &component_id))
290            } else {
291                None
292            },
293        });
294    }
295
296    fn on_node_end(&self, event: &NodeEvent<'_>, output: &Value) {
297        if self.config.mode == TraceMode::Off {
298            return;
299        }
300        let output_hash = hash_value(output);
301        let mut state = self.state.lock();
302        let step = if let Some(in_flight) = state.in_flight.take() {
303            TraceStep {
304                node_id: in_flight.node_id,
305                component_id: in_flight.component_id,
306                operation: in_flight.operation,
307                input_hash: in_flight.input_hash,
308                invocation_json: in_flight.invocation_json,
309                invocation_path: None,
310                output_hash: Some(output_hash),
311                state_delta_hash: None,
312                duration_ms: in_flight.started_at.elapsed().as_millis() as u64,
313                validation_issues: in_flight.validation_issues,
314                error: None,
315            }
316        } else {
317            TraceStep {
318                node_id: event.node_id.to_string(),
319                component_id: event.node.component_id().to_string(),
320                operation: event.node.operation_name().unwrap_or("unknown").to_string(),
321                input_hash: hash_value(event.payload),
322                invocation_json: if self.config.capture_inputs {
323                    Some(build_invocation(event, event.node.component_id()))
324                } else {
325                    None
326                },
327                invocation_path: None,
328                output_hash: Some(output_hash),
329                state_delta_hash: None,
330                duration_ms: 0,
331                validation_issues: Vec::new(),
332                error: None,
333            }
334        };
335        self.emit_audit(event, &step, Outcome::Ok);
336        state.buffer.push_back(step);
337        while state.buffer.len() > self.config.buffer_size {
338            state.buffer.pop_front();
339        }
340    }
341
342    fn on_node_error(&self, event: &NodeEvent<'_>, error: &dyn std::error::Error) {
343        if self.config.mode == TraceMode::Off {
344            return;
345        }
346        let mut state = self.state.lock();
347        let step = if let Some(in_flight) = state.in_flight.take() {
348            TraceStep {
349                node_id: in_flight.node_id,
350                component_id: in_flight.component_id,
351                operation: in_flight.operation,
352                input_hash: in_flight.input_hash,
353                invocation_json: in_flight.invocation_json,
354                invocation_path: None,
355                output_hash: None,
356                state_delta_hash: None,
357                duration_ms: in_flight.started_at.elapsed().as_millis() as u64,
358                validation_issues: in_flight.validation_issues,
359                error: Some(TraceError {
360                    code: "node_error".to_string(),
361                    message: error.to_string(),
362                    details: Value::Null,
363                }),
364            }
365        } else {
366            TraceStep {
367                node_id: event.node_id.to_string(),
368                component_id: event.node.component_id().to_string(),
369                operation: event.node.operation_name().unwrap_or("unknown").to_string(),
370                input_hash: hash_value(event.payload),
371                invocation_json: if self.config.capture_inputs {
372                    Some(build_invocation(event, event.node.component_id()))
373                } else {
374                    None
375                },
376                invocation_path: None,
377                output_hash: None,
378                state_delta_hash: None,
379                duration_ms: 0,
380                validation_issues: Vec::new(),
381                error: Some(TraceError {
382                    code: "node_error".to_string(),
383                    message: error.to_string(),
384                    details: Value::Null,
385                }),
386            }
387        };
388        self.emit_audit(event, &step, Outcome::Error);
389        state.buffer.push_back(step);
390        while state.buffer.len() > self.config.buffer_size {
391            state.buffer.pop_front();
392        }
393        drop(state);
394        if let Err(err) = self.flush_buffer() {
395            tracing::warn!(error = %err, "failed to write trace");
396        }
397    }
398
399    fn on_validation(&self, _event: &NodeEvent<'_>, issues: &[ValidationIssue]) {
400        if self.config.mode == TraceMode::Off || issues.is_empty() {
401            return;
402        }
403        let mut state = self.state.lock();
404        if let Some(in_flight) = state.in_flight.as_mut() {
405            in_flight.validation_issues.extend_from_slice(issues);
406        }
407    }
408}
409
410fn hash_value(value: &Value) -> TraceHash {
411    let bytes = serde_json::to_vec(value).unwrap_or_default();
412    let digest = blake3::hash(&bytes).to_hex().to_string();
413    TraceHash {
414        algorithm: HASH_ALGORITHM.to_string(),
415        value: digest,
416    }
417}
418
419/// Generates a random hex id for the audit event's `id` field. No `uuid` dep
420/// exists in this crate; mirrors `runner::engine::generate_correlation_id`'s
421/// random-bytes-then-hex-encode pattern rather than adding one.
422///
423/// `pub(crate)` so `trace::agent_audit`'s `AgentAuditObserver` can reuse the
424/// same id source for agent-step audit events.
425pub(crate) fn generate_audit_event_id() -> String {
426    let mut bytes = [0u8; 16];
427    rng().fill(&mut bytes);
428    hex::encode(bytes)
429}
430
431fn build_invocation(event: &NodeEvent<'_>, component_id: &str) -> Value {
432    serde_json::json!({
433        "component_id": component_id,
434        "operation": event
435            .node
436            .operation_name()
437            .or_else(|| event.node.operation_in_mapping())
438            .unwrap_or("unknown")
439            .to_string(),
440        "payload": event.payload,
441    })
442}
443
444fn write_trace_atomic(path: &Path, trace: &TraceEnvelope) -> Result<()> {
445    if let Some(parent) = path.parent() {
446        fs::create_dir_all(parent)
447            .with_context(|| format!("failed to create {}", parent.display()))?;
448    }
449    let file_name = path
450        .file_name()
451        .and_then(|name| name.to_str())
452        .unwrap_or(DEFAULT_TRACE_FILE);
453    let tmp = path.with_file_name(format!("{file_name}.tmp"));
454    let payload = serde_json::to_vec_pretty(trace).context("serialize trace")?;
455    fs::write(&tmp, payload).with_context(|| format!("write {}", tmp.display()))?;
456    fs::rename(&tmp, path)
457        .with_context(|| format!("rename {} -> {}", tmp.display(), path.display()))?;
458    Ok(())
459}
460
461fn git_sha() -> Option<String> {
462    env::var("GIT_SHA")
463        .ok()
464        .or_else(|| env::var("GITHUB_SHA").ok())
465        .map(|value| value.trim().to_string())
466        .filter(|value| !value.is_empty())
467}
468
469#[cfg(test)]
470mod tests {
471    use super::*;
472    use crate::runner::engine::{FlowContext, HostNode, RetryConfig};
473    use greentic_types::{EnvId, TenantId};
474    use serde_json::json;
475    use tokio::sync::mpsc;
476
477    fn tenant_ctx() -> TenantCtx {
478        TenantCtx::new(
479            EnvId::try_from("prod").expect("valid env id"),
480            TenantId::try_from("t1").expect("valid tenant id"),
481        )
482    }
483
484    fn trace_config(out_path: PathBuf) -> TraceConfig {
485        TraceConfig {
486            mode: TraceMode::Always,
487            out_path,
488            buffer_size: 20,
489            capture_inputs: false,
490        }
491    }
492
493    fn trace_context() -> TraceContext {
494        TraceContext {
495            pack_ref: "pack1".to_string(),
496            resolved_digest: None,
497            flow_id: "flow1".to_string(),
498            flow_version: "1".to_string(),
499        }
500    }
501
502    fn flow_context(session_id: Option<&str>) -> FlowContext<'_> {
503        FlowContext {
504            tenant: "t1",
505            pack_id: "pack1",
506            flow_id: "flow1",
507            node_id: Some("n1"),
508            tool: None,
509            action: None,
510            session_id,
511            provider_id: None,
512            reply_scope: None,
513            retry_config: RetryConfig {
514                max_attempts: 1,
515                base_delay_ms: 1,
516            },
517            attempt: 1,
518            observer: None,
519            mocks: None,
520        }
521    }
522
523    fn audit_sink() -> (AuditSink, mpsc::Receiver<(String, Vec<u8>)>) {
524        let (tx, rx) = mpsc::channel::<(String, Vec<u8>)>(8);
525        (AuditSink::from_sender(tx), rx)
526    }
527
528    #[test]
529    fn on_node_end_emits_exactly_one_audit_event_with_correct_subject() {
530        let dir = tempfile::tempdir().unwrap();
531        let (sink, mut rx) = audit_sink();
532        let recorder = TraceRecorder::new_with_audit(
533            trace_config(dir.path().join("trace.json")),
534            trace_context(),
535            Some(sink),
536            Some(tenant_ctx()),
537        );
538
539        let ctx = flow_context(Some("s1"));
540        let node = HostNode::for_test("greentic:http", Some("call"));
541        let payload = json!({"foo": "bar"});
542        let event = NodeEvent {
543            context: &ctx,
544            node_id: "n1",
545            node: &node,
546            payload: &payload,
547        };
548        recorder.on_node_start(&event);
549        recorder.on_node_end(&event, &json!({"ok": true}));
550
551        let (subject, bytes) = rx.try_recv().expect("exactly one audit event enqueued");
552        assert_eq!(subject, "audit.t1.flow.node_end");
553        let value: Value = serde_json::from_slice(&bytes).expect("valid JSON");
554        assert_eq!(
555            value.get("type").and_then(Value::as_str),
556            Some("greentic.runner.flow.node_end")
557        );
558        assert!(
559            rx.try_recv().is_err(),
560            "expected exactly one audit event, found a second"
561        );
562    }
563
564    #[test]
565    fn on_node_start_emits_no_audit_event() {
566        let dir = tempfile::tempdir().unwrap();
567        let (sink, mut rx) = audit_sink();
568        let recorder = TraceRecorder::new_with_audit(
569            trace_config(dir.path().join("trace.json")),
570            trace_context(),
571            Some(sink),
572            Some(tenant_ctx()),
573        );
574
575        let ctx = flow_context(Some("s1"));
576        let node = HostNode::for_test("greentic:http", Some("call"));
577        let payload = json!({"foo": "bar"});
578        let event = NodeEvent {
579            context: &ctx,
580            node_id: "n1",
581            node: &node,
582            payload: &payload,
583        };
584        recorder.on_node_start(&event);
585
586        assert!(
587            rx.try_recv().is_err(),
588            "on_node_start must never emit an audit event"
589        );
590    }
591
592    #[test]
593    fn on_node_error_emits_one_audit_event_with_error_message() {
594        let dir = tempfile::tempdir().unwrap();
595        let (sink, mut rx) = audit_sink();
596        let recorder = TraceRecorder::new_with_audit(
597            trace_config(dir.path().join("trace.json")),
598            trace_context(),
599            Some(sink),
600            Some(tenant_ctx()),
601        );
602
603        let ctx = flow_context(Some("s1"));
604        let node = HostNode::for_test("greentic:http", Some("call"));
605        let payload = json!({"foo": "bar"});
606        let event = NodeEvent {
607            context: &ctx,
608            node_id: "n1",
609            node: &node,
610            payload: &payload,
611        };
612        recorder.on_node_start(&event);
613        let err = anyhow::anyhow!("boom");
614        recorder.on_node_error(&event, err.as_ref());
615
616        let (subject, bytes) = rx.try_recv().expect("one audit error event enqueued");
617        assert_eq!(subject, "audit.t1.flow.node_error");
618        let value: Value = serde_json::from_slice(&bytes).expect("valid JSON");
619        assert_eq!(
620            value.get("type").and_then(Value::as_str),
621            Some("greentic.runner.flow.node_error")
622        );
623        assert_eq!(
624            value
625                .get("payload")
626                .and_then(|p| p.get("error"))
627                .and_then(Value::as_str),
628            Some("boom")
629        );
630    }
631
632    #[test]
633    fn without_audit_sink_file_trace_still_recorded() {
634        let dir = tempfile::tempdir().unwrap();
635        let out_path = dir.path().join("trace.json");
636        let recorder = TraceRecorder::new(trace_config(out_path.clone()), trace_context());
637
638        let ctx = flow_context(Some("s1"));
639        let node = HostNode::for_test("greentic:http", Some("call"));
640        let payload = json!({"foo": "bar"});
641        let event = NodeEvent {
642            context: &ctx,
643            node_id: "n1",
644            node: &node,
645            payload: &payload,
646        };
647        recorder.on_node_start(&event);
648        recorder.on_node_end(&event, &json!({"ok": true}));
649        recorder.flush_success().expect("flush should succeed");
650
651        let written = fs::read_to_string(&out_path).expect("trace file was written");
652        assert!(written.contains("\"node_id\": \"n1\""));
653    }
654
655    #[test]
656    fn file_buffering_is_unchanged_with_audit_sink_attached() {
657        let dir = tempfile::tempdir().unwrap();
658        let out_path = dir.path().join("trace.json");
659        let (sink, _rx) = audit_sink();
660        let recorder = TraceRecorder::new_with_audit(
661            trace_config(out_path.clone()),
662            trace_context(),
663            Some(sink),
664            Some(tenant_ctx()),
665        );
666
667        let ctx = flow_context(Some("s1"));
668        let node = HostNode::for_test("greentic:http", Some("call"));
669        let payload = json!({"foo": "bar"});
670        let event = NodeEvent {
671            context: &ctx,
672            node_id: "n1",
673            node: &node,
674            payload: &payload,
675        };
676        recorder.on_node_start(&event);
677        recorder.on_node_end(&event, &json!({"ok": true}));
678        recorder.flush_success().expect("flush should succeed");
679
680        let written = fs::read_to_string(&out_path).expect("trace file was written");
681        assert!(written.contains("\"node_id\": \"n1\""));
682    }
683}
greentic_runner_host/trace/recorder.rs

greentic_runner_host/trace/
recorder.rs
