harn_vm/orchestration/
mod.rs

1use std::path::PathBuf;
2use std::{cell::RefCell, thread_local};
3
4use serde::{Deserialize, Serialize};
5
6use crate::llm::vm_value_to_json;
7use crate::value::{VmError, VmValue};
8
9pub(crate) fn now_rfc3339() -> String {
10    use std::time::{SystemTime, UNIX_EPOCH};
11    let ts = SystemTime::now()
12        .duration_since(UNIX_EPOCH)
13        .unwrap_or_default()
14        .as_secs();
15    format!("{ts}")
16}
17
18pub(crate) fn new_id(prefix: &str) -> String {
19    format!("{prefix}_{}", uuid::Uuid::now_v7())
20}
21
22pub(crate) fn default_run_dir() -> PathBuf {
23    let base = std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."));
24    crate::runtime_paths::run_root(&base)
25}
26
27mod hooks;
28pub use hooks::*;
29#[cfg(test)]
30mod tests_lazy_hooks;
31
32mod pipeline_lifecycle;
33pub use pipeline_lifecycle::*;
34
35mod settlement_agent;
36pub use settlement_agent::*;
37
38mod lifecycle_receipts;
39pub use lifecycle_receipts::*;
40
41mod command_policy;
42pub use command_policy::*;
43
44mod compaction;
45pub use compaction::*;
46
47mod repair_ledger;
48
49mod compact_lifecycle;
50pub use compact_lifecycle::*;
51
52mod compaction_policy_registry;
53pub use compaction_policy_registry::*;
54
55pub mod agent_inbox;
56
57mod artifacts;
58pub use artifacts::*;
59
60mod assemble;
61pub use assemble::*;
62
63mod handoffs;
64pub use handoffs::*;
65
66mod friction;
67pub use friction::*;
68
69mod crystallize;
70pub use crystallize::*;
71
72mod release_fixture;
73pub use release_fixture::*;
74
75mod replay_oracle;
76pub use replay_oracle::*;
77
78mod replay_bench;
79pub use replay_bench::*;
80
81mod policy;
82pub use policy::*;
83
84mod ambient_scope;
85pub(crate) use ambient_scope::{scope_ambient, AmbientExecutionScope};
86
87mod stage_options;
88pub use stage_options::*;
89
90mod workflow;
91pub use workflow::*;
92
93mod workflow_bundle;
94pub use workflow_bundle::*;
95
96mod workflow_patch;
97pub use workflow_patch::*;
98
99mod safe_function_tools;
100pub use safe_function_tools::*;
101
102mod nested_invocation;
103pub use nested_invocation::*;
104
105#[cfg(test)]
106mod workflow_test_fixtures;
107
108mod records;
109pub use records::*;
110
111mod context_eval;
112pub use context_eval::*;
113
114mod skill_gate;
115pub use skill_gate::*;
116
117mod merge_captain_audit;
118pub use merge_captain_audit::*;
119
120mod merge_captain_driver;
121pub use merge_captain_driver::*;
122
123mod merge_captain_ladder;
124pub use merge_captain_ladder::*;
125
126mod merge_captain_iteration;
127pub use merge_captain_iteration::*;
128
129pub mod playground;
130
131thread_local! {
132    static CURRENT_MUTATION_SESSION: RefCell<Option<MutationSessionRecord>> = const { RefCell::new(None) };
133    /// Workflow-level skill context, installed by `workflow_execute` so
134    /// every per-node agent loop constructed inside `execute_stage_node`
135    /// can pick up the same `skills:` / `skill_match:` registry without
136    /// threading a new parameter through every helper. Cleared on
137    /// workflow exit (success or error) by `WorkflowSkillContextGuard`.
138    static CURRENT_WORKFLOW_SKILL_CONTEXT: RefCell<Option<WorkflowSkillContext>> = const { RefCell::new(None) };
139}
140
141/// Skill wiring threaded from `workflow_execute` into the per-stage
142/// agent loops via thread-local context. The workflow runner pins itself
143/// to one task via `LocalSet`, so every stage observes the same context
144/// without cross-task synchronization.
145#[derive(Clone, Default)]
146pub struct WorkflowSkillContext {
147    pub registry: Option<VmValue>,
148    pub match_config: Option<VmValue>,
149}
150
151pub fn install_workflow_skill_context(context: Option<WorkflowSkillContext>) {
152    CURRENT_WORKFLOW_SKILL_CONTEXT.with(|slot| {
153        *slot.borrow_mut() = context;
154    });
155}
156
157pub fn current_workflow_skill_context() -> Option<WorkflowSkillContext> {
158    CURRENT_WORKFLOW_SKILL_CONTEXT.with(|slot| slot.borrow().clone())
159}
160
161/// RAII guard that clears the workflow skill context on drop. Paired
162/// with `install_workflow_skill_context` at the top of `execute_workflow`
163/// so the context never leaks past a workflow's scope.
164pub struct WorkflowSkillContextGuard;
165
166impl Drop for WorkflowSkillContextGuard {
167    fn drop(&mut self) {
168        install_workflow_skill_context(None);
169    }
170}
171
172#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
173#[serde(default)]
174pub struct MutationSessionRecord {
175    pub session_id: String,
176    pub parent_session_id: Option<String>,
177    pub run_id: Option<String>,
178    pub worker_id: Option<String>,
179    pub execution_kind: Option<String>,
180    pub mutation_scope: String,
181    /// Declarative per-tool approval policy for this session. When `None`,
182    /// no policy-driven approval is requested; the session update stream
183    /// remains the only host-observable surface for tool dispatch.
184    pub approval_policy: Option<ToolApprovalPolicy>,
185}
186
187impl MutationSessionRecord {
188    pub fn normalize(mut self) -> Self {
189        if self.session_id.is_empty() {
190            self.session_id = new_id("session");
191        }
192        if self.mutation_scope.is_empty() {
193            self.mutation_scope = "read_only".to_string();
194        }
195        self
196    }
197}
198
199pub fn install_current_mutation_session(session: Option<MutationSessionRecord>) {
200    CURRENT_MUTATION_SESSION.with(|slot| {
201        *slot.borrow_mut() = session.map(MutationSessionRecord::normalize);
202    });
203}
204
205pub fn current_mutation_session() -> Option<MutationSessionRecord> {
206    CURRENT_MUTATION_SESSION.with(|slot| slot.borrow().clone())
207}
208
209/// Per-task ambient-scope swap of the current mutation session. See
210/// `orchestration::ambient_scope`: the mutation session attributes audit
211/// records, `run_id`, approval policy, and secret-access scope to the running
212/// task, so a worker holding it across an `.await` must keep its OWN copy rather
213/// than read whatever a cooperatively-scheduled fan-out sibling left behind. The
214/// helper is `pub(crate)` — only the ambient combinator moves whole sessions;
215/// ordinary code uses `install_current_mutation_session`/`current_mutation_session`.
216pub(crate) fn swap_mutation_session(
217    next: Option<MutationSessionRecord>,
218) -> Option<MutationSessionRecord> {
219    CURRENT_MUTATION_SESSION.with(|slot| std::mem::replace(&mut *slot.borrow_mut(), next))
220}
221pub(crate) fn parse_json_payload<T: for<'de> Deserialize<'de>>(
222    json: serde_json::Value,
223    label: &str,
224) -> Result<T, VmError> {
225    let payload = json.to_string();
226    let mut deserializer = serde_json::Deserializer::from_str(&payload);
227    let mut tracker = serde_path_to_error::Track::new();
228    let path_deserializer = serde_path_to_error::Deserializer::new(&mut deserializer, &mut tracker);
229    T::deserialize(path_deserializer).map_err(|error| {
230        let snippet = if payload.len() > 600 {
231            format!("{}...", &payload[..600])
232        } else {
233            payload.clone()
234        };
235        VmError::Runtime(format!(
236            "{label} parse error at {}: {} | payload={}",
237            tracker.path(),
238            error,
239            snippet
240        ))
241    })
242}
243
244pub(crate) fn parse_json_value<T: for<'de> Deserialize<'de>>(
245    value: &VmValue,
246) -> Result<T, VmError> {
247    parse_json_payload(vm_value_to_json(value), "orchestration")
248}
249
250#[cfg(test)]
251mod tests;
252
253#[cfg(test)]
254mod typed_options_parity;
harn_vm/orchestration/mod.rs

harn_vm/orchestration/
mod.rs
