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
30mod compaction;
31pub use compaction::*;
32
33mod artifacts;
34pub use artifacts::*;
35
36mod policy;
37pub use policy::*;
38
39mod workflow;
40pub use workflow::*;
41
42mod records;
43pub use records::*;
44
45thread_local! {
46    static CURRENT_MUTATION_SESSION: RefCell<Option<MutationSessionRecord>> = const { RefCell::new(None) };
47    /// Workflow-level skill context, installed by `workflow_execute` so
48    /// every per-node agent loop constructed inside `execute_stage_node`
49    /// can pick up the same `skills:` / `skill_match:` registry without
50    /// threading a new parameter through every helper. Cleared on
51    /// workflow exit (success or error) by `WorkflowSkillContextGuard`.
52    static CURRENT_WORKFLOW_SKILL_CONTEXT: RefCell<Option<WorkflowSkillContext>> = const { RefCell::new(None) };
53}
54
55/// Skill wiring threaded from `workflow_execute` into the per-stage
56/// agent loops via thread-local context. `VmValue` wraps `Rc` and is
57/// not `Send`, so we store it in a thread-local rather than a mutex —
58/// the workflow runner pins itself to one task via `LocalSet`, so
59/// every stage observes the same context.
60#[derive(Clone, Default)]
61pub struct WorkflowSkillContext {
62    pub registry: Option<VmValue>,
63    pub match_config: Option<VmValue>,
64}
65
66pub fn install_workflow_skill_context(context: Option<WorkflowSkillContext>) {
67    CURRENT_WORKFLOW_SKILL_CONTEXT.with(|slot| {
68        *slot.borrow_mut() = context;
69    });
70}
71
72pub fn current_workflow_skill_context() -> Option<WorkflowSkillContext> {
73    CURRENT_WORKFLOW_SKILL_CONTEXT.with(|slot| slot.borrow().clone())
74}
75
76/// RAII guard that clears the workflow skill context on drop. Paired
77/// with `install_workflow_skill_context` at the top of `execute_workflow`
78/// so the context never leaks past a workflow's scope.
79pub struct WorkflowSkillContextGuard;
80
81impl Drop for WorkflowSkillContextGuard {
82    fn drop(&mut self) {
83        install_workflow_skill_context(None);
84    }
85}
86
87#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
88#[serde(default)]
89pub struct MutationSessionRecord {
90    pub session_id: String,
91    pub parent_session_id: Option<String>,
92    pub run_id: Option<String>,
93    pub worker_id: Option<String>,
94    pub execution_kind: Option<String>,
95    pub mutation_scope: String,
96    /// Declarative per-tool approval policy for this session. When `None`,
97    /// no policy-driven approval is requested; the session update stream
98    /// remains the only host-observable surface for tool dispatch.
99    pub approval_policy: Option<ToolApprovalPolicy>,
100}
101
102impl MutationSessionRecord {
103    pub fn normalize(mut self) -> Self {
104        if self.session_id.is_empty() {
105            self.session_id = new_id("session");
106        }
107        if self.mutation_scope.is_empty() {
108            self.mutation_scope = "read_only".to_string();
109        }
110        self
111    }
112}
113
114pub fn install_current_mutation_session(session: Option<MutationSessionRecord>) {
115    CURRENT_MUTATION_SESSION.with(|slot| {
116        *slot.borrow_mut() = session.map(MutationSessionRecord::normalize);
117    });
118}
119
120pub fn current_mutation_session() -> Option<MutationSessionRecord> {
121    CURRENT_MUTATION_SESSION.with(|slot| slot.borrow().clone())
122}
123pub(crate) fn parse_json_payload<T: for<'de> Deserialize<'de>>(
124    json: serde_json::Value,
125    label: &str,
126) -> Result<T, VmError> {
127    let payload = json.to_string();
128    let mut deserializer = serde_json::Deserializer::from_str(&payload);
129    let mut tracker = serde_path_to_error::Track::new();
130    let path_deserializer = serde_path_to_error::Deserializer::new(&mut deserializer, &mut tracker);
131    T::deserialize(path_deserializer).map_err(|error| {
132        let snippet = if payload.len() > 600 {
133            format!("{}...", &payload[..600])
134        } else {
135            payload.clone()
136        };
137        VmError::Runtime(format!(
138            "{label} parse error at {}: {} | payload={}",
139            tracker.path(),
140            error,
141            snippet
142        ))
143    })
144}
145
146pub(crate) fn parse_json_value<T: for<'de> Deserialize<'de>>(
147    value: &VmValue,
148) -> Result<T, VmError> {
149    parse_json_payload(vm_value_to_json(value), "orchestration")
150}
151
152#[cfg(test)]
153mod tests;
harn_vm/orchestration/mod.rs

harn_vm/orchestration/
mod.rs
