harn_vm/vm/
callable_entry.rs

1use std::sync::Arc;
2use std::time::Duration;
3
4use crate::chunk::ChunkRef;
5use crate::value::{VmError, VmValue};
6
7use super::{ScopeSpan, Vm};
8
9pub(super) enum TopLevelEntry {
10    Chunk(ChunkRef),
11    Callable {
12        bootstrap: ChunkRef,
13        has_fixture: bool,
14        fixture_expects_harness: bool,
15        expects_harness: bool,
16        args: Vec<VmValue>,
17    },
18}
19
20impl TopLevelEntry {
21    pub(super) async fn run(self, vm: &mut Vm) -> Result<VmValue, VmError> {
22        match self {
23            Self::Chunk(chunk) => vm.run_chunk(chunk).await,
24            Self::Callable {
25                bootstrap,
26                has_fixture,
27                fixture_expects_harness,
28                expects_harness,
29                mut args,
30            } => {
31                let value = vm.run_chunk(bootstrap).await?;
32                let target = if has_fixture {
33                    let VmValue::List(callables) = value else {
34                        return Err(VmError::Runtime(
35                            "callable entry bootstrap did not return [fixture, target]".to_string(),
36                        ));
37                    };
38                    let [VmValue::Closure(fixture), VmValue::Closure(target)] =
39                        callables.as_slice()
40                    else {
41                        return Err(VmError::Runtime(
42                            "callable entry bootstrap returned invalid fixture callables"
43                                .to_string(),
44                        ));
45                    };
46                    let fixture_args = if fixture_expects_harness {
47                        vec![vm.root_harness_value().ok_or_else(|| {
48                            VmError::Runtime(
49                                "test fixture requires Harness, but no root Harness is installed"
50                                    .to_string(),
51                            )
52                        })?]
53                    } else {
54                        Vec::new()
55                    };
56                    let fixture_value = vm.call_closure_pub(fixture, &fixture_args).await?;
57                    args.insert(0, fixture_value);
58                    Arc::clone(target)
59                } else {
60                    let VmValue::Closure(target) = value else {
61                        return Err(VmError::Runtime(
62                            "callable entry bootstrap did not return a callable".to_string(),
63                        ));
64                    };
65                    target
66                };
67                if expects_harness {
68                    let harness = vm.root_harness_value().ok_or_else(|| {
69                        VmError::Runtime(
70                            "callable entry requires Harness, but no root Harness is installed"
71                                .to_string(),
72                        )
73                    })?;
74                    args.insert(0, harness);
75                }
76                vm.call_closure_pub(&target, &args).await
77            }
78        }
79    }
80}
81
82impl Vm {
83    pub(super) fn prepare_top_level_ambient(
84        &mut self,
85    ) -> crate::orchestration::AmbientExecutionScope {
86        self.prepare_execution_for_top_level();
87        crate::orchestration::AmbientExecutionScope::capture_for_top_level_execution(
88            self.execution_id.clone(),
89            self.llm_mock_context.clone(),
90            self.worker_registry.clone(),
91            self.daemon_registry.clone(),
92            self.trigger_registry.clone(),
93            self.session_runtime.clone(),
94            self.tracing_runtime.clone(),
95            self.agent_host_session_runtime.clone(),
96        )
97    }
98
99    /// Execute a compiled callable entry with explicit values under a host
100    /// wall-clock limit.
101    ///
102    /// The entry bootstrap initializes top-level state once. A bundled fixture
103    /// is then invoked with no arguments and its result is prepended to `args`;
104    /// the target callable is invoked through ordinary arity/type guards.
105    /// Pipeline-finish hooks wrap the complete operation exactly once.
106    pub async fn execute_callable_entry_with_timeout(
107        &mut self,
108        entry: &crate::CompiledCallableEntry,
109        args: &[VmValue],
110        timeout: Duration,
111    ) -> Result<VmValue, VmError> {
112        self.execute_top_level_with_timeout(
113            TopLevelEntry::Callable {
114                bootstrap: Arc::new(entry.bootstrap.clone()),
115                has_fixture: entry.has_fixture,
116                fixture_expects_harness: entry.fixture_expects_harness,
117                expects_harness: entry.expects_harness,
118                args: args.to_vec(),
119            },
120            timeout,
121        )
122        .await
123    }
124
125    pub(super) async fn execute_top_level(
126        &mut self,
127        entry: TopLevelEntry,
128    ) -> Result<VmValue, VmError> {
129        self.ensure_execution_available()?;
130        let registry = self.pool_registry.clone();
131        let ambient = self.prepare_top_level_ambient();
132        let execution = crate::stdlib::pool::with_pool_registry_scope(registry, async {
133            self.execute_entry_scoped(entry).await
134        });
135        let result = crate::orchestration::scope_ambient(ambient, execution).await;
136        if self.owns_execution {
137            if let Some(recorder) = self.flight_recorder.as_ref() {
138                match &result {
139                    Ok(_) => recorder.finish("returned", None),
140                    Err(error) => recorder.finish(
141                        if error.process_exit_code().is_some() {
142                            "process_exited"
143                        } else {
144                            "failed"
145                        },
146                        error.process_exit_code(),
147                    ),
148                }
149            }
150        }
151        result
152    }
153
154    async fn execute_entry_scoped(&mut self, entry: TopLevelEntry) -> Result<VmValue, VmError> {
155        let _execution_activity = self
156            .wait_for_graph
157            .register_task(self.runtime_context.task_id.clone());
158        let _span = ScopeSpan::new(crate::tracing::SpanKind::Pipeline, "main".into());
159        match entry.run(self).await {
160            Ok(value) => self.run_pipeline_finish_lifecycle(value).await,
161            Err(error) => {
162                crate::orchestration::clear_pipeline_on_finish();
163                Err(error)
164            }
165        }
166    }
167}
harn_vm/vm/callable_entry.rs

harn_vm/vm/
callable_entry.rs
