1use std::collections::HashSet;
2use std::fs;
3use std::io::{self, Write};
4use std::path::{Path, PathBuf};
5use std::process;
6use std::sync::atomic::{AtomicBool, Ordering};
7use std::sync::{Arc, Mutex};
8use std::time::{Duration, Instant};
9
10use harn_parser::DiagnosticSeverity;
11use harn_vm::event_log::EventLog;
12use serde::Serialize;
13
14use crate::commands::time::{self, PhaseRecord, RunTiming};
15use crate::package;
16use crate::skill_loader::{
17 canonicalize_cli_dirs, emit_loader_warnings, install_skills_global, load_skills,
18 SkillLoaderInputs,
19};
20
21mod eval_source;
22mod explain_cost;
23pub mod harnpack;
24mod interrupts;
25pub mod json_events;
26mod lifecycle;
27mod llm_mock;
28mod manifest_runtime;
29mod sandbox;
30
31use self::eval_source::create_eval_temp_file;
32pub(crate) use self::eval_source::prepare_eval_temp_file;
33#[cfg(test)]
34use self::eval_source::{eval_source_for_code, split_eval_header};
35use self::harnpack::{HarnpackError, HarnpackRunOptions, PreparedHarnpack};
36use self::interrupts::{
37 install_signal_shutdown_handler, start_run_deadline_watchdog, RunDeadlineGuard,
38};
39use self::json_events::NdjsonEmitter;
40pub use self::lifecycle::RunProfileOptions;
41use self::lifecycle::{RunExecution, TerminalRun};
42pub use self::llm_mock::*;
43pub(crate) use self::manifest_runtime::connect_mcp_servers;
44#[cfg(test)]
45use self::sandbox::default_run_capability_policy;
46pub use self::sandbox::RunSandboxOptions;
47use self::sandbox::{
48 default_run_workspace_root, install_run_sandbox_scope, run_sandbox_attestation,
49};
50
51#[derive(Clone, Default)]
53pub struct RunJsonOptions {
54 pub quiet: bool,
57}
58
59#[derive(Clone, Debug)]
61pub struct RunSummaryOptions {
62 pub sink: RunJsonSink,
63}
64
65#[derive(Clone, Debug)]
66pub struct RunPhaseOptions {
67 pub sink: RunJsonSink,
68}
69
70#[derive(Clone, Debug)]
71pub struct RunRusageOptions {
72 pub sink: RunJsonSink,
73}
74
75#[derive(Clone, Debug, Default)]
76pub struct RunAuxOptions {
77 pub summary: Option<RunSummaryOptions>,
78 pub phase: Option<RunPhaseOptions>,
79 pub rusage: Option<RunRusageOptions>,
80}
81
82#[derive(Clone, Debug, Default)]
83pub struct RunControlOptions {
84 pub timeout: Option<Duration>,
85}
86
87#[derive(Clone, Debug)]
88pub struct RunJsonSink {
89 pub target: RunJsonSinkTarget,
90 pub fd_flag: &'static str,
91}
92
93#[derive(Clone, Debug)]
94pub enum RunJsonSinkTarget {
95 Stderr,
99 File(PathBuf),
100 Fd(i32),
101}
102
103#[derive(Serialize)]
104struct RunSummary<'a> {
105 schema_version: u32,
106 event: &'static str,
107 wall_time_ms: u64,
108 exit_code: i32,
109 llm: RunSummaryLlm,
110 #[serde(skip_serializing_if = "Option::is_none")]
111 profile: Option<&'a harn_vm::profile::RunProfile>,
112}
113
114#[derive(Serialize)]
115struct RunSummaryLlm {
116 call_count: i64,
117 input_tokens: i64,
118 output_tokens: i64,
119 time_ms: i64,
120 cost_usd: f64,
121}
122
123pub const RUN_SUMMARY_SCHEMA_VERSION: u32 = 1;
124pub const RUN_PHASE_SCHEMA_VERSION: u32 = 2;
125pub const RUN_RUSAGE_SCHEMA_VERSION: u32 = 1;
126
127#[derive(Serialize)]
128struct RunPhaseEvent {
129 schema_version: u32,
130 event: &'static str,
131 phases: Vec<PhaseRecord>,
132}
133
134#[derive(Serialize)]
135struct RunRusageEvent {
136 schema_version: u32,
137 event: &'static str,
138 cpu_ms: u64,
139}
140
141pub(crate) fn run_summary_options_from_args(
142 args: &crate::cli::RunArgs,
143) -> Option<RunSummaryOptions> {
144 args.emit_summary_json.then(|| RunSummaryOptions {
145 sink: build_run_json_sink(args.summary_file.clone(), args.summary_fd, "--summary-fd"),
146 })
147}
148
149pub(crate) fn run_aux_options_from_args(args: &crate::cli::RunArgs) -> RunAuxOptions {
150 RunAuxOptions {
151 summary: run_summary_options_from_args(args),
152 phase: run_phase_options_from_args(args),
153 rusage: run_rusage_options_from_args(args),
154 }
155}
156
157pub(crate) fn run_control_options_from_args(args: &crate::cli::RunArgs) -> RunControlOptions {
158 RunControlOptions {
159 timeout: args.timeout,
160 }
161}
162
163pub(crate) fn run_phase_options_from_args(args: &crate::cli::RunArgs) -> Option<RunPhaseOptions> {
164 args.emit_phase_json.then(|| RunPhaseOptions {
165 sink: build_run_json_sink(args.phase_file.clone(), args.phase_fd, "--phase-fd"),
166 })
167}
168
169pub(crate) fn run_rusage_options_from_args(args: &crate::cli::RunArgs) -> Option<RunRusageOptions> {
170 args.emit_rusage_json.then(|| RunRusageOptions {
171 sink: build_run_json_sink(args.rusage_file.clone(), args.rusage_fd, "--rusage-fd"),
172 })
173}
174
175fn build_run_json_sink(
176 file: Option<PathBuf>,
177 fd: Option<i32>,
178 fd_flag: &'static str,
179) -> RunJsonSink {
180 RunJsonSink {
181 target: if let Some(path) = file {
182 RunJsonSinkTarget::File(path)
183 } else if let Some(fd) = fd {
184 RunJsonSinkTarget::Fd(fd)
185 } else {
186 RunJsonSinkTarget::Stderr
187 },
188 fd_flag,
189 }
190}
191
192pub(crate) enum RunFileMcpServeMode {
193 Stdio,
194 Http(Box<RunFileMcpServeHttp>),
195}
196
197pub(crate) struct RunFileMcpServeHttp {
198 pub options: harn_serve::McpHttpServeOptions,
199 pub auth_policy: harn_serve::AuthPolicy,
200}
201
202const CORE_BUILTINS: &[&str] = &[
204 "println",
205 "print",
206 "log",
207 "type_of",
208 "to_string",
209 "to_int",
210 "to_float",
211 "len",
212 "assert",
213 "assert_eq",
214 "assert_ne",
215 "json_parse",
216 "json_stringify",
217 "runtime_context",
218 "task_current",
219 "runtime_context_values",
220 "runtime_context_get",
221 "runtime_context_set",
222 "runtime_context_clear",
223];
224
225pub(crate) fn build_denied_builtins(
230 deny_csv: Option<&str>,
231 allow_csv: Option<&str>,
232) -> HashSet<String> {
233 if let Some(csv) = deny_csv {
234 csv.split(',')
235 .map(|s| s.trim().to_string())
236 .filter(|s| !s.is_empty())
237 .collect()
238 } else if let Some(csv) = allow_csv {
239 let allowed: HashSet<String> = csv
242 .split(',')
243 .map(|s| s.trim().to_string())
244 .filter(|s| !s.is_empty())
245 .collect();
246 let core: HashSet<&str> = CORE_BUILTINS.iter().copied().collect();
247
248 let mut tmp = harn_vm::Vm::new();
250 harn_vm::register_vm_stdlib(&mut tmp);
251 harn_vm::register_store_builtins(&mut tmp, std::path::Path::new("."));
252 harn_vm::register_metadata_builtins(&mut tmp, std::path::Path::new("."));
253
254 tmp.builtin_names()
255 .into_iter()
256 .filter(|name| !allowed.contains(name) && !core.contains(name.as_str()))
257 .collect()
258 } else {
259 HashSet::new()
260 }
261}
262
263pub(crate) struct LoadedChunk {
267 pub(crate) source: String,
268 pub(crate) chunk: harn_vm::Chunk,
269}
270
271pub(crate) fn compile_or_load_chunk_for_run(
283 path: &str,
284 stderr: &mut String,
285) -> Option<LoadedChunk> {
286 compile_or_load_chunk_with_timing(path, stderr, None)
287}
288
289#[allow(clippy::needless_option_as_deref)]
299pub(crate) fn compile_or_load_chunk_with_timing(
300 path: &str,
301 stderr: &mut String,
302 mut timing: Option<&mut RunTiming>,
303) -> Option<LoadedChunk> {
304 let source = match fs::read_to_string(path) {
305 Ok(s) => s,
306 Err(e) => {
307 stderr.push_str(&format!("Error reading {path}: {e}\n"));
308 return None;
309 }
310 };
311 if let Some(t) = timing.as_deref_mut() {
312 t.input_bytes = source.len() as u64;
313 }
314
315 let compile_phase_start = Instant::now();
316 let lookup = harn_vm::bytecode_cache::load(Path::new(path), &source);
317 if let Some(chunk) = lookup.chunk {
318 if let Some(t) = timing.as_deref_mut() {
319 t.cache_hit = true;
320 t.bytecode_compile = compile_phase_start.elapsed();
321 }
322 return Some(LoadedChunk { source, chunk });
323 }
324 if let Some(t) = timing.as_deref_mut() {
325 t.cache_hit = false;
326 }
327
328 let parse_start = Instant::now();
329 let program = parse_source_for_run(path, &source, stderr)?;
330 if let Some(t) = timing.as_deref_mut() {
331 t.parse = parse_start.elapsed();
332 }
333
334 let typecheck_start = Instant::now();
335 let mut had_type_error = false;
336 let type_diagnostics = match typecheck_with_imports(&program, Path::new(path), &source) {
337 Ok(diagnostics) => diagnostics,
338 Err(error) => {
339 stderr.push_str(&format!("error: {error}\n"));
340 return None;
341 }
342 };
343 for diag in &type_diagnostics {
344 let rendered = harn_parser::diagnostic::render_type_diagnostic(&source, path, diag);
345 if matches!(diag.severity, DiagnosticSeverity::Error) {
346 had_type_error = true;
347 }
348 stderr.push_str(&rendered);
349 }
350 if let Some(t) = timing.as_deref_mut() {
351 t.typecheck = typecheck_start.elapsed();
352 }
353 if had_type_error {
354 return None;
355 }
356
357 let compile_step_start = Instant::now();
358 let chunk = match crate::compiler_for_source(Path::new(path), &source).compile(&program) {
359 Ok(c) => c,
360 Err(e) => {
361 stderr.push_str(&format!("error: compile error: {e}\n"));
362 return None;
363 }
364 };
365
366 if let Err(err) = harn_vm::bytecode_cache::store(&lookup.key, &chunk) {
371 if std::env::var_os(crate::dispatch::CACHE_DEBUG_ENV).is_some() {
372 eprintln!("[harn] bytecode cache write skipped: {err}");
373 }
374 }
375 if let Some(t) = timing.as_deref_mut() {
376 t.bytecode_compile = compile_step_start.elapsed();
377 }
378
379 Some(LoadedChunk { source, chunk })
380}
381
382fn parse_source_for_run(
383 path: &str,
384 source: &str,
385 stderr: &mut String,
386) -> Option<Vec<harn_parser::SNode>> {
387 crate::ensure_builtin_signatures_installed();
388
389 let mut lexer = harn_lexer::Lexer::new(source);
390 let tokens = match lexer.tokenize() {
391 Ok(tokens) => tokens,
392 Err(error) => {
393 let diagnostic = harn_parser::diagnostic::render_diagnostic_with_code(
394 source,
395 path,
396 &error_span_from_lex(&error),
397 "error",
398 harn_parser::diagnostic::lexer_error_code(&error),
399 &error.to_string(),
400 Some("here"),
401 None,
402 );
403 stderr.push_str(&diagnostic);
404 return None;
405 }
406 };
407
408 let mut parser = harn_parser::Parser::new(tokens);
409 match parser.parse() {
410 Ok(program) => Some(program),
411 Err(error) => {
412 if parser.all_errors().is_empty() {
413 render_parse_error(path, source, &error, stderr);
414 } else {
415 for error in parser.all_errors() {
416 render_parse_error(path, source, error, stderr);
417 }
418 }
419 None
420 }
421 }
422}
423
424fn render_parse_error(
425 path: &str,
426 source: &str,
427 error: &harn_parser::ParserError,
428 stderr: &mut String,
429) {
430 let span = error_span_from_parse(error);
431 let diagnostic = harn_parser::diagnostic::render_diagnostic_with_code(
432 source,
433 path,
434 &span,
435 "error",
436 harn_parser::diagnostic::parser_error_code(error),
437 &harn_parser::diagnostic::parser_error_message(error),
438 Some(harn_parser::diagnostic::parser_error_label(error)),
439 harn_parser::diagnostic::parser_error_help(error),
440 );
441 stderr.push_str(&diagnostic);
442}
443
444fn error_span_from_lex(error: &harn_lexer::LexerError) -> harn_lexer::Span {
445 match error {
446 harn_lexer::LexerError::UnexpectedCharacter(_, span)
447 | harn_lexer::LexerError::UnterminatedString(span)
448 | harn_lexer::LexerError::IntegerLiteralOutOfRange(_, span)
449 | harn_lexer::LexerError::UnterminatedBlockComment(span) => *span,
450 }
451}
452
453fn error_span_from_parse(error: &harn_parser::ParserError) -> harn_lexer::Span {
454 match error {
455 harn_parser::ParserError::Unexpected { span, .. } => *span,
456 harn_parser::ParserError::UnexpectedEof { span, .. } => *span,
457 }
458}
459
460fn typecheck_with_imports(
465 program: &[harn_parser::SNode],
466 path: &Path,
467 source: &str,
468) -> Result<Vec<harn_parser::TypeDiagnostic>, String> {
469 package::ensure_dependencies_materialized(path)?;
470 let checker = crate::typecheck_imports::checker_with_resolved_imports(
471 harn_parser::TypeChecker::new(),
472 path,
473 );
474 Ok(checker.check_with_source(program, source))
475}
476
477#[derive(Clone, Debug, Default, PartialEq, Eq)]
478pub struct RunAttestationOptions {
479 pub receipt_out: Option<PathBuf>,
480 pub agent_id: Option<String>,
481}
482
483#[derive(Clone)]
484pub struct RunInterruptTokens {
485 pub cancel_token: Arc<AtomicBool>,
486 pub signal_token: Arc<Mutex<Option<String>>>,
487}
488
489struct ExecuteRunInputs<'a> {
490 path: &'a str,
491 trace: bool,
492 denied_builtins: HashSet<String>,
493 script_argv: Vec<String>,
494 skill_dirs_raw: Vec<String>,
495 llm_mock_mode: CliLlmMockMode,
496 attestation: Option<RunAttestationOptions>,
497 profile: RunProfileOptions,
498 sandbox: RunSandboxOptions,
499 interrupt_tokens: Option<RunInterruptTokens>,
500 json: Option<(RunJsonOptions, Box<dyn io::Write + Send>)>,
501 aux: RunAuxOptions,
502 timing: Option<&'a mut RunTiming>,
503 harnpack: HarnpackRunOptions,
504}
505
506#[derive(Clone, Debug, Default)]
510pub struct RunOutcome {
511 pub stdout: String,
512 pub stderr: String,
513 pub exit_code: i32,
514}
515
516pub(crate) async fn run_file(
517 path: &str,
518 trace: bool,
519 denied_builtins: HashSet<String>,
520 script_argv: Vec<String>,
521 llm_mock_mode: CliLlmMockMode,
522 attestation: Option<RunAttestationOptions>,
523 profile: RunProfileOptions,
524) {
525 let exit_code = run_file_with_skill_dirs(
526 path,
527 trace,
528 denied_builtins,
529 script_argv,
530 Vec::new(),
531 llm_mock_mode,
532 attestation,
533 profile,
534 RunSandboxOptions::default(),
535 None,
536 RunAuxOptions::default(),
537 RunControlOptions::default(),
538 HarnpackRunOptions::default(),
539 )
540 .await;
541 if exit_code != 0 {
542 process::exit(exit_code);
543 }
544}
545
546pub(crate) fn run_explain_cost_file_with_skill_dirs(path: &str) -> i32 {
547 let outcome = execute_explain_cost(path);
548 if !outcome.stderr.is_empty() {
549 io::stderr().write_all(outcome.stderr.as_bytes()).ok();
550 }
551 if !outcome.stdout.is_empty() {
552 io::stdout().write_all(outcome.stdout.as_bytes()).ok();
553 }
554 outcome.exit_code
555}
556
557#[allow(clippy::too_many_arguments)]
558pub(crate) async fn run_file_with_skill_dirs(
559 path: &str,
560 trace: bool,
561 denied_builtins: HashSet<String>,
562 script_argv: Vec<String>,
563 skill_dirs_raw: Vec<String>,
564 llm_mock_mode: CliLlmMockMode,
565 attestation: Option<RunAttestationOptions>,
566 profile: RunProfileOptions,
567 sandbox: RunSandboxOptions,
568 json: Option<RunJsonOptions>,
569 aux: RunAuxOptions,
570 control: RunControlOptions,
571 harnpack: HarnpackRunOptions,
572) -> i32 {
573 let interrupt_tokens = install_signal_shutdown_handler();
575 let deadline_guard = control
576 .timeout
577 .map(|timeout| start_run_deadline_watchdog(timeout, interrupt_tokens.clone()));
578
579 let _stdout_passthrough = StdoutPassthroughGuard::enable();
580 let json_with_stdout =
581 json.map(|opts| (opts, Box::new(io::stdout()) as Box<dyn io::Write + Send>));
582 let outcome = execute_run_inner(ExecuteRunInputs {
583 path,
584 trace,
585 denied_builtins,
586 script_argv,
587 skill_dirs_raw,
588 llm_mock_mode,
589 attestation,
590 profile,
591 sandbox,
592 interrupt_tokens: Some(interrupt_tokens.clone()),
593 json: json_with_stdout,
594 aux,
595 timing: None,
596 harnpack,
597 })
598 .await;
599 if let Some(guard) = &deadline_guard {
600 guard.finish();
601 }
602
603 if !outcome.stderr.is_empty() {
606 io::stderr().write_all(outcome.stderr.as_bytes()).ok();
607 }
608 if !outcome.stdout.is_empty() {
609 io::stdout().write_all(outcome.stdout.as_bytes()).ok();
610 }
611
612 let mut exit_code = outcome.exit_code;
613 if deadline_guard
614 .as_ref()
615 .is_some_and(RunDeadlineGuard::timed_out)
616 || (exit_code != 0 && interrupt_tokens.cancel_token.load(Ordering::SeqCst))
617 {
618 exit_code = 124;
619 }
620 exit_code
621}
622
623#[allow(clippy::too_many_arguments)]
624pub(crate) async fn run_resume_with_skill_dirs(
625 target: &str,
626 trace: bool,
627 denied_builtins: HashSet<String>,
628 resume_argv: Vec<String>,
629 skill_dirs_raw: Vec<String>,
630 llm_mock_mode: CliLlmMockMode,
631 attestation: Option<RunAttestationOptions>,
632 profile: RunProfileOptions,
633 sandbox: RunSandboxOptions,
634 json: Option<RunJsonOptions>,
635 aux: RunAuxOptions,
636 control: RunControlOptions,
637) -> i32 {
638 let source = r#"import { resume_agent, wait_agent } from "std/agent/workers"
639
640pipeline main(task) {
641 const input = if len(argv) > 1 {
642 argv[1]
643 } else {
644 nil
645 }
646 const handle = resume_agent(argv[0], input, true)
647 return wait_agent(handle)
648}
649"#;
650 let tmp = match create_eval_temp_file() {
651 Ok(tmp) => tmp,
652 Err(error) => {
653 eprintln!("error: {error}");
654 return 1;
655 }
656 };
657 let tmp_path = tmp.path().to_path_buf();
658 if let Err(error) = fs::write(&tmp_path, source) {
659 eprintln!("error: failed to write temp file for --resume: {error}");
660 return 1;
661 }
662 let mut argv = Vec::with_capacity(resume_argv.len() + 1);
663 argv.push(target.to_string());
664 argv.extend(resume_argv);
665 let tmp_str = tmp_path.to_string_lossy().into_owned();
666 run_file_with_skill_dirs(
667 &tmp_str,
668 trace,
669 denied_builtins,
670 argv,
671 skill_dirs_raw,
672 llm_mock_mode,
673 attestation,
674 profile,
675 sandbox,
676 json,
677 aux,
678 control,
679 HarnpackRunOptions::default(),
680 )
681 .await
682}
683
684pub fn execute_explain_cost(path: &str) -> RunOutcome {
685 let stdout = String::new();
686 let mut stderr = String::new();
687
688 let source = match fs::read_to_string(path) {
689 Ok(source) => source,
690 Err(error) => {
691 stderr.push_str(&format!("Error reading {path}: {error}\n"));
692 return RunOutcome {
693 stdout,
694 stderr,
695 exit_code: 1,
696 };
697 }
698 };
699 let program = match parse_source_for_run(path, &source, &mut stderr) {
700 Some(program) => program,
701 None => {
702 return RunOutcome {
703 stdout,
704 stderr,
705 exit_code: 1,
706 };
707 }
708 };
709
710 let mut had_type_error = false;
711 let type_diagnostics = match typecheck_with_imports(&program, Path::new(path), &source) {
712 Ok(diagnostics) => diagnostics,
713 Err(error) => {
714 stderr.push_str(&format!("error: {error}\n"));
715 return RunOutcome {
716 stdout,
717 stderr,
718 exit_code: 1,
719 };
720 }
721 };
722 for diag in &type_diagnostics {
723 let rendered = harn_parser::diagnostic::render_type_diagnostic(&source, path, diag);
724 if matches!(diag.severity, DiagnosticSeverity::Error) {
725 had_type_error = true;
726 }
727 stderr.push_str(&rendered);
728 }
729 if had_type_error {
730 return RunOutcome {
731 stdout,
732 stderr,
733 exit_code: 1,
734 };
735 }
736
737 let extensions = package::load_runtime_extensions(Path::new(path));
738 package::install_runtime_extensions(&extensions);
739 RunOutcome {
740 stdout: explain_cost::render_explain_cost(path, &program),
741 stderr,
742 exit_code: 0,
743 }
744}
745
746pub(crate) struct StdoutPassthroughGuard {
747 previous: bool,
748}
749
750impl StdoutPassthroughGuard {
751 pub(crate) fn enable() -> Self {
752 Self {
753 previous: harn_vm::set_stdout_passthrough(true),
754 }
755 }
756}
757
758impl Drop for StdoutPassthroughGuard {
759 fn drop(&mut self) {
760 harn_vm::set_stdout_passthrough(self.previous);
761 }
762}
763
764pub async fn execute_run(
778 path: &str,
779 trace: bool,
780 denied_builtins: HashSet<String>,
781 script_argv: Vec<String>,
782 skill_dirs_raw: Vec<String>,
783 llm_mock_mode: CliLlmMockMode,
784 attestation: Option<RunAttestationOptions>,
785 profile: RunProfileOptions,
786) -> RunOutcome {
787 crate::ensure_builtin_signatures_installed();
788 execute_run_with_harnpack_and_sandbox_options(
789 path,
790 trace,
791 denied_builtins,
792 script_argv,
793 skill_dirs_raw,
794 llm_mock_mode,
795 attestation,
796 profile,
797 RunSandboxOptions::default(),
798 HarnpackRunOptions::default(),
799 )
800 .await
801}
802
803#[allow(clippy::too_many_arguments)]
807pub async fn execute_run_with_sandbox_options(
808 path: &str,
809 trace: bool,
810 denied_builtins: HashSet<String>,
811 script_argv: Vec<String>,
812 skill_dirs_raw: Vec<String>,
813 llm_mock_mode: CliLlmMockMode,
814 attestation: Option<RunAttestationOptions>,
815 profile: RunProfileOptions,
816 sandbox: RunSandboxOptions,
817) -> RunOutcome {
818 execute_run_with_harnpack_and_sandbox_options(
819 path,
820 trace,
821 denied_builtins,
822 script_argv,
823 skill_dirs_raw,
824 llm_mock_mode,
825 attestation,
826 profile,
827 sandbox,
828 HarnpackRunOptions::default(),
829 )
830 .await
831}
832
833#[allow(clippy::too_many_arguments)]
838pub async fn execute_run_with_harnpack_options(
839 path: &str,
840 trace: bool,
841 denied_builtins: HashSet<String>,
842 script_argv: Vec<String>,
843 skill_dirs_raw: Vec<String>,
844 llm_mock_mode: CliLlmMockMode,
845 attestation: Option<RunAttestationOptions>,
846 profile: RunProfileOptions,
847 harnpack: HarnpackRunOptions,
848) -> RunOutcome {
849 execute_run_with_harnpack_and_sandbox_options(
850 path,
851 trace,
852 denied_builtins,
853 script_argv,
854 skill_dirs_raw,
855 llm_mock_mode,
856 attestation,
857 profile,
858 RunSandboxOptions::default(),
859 harnpack,
860 )
861 .await
862}
863
864#[allow(clippy::too_many_arguments)]
865async fn execute_run_with_harnpack_and_sandbox_options(
866 path: &str,
867 trace: bool,
868 denied_builtins: HashSet<String>,
869 script_argv: Vec<String>,
870 skill_dirs_raw: Vec<String>,
871 llm_mock_mode: CliLlmMockMode,
872 attestation: Option<RunAttestationOptions>,
873 profile: RunProfileOptions,
874 sandbox: RunSandboxOptions,
875 harnpack: HarnpackRunOptions,
876) -> RunOutcome {
877 execute_run_inner(ExecuteRunInputs {
878 path,
879 trace,
880 denied_builtins,
881 script_argv,
882 skill_dirs_raw,
883 llm_mock_mode,
884 attestation,
885 profile,
886 sandbox,
887 interrupt_tokens: None,
888 json: None,
889 aux: RunAuxOptions::default(),
890 timing: None,
891 harnpack,
892 })
893 .await
894}
895
896#[allow(clippy::too_many_arguments)]
902pub async fn execute_run_json(
903 path: &str,
904 trace: bool,
905 denied_builtins: HashSet<String>,
906 script_argv: Vec<String>,
907 skill_dirs_raw: Vec<String>,
908 llm_mock_mode: CliLlmMockMode,
909 attestation: Option<RunAttestationOptions>,
910 profile: RunProfileOptions,
911 out: Box<dyn io::Write + Send>,
912 options: RunJsonOptions,
913) -> RunOutcome {
914 execute_run_inner(ExecuteRunInputs {
915 path,
916 trace,
917 denied_builtins,
918 script_argv,
919 skill_dirs_raw,
920 llm_mock_mode,
921 attestation,
922 profile,
923 sandbox: RunSandboxOptions::default(),
924 interrupt_tokens: None,
925 json: Some((options, out)),
926 aux: RunAuxOptions::default(),
927 timing: None,
928 harnpack: HarnpackRunOptions::default(),
929 })
930 .await
931}
932
933pub(crate) async fn execute_run_with_timing(
937 path: &str,
938 script_argv: Vec<String>,
939 timing: Option<&mut RunTiming>,
940 sandbox: RunSandboxOptions,
941) -> RunOutcome {
942 execute_run_inner(ExecuteRunInputs {
943 path,
944 trace: false,
945 denied_builtins: HashSet::new(),
946 script_argv,
947 skill_dirs_raw: Vec::new(),
948 llm_mock_mode: CliLlmMockMode::Off,
949 attestation: None,
950 profile: RunProfileOptions::default(),
951 sandbox,
952 interrupt_tokens: None,
953 json: None,
954 aux: RunAuxOptions::default(),
955 timing,
956 harnpack: HarnpackRunOptions::default(),
957 })
958 .await
959}
960
961fn entry_source_dir(path: &str) -> std::path::PathBuf {
975 match std::path::Path::new(path).parent() {
976 Some(parent) if !parent.as_os_str().is_empty() => parent.to_path_buf(),
977 _ => std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")),
978 }
979}
980
981#[allow(clippy::needless_option_as_deref)]
984async fn execute_run_inner(inputs: ExecuteRunInputs<'_>) -> RunOutcome {
985 let ExecuteRunInputs {
986 path,
987 trace,
988 denied_builtins,
989 script_argv,
990 skill_dirs_raw,
991 llm_mock_mode,
992 attestation,
993 profile,
994 sandbox,
995 interrupt_tokens,
996 json,
997 aux,
998 timing,
999 harnpack,
1000 } = inputs;
1001 let RunAuxOptions {
1002 summary,
1003 phase,
1004 rusage,
1005 } = aux;
1006 let run_started = Instant::now();
1007 let cpu_started_ms = rusage.as_ref().map(|_| time::cpu_ms());
1008 let mut owned_timing = if timing.is_none() && (phase.is_some() || rusage.is_some()) {
1009 Some(RunTiming::default())
1010 } else {
1011 None
1012 };
1013 let mut timing = timing.or(owned_timing.as_mut());
1014
1015 let json_session = json.map(|(options, out)| JsonRunSession::install(options, out));
1021
1022 let mut stderr = String::new();
1023 let mut stdout = String::new();
1024
1025 let owned_run_path: String;
1030 let resolved_path: &str = if harnpack::looks_like_harnpack(Path::new(path)) {
1031 let outcome = match harnpack::prepare_harnpack(Path::new(path), &harnpack, &mut stderr) {
1032 Ok(prepared) => prepared,
1033 Err(err) => {
1034 return finalize_harnpack_error(
1035 stderr,
1036 json_session,
1037 summary.as_ref(),
1038 phase.as_ref(),
1039 rusage.as_ref(),
1040 run_started,
1041 err,
1042 );
1043 }
1044 };
1045 harn_vm::run_events::emit(harn_vm::run_events::RunEvent::PackRun {
1046 bundle_hash: outcome.bundle_hash.clone(),
1047 signature_verified: outcome.signature_verified,
1048 key_id: outcome.key_id.clone(),
1049 cache_hit: outcome.cache_hit,
1050 dry_run_verify: harnpack.dry_run_verify,
1051 });
1052 if harnpack.dry_run_verify {
1053 return finalize_harnpack_dry_run(
1054 stderr,
1055 json_session,
1056 summary.as_ref(),
1057 phase.as_ref(),
1058 rusage.as_ref(),
1059 run_started,
1060 cpu_started_ms.map(|start| time::cpu_ms().saturating_sub(start)),
1061 &outcome,
1062 );
1063 }
1064 owned_run_path = outcome.entrypoint_path.to_string_lossy().into_owned();
1065 owned_run_path.as_str()
1066 } else {
1067 path
1068 };
1069
1070 let Some(LoadedChunk { source, chunk }) =
1071 compile_or_load_chunk_with_timing(resolved_path, &mut stderr, timing.as_deref_mut())
1072 else {
1073 let message = stderr.clone();
1074 return finalize_run_error(
1075 stdout,
1076 stderr,
1077 json_session,
1078 summary.as_ref(),
1079 phase.as_ref(),
1080 rusage.as_ref(),
1081 run_started,
1082 None,
1083 timing.as_deref(),
1084 0,
1085 cpu_started_ms.map(|start| time::cpu_ms().saturating_sub(start)),
1086 "compile_error",
1087 message,
1088 );
1089 };
1090 let path = resolved_path;
1091
1092 let setup_start = Instant::now();
1093 if trace || summary.is_some() {
1094 harn_vm::llm::enable_tracing();
1095 }
1096 if profile.is_enabled() || phase.is_some() {
1097 harn_vm::tracing::set_tracing_enabled(true);
1098 }
1099 if profile.is_enabled() {
1100 harn_vm::builtin_profile::enable();
1105 }
1106 if let Err(error) = install_cli_llm_mock_mode(&llm_mock_mode) {
1107 stderr.push_str(&format!("error: {error}\n"));
1108 time::record_run_setup_elapsed(timing.as_deref_mut(), setup_start);
1109 return finalize_run_error(
1110 stdout,
1111 stderr,
1112 json_session,
1113 summary.as_ref(),
1114 phase.as_ref(),
1115 rusage.as_ref(),
1116 run_started,
1117 None,
1118 timing.as_deref(),
1119 0,
1120 cpu_started_ms.map(|start| time::cpu_ms().saturating_sub(start)),
1121 "llm_mock_install",
1122 error,
1123 );
1124 }
1125
1126 let mut vm = harn_vm::Vm::new();
1127 if let Some(timing) = timing.as_deref_mut() {
1128 timing.module_phases = Some(vm.enable_module_phase_timing());
1129 }
1130 if let Some(interrupt_tokens) = interrupt_tokens {
1131 vm.install_interrupt_signal_token(interrupt_tokens.signal_token);
1132 vm.install_cancel_token(interrupt_tokens.cancel_token);
1133 }
1134 harn_vm::register_vm_stdlib(&mut vm);
1135 crate::install_default_hostlib(&mut vm);
1136 let source_parent = std::path::Path::new(path)
1137 .parent()
1138 .unwrap_or(std::path::Path::new("."));
1139 let project_root = harn_vm::stdlib::process::find_project_root(source_parent);
1141 let store_base = project_root.as_deref().unwrap_or(source_parent);
1142 let sandbox_root = sandbox
1143 .workspace_root
1144 .clone()
1145 .unwrap_or_else(|| default_run_workspace_root(project_root.as_deref(), source_parent));
1146 let _sandbox_scope = install_run_sandbox_scope(&sandbox, &sandbox_root, &mut stderr);
1147 let attestation_started_at_ms = now_ms();
1148 let attestation_log = if attestation.is_some() {
1149 Some(harn_vm::event_log::install_memory_for_current_thread(256))
1150 } else {
1151 None
1152 };
1153 if let Some(log) = attestation_log.as_ref() {
1154 append_run_provenance_event(
1155 log,
1156 "started",
1157 serde_json::json!({
1158 "pipeline": path,
1159 "argv": &script_argv,
1160 "project_root": store_base.display().to_string(),
1161 "sandbox": run_sandbox_attestation(&sandbox),
1162 }),
1163 )
1164 .await;
1165 }
1166 harn_vm::register_store_builtins(&mut vm, store_base);
1167 harn_vm::register_metadata_builtins(&mut vm, store_base);
1168 let pipeline_name = std::path::Path::new(path)
1169 .file_stem()
1170 .and_then(|s| s.to_str())
1171 .unwrap_or("default");
1172 harn_vm::register_checkpoint_builtins(&mut vm, store_base, pipeline_name);
1173 vm.set_source_info(path, &source);
1174 let lazy_manifest_handlers = !denied_builtins.is_empty();
1175 if lazy_manifest_handlers {
1176 vm.set_denied_builtins(denied_builtins);
1177 }
1178 if let Some(ref root) = project_root {
1179 vm.set_project_root(root);
1180 }
1181
1182 vm.set_source_dir(&entry_source_dir(path));
1187
1188 let cli_dirs = canonicalize_cli_dirs(&skill_dirs_raw, None);
1191 let loaded = load_skills(&SkillLoaderInputs {
1192 cli_dirs,
1193 source_path: Some(std::path::PathBuf::from(path)),
1194 });
1195 emit_loader_warnings(&loaded.loader_warnings);
1196 install_skills_global(&mut vm, &loaded);
1197
1198 let argv_values: Vec<harn_vm::VmValue> = script_argv
1201 .iter()
1202 .map(|s| harn_vm::VmValue::String(arcstr::ArcStr::from(s.as_str())))
1203 .collect();
1204 vm.set_global(
1205 "argv",
1206 harn_vm::VmValue::List(std::sync::Arc::new(argv_values)),
1207 );
1208
1209 let runtime_harness =
1213 match crate::default_harness_for_manifest_or_base_dir(Path::new(path), store_base) {
1214 Ok(harness) => harness,
1215 Err(error) => {
1216 stderr.push_str(&format!(
1217 "error: failed to configure harness secret provider: {error}\n"
1218 ));
1219 time::record_run_setup_elapsed(timing.as_deref_mut(), setup_start);
1220 return finalize_run_error(
1221 stdout,
1222 stderr,
1223 json_session,
1224 summary.as_ref(),
1225 phase.as_ref(),
1226 rusage.as_ref(),
1227 run_started,
1228 None,
1229 timing.as_deref(),
1230 0,
1231 cpu_started_ms.map(|start| time::cpu_ms().saturating_sub(start)),
1232 "harness_secret_provider",
1233 error,
1234 );
1235 }
1236 };
1237 vm.set_harness(runtime_harness);
1238
1239 if let Err(error) =
1242 manifest_runtime::install_manifest_runtime(Path::new(path), &mut vm, lazy_manifest_handlers)
1243 .await
1244 {
1245 stderr.push_str(&format!(
1246 "error: failed to install {}: {error}\n",
1247 error.label()
1248 ));
1249 time::record_run_setup_elapsed(timing.as_deref_mut(), setup_start);
1250 return finalize_run_error(
1251 stdout,
1252 stderr,
1253 json_session,
1254 summary.as_ref(),
1255 phase.as_ref(),
1256 rusage.as_ref(),
1257 run_started,
1258 None,
1259 timing.as_deref(),
1260 0,
1261 cpu_started_ms.map(|start| time::cpu_ms().saturating_sub(start)),
1262 error.phase(),
1263 error.to_string(),
1264 );
1265 }
1266
1267 let local = tokio::task::LocalSet::new();
1268 time::record_run_setup_elapsed(timing.as_deref_mut(), setup_start);
1269 let main_start = Instant::now();
1270 vm.set_source_dir(&entry_source_dir(path));
1278 let execution = local
1279 .run_until(async {
1280 match vm.execute(&chunk).await {
1281 Ok(value) => RunExecution::Terminal(TerminalRun::Returned(value)),
1282 Err(error) => match error.process_exit_code() {
1283 Some(code) => RunExecution::Terminal(TerminalRun::ProcessExited(code)),
1284 None => RunExecution::Failed(vm.format_runtime_error(&error)),
1285 },
1286 }
1287 })
1288 .await;
1289 let output = vm.output();
1290 if let Some(t) = timing.as_deref_mut() {
1291 t.run_main = main_start.elapsed();
1292 }
1293 if let Err(error) = persist_cli_llm_mock_recording(&llm_mock_mode) {
1294 stderr.push_str(&format!("error: {error}\n"));
1295 let profile_rollup = if profile.is_enabled() {
1296 Some(harn_vm::profile::build(&harn_vm::tracing::peek_spans()))
1297 } else {
1298 None
1299 };
1300 return finalize_run_error(
1301 stdout,
1302 stderr,
1303 json_session,
1304 summary.as_ref(),
1305 phase.as_ref(),
1306 rusage.as_ref(),
1307 run_started,
1308 profile_rollup.as_ref(),
1309 timing.as_deref(),
1310 harn_vm::tracing::peek_spans().len() as u64,
1311 cpu_started_ms.map(|start| time::cpu_ms().saturating_sub(start)),
1312 "llm_mock_record",
1313 error,
1314 );
1315 }
1316
1317 let buffered_stderr = harn_vm::take_stderr_buffer();
1319 stderr.push_str(&buffered_stderr);
1320
1321 let exit_code = match &execution {
1322 RunExecution::Terminal(terminal) => terminal.exit_code(),
1323 RunExecution::Failed(_) => 1,
1324 };
1325
1326 if let (Some(options), Some(log)) = (attestation.as_ref(), attestation_log.as_ref()) {
1327 if let Err(error) = emit_run_attestation(
1328 log,
1329 path,
1330 store_base,
1331 attestation_started_at_ms,
1332 exit_code,
1333 options,
1334 &mut stderr,
1335 )
1336 .await
1337 {
1338 stderr.push_str(&format!(
1339 "error: failed to emit provenance receipt: {error}\n"
1340 ));
1341 let profile_rollup = if profile.is_enabled() {
1342 Some(harn_vm::profile::build(&harn_vm::tracing::peek_spans()))
1343 } else {
1344 None
1345 };
1346 return finalize_run_error(
1347 stdout,
1348 stderr,
1349 json_session,
1350 summary.as_ref(),
1351 phase.as_ref(),
1352 rusage.as_ref(),
1353 run_started,
1354 profile_rollup.as_ref(),
1355 timing.as_deref(),
1356 harn_vm::tracing::peek_spans().len() as u64,
1357 cpu_started_ms.map(|start| time::cpu_ms().saturating_sub(start)),
1358 "attestation",
1359 error,
1360 );
1361 }
1362 harn_vm::event_log::reset_active_event_log();
1363 }
1364
1365 match execution {
1366 RunExecution::Terminal(terminal) => {
1367 stdout.push_str(output);
1368 let main_events = harn_vm::tracing::peek_spans().len() as u64;
1369 let cpu_ms_total = cpu_started_ms.map(|start| time::cpu_ms().saturating_sub(start));
1370 let profile_rollup = if profile.is_enabled() {
1371 Some(harn_vm::profile::build(&harn_vm::tracing::peek_spans()))
1372 } else {
1373 None
1374 };
1375 let summary_llm = summary.as_ref().map(|_| run_summary_llm_snapshot());
1376 if trace {
1377 stderr.push_str(&render_trace_summary());
1378 }
1379 if let Some(profile_rollup) = profile_rollup.as_ref() {
1380 if let Err(error) =
1381 render_and_persist_profile_rollup(&profile, profile_rollup, &mut stderr)
1382 {
1383 stderr.push_str(&format!("warning: failed to write profile: {error}\n"));
1384 }
1385 }
1386 if let Some(diagnostic) = terminal.nonzero_return_diagnostic() {
1387 stderr.push_str(&diagnostic);
1388 }
1389 let aux_emission = emit_run_aux_for_exit(
1390 summary.as_ref(),
1391 phase.as_ref(),
1392 rusage.as_ref(),
1393 run_started,
1394 exit_code,
1395 profile_rollup.as_ref(),
1396 summary_llm,
1397 timing.as_deref(),
1398 main_events,
1399 cpu_ms_total,
1400 json_session.is_some(),
1401 &mut stderr,
1402 );
1403 if let Some(session) = json_session {
1404 if let Some(error) = aux_emission.error {
1405 let mut outcome = session.finalize_error(
1406 "run_aux",
1407 format!("failed to emit auxiliary run JSON: {error}"),
1408 1,
1409 );
1410 outcome.stderr = aux_emission.stderr;
1411 return outcome;
1412 }
1413 let value = terminal.json_value();
1414 let mut outcome = session.finalize_result(value, aux_emission.exit_code);
1415 outcome.stderr = aux_emission.stderr;
1416 return outcome;
1417 }
1418 RunOutcome {
1419 stdout,
1420 stderr,
1421 exit_code: aux_emission.exit_code,
1422 }
1423 }
1424 RunExecution::Failed(rendered_error) => {
1425 stderr.push_str(&rendered_error);
1426 let main_events = harn_vm::tracing::peek_spans().len() as u64;
1427 let cpu_ms_total = cpu_started_ms.map(|start| time::cpu_ms().saturating_sub(start));
1428 let profile_rollup = if profile.is_enabled() {
1429 Some(harn_vm::profile::build(&harn_vm::tracing::peek_spans()))
1430 } else {
1431 None
1432 };
1433 if let Some(profile_rollup) = profile_rollup.as_ref() {
1434 if let Err(error) =
1435 render_and_persist_profile_rollup(&profile, profile_rollup, &mut stderr)
1436 {
1437 stderr.push_str(&format!("warning: failed to write profile: {error}\n"));
1438 }
1439 }
1440 let aux_emission = emit_run_aux_for_exit(
1441 summary.as_ref(),
1442 phase.as_ref(),
1443 rusage.as_ref(),
1444 run_started,
1445 1,
1446 profile_rollup.as_ref(),
1447 None,
1448 timing.as_deref(),
1449 main_events,
1450 cpu_ms_total,
1451 json_session.is_some(),
1452 &mut stderr,
1453 );
1454 if let Some(session) = json_session {
1455 let mut outcome =
1456 session.finalize_error("runtime", rendered_error, aux_emission.exit_code);
1457 outcome.stderr = aux_emission.stderr;
1458 return outcome;
1459 }
1460 RunOutcome {
1461 stdout,
1462 stderr,
1463 exit_code: aux_emission.exit_code,
1464 }
1465 }
1466 }
1467}
1468
1469fn render_and_persist_profile_rollup(
1470 options: &RunProfileOptions,
1471 profile: &harn_vm::profile::RunProfile,
1472 stderr: &mut String,
1473) -> Result<(), String> {
1474 if options.text {
1475 stderr.push_str(&harn_vm::profile::render(profile));
1476 }
1477 if let Some(path) = options.json_path.as_ref() {
1478 if let Some(parent) = path.parent() {
1479 if !parent.as_os_str().is_empty() {
1480 fs::create_dir_all(parent)
1481 .map_err(|error| format!("create {}: {error}", parent.display()))?;
1482 }
1483 }
1484 let json = serde_json::to_string_pretty(profile)
1485 .map_err(|error| format!("serialize profile: {error}"))?;
1486 fs::write(path, json).map_err(|error| format!("write {}: {error}", path.display()))?;
1487 }
1488 Ok(())
1489}
1490
1491fn build_run_summary<'a>(
1492 started: Instant,
1493 exit_code: i32,
1494 profile: Option<&'a harn_vm::profile::RunProfile>,
1495 llm: RunSummaryLlm,
1496) -> RunSummary<'a> {
1497 RunSummary {
1498 schema_version: RUN_SUMMARY_SCHEMA_VERSION,
1499 event: "run_summary",
1500 wall_time_ms: started.elapsed().as_millis().min(u128::from(u64::MAX)) as u64,
1501 exit_code,
1502 llm,
1503 profile,
1504 }
1505}
1506
1507fn run_summary_llm_snapshot() -> RunSummaryLlm {
1508 let (input_tokens, output_tokens, time_ms, call_count) = harn_vm::llm::peek_trace_summary();
1509 let cost_usd = harn_vm::llm::peek_total_cost();
1510 RunSummaryLlm {
1511 call_count,
1512 input_tokens,
1513 output_tokens,
1514 time_ms,
1515 cost_usd: if cost_usd.is_finite() { cost_usd } else { 0.0 },
1516 }
1517}
1518
1519struct RunAuxEmission {
1520 stderr: String,
1521 exit_code: i32,
1522 error: Option<String>,
1523}
1524
1525#[allow(clippy::too_many_arguments)]
1526fn emit_run_aux_for_exit(
1527 summary: Option<&RunSummaryOptions>,
1528 phase: Option<&RunPhaseOptions>,
1529 rusage: Option<&RunRusageOptions>,
1530 started: Instant,
1531 exit_code: i32,
1532 profile: Option<&harn_vm::profile::RunProfile>,
1533 llm: Option<RunSummaryLlm>,
1534 timing: Option<&RunTiming>,
1535 main_events: u64,
1536 cpu_ms_total: Option<u64>,
1537 json_mode: bool,
1538 stderr: &mut String,
1539) -> RunAuxEmission {
1540 let mut aux_stderr = String::new();
1541 let mut final_exit_code = exit_code;
1542 let mut aux_error = None;
1543 let aux_target = if json_mode { &mut aux_stderr } else { stderr };
1544 let default_timing = RunTiming::default();
1545 let timing = timing.unwrap_or(&default_timing);
1546
1547 if let Some(options) = summary {
1548 let llm = llm.unwrap_or_else(run_summary_llm_snapshot);
1549 let summary = build_run_summary(started, exit_code, profile, llm);
1550 if let Err(error) = emit_raw_json_line(&options.sink, &summary, "run summary", aux_target) {
1551 record_aux_error(
1552 &mut final_exit_code,
1553 &mut aux_error,
1554 aux_target,
1555 "run summary",
1556 error,
1557 );
1558 }
1559 }
1560 if let Some(options) = phase {
1561 let phase_event = RunPhaseEvent {
1562 schema_version: RUN_PHASE_SCHEMA_VERSION,
1563 event: "run_phase",
1564 phases: time::build_phase_records(timing, main_events),
1565 };
1566 if let Err(error) = emit_raw_json_line(&options.sink, &phase_event, "run phase", aux_target)
1567 {
1568 record_aux_error(
1569 &mut final_exit_code,
1570 &mut aux_error,
1571 aux_target,
1572 "run phase",
1573 error,
1574 );
1575 }
1576 }
1577 if let Some(options) = rusage {
1578 let rusage_event = RunRusageEvent {
1579 schema_version: RUN_RUSAGE_SCHEMA_VERSION,
1580 event: "run_rusage",
1581 cpu_ms: cpu_ms_total.unwrap_or(0),
1582 };
1583 if let Err(error) =
1584 emit_raw_json_line(&options.sink, &rusage_event, "run rusage", aux_target)
1585 {
1586 record_aux_error(
1587 &mut final_exit_code,
1588 &mut aux_error,
1589 aux_target,
1590 "run rusage",
1591 error,
1592 );
1593 }
1594 }
1595
1596 RunAuxEmission {
1597 stderr: aux_stderr,
1598 exit_code: final_exit_code,
1599 error: aux_error,
1600 }
1601}
1602
1603fn record_aux_error(
1604 final_exit_code: &mut i32,
1605 aux_error: &mut Option<String>,
1606 stderr: &mut String,
1607 label: &str,
1608 error: String,
1609) {
1610 stderr.push_str(&format!("error: failed to emit {label}: {error}\n"));
1611 if *final_exit_code == 0 {
1612 *final_exit_code = 1;
1613 }
1614 if aux_error.is_none() {
1615 *aux_error = Some(error);
1616 }
1617}
1618
1619fn emit_raw_json_line(
1620 sink: &RunJsonSink,
1621 value: &impl Serialize,
1622 label: &str,
1623 stderr: &mut String,
1624) -> Result<(), String> {
1625 let line =
1626 serde_json::to_string(value).map_err(|error| format!("serialize {label}: {error}"))? + "\n";
1627 match &sink.target {
1628 RunJsonSinkTarget::Stderr => {
1629 stderr.push_str(&line);
1630 Ok(())
1631 }
1632 RunJsonSinkTarget::File(path) => write_raw_json_file(path, &line),
1633 RunJsonSinkTarget::Fd(fd) => write_raw_json_fd(*fd, &line, sink.fd_flag),
1634 }
1635}
1636
1637fn write_raw_json_file(path: &Path, line: &str) -> Result<(), String> {
1638 if let Some(parent) = path.parent() {
1639 if !parent.as_os_str().is_empty() {
1640 fs::create_dir_all(parent)
1641 .map_err(|error| format!("create {}: {error}", parent.display()))?;
1642 }
1643 }
1644 fs::write(path, line).map_err(|error| format!("write {}: {error}", path.display()))
1645}
1646
1647#[cfg(unix)]
1648fn write_raw_json_fd(fd: i32, line: &str, flag: &str) -> Result<(), String> {
1649 use std::fs::File;
1650 use std::os::unix::io::FromRawFd;
1651
1652 if fd < 0 {
1653 return Err(format!("invalid {flag} {fd}: must be non-negative"));
1654 }
1655 let duped = unsafe { libc::dup(fd) };
1656 if duped < 0 {
1657 return Err(format!(
1658 "duplicate {flag} {fd}: {}",
1659 io::Error::last_os_error()
1660 ));
1661 }
1662 let mut file = unsafe { File::from_raw_fd(duped) };
1663 file.write_all(line.as_bytes())
1664 .and_then(|_| file.flush())
1665 .map_err(|error| format!("write {flag} {fd}: {error}"))
1666}
1667
1668#[cfg(not(unix))]
1669fn write_raw_json_fd(_fd: i32, _line: &str, flag: &str) -> Result<(), String> {
1670 Err(format!("{flag} is only supported on Unix platforms"))
1671}
1672
1673async fn append_run_provenance_event(
1674 log: &Arc<harn_vm::event_log::AnyEventLog>,
1675 kind: &str,
1676 payload: serde_json::Value,
1677) {
1678 let Ok(topic) = harn_vm::event_log::Topic::new("run.provenance") else {
1679 return;
1680 };
1681 let _ = log
1682 .append(&topic, harn_vm::event_log::LogEvent::new(kind, payload))
1683 .await;
1684}
1685
1686async fn emit_run_attestation(
1687 log: &Arc<harn_vm::event_log::AnyEventLog>,
1688 path: &str,
1689 store_base: &Path,
1690 started_at_ms: i64,
1691 exit_code: i32,
1692 options: &RunAttestationOptions,
1693 stderr: &mut String,
1694) -> Result<(), String> {
1695 let finished_at_ms = now_ms();
1696 let status = if exit_code == 0 { "success" } else { "failure" };
1697 append_run_provenance_event(
1698 log,
1699 "finished",
1700 serde_json::json!({
1701 "pipeline": path,
1702 "status": status,
1703 "exit_code": exit_code,
1704 }),
1705 )
1706 .await;
1707 log.flush()
1708 .await
1709 .map_err(|error| format!("failed to flush attestation event log: {error}"))?;
1710 let secret_provider = harn_vm::secrets::configured_default_chain("harn.provenance")
1711 .map_err(|error| format!("failed to configure provenance secrets: {error}"))?;
1712 let (signing_key, key_id) =
1713 harn_vm::load_or_generate_agent_signing_key(&secret_provider, options.agent_id.as_deref())
1714 .await
1715 .map_err(|error| format!("failed to load provenance signing key: {error}"))?;
1716 let receipt = harn_vm::build_signed_receipt(
1717 log,
1718 harn_vm::ReceiptBuildOptions {
1719 pipeline: path.to_string(),
1720 status: status.to_string(),
1721 started_at_ms,
1722 finished_at_ms,
1723 exit_code,
1724 producer_name: "harn-cli".to_string(),
1725 producer_version: env!("CARGO_PKG_VERSION").to_string(),
1726 },
1727 &signing_key,
1728 key_id,
1729 )
1730 .await
1731 .map_err(|error| format!("failed to build provenance receipt: {error}"))?;
1732 let receipt_path = receipt_output_path(store_base, options, &receipt.receipt_id);
1733 if let Some(parent) = receipt_path.parent() {
1734 fs::create_dir_all(parent)
1735 .map_err(|error| format!("failed to create {}: {error}", parent.display()))?;
1736 }
1737 let encoded = serde_json::to_vec_pretty(&receipt)
1738 .map_err(|error| format!("failed to encode provenance receipt: {error}"))?;
1739 fs::write(&receipt_path, encoded)
1740 .map_err(|error| format!("failed to write {}: {error}", receipt_path.display()))?;
1741 stderr.push_str(&format!("provenance receipt: {}\n", receipt_path.display()));
1742 Ok(())
1743}
1744
1745fn receipt_output_path(
1746 store_base: &Path,
1747 options: &RunAttestationOptions,
1748 receipt_id: &str,
1749) -> PathBuf {
1750 if let Some(path) = options.receipt_out.as_ref() {
1751 return path.clone();
1752 }
1753 harn_vm::runtime_paths::state_root(store_base)
1754 .join("receipts")
1755 .join(format!("{receipt_id}.json"))
1756}
1757
1758fn now_ms() -> i64 {
1759 std::time::SystemTime::now()
1760 .duration_since(std::time::UNIX_EPOCH)
1761 .map(|duration| duration.as_millis() as i64)
1762 .unwrap_or(0)
1763}
1764
1765fn exit_code_from_return_value(value: &harn_vm::VmValue) -> i32 {
1772 use harn_vm::VmValue;
1773 match value {
1774 VmValue::Int(n) => (*n).clamp(0, 255) as i32,
1775 VmValue::EnumVariant(enum_variant) if enum_variant.is_variant("Result", "Err") => 1,
1776 _ => 0,
1777 }
1778}
1779
1780struct JsonRunSession {
1794 emitter: self::json_events::NdjsonEmitter,
1795 prior_sink: Option<Arc<dyn harn_vm::run_events::RunEventSink>>,
1796}
1797
1798impl JsonRunSession {
1799 fn install(options: RunJsonOptions, out: Box<dyn io::Write + Send>) -> Self {
1800 let emitter = NdjsonEmitter::new(out, options.quiet);
1801 let prior_sink = harn_vm::run_events::install_sink(emitter.sink());
1802 Self {
1803 emitter,
1804 prior_sink,
1805 }
1806 }
1807
1808 fn finalize_result(self, value: serde_json::Value, exit_code: i32) -> RunOutcome {
1809 self.emitter.emit_result(value, exit_code);
1810 RunOutcome {
1811 stdout: String::new(),
1812 stderr: String::new(),
1813 exit_code,
1814 }
1815 }
1816
1817 fn finalize_error(
1818 self,
1819 code: impl Into<String>,
1820 message: impl Into<String>,
1821 exit_code: i32,
1822 ) -> RunOutcome {
1823 self.emitter.emit_error(code, message);
1824 RunOutcome {
1825 stdout: String::new(),
1826 stderr: String::new(),
1827 exit_code,
1828 }
1829 }
1830}
1831
1832impl Drop for JsonRunSession {
1833 fn drop(&mut self) {
1834 match self.prior_sink.take() {
1835 Some(prior) => {
1836 harn_vm::run_events::install_sink(prior);
1837 }
1838 None => harn_vm::run_events::clear_sink(),
1839 }
1840 }
1841}
1842
1843#[allow(clippy::too_many_arguments)]
1844fn finalize_run_error(
1845 stdout: String,
1846 mut stderr: String,
1847 json_session: Option<JsonRunSession>,
1848 summary: Option<&RunSummaryOptions>,
1849 phase: Option<&RunPhaseOptions>,
1850 rusage: Option<&RunRusageOptions>,
1851 started: Instant,
1852 profile: Option<&harn_vm::profile::RunProfile>,
1853 timing: Option<&RunTiming>,
1854 main_events: u64,
1855 cpu_ms_total: Option<u64>,
1856 code: impl Into<String>,
1857 message: impl Into<String>,
1858) -> RunOutcome {
1859 let aux_emission = emit_run_aux_for_exit(
1860 summary,
1861 phase,
1862 rusage,
1863 started,
1864 1,
1865 profile,
1866 None,
1867 timing,
1868 main_events,
1869 cpu_ms_total,
1870 json_session.is_some(),
1871 &mut stderr,
1872 );
1873 if let Some(session) = json_session {
1874 let mut outcome = session.finalize_error(code, message, aux_emission.exit_code);
1875 outcome.stderr = aux_emission.stderr;
1876 return outcome;
1877 }
1878 RunOutcome {
1879 stdout,
1880 stderr,
1881 exit_code: aux_emission.exit_code,
1882 }
1883}
1884
1885fn finalize_harnpack_error(
1890 mut stderr: String,
1891 json_session: Option<JsonRunSession>,
1892 summary: Option<&RunSummaryOptions>,
1893 phase: Option<&RunPhaseOptions>,
1894 rusage: Option<&RunRusageOptions>,
1895 started: Instant,
1896 err: HarnpackError,
1897) -> RunOutcome {
1898 let code = err.code;
1899 let message = err.message;
1900 stderr.push_str(&format!("error: {message}\n"));
1901 finalize_run_error(
1902 String::new(),
1903 stderr,
1904 json_session,
1905 summary,
1906 phase,
1907 rusage,
1908 started,
1909 None,
1910 None,
1911 0,
1912 None,
1913 code,
1914 message,
1915 )
1916}
1917
1918fn finalize_harnpack_dry_run(
1923 mut stderr: String,
1924 json_session: Option<JsonRunSession>,
1925 summary_options: Option<&RunSummaryOptions>,
1926 phase_options: Option<&RunPhaseOptions>,
1927 rusage_options: Option<&RunRusageOptions>,
1928 started: Instant,
1929 cpu_ms_total: Option<u64>,
1930 prepared: &PreparedHarnpack,
1931) -> RunOutcome {
1932 let summary = format!(
1933 "[harn] harnpack verify ok: bundle_hash={}, signature_verified={}, cache_hit={}\n",
1934 prepared.bundle_hash, prepared.signature_verified, prepared.cache_hit
1935 );
1936 stderr.push_str(&summary);
1937 let aux_emission = emit_run_aux_for_exit(
1938 summary_options,
1939 phase_options,
1940 rusage_options,
1941 started,
1942 0,
1943 None,
1944 None,
1945 None,
1946 0,
1947 cpu_ms_total,
1948 json_session.is_some(),
1949 &mut stderr,
1950 );
1951 if let Some(session) = json_session {
1952 if let Some(error) = aux_emission.error {
1953 let mut outcome = session.finalize_error(
1954 "run_aux",
1955 format!("failed to emit auxiliary run JSON: {error}"),
1956 1,
1957 );
1958 outcome.stderr = aux_emission.stderr;
1959 return outcome;
1960 }
1961 let value = serde_json::json!({
1962 "bundle_hash": prepared.bundle_hash,
1963 "signature_verified": prepared.signature_verified,
1964 "key_id": prepared.key_id,
1965 "cache_hit": prepared.cache_hit,
1966 "dry_run_verify": true,
1967 });
1968 let mut outcome = session.finalize_result(value, aux_emission.exit_code);
1969 outcome.stderr = aux_emission.stderr;
1970 return outcome;
1971 }
1972 RunOutcome {
1973 stdout: String::new(),
1974 stderr,
1975 exit_code: aux_emission.exit_code,
1976 }
1977}
1978
1979fn render_return_value_error(value: &harn_vm::VmValue) -> String {
1980 let harn_vm::VmValue::EnumVariant(enum_variant) = value else {
1981 return String::new();
1982 };
1983 if !enum_variant.is_variant("Result", "Err") {
1984 return String::new();
1985 }
1986 let rendered = enum_variant
1987 .fields
1988 .first()
1989 .map(|p| p.display())
1990 .unwrap_or_default();
1991 if rendered.is_empty() {
1992 "error\n".to_string()
1993 } else if rendered.ends_with('\n') {
1994 rendered
1995 } else {
1996 format!("{rendered}\n")
1997 }
1998}
1999
2000pub(crate) fn render_trace_summary() -> String {
2001 use std::fmt::Write;
2002 let entries = harn_vm::llm::take_trace();
2003 if entries.is_empty() {
2004 return String::new();
2005 }
2006 let mut out = String::new();
2007 let _ = writeln!(out, "\n\x1b[2m─── LLM trace ───\x1b[0m");
2008 let mut total_input = 0i64;
2009 let mut total_output = 0i64;
2010 let mut total_ms = 0u64;
2011 for (i, entry) in entries.iter().enumerate() {
2012 let _ = writeln!(
2013 out,
2014 " #{}: {} | {} in + {} out tokens | {} ms",
2015 i + 1,
2016 entry.model,
2017 entry.input_tokens,
2018 entry.output_tokens,
2019 entry.duration_ms,
2020 );
2021 total_input += entry.input_tokens;
2022 total_output += entry.output_tokens;
2023 total_ms += entry.duration_ms;
2024 }
2025 let total_tokens = total_input + total_output;
2026 let cost = (total_input as f64 * 3.0 + total_output as f64 * 15.0) / 1_000_000.0;
2028 let _ = writeln!(
2029 out,
2030 " \x1b[1m{} call{}, {} tokens ({}in + {}out), {} ms, ~${:.4}\x1b[0m",
2031 entries.len(),
2032 if entries.len() == 1 { "" } else { "s" },
2033 total_tokens,
2034 total_input,
2035 total_output,
2036 total_ms,
2037 cost,
2038 );
2039 out
2040}
2041
2042pub(crate) async fn run_file_mcp_serve(
2056 path: &str,
2057 card_source: Option<&str>,
2058 mode: RunFileMcpServeMode,
2059) {
2060 let mut diagnostics = String::new();
2061 let Some(LoadedChunk { source, chunk }) = compile_or_load_chunk_for_run(path, &mut diagnostics)
2062 else {
2063 eprint!("{diagnostics}");
2064 process::exit(1);
2065 };
2066 if !diagnostics.is_empty() {
2067 eprint!("{diagnostics}");
2068 }
2069
2070 let mut vm = harn_vm::Vm::new();
2071 harn_vm::register_vm_stdlib(&mut vm);
2072 crate::install_default_hostlib(&mut vm);
2073 let source_parent = std::path::Path::new(path)
2074 .parent()
2075 .unwrap_or(std::path::Path::new("."));
2076 let project_root = harn_vm::stdlib::process::find_project_root(source_parent);
2077 let store_base = project_root.as_deref().unwrap_or(source_parent);
2078 harn_vm::register_store_builtins(&mut vm, store_base);
2079 harn_vm::register_metadata_builtins(&mut vm, store_base);
2080 let pipeline_name = std::path::Path::new(path)
2081 .file_stem()
2082 .and_then(|s| s.to_str())
2083 .unwrap_or("default");
2084 harn_vm::register_checkpoint_builtins(&mut vm, store_base, pipeline_name);
2085 vm.set_source_info(path, &source);
2086 if let Some(ref root) = project_root {
2087 vm.set_project_root(root);
2088 }
2089 vm.set_source_dir(&entry_source_dir(path));
2093
2094 let loaded = load_skills(&SkillLoaderInputs {
2096 cli_dirs: Vec::new(),
2097 source_path: Some(std::path::PathBuf::from(path)),
2098 });
2099 emit_loader_warnings(&loaded.loader_warnings);
2100 install_skills_global(&mut vm, &loaded);
2101
2102 if let Err(error) =
2103 manifest_runtime::install_manifest_runtime(Path::new(path), &mut vm, false).await
2104 {
2105 eprintln!("error: failed to install {}: {error}", error.label());
2106 process::exit(1);
2107 }
2108
2109 vm.set_source_dir(&entry_source_dir(path));
2114 let local = tokio::task::LocalSet::new();
2115 local
2116 .run_until(async {
2117 match vm.execute(&chunk).await {
2118 Ok(_) => {}
2119 Err(error) => crate::commands::serve::exit_after_mcp_pipeline_error(&vm, &error),
2120 }
2121
2122 let output = vm.output();
2124 if !output.is_empty() {
2125 eprint!("{output}");
2126 }
2127
2128 let registry = match harn_vm::take_mcp_serve_registry() {
2129 Some(r) => r,
2130 None => {
2131 eprintln!("error: pipeline did not call mcp_serve(registry)");
2132 eprintln!("hint: call mcp_serve(tools) at the end of your pipeline");
2133 process::exit(1);
2134 }
2135 };
2136
2137 let tools = match harn_vm::tool_registry_to_mcp_tools(®istry) {
2138 Ok(t) => t,
2139 Err(e) => {
2140 eprintln!("error: {e}");
2141 process::exit(1);
2142 }
2143 };
2144
2145 let resources = harn_vm::take_mcp_serve_resources();
2146 let resource_templates = harn_vm::take_mcp_serve_resource_templates();
2147 let prompts = harn_vm::take_mcp_serve_prompts();
2148 let metadata = harn_vm::take_mcp_serve_metadata();
2149
2150 let mut server_name = std::path::Path::new(path)
2151 .file_stem()
2152 .and_then(|s| s.to_str())
2153 .unwrap_or("harn")
2154 .to_string();
2155 if let Some(name) = metadata
2156 .as_ref()
2157 .and_then(|metadata| metadata.name.as_ref())
2158 {
2159 server_name = name.clone();
2160 }
2161
2162 let mut caps = Vec::new();
2163 if !tools.is_empty() {
2164 caps.push(format!(
2165 "{} tool{}",
2166 tools.len(),
2167 if tools.len() == 1 { "" } else { "s" }
2168 ));
2169 }
2170 let total_resources = resources.len() + resource_templates.len();
2171 if total_resources > 0 {
2172 caps.push(format!(
2173 "{total_resources} resource{}",
2174 if total_resources == 1 { "" } else { "s" }
2175 ));
2176 }
2177 if !prompts.is_empty() {
2178 caps.push(format!(
2179 "{} prompt{}",
2180 prompts.len(),
2181 if prompts.len() == 1 { "" } else { "s" }
2182 ));
2183 }
2184 eprintln!(
2185 "[harn] serve mcp: serving {} as '{server_name}'",
2186 caps.join(", ")
2187 );
2188
2189 let mut server =
2190 harn_vm::McpServer::new(server_name, tools, resources, resource_templates, prompts);
2191 if let Some(metadata) = metadata {
2192 server = server.with_metadata(metadata);
2193 }
2194 if let Some(source) = card_source {
2195 match resolve_card_source(source) {
2196 Ok(card) => server = server.with_server_card(card),
2197 Err(e) => {
2198 eprintln!("error: --card: {e}");
2199 process::exit(1);
2200 }
2201 }
2202 }
2203 match mode {
2204 RunFileMcpServeMode::Stdio => {
2205 if let Err(e) = server.run(&mut vm).await {
2206 eprintln!("error: MCP server error: {e}");
2207 process::exit(1);
2208 }
2209 }
2210 RunFileMcpServeMode::Http(http) => {
2211 let RunFileMcpServeHttp {
2212 options,
2213 auth_policy,
2214 } = *http;
2215 if let Err(e) = crate::commands::serve::run_script_mcp_http_server(
2216 server,
2217 vm,
2218 options,
2219 auth_policy,
2220 )
2221 .await
2222 {
2223 eprintln!("error: MCP server error: {e}");
2224 process::exit(1);
2225 }
2226 }
2227 }
2228 })
2229 .await;
2230}
2231
2232pub(crate) fn resolve_card_source(source: &str) -> Result<serde_json::Value, String> {
2237 let trimmed = source.trim_start();
2238 if trimmed.starts_with('{') || trimmed.starts_with('[') {
2239 return serde_json::from_str(source).map_err(|e| format!("inline JSON parse error: {e}"));
2240 }
2241 let path = std::path::Path::new(source);
2242 harn_vm::load_server_card_from_path(path).map_err(|e| format!("{e}"))
2243}
2244
2245pub(crate) async fn run_watch(path: &str, denied_builtins: HashSet<String>) {
2246 use notify::{Event, EventKind, RecursiveMode, Watcher};
2247
2248 let abs_path = std::fs::canonicalize(path).unwrap_or_else(|e| {
2249 eprintln!("Error: {e}");
2250 process::exit(1);
2251 });
2252 let watch_dir = abs_path.parent().unwrap_or(Path::new("."));
2253
2254 eprintln!("\x1b[2m[watch] running {path}...\x1b[0m");
2255 run_file(
2256 path,
2257 false,
2258 denied_builtins.clone(),
2259 Vec::new(),
2260 CliLlmMockMode::Off,
2261 None,
2262 RunProfileOptions::default(),
2263 )
2264 .await;
2265
2266 let (tx, mut rx) = tokio::sync::mpsc::channel::<()>(1);
2267 let _watcher = {
2268 let tx = tx.clone();
2269 let mut watcher = notify::recommended_watcher(move |res: Result<Event, _>| {
2270 if let Ok(event) = res {
2271 if matches!(
2272 event.kind,
2273 EventKind::Modify(_) | EventKind::Create(_) | EventKind::Remove(_)
2274 ) {
2275 let has_harn = event
2276 .paths
2277 .iter()
2278 .any(|p| p.extension().is_some_and(|ext| ext == "harn"));
2279 if has_harn {
2280 let _ = tx.blocking_send(());
2281 }
2282 }
2283 }
2284 })
2285 .unwrap_or_else(|e| {
2286 eprintln!("Error setting up file watcher: {e}");
2287 process::exit(1);
2288 });
2289 watcher
2290 .watch(watch_dir, RecursiveMode::Recursive)
2291 .unwrap_or_else(|e| {
2292 eprintln!("Error watching directory: {e}");
2293 process::exit(1);
2294 });
2295 watcher };
2297
2298 eprintln!(
2299 "\x1b[2m[watch] watching {} for .harn changes (ctrl-c to stop)\x1b[0m",
2300 watch_dir.display()
2301 );
2302
2303 loop {
2304 rx.recv().await;
2305 tokio::time::sleep(std::time::Duration::from_millis(200)).await;
2307 while rx.try_recv().is_ok() {}
2308
2309 eprintln!();
2310 eprintln!("\x1b[2m[watch] change detected, re-running {path}...\x1b[0m");
2311 run_file(
2312 path,
2313 false,
2314 denied_builtins.clone(),
2315 Vec::new(),
2316 CliLlmMockMode::Off,
2317 None,
2318 RunProfileOptions::default(),
2319 )
2320 .await;
2321 }
2322}
2323
2324#[cfg(test)]
2325mod tests;