harn_cli/commands/run/
reporting.rs

1//! Auxiliary run output, profiling, and provenance receipts.
2//!
3//! The execution path owns when these artifacts are emitted; this module owns
4//! their wire shapes and sinks so the runner itself stays focused on lifecycle.
5
6use std::fs;
7#[cfg(unix)]
8use std::io::{self, Write};
9use std::path::{Path, PathBuf};
10use std::sync::Arc;
11use std::time::{Duration, Instant};
12
13use serde::Serialize;
14
15use crate::commands::time::{self, PhaseRecord, RunTiming};
16use harn_vm::clock::{now_wall_ms, RealClock};
17use harn_vm::event_log::EventLog;
18
19use super::{RunAttestationOptions, RunProfileOptions};
20
21/// JSON event-stream configuration for `--json` runs.
22#[derive(Clone, Default)]
23pub struct RunJsonOptions {
24    /// Suppress `stdout` / `stderr` events. Transcript, tool, hook,
25    /// persona, and the terminal result/error events still flow.
26    pub quiet: bool,
27}
28
29/// Post-run summary configuration for `harn run --emit-summary-json`.
30#[derive(Clone, Debug)]
31pub struct RunSummaryOptions {
32    pub sink: RunJsonSink,
33}
34
35#[derive(Clone, Debug)]
36pub struct RunPhaseOptions {
37    pub sink: RunJsonSink,
38}
39
40#[derive(Clone, Debug)]
41pub struct RunRusageOptions {
42    pub sink: RunJsonSink,
43}
44
45#[derive(Clone, Debug, Default)]
46pub struct RunAuxOptions {
47    pub summary: Option<RunSummaryOptions>,
48    pub phase: Option<RunPhaseOptions>,
49    pub rusage: Option<RunRusageOptions>,
50}
51
52#[derive(Clone, Debug, Default)]
53pub struct RunControlOptions {
54    pub timeout: Option<Duration>,
55    pub defer_project_handlers: bool,
56}
57
58#[derive(Clone, Debug)]
59pub struct RunJsonSink {
60    pub target: RunJsonSinkTarget,
61    pub fd_flag: &'static str,
62}
63
64#[derive(Clone, Debug)]
65pub enum RunJsonSinkTarget {
66    /// Append the summary to the captured stderr buffer so it remains
67    /// terminal after all diagnostics that `run_file_with_skill_dirs`
68    /// flushes on return.
69    Stderr,
70    File(PathBuf),
71    Fd(i32),
72}
73
74#[derive(Serialize)]
75struct RunSummary<'a> {
76    schema_version: u32,
77    event: &'static str,
78    wall_time_ms: u64,
79    exit_code: i32,
80    llm: RunSummaryLlm,
81    #[serde(skip_serializing_if = "Option::is_none")]
82    profile: Option<&'a harn_vm::profile::RunProfile>,
83}
84
85#[derive(Serialize)]
86pub(super) struct RunSummaryLlm {
87    call_count: i64,
88    input_tokens: i64,
89    output_tokens: i64,
90    time_ms: i64,
91    cost_usd: f64,
92}
93
94pub const RUN_SUMMARY_SCHEMA_VERSION: u32 = 1;
95pub const RUN_PHASE_SCHEMA_VERSION: u32 = 2;
96pub const RUN_RUSAGE_SCHEMA_VERSION: u32 = 1;
97
98#[derive(Serialize)]
99struct RunPhaseEvent {
100    schema_version: u32,
101    event: &'static str,
102    phases: Vec<PhaseRecord>,
103}
104
105#[derive(Serialize)]
106struct RunRusageEvent {
107    schema_version: u32,
108    event: &'static str,
109    cpu_ms: u64,
110}
111
112fn run_summary_options_from_args(args: &crate::cli::RunArgs) -> Option<RunSummaryOptions> {
113    args.emit_summary_json.then(|| RunSummaryOptions {
114        sink: build_run_json_sink(args.summary_file.clone(), args.summary_fd, "--summary-fd"),
115    })
116}
117
118pub(crate) fn run_aux_options_from_args(args: &crate::cli::RunArgs) -> RunAuxOptions {
119    RunAuxOptions {
120        summary: run_summary_options_from_args(args),
121        phase: run_phase_options_from_args(args),
122        rusage: run_rusage_options_from_args(args),
123    }
124}
125
126pub(crate) fn run_control_options_from_args(args: &crate::cli::RunArgs) -> RunControlOptions {
127    RunControlOptions {
128        timeout: args.timeout,
129        defer_project_handlers: args.defer_project_handlers,
130    }
131}
132
133fn run_phase_options_from_args(args: &crate::cli::RunArgs) -> Option<RunPhaseOptions> {
134    args.emit_phase_json.then(|| RunPhaseOptions {
135        sink: build_run_json_sink(args.phase_file.clone(), args.phase_fd, "--phase-fd"),
136    })
137}
138
139fn run_rusage_options_from_args(args: &crate::cli::RunArgs) -> Option<RunRusageOptions> {
140    args.emit_rusage_json.then(|| RunRusageOptions {
141        sink: build_run_json_sink(args.rusage_file.clone(), args.rusage_fd, "--rusage-fd"),
142    })
143}
144
145fn build_run_json_sink(
146    file: Option<PathBuf>,
147    fd: Option<i32>,
148    fd_flag: &'static str,
149) -> RunJsonSink {
150    RunJsonSink {
151        target: if let Some(path) = file {
152            RunJsonSinkTarget::File(path)
153        } else if let Some(fd) = fd {
154            RunJsonSinkTarget::Fd(fd)
155        } else {
156            RunJsonSinkTarget::Stderr
157        },
158        fd_flag,
159    }
160}
161
162pub(super) fn render_and_persist_profile_rollup(
163    options: &RunProfileOptions,
164    profile: &harn_vm::profile::RunProfile,
165    stderr: &mut String,
166) -> Result<(), String> {
167    if options.text {
168        stderr.push_str(&harn_vm::profile::render(profile));
169    }
170    if let Some(path) = options.json_path.as_ref() {
171        if let Some(parent) = path.parent() {
172            if !parent.as_os_str().is_empty() {
173                fs::create_dir_all(parent)
174                    .map_err(|error| format!("create {}: {error}", parent.display()))?;
175            }
176        }
177        let json = serde_json::to_string_pretty(profile)
178            .map_err(|error| format!("serialize profile: {error}"))?;
179        fs::write(path, json).map_err(|error| format!("write {}: {error}", path.display()))?;
180    }
181    Ok(())
182}
183
184fn build_run_summary<'a>(
185    started: Instant,
186    exit_code: i32,
187    profile: Option<&'a harn_vm::profile::RunProfile>,
188    llm: RunSummaryLlm,
189) -> RunSummary<'a> {
190    RunSummary {
191        schema_version: RUN_SUMMARY_SCHEMA_VERSION,
192        event: "run_summary",
193        wall_time_ms: started.elapsed().as_millis().min(u128::from(u64::MAX)) as u64,
194        exit_code,
195        llm,
196        profile,
197    }
198}
199
200pub(super) fn run_summary_llm_snapshot() -> RunSummaryLlm {
201    let (input_tokens, output_tokens, time_ms, call_count) = harn_vm::llm::peek_trace_summary();
202    let cost_usd = harn_vm::llm::peek_total_cost();
203    RunSummaryLlm {
204        call_count,
205        input_tokens,
206        output_tokens,
207        time_ms,
208        cost_usd: if cost_usd.is_finite() { cost_usd } else { 0.0 },
209    }
210}
211
212pub(super) struct RunAuxEmission {
213    pub stderr: String,
214    pub exit_code: i32,
215    pub error: Option<String>,
216}
217
218#[allow(clippy::too_many_arguments)]
219pub(super) fn emit_run_aux_for_exit(
220    summary: Option<&RunSummaryOptions>,
221    phase: Option<&RunPhaseOptions>,
222    rusage: Option<&RunRusageOptions>,
223    started: Instant,
224    exit_code: i32,
225    profile: Option<&harn_vm::profile::RunProfile>,
226    llm: Option<RunSummaryLlm>,
227    timing: Option<&RunTiming>,
228    main_events: u64,
229    cpu_ms_total: Option<u64>,
230    json_mode: bool,
231    stderr: &mut String,
232) -> RunAuxEmission {
233    let mut aux_stderr = String::new();
234    let mut final_exit_code = exit_code;
235    let mut aux_error = None;
236    let aux_target = if json_mode { &mut aux_stderr } else { stderr };
237    let default_timing = RunTiming::default();
238    let timing = timing.unwrap_or(&default_timing);
239
240    if let Some(options) = summary {
241        let llm = llm.unwrap_or_else(run_summary_llm_snapshot);
242        let summary = build_run_summary(started, exit_code, profile, llm);
243        if let Err(error) = emit_raw_json_line(&options.sink, &summary, "run summary", aux_target) {
244            record_aux_error(
245                &mut final_exit_code,
246                &mut aux_error,
247                aux_target,
248                "run summary",
249                error,
250            );
251        }
252    }
253    if let Some(options) = phase {
254        let phase_event = RunPhaseEvent {
255            schema_version: RUN_PHASE_SCHEMA_VERSION,
256            event: "run_phase",
257            phases: time::build_phase_records(timing, main_events),
258        };
259        if let Err(error) = emit_raw_json_line(&options.sink, &phase_event, "run phase", aux_target)
260        {
261            record_aux_error(
262                &mut final_exit_code,
263                &mut aux_error,
264                aux_target,
265                "run phase",
266                error,
267            );
268        }
269    }
270    if let Some(options) = rusage {
271        let rusage_event = RunRusageEvent {
272            schema_version: RUN_RUSAGE_SCHEMA_VERSION,
273            event: "run_rusage",
274            cpu_ms: cpu_ms_total.unwrap_or(0),
275        };
276        if let Err(error) =
277            emit_raw_json_line(&options.sink, &rusage_event, "run rusage", aux_target)
278        {
279            record_aux_error(
280                &mut final_exit_code,
281                &mut aux_error,
282                aux_target,
283                "run rusage",
284                error,
285            );
286        }
287    }
288
289    RunAuxEmission {
290        stderr: aux_stderr,
291        exit_code: final_exit_code,
292        error: aux_error,
293    }
294}
295
296fn record_aux_error(
297    final_exit_code: &mut i32,
298    aux_error: &mut Option<String>,
299    stderr: &mut String,
300    label: &str,
301    error: String,
302) {
303    stderr.push_str(&format!("error: failed to emit {label}: {error}\n"));
304    if *final_exit_code == 0 {
305        *final_exit_code = 1;
306    }
307    if aux_error.is_none() {
308        *aux_error = Some(error);
309    }
310}
311
312fn emit_raw_json_line(
313    sink: &RunJsonSink,
314    value: &impl Serialize,
315    label: &str,
316    stderr: &mut String,
317) -> Result<(), String> {
318    let line =
319        serde_json::to_string(value).map_err(|error| format!("serialize {label}: {error}"))? + "\n";
320    match &sink.target {
321        RunJsonSinkTarget::Stderr => {
322            stderr.push_str(&line);
323            Ok(())
324        }
325        RunJsonSinkTarget::File(path) => write_raw_json_file(path, &line),
326        RunJsonSinkTarget::Fd(fd) => write_raw_json_fd(*fd, &line, sink.fd_flag),
327    }
328}
329
330fn write_raw_json_file(path: &Path, line: &str) -> Result<(), String> {
331    if let Some(parent) = path.parent() {
332        if !parent.as_os_str().is_empty() {
333            fs::create_dir_all(parent)
334                .map_err(|error| format!("create {}: {error}", parent.display()))?;
335        }
336    }
337    fs::write(path, line).map_err(|error| format!("write {}: {error}", path.display()))
338}
339
340#[cfg(unix)]
341fn write_raw_json_fd(fd: i32, line: &str, flag: &str) -> Result<(), String> {
342    use std::fs::File;
343    use std::os::unix::io::FromRawFd;
344
345    if fd < 0 {
346        return Err(format!("invalid {flag} {fd}: must be non-negative"));
347    }
348    let duped = unsafe { libc::dup(fd) };
349    if duped < 0 {
350        return Err(format!(
351            "duplicate {flag} {fd}: {}",
352            io::Error::last_os_error()
353        ));
354    }
355    let mut file = unsafe { File::from_raw_fd(duped) };
356    file.write_all(line.as_bytes())
357        .and_then(|_| file.flush())
358        .map_err(|error| format!("write {flag} {fd}: {error}"))
359}
360
361#[cfg(not(unix))]
362fn write_raw_json_fd(_fd: i32, _line: &str, flag: &str) -> Result<(), String> {
363    Err(format!("{flag} is only supported on Unix platforms"))
364}
365
366pub(super) async fn append_run_provenance_event(
367    log: &Arc<harn_vm::event_log::AnyEventLog>,
368    kind: &str,
369    payload: serde_json::Value,
370) {
371    let Ok(topic) = harn_vm::event_log::Topic::new("run.provenance") else {
372        return;
373    };
374    let _ = log
375        .append(&topic, harn_vm::event_log::LogEvent::new(kind, payload))
376        .await;
377}
378
379pub(super) async fn emit_run_attestation(
380    log: &Arc<harn_vm::event_log::AnyEventLog>,
381    path: &str,
382    store_base: &Path,
383    started_at_ms: i64,
384    exit_code: i32,
385    options: &RunAttestationOptions,
386    stderr: &mut String,
387) -> Result<(), String> {
388    let finished_at_ms = now_ms();
389    let status = if exit_code == 0 { "success" } else { "failure" };
390    append_run_provenance_event(
391        log,
392        "finished",
393        serde_json::json!({
394            "pipeline": path,
395            "status": status,
396            "exit_code": exit_code,
397        }),
398    )
399    .await;
400    log.flush()
401        .await
402        .map_err(|error| format!("failed to flush attestation event log: {error}"))?;
403    let secret_provider = harn_vm::secrets::configured_default_chain("harn.provenance")
404        .map_err(|error| format!("failed to configure provenance secrets: {error}"))?;
405    let (signing_key, key_id) =
406        harn_vm::load_or_generate_agent_signing_key(&secret_provider, options.agent_id.as_deref())
407            .await
408            .map_err(|error| format!("failed to load provenance signing key: {error}"))?;
409    let receipt = harn_vm::build_signed_receipt(
410        log,
411        harn_vm::ReceiptBuildOptions {
412            pipeline: path.to_string(),
413            status: status.to_string(),
414            started_at_ms,
415            finished_at_ms,
416            exit_code,
417            producer_name: "harn-cli".to_string(),
418            producer_version: env!("CARGO_PKG_VERSION").to_string(),
419        },
420        &signing_key,
421        key_id,
422    )
423    .await
424    .map_err(|error| format!("failed to build provenance receipt: {error}"))?;
425    let receipt_path = receipt_output_path(store_base, options, &receipt.receipt_id);
426    if let Some(parent) = receipt_path.parent() {
427        fs::create_dir_all(parent)
428            .map_err(|error| format!("failed to create {}: {error}", parent.display()))?;
429    }
430    let encoded = serde_json::to_vec_pretty(&receipt)
431        .map_err(|error| format!("failed to encode provenance receipt: {error}"))?;
432    fs::write(&receipt_path, encoded)
433        .map_err(|error| format!("failed to write {}: {error}", receipt_path.display()))?;
434    stderr.push_str(&format!("provenance receipt: {}\n", receipt_path.display()));
435    Ok(())
436}
437
438fn receipt_output_path(
439    store_base: &Path,
440    options: &RunAttestationOptions,
441    receipt_id: &str,
442) -> PathBuf {
443    if let Some(path) = options.receipt_out.as_ref() {
444        return path.clone();
445    }
446    harn_vm::runtime_paths::state_root(store_base)
447        .join("receipts")
448        .join(format!("{receipt_id}.json"))
449}
450
451pub(super) fn now_ms() -> i64 {
452    now_wall_ms(&RealClock::new())
453}
454
455/// Map a script's top-level return value to a process exit code.
456///
457/// - `int n`             → exit n (clamped to 0..=255)
458/// - `Result::Ok(_)`     → exit 0
459/// - `Result::Err(_)`    → exit 1
460/// - anything else       → exit 0
461pub(super) fn exit_code_from_return_value(value: &harn_vm::VmValue) -> i32 {
462    use harn_vm::VmValue;
463    match value {
464        VmValue::Int(n) => (*n).clamp(0, 255) as i32,
465        VmValue::EnumVariant(enum_variant) if enum_variant.is_variant("Result", "Err") => 1,
466        _ => 0,
467    }
468}
469
470pub(crate) fn render_trace_summary() -> String {
471    use std::fmt::Write;
472
473    let entries = harn_vm::llm::take_trace();
474    if entries.is_empty() {
475        return String::new();
476    }
477    let mut out = String::new();
478    let _ = writeln!(out, "\n\x1b[2m─── LLM trace ───\x1b[0m");
479    let mut total_input = 0i64;
480    let mut total_output = 0i64;
481    let mut total_ms = 0u64;
482    for (index, entry) in entries.iter().enumerate() {
483        let _ = writeln!(
484            out,
485            "  #{}: {} | {} in + {} out tokens | {} ms",
486            index + 1,
487            entry.model,
488            entry.input_tokens,
489            entry.output_tokens,
490            entry.duration_ms,
491        );
492        total_input += entry.input_tokens;
493        total_output += entry.output_tokens;
494        total_ms += entry.duration_ms;
495    }
496    let total_tokens = total_input + total_output;
497    // Rough cost estimate using Sonnet 4 pricing ($3/MTok in, $15/MTok out).
498    let cost = (total_input as f64 * 3.0 + total_output as f64 * 15.0) / 1_000_000.0;
499    let _ = writeln!(
500        out,
501        "  \x1b[1m{} call{}, {} tokens ({}in + {}out), {} ms, ~${:.4}\x1b[0m",
502        entries.len(),
503        if entries.len() == 1 { "" } else { "s" },
504        total_tokens,
505        total_input,
506        total_output,
507        total_ms,
508        cost,
509    );
510    out
511}
harn_cli/commands/run/reporting.rs

harn_cli/commands/run/
reporting.rs
