ncp_runtime/
trace.rs

1use std::io::Write;
2
3use sha2::{Digest, Sha256};
4
5use crate::result::BrickResult;
6
7/// Trace schema version — included in runtime_info and every invoke record.
8const TRACE_SCHEMA_VERSION: &str = "ncp-trace-0.1";
9
10// ── TraceSink trait ────────────────────────────────────────────────
11
12/// Pluggable trace emission. JsonlTraceWriter for CLI; NullTrace for bench.
13pub trait TraceSink: Send {
14    /// Returns false to skip all trace work (timestamp allocation, serialization).
15    /// NullTrace returns false; real writers return true (default).
16    fn enabled(&self) -> bool {
17        true
18    }
19
20    fn emit_runtime_info(&mut self, runtime_version: &str, wasmtime_version: &str, timestamp: &str);
21
22    // Rationale: invoke trace record carries every field of the §11.1 schema;
23    // the trait signature mirrors the schema 1:1 to keep call sites honest.
24    // Refactor into a TraceRecord struct in Phase 3B/3C.
25    #[allow(clippy::too_many_arguments)]
26    fn emit_invoke(
27        &mut self,
28        trace_id: &str,
29        session_id: &str,
30        step: u64,
31        graph_id: &str,
32        graph_version: &str,
33        brick_id: &str,
34        brick_version: &str,
35        bundle_digest: &str,
36        node_id: &str,
37        envelope_bytes: &[u8],
38        trigger_source_node_id: &str,
39        trigger_source_step: u64,
40        trigger_edge_id: &str,
41        trigger_routing_reason: &str,
42        result: &BrickResult,
43        result_bytes: Option<&[u8]>,
44        latency_ms: f64,
45        timestamp: &str,
46    );
47}
48
49/// No-op trace sink for benchmarks — zero overhead.
50pub struct NullTrace;
51
52impl TraceSink for NullTrace {
53    fn enabled(&self) -> bool {
54        false
55    }
56    fn emit_runtime_info(&mut self, _: &str, _: &str, _: &str) {}
57    fn emit_invoke(
58        &mut self,
59        _: &str,
60        _: &str,
61        _: u64,
62        _: &str,
63        _: &str,
64        _: &str,
65        _: &str,
66        _: &str,
67        _: &str,
68        _: &[u8],
69        _: &str,
70        _: u64,
71        _: &str,
72        _: &str,
73        _: &BrickResult,
74        _: Option<&[u8]>,
75        _: f64,
76        _: &str,
77    ) {
78    }
79}
80
81// ── JsonlTraceWriter ───────────────────────────────────────────────
82
83/// JSON Lines trace writer — emits to stderr or a file.
84pub struct JsonlTraceWriter {
85    dest: Box<dyn Write + Send>,
86}
87
88impl JsonlTraceWriter {
89    /// Create a trace writer that writes to stderr.
90    pub fn stderr() -> Self {
91        Self {
92            dest: Box::new(std::io::stderr()),
93        }
94    }
95
96    /// Create a trace writer that writes to a file.
97    pub fn file(path: &std::path::Path) -> std::io::Result<Self> {
98        let file = std::fs::File::create(path)?;
99        Ok(Self {
100            dest: Box::new(std::io::BufWriter::new(file)),
101        })
102    }
103
104    fn write_line(&mut self, value: &serde_json::Value) {
105        let _ = serde_json::to_writer(&mut self.dest, value);
106        let _ = self.dest.write_all(b"\n");
107    }
108}
109
110impl TraceSink for JsonlTraceWriter {
111    fn emit_runtime_info(
112        &mut self,
113        runtime_version: &str,
114        wasmtime_version: &str,
115        timestamp: &str,
116    ) {
117        let record = serde_json::json!({
118            "type": "runtime_info",
119            "trace_schema_version": TRACE_SCHEMA_VERSION,
120            "runtime_version": runtime_version,
121            "wasmtime_version": wasmtime_version,
122            "timestamp": timestamp,
123        });
124        self.write_line(&record);
125    }
126
127    fn emit_invoke(
128        &mut self,
129        trace_id: &str,
130        session_id: &str,
131        step: u64,
132        graph_id: &str,
133        graph_version: &str,
134        brick_id: &str,
135        brick_version: &str,
136        bundle_digest: &str,
137        node_id: &str,
138        envelope_bytes: &[u8],
139        trigger_source_node_id: &str,
140        trigger_source_step: u64,
141        trigger_edge_id: &str,
142        trigger_routing_reason: &str,
143        result: &BrickResult,
144        result_bytes: Option<&[u8]>,
145        latency_ms: f64,
146        timestamp: &str,
147    ) {
148        let input_hash = sha256_prefixed(envelope_bytes);
149        let output_hash = result_bytes.map(sha256_prefixed);
150        let result_len = result_bytes.map(|b| b.len());
151
152        let mut record = serde_json::json!({
153            "type": "invoke",
154            "trace_schema_version": TRACE_SCHEMA_VERSION,
155            "timestamp": timestamp,
156            "trace_id": trace_id,
157            "session_id": session_id,
158            "step": step,
159            "graph_id": graph_id,
160            "graph_version": graph_version,
161            "brick_id": brick_id,
162            "brick_version": brick_version,
163            "bundle_digest": bundle_digest,
164            "node_id": node_id,
165            "trigger": {
166                "source_node_id": trigger_source_node_id,
167                "source_step": trigger_source_step,
168                "edge_id": trigger_edge_id,
169                "routing_reason": trigger_routing_reason,
170            },
171            "input_hash": input_hash,
172            "carry_state_hash": serde_json::Value::Null,
173            "output_hash": output_hash,
174            "result_len": result_len,
175            "result_type": result.result_type(),
176            "latency_ms": latency_ms,
177        });
178
179        if let Some(error) = result.error() {
180            let map = record.as_object_mut().unwrap();
181            map.insert(
182                "error_class".into(),
183                serde_json::Value::String(error.error_class.clone()),
184            );
185        }
186
187        self.write_line(&record);
188    }
189}
190
191fn sha256_prefixed(data: &[u8]) -> String {
192    let mut hasher = Sha256::new();
193    hasher.update(data);
194    format!("sha256:{}", hex::encode(hasher.finalize()))
195}
ncp_runtime/trace.rs

ncp_runtime/
trace.rs
