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;
29pub(crate) mod 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<JsonRunSession>,
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_session = json.map(|options| {
581 JsonRunSession::new(options, Box::new(io::stdout()) as Box<dyn io::Write + Send>)
582 });
583 let outcome = execute_run_inner(ExecuteRunInputs {
584 path,
585 trace,
586 denied_builtins,
587 script_argv,
588 skill_dirs_raw,
589 llm_mock_mode,
590 attestation,
591 profile,
592 sandbox,
593 interrupt_tokens: Some(interrupt_tokens.clone()),
594 json: json_session,
595 aux,
596 timing: None,
597 harnpack,
598 })
599 .await;
600 if let Some(guard) = &deadline_guard {
601 guard.finish();
602 }
603
604 if !outcome.stderr.is_empty() {
607 io::stderr().write_all(outcome.stderr.as_bytes()).ok();
608 }
609 if !outcome.stdout.is_empty() {
610 io::stdout().write_all(outcome.stdout.as_bytes()).ok();
611 }
612
613 let mut exit_code = outcome.exit_code;
614 if deadline_guard
615 .as_ref()
616 .is_some_and(RunDeadlineGuard::timed_out)
617 || (exit_code != 0 && interrupt_tokens.cancel_token.load(Ordering::SeqCst))
618 {
619 exit_code = 124;
620 }
621 exit_code
622}
623
624#[allow(clippy::too_many_arguments)]
625pub(crate) async fn run_resume_with_skill_dirs(
626 target: &str,
627 trace: bool,
628 denied_builtins: HashSet<String>,
629 resume_argv: Vec<String>,
630 skill_dirs_raw: Vec<String>,
631 llm_mock_mode: CliLlmMockMode,
632 attestation: Option<RunAttestationOptions>,
633 profile: RunProfileOptions,
634 sandbox: RunSandboxOptions,
635 json: Option<RunJsonOptions>,
636 aux: RunAuxOptions,
637 control: RunControlOptions,
638) -> i32 {
639 let source = r#"import { resume_agent, wait_agent } from "std/agent/workers"
640
641pipeline main(task) {
642 const input = if len(argv) > 1 {
643 argv[1]
644 } else {
645 nil
646 }
647 const handle = resume_agent(argv[0], input, true)
648 return wait_agent(handle)
649}
650"#;
651 let tmp = match create_eval_temp_file() {
652 Ok(tmp) => tmp,
653 Err(error) => {
654 eprintln!("error: {error}");
655 return 1;
656 }
657 };
658 let tmp_path = tmp.path().to_path_buf();
659 if let Err(error) = fs::write(&tmp_path, source) {
660 eprintln!("error: failed to write temp file for --resume: {error}");
661 return 1;
662 }
663 let mut argv = Vec::with_capacity(resume_argv.len() + 1);
664 argv.push(target.to_string());
665 argv.extend(resume_argv);
666 let tmp_str = tmp_path.to_string_lossy().into_owned();
667 run_file_with_skill_dirs(
668 &tmp_str,
669 trace,
670 denied_builtins,
671 argv,
672 skill_dirs_raw,
673 llm_mock_mode,
674 attestation,
675 profile,
676 sandbox,
677 json,
678 aux,
679 control,
680 HarnpackRunOptions::default(),
681 )
682 .await
683}
684
685pub fn execute_explain_cost(path: &str) -> RunOutcome {
686 let stdout = String::new();
687 let mut stderr = String::new();
688
689 let source = match fs::read_to_string(path) {
690 Ok(source) => source,
691 Err(error) => {
692 stderr.push_str(&format!("Error reading {path}: {error}\n"));
693 return RunOutcome {
694 stdout,
695 stderr,
696 exit_code: 1,
697 };
698 }
699 };
700 let program = match parse_source_for_run(path, &source, &mut stderr) {
701 Some(program) => program,
702 None => {
703 return RunOutcome {
704 stdout,
705 stderr,
706 exit_code: 1,
707 };
708 }
709 };
710
711 let mut had_type_error = false;
712 let type_diagnostics = match typecheck_with_imports(&program, Path::new(path), &source) {
713 Ok(diagnostics) => diagnostics,
714 Err(error) => {
715 stderr.push_str(&format!("error: {error}\n"));
716 return RunOutcome {
717 stdout,
718 stderr,
719 exit_code: 1,
720 };
721 }
722 };
723 for diag in &type_diagnostics {
724 let rendered = harn_parser::diagnostic::render_type_diagnostic(&source, path, diag);
725 if matches!(diag.severity, DiagnosticSeverity::Error) {
726 had_type_error = true;
727 }
728 stderr.push_str(&rendered);
729 }
730 if had_type_error {
731 return RunOutcome {
732 stdout,
733 stderr,
734 exit_code: 1,
735 };
736 }
737
738 let extensions = package::load_runtime_extensions(Path::new(path));
739 package::install_runtime_extensions(&extensions);
740 RunOutcome {
741 stdout: explain_cost::render_explain_cost(path, &program),
742 stderr,
743 exit_code: 0,
744 }
745}
746
747pub(crate) struct StdoutPassthroughGuard {
748 previous: bool,
749}
750
751impl StdoutPassthroughGuard {
752 pub(crate) fn enable() -> Self {
753 Self {
754 previous: harn_vm::set_stdout_passthrough(true),
755 }
756 }
757}
758
759impl Drop for StdoutPassthroughGuard {
760 fn drop(&mut self) {
761 harn_vm::set_stdout_passthrough(self.previous);
762 }
763}
764
765pub async fn execute_run(
779 path: &str,
780 trace: bool,
781 denied_builtins: HashSet<String>,
782 script_argv: Vec<String>,
783 skill_dirs_raw: Vec<String>,
784 llm_mock_mode: CliLlmMockMode,
785 attestation: Option<RunAttestationOptions>,
786 profile: RunProfileOptions,
787) -> RunOutcome {
788 crate::ensure_builtin_signatures_installed();
789 execute_run_with_harnpack_and_sandbox_options(
790 path,
791 trace,
792 denied_builtins,
793 script_argv,
794 skill_dirs_raw,
795 llm_mock_mode,
796 attestation,
797 profile,
798 RunSandboxOptions::default(),
799 HarnpackRunOptions::default(),
800 )
801 .await
802}
803
804#[allow(clippy::too_many_arguments)]
808pub async fn execute_run_with_sandbox_options(
809 path: &str,
810 trace: bool,
811 denied_builtins: HashSet<String>,
812 script_argv: Vec<String>,
813 skill_dirs_raw: Vec<String>,
814 llm_mock_mode: CliLlmMockMode,
815 attestation: Option<RunAttestationOptions>,
816 profile: RunProfileOptions,
817 sandbox: RunSandboxOptions,
818) -> RunOutcome {
819 execute_run_with_harnpack_and_sandbox_options(
820 path,
821 trace,
822 denied_builtins,
823 script_argv,
824 skill_dirs_raw,
825 llm_mock_mode,
826 attestation,
827 profile,
828 sandbox,
829 HarnpackRunOptions::default(),
830 )
831 .await
832}
833
834#[allow(clippy::too_many_arguments)]
839pub async fn execute_run_with_harnpack_options(
840 path: &str,
841 trace: bool,
842 denied_builtins: HashSet<String>,
843 script_argv: Vec<String>,
844 skill_dirs_raw: Vec<String>,
845 llm_mock_mode: CliLlmMockMode,
846 attestation: Option<RunAttestationOptions>,
847 profile: RunProfileOptions,
848 harnpack: HarnpackRunOptions,
849) -> RunOutcome {
850 execute_run_with_harnpack_and_sandbox_options(
851 path,
852 trace,
853 denied_builtins,
854 script_argv,
855 skill_dirs_raw,
856 llm_mock_mode,
857 attestation,
858 profile,
859 RunSandboxOptions::default(),
860 harnpack,
861 )
862 .await
863}
864
865#[allow(clippy::too_many_arguments)]
866async fn execute_run_with_harnpack_and_sandbox_options(
867 path: &str,
868 trace: bool,
869 denied_builtins: HashSet<String>,
870 script_argv: Vec<String>,
871 skill_dirs_raw: Vec<String>,
872 llm_mock_mode: CliLlmMockMode,
873 attestation: Option<RunAttestationOptions>,
874 profile: RunProfileOptions,
875 sandbox: RunSandboxOptions,
876 harnpack: HarnpackRunOptions,
877) -> RunOutcome {
878 execute_run_inner(ExecuteRunInputs {
879 path,
880 trace,
881 denied_builtins,
882 script_argv,
883 skill_dirs_raw,
884 llm_mock_mode,
885 attestation,
886 profile,
887 sandbox,
888 interrupt_tokens: None,
889 json: None,
890 aux: RunAuxOptions::default(),
891 timing: None,
892 harnpack,
893 })
894 .await
895}
896
897#[allow(clippy::too_many_arguments)]
903pub async fn execute_run_json(
904 path: &str,
905 trace: bool,
906 denied_builtins: HashSet<String>,
907 script_argv: Vec<String>,
908 skill_dirs_raw: Vec<String>,
909 llm_mock_mode: CliLlmMockMode,
910 attestation: Option<RunAttestationOptions>,
911 profile: RunProfileOptions,
912 out: Box<dyn io::Write + Send>,
913 options: RunJsonOptions,
914) -> RunOutcome {
915 execute_run_inner(ExecuteRunInputs {
916 path,
917 trace,
918 denied_builtins,
919 script_argv,
920 skill_dirs_raw,
921 llm_mock_mode,
922 attestation,
923 profile,
924 sandbox: RunSandboxOptions::default(),
925 interrupt_tokens: None,
926 json: Some(JsonRunSession::new(options, out)),
927 aux: RunAuxOptions::default(),
928 timing: None,
929 harnpack: HarnpackRunOptions::default(),
930 })
931 .await
932}
933
934pub(crate) async fn execute_run_with_timing(
938 path: &str,
939 script_argv: Vec<String>,
940 timing: Option<&mut RunTiming>,
941 sandbox: RunSandboxOptions,
942) -> RunOutcome {
943 execute_run_inner(ExecuteRunInputs {
944 path,
945 trace: false,
946 denied_builtins: HashSet::new(),
947 script_argv,
948 skill_dirs_raw: Vec::new(),
949 llm_mock_mode: CliLlmMockMode::Off,
950 attestation: None,
951 profile: RunProfileOptions::default(),
952 sandbox,
953 interrupt_tokens: None,
954 json: None,
955 aux: RunAuxOptions::default(),
956 timing,
957 harnpack: HarnpackRunOptions::default(),
958 })
959 .await
960}
961
962fn entry_source_dir(path: &str) -> std::path::PathBuf {
976 match std::path::Path::new(path).parent() {
977 Some(parent) if !parent.as_os_str().is_empty() => parent.to_path_buf(),
978 _ => std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")),
979 }
980}
981
982#[allow(clippy::needless_option_as_deref)]
985async fn execute_run_inner(inputs: ExecuteRunInputs<'_>) -> RunOutcome {
986 let mut inputs = inputs;
987 let json_session = inputs.json.take();
988 let Some(json_session) = json_session else {
989 return execute_run_inner_scoped(inputs, None).await;
990 };
991 let sink = json_session.sink();
992 harn_vm::run_events::scope(sink, execute_run_inner_scoped(inputs, Some(json_session))).await
993}
994
995async fn execute_run_inner_scoped(
996 inputs: ExecuteRunInputs<'_>,
997 json_session: Option<JsonRunSession>,
998) -> RunOutcome {
999 let ExecuteRunInputs {
1000 path,
1001 trace,
1002 denied_builtins,
1003 script_argv,
1004 skill_dirs_raw,
1005 llm_mock_mode,
1006 attestation,
1007 profile,
1008 sandbox,
1009 interrupt_tokens,
1010 json: _,
1011 aux,
1012 timing,
1013 harnpack,
1014 } = inputs;
1015 let RunAuxOptions {
1016 summary,
1017 phase,
1018 rusage,
1019 } = aux;
1020 let run_started = Instant::now();
1021 let cpu_started_ms = rusage.as_ref().map(|_| time::cpu_ms());
1022 let mut owned_timing = if timing.is_none() && (phase.is_some() || rusage.is_some()) {
1023 Some(RunTiming::default())
1024 } else {
1025 None
1026 };
1027 let mut timing = timing.or(owned_timing.as_mut());
1028
1029 let mut stderr = String::new();
1030 let mut stdout = String::new();
1031
1032 let owned_run_path: String;
1037 let resolved_path: &str = if harnpack::looks_like_harnpack(Path::new(path)) {
1038 let outcome = match harnpack::prepare_harnpack(Path::new(path), &harnpack, &mut stderr) {
1039 Ok(prepared) => prepared,
1040 Err(err) => {
1041 return finalize_harnpack_error(
1042 stderr,
1043 json_session,
1044 summary.as_ref(),
1045 phase.as_ref(),
1046 rusage.as_ref(),
1047 run_started,
1048 err,
1049 );
1050 }
1051 };
1052 harn_vm::run_events::emit(harn_vm::run_events::RunEvent::PackRun {
1053 bundle_hash: outcome.bundle_hash.clone(),
1054 signature_verified: outcome.signature_verified,
1055 key_id: outcome.key_id.clone(),
1056 cache_hit: outcome.cache_hit,
1057 dry_run_verify: harnpack.dry_run_verify,
1058 });
1059 if harnpack.dry_run_verify {
1060 return finalize_harnpack_dry_run(
1061 stderr,
1062 json_session,
1063 summary.as_ref(),
1064 phase.as_ref(),
1065 rusage.as_ref(),
1066 run_started,
1067 cpu_started_ms.map(|start| time::cpu_ms().saturating_sub(start)),
1068 &outcome,
1069 );
1070 }
1071 owned_run_path = outcome.entrypoint_path.to_string_lossy().into_owned();
1072 owned_run_path.as_str()
1073 } else {
1074 path
1075 };
1076
1077 let Some(LoadedChunk { source, chunk }) =
1078 compile_or_load_chunk_with_timing(resolved_path, &mut stderr, timing.as_deref_mut())
1079 else {
1080 let message = stderr.clone();
1081 return finalize_run_error(
1082 stdout,
1083 stderr,
1084 json_session,
1085 summary.as_ref(),
1086 phase.as_ref(),
1087 rusage.as_ref(),
1088 run_started,
1089 None,
1090 timing.as_deref(),
1091 0,
1092 cpu_started_ms.map(|start| time::cpu_ms().saturating_sub(start)),
1093 "compile_error",
1094 message,
1095 );
1096 };
1097 let path = resolved_path;
1098
1099 let setup_start = Instant::now();
1100 if trace || summary.is_some() {
1101 harn_vm::llm::enable_tracing();
1102 }
1103 if profile.is_enabled() || phase.is_some() {
1104 harn_vm::tracing::set_tracing_enabled(true);
1105 }
1106 if profile.is_enabled() {
1107 harn_vm::builtin_profile::enable();
1112 }
1113 if let Err(error) = install_cli_llm_mock_mode(&llm_mock_mode) {
1114 stderr.push_str(&format!("error: {error}\n"));
1115 time::record_run_setup_elapsed(timing.as_deref_mut(), setup_start);
1116 return finalize_run_error(
1117 stdout,
1118 stderr,
1119 json_session,
1120 summary.as_ref(),
1121 phase.as_ref(),
1122 rusage.as_ref(),
1123 run_started,
1124 None,
1125 timing.as_deref(),
1126 0,
1127 cpu_started_ms.map(|start| time::cpu_ms().saturating_sub(start)),
1128 "llm_mock_install",
1129 error,
1130 );
1131 }
1132
1133 let mut vm = harn_vm::Vm::new();
1134 if let Some(timing) = timing.as_deref_mut() {
1135 timing.module_phases = Some(vm.enable_module_phase_timing());
1136 }
1137 if let Some(interrupt_tokens) = interrupt_tokens {
1138 vm.install_interrupt_signal_token(interrupt_tokens.signal_token);
1139 vm.install_cancel_token(interrupt_tokens.cancel_token);
1140 }
1141 harn_vm::register_vm_stdlib(&mut vm);
1142 crate::install_default_hostlib(&mut vm);
1143 let source_parent = std::path::Path::new(path)
1144 .parent()
1145 .unwrap_or(std::path::Path::new("."));
1146 let project_root = harn_vm::stdlib::process::find_project_root(source_parent);
1148 let store_base = project_root.as_deref().unwrap_or(source_parent);
1149 let sandbox_root = sandbox
1150 .workspace_root
1151 .clone()
1152 .unwrap_or_else(|| default_run_workspace_root(project_root.as_deref(), source_parent));
1153 let _sandbox_scope = install_run_sandbox_scope(&sandbox, &sandbox_root, &mut stderr);
1154 let attestation_started_at_ms = now_ms();
1155 let attestation_log = if attestation.is_some() {
1156 Some(harn_vm::event_log::install_memory_for_current_thread(256))
1157 } else {
1158 None
1159 };
1160 if let Some(log) = attestation_log.as_ref() {
1161 append_run_provenance_event(
1162 log,
1163 "started",
1164 serde_json::json!({
1165 "pipeline": path,
1166 "argv": &script_argv,
1167 "project_root": store_base.display().to_string(),
1168 "sandbox": run_sandbox_attestation(&sandbox),
1169 }),
1170 )
1171 .await;
1172 }
1173 harn_vm::register_store_builtins(&mut vm, store_base);
1174 harn_vm::register_metadata_builtins(&mut vm, store_base);
1175 let pipeline_name = std::path::Path::new(path)
1176 .file_stem()
1177 .and_then(|s| s.to_str())
1178 .unwrap_or("default");
1179 harn_vm::register_checkpoint_builtins(&mut vm, store_base, pipeline_name);
1180 vm.set_source_info(path, &source);
1181 let lazy_manifest_handlers = !denied_builtins.is_empty();
1182 if lazy_manifest_handlers {
1183 vm.set_denied_builtins(denied_builtins);
1184 }
1185 if let Some(ref root) = project_root {
1186 vm.set_project_root(root);
1187 }
1188
1189 vm.set_source_dir(&entry_source_dir(path));
1194
1195 let cli_dirs = canonicalize_cli_dirs(&skill_dirs_raw, None);
1198 let loaded = load_skills(&SkillLoaderInputs {
1199 cli_dirs,
1200 source_path: Some(std::path::PathBuf::from(path)),
1201 });
1202 emit_loader_warnings(&loaded.loader_warnings);
1203 install_skills_global(&mut vm, &loaded);
1204
1205 let argv_values: Vec<harn_vm::VmValue> = script_argv
1208 .iter()
1209 .map(|s| harn_vm::VmValue::String(arcstr::ArcStr::from(s.as_str())))
1210 .collect();
1211 vm.set_global(
1212 "argv",
1213 harn_vm::VmValue::List(std::sync::Arc::new(argv_values)),
1214 );
1215
1216 let runtime_harness =
1220 match crate::default_harness_for_manifest_or_base_dir(Path::new(path), store_base) {
1221 Ok(harness) => harness,
1222 Err(error) => {
1223 stderr.push_str(&format!(
1224 "error: failed to configure harness secret provider: {error}\n"
1225 ));
1226 time::record_run_setup_elapsed(timing.as_deref_mut(), setup_start);
1227 return finalize_run_error(
1228 stdout,
1229 stderr,
1230 json_session,
1231 summary.as_ref(),
1232 phase.as_ref(),
1233 rusage.as_ref(),
1234 run_started,
1235 None,
1236 timing.as_deref(),
1237 0,
1238 cpu_started_ms.map(|start| time::cpu_ms().saturating_sub(start)),
1239 "harness_secret_provider",
1240 error,
1241 );
1242 }
1243 };
1244 vm.set_harness(runtime_harness);
1245
1246 if let Err(error) =
1249 manifest_runtime::install_manifest_runtime(Path::new(path), &mut vm, lazy_manifest_handlers)
1250 .await
1251 {
1252 stderr.push_str(&format!(
1253 "error: failed to install {}: {error}\n",
1254 error.label()
1255 ));
1256 time::record_run_setup_elapsed(timing.as_deref_mut(), setup_start);
1257 return finalize_run_error(
1258 stdout,
1259 stderr,
1260 json_session,
1261 summary.as_ref(),
1262 phase.as_ref(),
1263 rusage.as_ref(),
1264 run_started,
1265 None,
1266 timing.as_deref(),
1267 0,
1268 cpu_started_ms.map(|start| time::cpu_ms().saturating_sub(start)),
1269 error.phase(),
1270 error.to_string(),
1271 );
1272 }
1273
1274 let local = tokio::task::LocalSet::new();
1275 time::record_run_setup_elapsed(timing.as_deref_mut(), setup_start);
1276 let main_start = Instant::now();
1277 vm.set_source_dir(&entry_source_dir(path));
1285 let execution = local
1286 .run_until(async {
1287 match vm.execute(&chunk).await {
1288 Ok(value) => RunExecution::Terminal(TerminalRun::Returned(value)),
1289 Err(error) => match error.process_exit_code() {
1290 Some(code) => RunExecution::Terminal(TerminalRun::ProcessExited(code)),
1291 None => RunExecution::Failed(vm.format_runtime_error(&error)),
1292 },
1293 }
1294 })
1295 .await;
1296 let output = vm.output();
1297 if let Some(t) = timing.as_deref_mut() {
1298 t.run_main = main_start.elapsed();
1299 }
1300 if let Err(error) = persist_cli_llm_mock_recording(&llm_mock_mode) {
1301 stderr.push_str(&format!("error: {error}\n"));
1302 let profile_rollup = if profile.is_enabled() {
1303 Some(harn_vm::profile::build(&harn_vm::tracing::peek_spans()))
1304 } else {
1305 None
1306 };
1307 return finalize_run_error(
1308 stdout,
1309 stderr,
1310 json_session,
1311 summary.as_ref(),
1312 phase.as_ref(),
1313 rusage.as_ref(),
1314 run_started,
1315 profile_rollup.as_ref(),
1316 timing.as_deref(),
1317 harn_vm::tracing::peek_spans().len() as u64,
1318 cpu_started_ms.map(|start| time::cpu_ms().saturating_sub(start)),
1319 "llm_mock_record",
1320 error,
1321 );
1322 }
1323
1324 let buffered_stderr = harn_vm::take_stderr_buffer();
1326 stderr.push_str(&buffered_stderr);
1327
1328 let exit_code = match &execution {
1329 RunExecution::Terminal(terminal) => terminal.exit_code(),
1330 RunExecution::Failed(_) => 1,
1331 };
1332
1333 if let (Some(options), Some(log)) = (attestation.as_ref(), attestation_log.as_ref()) {
1334 if let Err(error) = emit_run_attestation(
1335 log,
1336 path,
1337 store_base,
1338 attestation_started_at_ms,
1339 exit_code,
1340 options,
1341 &mut stderr,
1342 )
1343 .await
1344 {
1345 stderr.push_str(&format!(
1346 "error: failed to emit provenance receipt: {error}\n"
1347 ));
1348 let profile_rollup = if profile.is_enabled() {
1349 Some(harn_vm::profile::build(&harn_vm::tracing::peek_spans()))
1350 } else {
1351 None
1352 };
1353 return finalize_run_error(
1354 stdout,
1355 stderr,
1356 json_session,
1357 summary.as_ref(),
1358 phase.as_ref(),
1359 rusage.as_ref(),
1360 run_started,
1361 profile_rollup.as_ref(),
1362 timing.as_deref(),
1363 harn_vm::tracing::peek_spans().len() as u64,
1364 cpu_started_ms.map(|start| time::cpu_ms().saturating_sub(start)),
1365 "attestation",
1366 error,
1367 );
1368 }
1369 harn_vm::event_log::reset_active_event_log();
1370 }
1371
1372 match execution {
1373 RunExecution::Terminal(terminal) => {
1374 stdout.push_str(output);
1375 let main_events = harn_vm::tracing::peek_spans().len() as u64;
1376 let cpu_ms_total = cpu_started_ms.map(|start| time::cpu_ms().saturating_sub(start));
1377 let profile_rollup = if profile.is_enabled() {
1378 Some(harn_vm::profile::build(&harn_vm::tracing::peek_spans()))
1379 } else {
1380 None
1381 };
1382 let summary_llm = summary.as_ref().map(|_| run_summary_llm_snapshot());
1383 if trace {
1384 stderr.push_str(&render_trace_summary());
1385 }
1386 if let Some(profile_rollup) = profile_rollup.as_ref() {
1387 if let Err(error) =
1388 render_and_persist_profile_rollup(&profile, profile_rollup, &mut stderr)
1389 {
1390 stderr.push_str(&format!("warning: failed to write profile: {error}\n"));
1391 }
1392 }
1393 if let Some(diagnostic) = terminal.nonzero_return_diagnostic() {
1394 stderr.push_str(&diagnostic);
1395 }
1396 let aux_emission = emit_run_aux_for_exit(
1397 summary.as_ref(),
1398 phase.as_ref(),
1399 rusage.as_ref(),
1400 run_started,
1401 exit_code,
1402 profile_rollup.as_ref(),
1403 summary_llm,
1404 timing.as_deref(),
1405 main_events,
1406 cpu_ms_total,
1407 json_session.is_some(),
1408 &mut stderr,
1409 );
1410 if let Some(session) = json_session {
1411 if let Some(error) = aux_emission.error {
1412 let mut outcome = session.finalize_error(
1413 "run_aux",
1414 format!("failed to emit auxiliary run JSON: {error}"),
1415 1,
1416 );
1417 outcome.stderr = aux_emission.stderr;
1418 return outcome;
1419 }
1420 let value = terminal.json_value();
1421 let mut outcome = session.finalize_result(value, aux_emission.exit_code);
1422 outcome.stderr = aux_emission.stderr;
1423 return outcome;
1424 }
1425 RunOutcome {
1426 stdout,
1427 stderr,
1428 exit_code: aux_emission.exit_code,
1429 }
1430 }
1431 RunExecution::Failed(rendered_error) => {
1432 stderr.push_str(&rendered_error);
1433 let main_events = harn_vm::tracing::peek_spans().len() as u64;
1434 let cpu_ms_total = cpu_started_ms.map(|start| time::cpu_ms().saturating_sub(start));
1435 let profile_rollup = if profile.is_enabled() {
1436 Some(harn_vm::profile::build(&harn_vm::tracing::peek_spans()))
1437 } else {
1438 None
1439 };
1440 if let Some(profile_rollup) = profile_rollup.as_ref() {
1441 if let Err(error) =
1442 render_and_persist_profile_rollup(&profile, profile_rollup, &mut stderr)
1443 {
1444 stderr.push_str(&format!("warning: failed to write profile: {error}\n"));
1445 }
1446 }
1447 let aux_emission = emit_run_aux_for_exit(
1448 summary.as_ref(),
1449 phase.as_ref(),
1450 rusage.as_ref(),
1451 run_started,
1452 1,
1453 profile_rollup.as_ref(),
1454 None,
1455 timing.as_deref(),
1456 main_events,
1457 cpu_ms_total,
1458 json_session.is_some(),
1459 &mut stderr,
1460 );
1461 if let Some(session) = json_session {
1462 let mut outcome =
1463 session.finalize_error("runtime", rendered_error, aux_emission.exit_code);
1464 outcome.stderr = aux_emission.stderr;
1465 return outcome;
1466 }
1467 RunOutcome {
1468 stdout,
1469 stderr,
1470 exit_code: aux_emission.exit_code,
1471 }
1472 }
1473 }
1474}
1475
1476fn render_and_persist_profile_rollup(
1477 options: &RunProfileOptions,
1478 profile: &harn_vm::profile::RunProfile,
1479 stderr: &mut String,
1480) -> Result<(), String> {
1481 if options.text {
1482 stderr.push_str(&harn_vm::profile::render(profile));
1483 }
1484 if let Some(path) = options.json_path.as_ref() {
1485 if let Some(parent) = path.parent() {
1486 if !parent.as_os_str().is_empty() {
1487 fs::create_dir_all(parent)
1488 .map_err(|error| format!("create {}: {error}", parent.display()))?;
1489 }
1490 }
1491 let json = serde_json::to_string_pretty(profile)
1492 .map_err(|error| format!("serialize profile: {error}"))?;
1493 fs::write(path, json).map_err(|error| format!("write {}: {error}", path.display()))?;
1494 }
1495 Ok(())
1496}
1497
1498fn build_run_summary<'a>(
1499 started: Instant,
1500 exit_code: i32,
1501 profile: Option<&'a harn_vm::profile::RunProfile>,
1502 llm: RunSummaryLlm,
1503) -> RunSummary<'a> {
1504 RunSummary {
1505 schema_version: RUN_SUMMARY_SCHEMA_VERSION,
1506 event: "run_summary",
1507 wall_time_ms: started.elapsed().as_millis().min(u128::from(u64::MAX)) as u64,
1508 exit_code,
1509 llm,
1510 profile,
1511 }
1512}
1513
1514fn run_summary_llm_snapshot() -> RunSummaryLlm {
1515 let (input_tokens, output_tokens, time_ms, call_count) = harn_vm::llm::peek_trace_summary();
1516 let cost_usd = harn_vm::llm::peek_total_cost();
1517 RunSummaryLlm {
1518 call_count,
1519 input_tokens,
1520 output_tokens,
1521 time_ms,
1522 cost_usd: if cost_usd.is_finite() { cost_usd } else { 0.0 },
1523 }
1524}
1525
1526struct RunAuxEmission {
1527 stderr: String,
1528 exit_code: i32,
1529 error: Option<String>,
1530}
1531
1532#[allow(clippy::too_many_arguments)]
1533fn emit_run_aux_for_exit(
1534 summary: Option<&RunSummaryOptions>,
1535 phase: Option<&RunPhaseOptions>,
1536 rusage: Option<&RunRusageOptions>,
1537 started: Instant,
1538 exit_code: i32,
1539 profile: Option<&harn_vm::profile::RunProfile>,
1540 llm: Option<RunSummaryLlm>,
1541 timing: Option<&RunTiming>,
1542 main_events: u64,
1543 cpu_ms_total: Option<u64>,
1544 json_mode: bool,
1545 stderr: &mut String,
1546) -> RunAuxEmission {
1547 let mut aux_stderr = String::new();
1548 let mut final_exit_code = exit_code;
1549 let mut aux_error = None;
1550 let aux_target = if json_mode { &mut aux_stderr } else { stderr };
1551 let default_timing = RunTiming::default();
1552 let timing = timing.unwrap_or(&default_timing);
1553
1554 if let Some(options) = summary {
1555 let llm = llm.unwrap_or_else(run_summary_llm_snapshot);
1556 let summary = build_run_summary(started, exit_code, profile, llm);
1557 if let Err(error) = emit_raw_json_line(&options.sink, &summary, "run summary", aux_target) {
1558 record_aux_error(
1559 &mut final_exit_code,
1560 &mut aux_error,
1561 aux_target,
1562 "run summary",
1563 error,
1564 );
1565 }
1566 }
1567 if let Some(options) = phase {
1568 let phase_event = RunPhaseEvent {
1569 schema_version: RUN_PHASE_SCHEMA_VERSION,
1570 event: "run_phase",
1571 phases: time::build_phase_records(timing, main_events),
1572 };
1573 if let Err(error) = emit_raw_json_line(&options.sink, &phase_event, "run phase", aux_target)
1574 {
1575 record_aux_error(
1576 &mut final_exit_code,
1577 &mut aux_error,
1578 aux_target,
1579 "run phase",
1580 error,
1581 );
1582 }
1583 }
1584 if let Some(options) = rusage {
1585 let rusage_event = RunRusageEvent {
1586 schema_version: RUN_RUSAGE_SCHEMA_VERSION,
1587 event: "run_rusage",
1588 cpu_ms: cpu_ms_total.unwrap_or(0),
1589 };
1590 if let Err(error) =
1591 emit_raw_json_line(&options.sink, &rusage_event, "run rusage", aux_target)
1592 {
1593 record_aux_error(
1594 &mut final_exit_code,
1595 &mut aux_error,
1596 aux_target,
1597 "run rusage",
1598 error,
1599 );
1600 }
1601 }
1602
1603 RunAuxEmission {
1604 stderr: aux_stderr,
1605 exit_code: final_exit_code,
1606 error: aux_error,
1607 }
1608}
1609
1610fn record_aux_error(
1611 final_exit_code: &mut i32,
1612 aux_error: &mut Option<String>,
1613 stderr: &mut String,
1614 label: &str,
1615 error: String,
1616) {
1617 stderr.push_str(&format!("error: failed to emit {label}: {error}\n"));
1618 if *final_exit_code == 0 {
1619 *final_exit_code = 1;
1620 }
1621 if aux_error.is_none() {
1622 *aux_error = Some(error);
1623 }
1624}
1625
1626fn emit_raw_json_line(
1627 sink: &RunJsonSink,
1628 value: &impl Serialize,
1629 label: &str,
1630 stderr: &mut String,
1631) -> Result<(), String> {
1632 let line =
1633 serde_json::to_string(value).map_err(|error| format!("serialize {label}: {error}"))? + "\n";
1634 match &sink.target {
1635 RunJsonSinkTarget::Stderr => {
1636 stderr.push_str(&line);
1637 Ok(())
1638 }
1639 RunJsonSinkTarget::File(path) => write_raw_json_file(path, &line),
1640 RunJsonSinkTarget::Fd(fd) => write_raw_json_fd(*fd, &line, sink.fd_flag),
1641 }
1642}
1643
1644fn write_raw_json_file(path: &Path, line: &str) -> Result<(), String> {
1645 if let Some(parent) = path.parent() {
1646 if !parent.as_os_str().is_empty() {
1647 fs::create_dir_all(parent)
1648 .map_err(|error| format!("create {}: {error}", parent.display()))?;
1649 }
1650 }
1651 fs::write(path, line).map_err(|error| format!("write {}: {error}", path.display()))
1652}
1653
1654#[cfg(unix)]
1655fn write_raw_json_fd(fd: i32, line: &str, flag: &str) -> Result<(), String> {
1656 use std::fs::File;
1657 use std::os::unix::io::FromRawFd;
1658
1659 if fd < 0 {
1660 return Err(format!("invalid {flag} {fd}: must be non-negative"));
1661 }
1662 let duped = unsafe { libc::dup(fd) };
1663 if duped < 0 {
1664 return Err(format!(
1665 "duplicate {flag} {fd}: {}",
1666 io::Error::last_os_error()
1667 ));
1668 }
1669 let mut file = unsafe { File::from_raw_fd(duped) };
1670 file.write_all(line.as_bytes())
1671 .and_then(|_| file.flush())
1672 .map_err(|error| format!("write {flag} {fd}: {error}"))
1673}
1674
1675#[cfg(not(unix))]
1676fn write_raw_json_fd(_fd: i32, _line: &str, flag: &str) -> Result<(), String> {
1677 Err(format!("{flag} is only supported on Unix platforms"))
1678}
1679
1680async fn append_run_provenance_event(
1681 log: &Arc<harn_vm::event_log::AnyEventLog>,
1682 kind: &str,
1683 payload: serde_json::Value,
1684) {
1685 let Ok(topic) = harn_vm::event_log::Topic::new("run.provenance") else {
1686 return;
1687 };
1688 let _ = log
1689 .append(&topic, harn_vm::event_log::LogEvent::new(kind, payload))
1690 .await;
1691}
1692
1693async fn emit_run_attestation(
1694 log: &Arc<harn_vm::event_log::AnyEventLog>,
1695 path: &str,
1696 store_base: &Path,
1697 started_at_ms: i64,
1698 exit_code: i32,
1699 options: &RunAttestationOptions,
1700 stderr: &mut String,
1701) -> Result<(), String> {
1702 let finished_at_ms = now_ms();
1703 let status = if exit_code == 0 { "success" } else { "failure" };
1704 append_run_provenance_event(
1705 log,
1706 "finished",
1707 serde_json::json!({
1708 "pipeline": path,
1709 "status": status,
1710 "exit_code": exit_code,
1711 }),
1712 )
1713 .await;
1714 log.flush()
1715 .await
1716 .map_err(|error| format!("failed to flush attestation event log: {error}"))?;
1717 let secret_provider = harn_vm::secrets::configured_default_chain("harn.provenance")
1718 .map_err(|error| format!("failed to configure provenance secrets: {error}"))?;
1719 let (signing_key, key_id) =
1720 harn_vm::load_or_generate_agent_signing_key(&secret_provider, options.agent_id.as_deref())
1721 .await
1722 .map_err(|error| format!("failed to load provenance signing key: {error}"))?;
1723 let receipt = harn_vm::build_signed_receipt(
1724 log,
1725 harn_vm::ReceiptBuildOptions {
1726 pipeline: path.to_string(),
1727 status: status.to_string(),
1728 started_at_ms,
1729 finished_at_ms,
1730 exit_code,
1731 producer_name: "harn-cli".to_string(),
1732 producer_version: env!("CARGO_PKG_VERSION").to_string(),
1733 },
1734 &signing_key,
1735 key_id,
1736 )
1737 .await
1738 .map_err(|error| format!("failed to build provenance receipt: {error}"))?;
1739 let receipt_path = receipt_output_path(store_base, options, &receipt.receipt_id);
1740 if let Some(parent) = receipt_path.parent() {
1741 fs::create_dir_all(parent)
1742 .map_err(|error| format!("failed to create {}: {error}", parent.display()))?;
1743 }
1744 let encoded = serde_json::to_vec_pretty(&receipt)
1745 .map_err(|error| format!("failed to encode provenance receipt: {error}"))?;
1746 fs::write(&receipt_path, encoded)
1747 .map_err(|error| format!("failed to write {}: {error}", receipt_path.display()))?;
1748 stderr.push_str(&format!("provenance receipt: {}\n", receipt_path.display()));
1749 Ok(())
1750}
1751
1752fn receipt_output_path(
1753 store_base: &Path,
1754 options: &RunAttestationOptions,
1755 receipt_id: &str,
1756) -> PathBuf {
1757 if let Some(path) = options.receipt_out.as_ref() {
1758 return path.clone();
1759 }
1760 harn_vm::runtime_paths::state_root(store_base)
1761 .join("receipts")
1762 .join(format!("{receipt_id}.json"))
1763}
1764
1765fn now_ms() -> i64 {
1766 std::time::SystemTime::now()
1767 .duration_since(std::time::UNIX_EPOCH)
1768 .map(|duration| duration.as_millis() as i64)
1769 .unwrap_or(0)
1770}
1771
1772fn exit_code_from_return_value(value: &harn_vm::VmValue) -> i32 {
1779 use harn_vm::VmValue;
1780 match value {
1781 VmValue::Int(n) => (*n).clamp(0, 255) as i32,
1782 VmValue::EnumVariant(enum_variant) if enum_variant.is_variant("Result", "Err") => 1,
1783 _ => 0,
1784 }
1785}
1786
1787struct JsonRunSession {
1797 emitter: self::json_events::NdjsonEmitter,
1798}
1799
1800impl JsonRunSession {
1801 fn new(options: RunJsonOptions, out: Box<dyn io::Write + Send>) -> Self {
1802 Self {
1803 emitter: NdjsonEmitter::new(out, options.quiet),
1804 }
1805 }
1806
1807 fn sink(&self) -> Arc<dyn harn_vm::run_events::RunEventSink> {
1808 self.emitter.sink()
1809 }
1810
1811 fn finalize_result(self, value: serde_json::Value, exit_code: i32) -> RunOutcome {
1812 self.emitter.emit_result(value, exit_code);
1813 RunOutcome {
1814 stdout: String::new(),
1815 stderr: String::new(),
1816 exit_code,
1817 }
1818 }
1819
1820 fn finalize_error(
1821 self,
1822 code: impl Into<String>,
1823 message: impl Into<String>,
1824 exit_code: i32,
1825 ) -> RunOutcome {
1826 self.emitter.emit_error(code, message);
1827 RunOutcome {
1828 stdout: String::new(),
1829 stderr: String::new(),
1830 exit_code,
1831 }
1832 }
1833}
1834
1835#[allow(clippy::too_many_arguments)]
1836fn finalize_run_error(
1837 stdout: String,
1838 mut stderr: String,
1839 json_session: Option<JsonRunSession>,
1840 summary: Option<&RunSummaryOptions>,
1841 phase: Option<&RunPhaseOptions>,
1842 rusage: Option<&RunRusageOptions>,
1843 started: Instant,
1844 profile: Option<&harn_vm::profile::RunProfile>,
1845 timing: Option<&RunTiming>,
1846 main_events: u64,
1847 cpu_ms_total: Option<u64>,
1848 code: impl Into<String>,
1849 message: impl Into<String>,
1850) -> RunOutcome {
1851 let aux_emission = emit_run_aux_for_exit(
1852 summary,
1853 phase,
1854 rusage,
1855 started,
1856 1,
1857 profile,
1858 None,
1859 timing,
1860 main_events,
1861 cpu_ms_total,
1862 json_session.is_some(),
1863 &mut stderr,
1864 );
1865 if let Some(session) = json_session {
1866 let mut outcome = session.finalize_error(code, message, aux_emission.exit_code);
1867 outcome.stderr = aux_emission.stderr;
1868 return outcome;
1869 }
1870 RunOutcome {
1871 stdout,
1872 stderr,
1873 exit_code: aux_emission.exit_code,
1874 }
1875}
1876
1877fn finalize_harnpack_error(
1882 mut stderr: String,
1883 json_session: Option<JsonRunSession>,
1884 summary: Option<&RunSummaryOptions>,
1885 phase: Option<&RunPhaseOptions>,
1886 rusage: Option<&RunRusageOptions>,
1887 started: Instant,
1888 err: HarnpackError,
1889) -> RunOutcome {
1890 let code = err.code;
1891 let message = err.message;
1892 stderr.push_str(&format!("error: {message}\n"));
1893 finalize_run_error(
1894 String::new(),
1895 stderr,
1896 json_session,
1897 summary,
1898 phase,
1899 rusage,
1900 started,
1901 None,
1902 None,
1903 0,
1904 None,
1905 code,
1906 message,
1907 )
1908}
1909
1910fn finalize_harnpack_dry_run(
1915 mut stderr: String,
1916 json_session: Option<JsonRunSession>,
1917 summary_options: Option<&RunSummaryOptions>,
1918 phase_options: Option<&RunPhaseOptions>,
1919 rusage_options: Option<&RunRusageOptions>,
1920 started: Instant,
1921 cpu_ms_total: Option<u64>,
1922 prepared: &PreparedHarnpack,
1923) -> RunOutcome {
1924 let summary = format!(
1925 "[harn] harnpack verify ok: bundle_hash={}, signature_verified={}, cache_hit={}\n",
1926 prepared.bundle_hash, prepared.signature_verified, prepared.cache_hit
1927 );
1928 stderr.push_str(&summary);
1929 let aux_emission = emit_run_aux_for_exit(
1930 summary_options,
1931 phase_options,
1932 rusage_options,
1933 started,
1934 0,
1935 None,
1936 None,
1937 None,
1938 0,
1939 cpu_ms_total,
1940 json_session.is_some(),
1941 &mut stderr,
1942 );
1943 if let Some(session) = json_session {
1944 if let Some(error) = aux_emission.error {
1945 let mut outcome = session.finalize_error(
1946 "run_aux",
1947 format!("failed to emit auxiliary run JSON: {error}"),
1948 1,
1949 );
1950 outcome.stderr = aux_emission.stderr;
1951 return outcome;
1952 }
1953 let value = serde_json::json!({
1954 "bundle_hash": prepared.bundle_hash,
1955 "signature_verified": prepared.signature_verified,
1956 "key_id": prepared.key_id,
1957 "cache_hit": prepared.cache_hit,
1958 "dry_run_verify": true,
1959 });
1960 let mut outcome = session.finalize_result(value, aux_emission.exit_code);
1961 outcome.stderr = aux_emission.stderr;
1962 return outcome;
1963 }
1964 RunOutcome {
1965 stdout: String::new(),
1966 stderr,
1967 exit_code: aux_emission.exit_code,
1968 }
1969}
1970
1971fn render_return_value_error(value: &harn_vm::VmValue) -> String {
1972 let harn_vm::VmValue::EnumVariant(enum_variant) = value else {
1973 return String::new();
1974 };
1975 if !enum_variant.is_variant("Result", "Err") {
1976 return String::new();
1977 }
1978 let rendered = enum_variant
1979 .fields
1980 .first()
1981 .map(|p| p.display())
1982 .unwrap_or_default();
1983 if rendered.is_empty() {
1984 "error\n".to_string()
1985 } else if rendered.ends_with('\n') {
1986 rendered
1987 } else {
1988 format!("{rendered}\n")
1989 }
1990}
1991
1992pub(crate) fn render_trace_summary() -> String {
1993 use std::fmt::Write;
1994 let entries = harn_vm::llm::take_trace();
1995 if entries.is_empty() {
1996 return String::new();
1997 }
1998 let mut out = String::new();
1999 let _ = writeln!(out, "\n\x1b[2m─── LLM trace ───\x1b[0m");
2000 let mut total_input = 0i64;
2001 let mut total_output = 0i64;
2002 let mut total_ms = 0u64;
2003 for (i, entry) in entries.iter().enumerate() {
2004 let _ = writeln!(
2005 out,
2006 " #{}: {} | {} in + {} out tokens | {} ms",
2007 i + 1,
2008 entry.model,
2009 entry.input_tokens,
2010 entry.output_tokens,
2011 entry.duration_ms,
2012 );
2013 total_input += entry.input_tokens;
2014 total_output += entry.output_tokens;
2015 total_ms += entry.duration_ms;
2016 }
2017 let total_tokens = total_input + total_output;
2018 let cost = (total_input as f64 * 3.0 + total_output as f64 * 15.0) / 1_000_000.0;
2020 let _ = writeln!(
2021 out,
2022 " \x1b[1m{} call{}, {} tokens ({}in + {}out), {} ms, ~${:.4}\x1b[0m",
2023 entries.len(),
2024 if entries.len() == 1 { "" } else { "s" },
2025 total_tokens,
2026 total_input,
2027 total_output,
2028 total_ms,
2029 cost,
2030 );
2031 out
2032}
2033
2034pub(crate) async fn run_file_mcp_serve(
2048 path: &str,
2049 card_source: Option<&str>,
2050 mode: RunFileMcpServeMode,
2051) {
2052 let mut diagnostics = String::new();
2053 let Some(LoadedChunk { source, chunk }) = compile_or_load_chunk_for_run(path, &mut diagnostics)
2054 else {
2055 eprint!("{diagnostics}");
2056 process::exit(1);
2057 };
2058 if !diagnostics.is_empty() {
2059 eprint!("{diagnostics}");
2060 }
2061
2062 let mut vm = harn_vm::Vm::new();
2063 harn_vm::register_vm_stdlib(&mut vm);
2064 crate::install_default_hostlib(&mut vm);
2065 let source_parent = std::path::Path::new(path)
2066 .parent()
2067 .unwrap_or(std::path::Path::new("."));
2068 let project_root = harn_vm::stdlib::process::find_project_root(source_parent);
2069 let store_base = project_root.as_deref().unwrap_or(source_parent);
2070 harn_vm::register_store_builtins(&mut vm, store_base);
2071 harn_vm::register_metadata_builtins(&mut vm, store_base);
2072 let pipeline_name = std::path::Path::new(path)
2073 .file_stem()
2074 .and_then(|s| s.to_str())
2075 .unwrap_or("default");
2076 harn_vm::register_checkpoint_builtins(&mut vm, store_base, pipeline_name);
2077 vm.set_source_info(path, &source);
2078 if let Some(ref root) = project_root {
2079 vm.set_project_root(root);
2080 }
2081 vm.set_source_dir(&entry_source_dir(path));
2085
2086 let loaded = load_skills(&SkillLoaderInputs {
2088 cli_dirs: Vec::new(),
2089 source_path: Some(std::path::PathBuf::from(path)),
2090 });
2091 emit_loader_warnings(&loaded.loader_warnings);
2092 install_skills_global(&mut vm, &loaded);
2093
2094 if let Err(error) =
2095 manifest_runtime::install_manifest_runtime(Path::new(path), &mut vm, false).await
2096 {
2097 eprintln!("error: failed to install {}: {error}", error.label());
2098 process::exit(1);
2099 }
2100
2101 vm.set_source_dir(&entry_source_dir(path));
2106 let local = tokio::task::LocalSet::new();
2107 local
2108 .run_until(async {
2109 match vm.execute(&chunk).await {
2110 Ok(_) => {}
2111 Err(error) => crate::commands::serve::exit_after_mcp_pipeline_error(&vm, &error),
2112 }
2113
2114 let output = vm.output();
2116 if !output.is_empty() {
2117 eprint!("{output}");
2118 }
2119
2120 let registry = match harn_vm::take_mcp_serve_registry() {
2121 Some(r) => r,
2122 None => {
2123 eprintln!("error: pipeline did not call mcp_serve(registry)");
2124 eprintln!("hint: call mcp_serve(tools) at the end of your pipeline");
2125 process::exit(1);
2126 }
2127 };
2128
2129 let tools = match harn_vm::tool_registry_to_mcp_tools(®istry) {
2130 Ok(t) => t,
2131 Err(e) => {
2132 eprintln!("error: {e}");
2133 process::exit(1);
2134 }
2135 };
2136
2137 let resources = harn_vm::take_mcp_serve_resources();
2138 let resource_templates = harn_vm::take_mcp_serve_resource_templates();
2139 let prompts = harn_vm::take_mcp_serve_prompts();
2140 let metadata = harn_vm::take_mcp_serve_metadata();
2141
2142 let mut server_name = std::path::Path::new(path)
2143 .file_stem()
2144 .and_then(|s| s.to_str())
2145 .unwrap_or("harn")
2146 .to_string();
2147 if let Some(name) = metadata
2148 .as_ref()
2149 .and_then(|metadata| metadata.name.as_ref())
2150 {
2151 server_name = name.clone();
2152 }
2153
2154 let mut caps = Vec::new();
2155 if !tools.is_empty() {
2156 caps.push(format!(
2157 "{} tool{}",
2158 tools.len(),
2159 if tools.len() == 1 { "" } else { "s" }
2160 ));
2161 }
2162 let total_resources = resources.len() + resource_templates.len();
2163 if total_resources > 0 {
2164 caps.push(format!(
2165 "{total_resources} resource{}",
2166 if total_resources == 1 { "" } else { "s" }
2167 ));
2168 }
2169 if !prompts.is_empty() {
2170 caps.push(format!(
2171 "{} prompt{}",
2172 prompts.len(),
2173 if prompts.len() == 1 { "" } else { "s" }
2174 ));
2175 }
2176 eprintln!(
2177 "[harn] serve mcp: serving {} as '{server_name}'",
2178 caps.join(", ")
2179 );
2180
2181 let mut server =
2182 harn_vm::McpServer::new(server_name, tools, resources, resource_templates, prompts);
2183 if let Some(metadata) = metadata {
2184 server = server.with_metadata(metadata);
2185 }
2186 if let Some(source) = card_source {
2187 match resolve_card_source(source) {
2188 Ok(card) => server = server.with_server_card(card),
2189 Err(e) => {
2190 eprintln!("error: --card: {e}");
2191 process::exit(1);
2192 }
2193 }
2194 }
2195 match mode {
2196 RunFileMcpServeMode::Stdio => {
2197 if let Err(e) = server.run(&mut vm).await {
2198 eprintln!("error: MCP server error: {e}");
2199 process::exit(1);
2200 }
2201 }
2202 RunFileMcpServeMode::Http(http) => {
2203 let RunFileMcpServeHttp {
2204 options,
2205 auth_policy,
2206 } = *http;
2207 if let Err(e) = crate::commands::serve::run_script_mcp_http_server(
2208 server,
2209 vm,
2210 options,
2211 auth_policy,
2212 )
2213 .await
2214 {
2215 eprintln!("error: MCP server error: {e}");
2216 process::exit(1);
2217 }
2218 }
2219 }
2220 })
2221 .await;
2222}
2223
2224pub(crate) fn resolve_card_source(source: &str) -> Result<serde_json::Value, String> {
2229 let trimmed = source.trim_start();
2230 if trimmed.starts_with('{') || trimmed.starts_with('[') {
2231 return serde_json::from_str(source).map_err(|e| format!("inline JSON parse error: {e}"));
2232 }
2233 let path = std::path::Path::new(source);
2234 harn_vm::load_server_card_from_path(path).map_err(|e| format!("{e}"))
2235}
2236
2237pub(crate) async fn run_watch(path: &str, denied_builtins: HashSet<String>) {
2238 use notify::{Event, EventKind, RecursiveMode, Watcher};
2239
2240 let abs_path = std::fs::canonicalize(path).unwrap_or_else(|e| {
2241 eprintln!("Error: {e}");
2242 process::exit(1);
2243 });
2244 let watch_dir = abs_path.parent().unwrap_or(Path::new("."));
2245
2246 eprintln!("\x1b[2m[watch] running {path}...\x1b[0m");
2247 run_file(
2248 path,
2249 false,
2250 denied_builtins.clone(),
2251 Vec::new(),
2252 CliLlmMockMode::Off,
2253 None,
2254 RunProfileOptions::default(),
2255 )
2256 .await;
2257
2258 let (tx, mut rx) = tokio::sync::mpsc::channel::<()>(1);
2259 let _watcher = {
2260 let tx = tx.clone();
2261 let mut watcher = notify::recommended_watcher(move |res: Result<Event, _>| {
2262 if let Ok(event) = res {
2263 if matches!(
2264 event.kind,
2265 EventKind::Modify(_) | EventKind::Create(_) | EventKind::Remove(_)
2266 ) {
2267 let has_harn = event
2268 .paths
2269 .iter()
2270 .any(|p| p.extension().is_some_and(|ext| ext == "harn"));
2271 if has_harn {
2272 let _ = tx.blocking_send(());
2273 }
2274 }
2275 }
2276 })
2277 .unwrap_or_else(|e| {
2278 eprintln!("Error setting up file watcher: {e}");
2279 process::exit(1);
2280 });
2281 watcher
2282 .watch(watch_dir, RecursiveMode::Recursive)
2283 .unwrap_or_else(|e| {
2284 eprintln!("Error watching directory: {e}");
2285 process::exit(1);
2286 });
2287 watcher };
2289
2290 eprintln!(
2291 "\x1b[2m[watch] watching {} for .harn changes (ctrl-c to stop)\x1b[0m",
2292 watch_dir.display()
2293 );
2294
2295 loop {
2296 rx.recv().await;
2297 tokio::time::sleep(std::time::Duration::from_millis(200)).await;
2299 while rx.try_recv().is_ok() {}
2300
2301 eprintln!();
2302 eprintln!("\x1b[2m[watch] change detected, re-running {path}...\x1b[0m");
2303 run_file(
2304 path,
2305 false,
2306 denied_builtins.clone(),
2307 Vec::new(),
2308 CliLlmMockMode::Off,
2309 None,
2310 RunProfileOptions::default(),
2311 )
2312 .await;
2313 }
2314}
2315
2316#[cfg(test)]
2317mod tests;