Vakyartha documentation

1/// Run a node's done-contract: every entry executes as brokered bash and
2/// must exit 0. First failure rejects the node with the failing check.
3async fn verify_accept(
4 node: &NodeDef,
5 deps: &ExecutorDeps,
6 cancel: &CancellationToken,
7) -> Result<(), String> {
8 if node.accept.is_empty() {
9 return Ok(());
10 }
11 let tool = deps
12 .tools
13 .iter()
14 .find(|t| t.name() == "bash")
15 .ok_or_else(|| "accept requires bash tool".to_string())?;
16 for check in &node.accept {
17 let command = check.trim_start_matches("verify:").trim();
18 let ctx = ToolContext {
19 cwd: deps.cwd.clone(),
20 cancel: cancel.child_token(),
21 sandbox: deps.sandbox.clone(),
22 sandbox_sink: None,
23 agent_id: None,
24 new_documents: Vec::new(),
25 };
26 let args = serde_json::json!({"command": command});
27 authorize_flow_tool("bash", &args, deps).await?;
28 let out = tool.execute(&args, &ctx).await;
29 if out.is_error {
30 return Err(format!(
31 "done-contract failed: `{command}`\n{}",
32 vak_tools::bounded(out.content)
33 ));
34 }
35 let _ = cancel.child_token();
36 }
37 Ok(())
38}
39
40use std::collections::BTreeMap;
41use std::path::PathBuf;
42use std::sync::Arc;
43
44use tokio_util::sync::CancellationToken;
45
46use vak_agent::{Agent, AgentConfig, ApprovalMode, Approver};
47use vak_llm::Provider;
48use vak_permission::{Decision, Mode, PermissionEngine};
49use vak_session::{SessionLog, SessionPath};
50use vak_tools::sandbox::Sandbox;
51use vak_tools::{Tool, ToolContext};
52
53use crate::parse::layers;
54use crate::types::{FlowDef, FlowState, NodeDef, NodeResult, NodeStatus};
55
56#[derive(Clone)]
57pub struct ExecutorDeps {
58 pub provider: Arc<dyn Provider>,
59 pub system_prompt: String,
60 pub prompt_layers: Vec<vak_session::types::PromptLayerDescriptor>,
61 pub model: String,
62 pub tools: Vec<Arc<dyn Tool>>,
63 pub read_only_tools: Vec<Arc<dyn Tool>>,
64 pub max_turns: usize,
65 pub outcome: Option<vak_intent::OutcomeSpec>,
66 pub max_retries: u32,
67 pub retry_base_backoff_ms: u64,
68 pub request_timeout: Option<std::time::Duration>,
69 pub circuit_breaker: Option<Arc<vak_agent::CircuitBreaker>>,
70 pub run_retry_attempts: u32,
71 pub run_retry_base_backoff_ms: u64,
72 pub dispatch_ceiling: u32,
73 pub spend_gate: Option<Arc<dyn vak_agent::SpendGate>>,
74 pub permission: Option<Arc<PermissionEngine>>,
75 pub mode: Mode,
76 pub approval_mode: ApprovalMode,
77 pub approver: Option<Arc<dyn Approver>>,
78 pub sandbox: Option<Arc<dyn Sandbox>>,
79 pub cwd: PathBuf,
80 pub sessions_home: PathBuf,
81 pub parent_session_id: String,
82 /// Where the run-state ledger is persisted.
83 pub state_path: PathBuf,
84 /// Parent Agent/conversation ownership carried into dynamic flow nodes.
85 pub agent_identity: Option<vak_session::types::AgentIdentity>,
86 pub conversation_context: Option<vak_session::ConversationContext>,
87 pub work: Option<FlowWorkContext>,
88}
89
90#[derive(Clone)]
91pub struct FlowWorkContext {
92 pub session: Arc<tokio::sync::Mutex<SessionLog>>,
93 pub contract_id: String,
94 pub work_item_id: String,
95}
96
97#[derive(Debug)]
98pub enum FlowOutcome {
99 Completed {
100 outputs: BTreeMap<String, String>,
101 },
102 Failed {
103 node: String,
104 reason: String,
105 outputs: BTreeMap<String, String>,
106 },
107 Aborted,
108}
109
110pub struct Executor {
111 deps: Arc<ExecutorDeps>,
112}
113
114impl Executor {
115 pub fn new(deps: ExecutorDeps) -> Self {
116 Executor {
117 deps: Arc::new(deps),
118 }
119 }
120
121 pub async fn run(
122 &self,
123 flow: &FlowDef,
124 state: &mut FlowState,
125 cancel: CancellationToken,
126 events: tokio::sync::mpsc::Sender<String>,
127 ) -> FlowOutcome {
128 if state.outcome.is_none() {
129 state.outcome = self.deps.outcome.clone();
130 }
131 let Ok(layer_list) = layers(flow) else {
132 return FlowOutcome::Failed {
133 node: "<flow>".into(),
134 reason: "invalid graph".into(),
135 outputs: BTreeMap::new(),
136 };
137 };
138
139 for id in &layer_list.iter().flatten().cloned().collect::<Vec<_>>() {
140 state.nodes.entry(id.clone()).or_insert(NodeResult {
141 status: NodeStatus::Pending,
142 output: String::new(),
143 });
144 }
145
146 let by_id: BTreeMap<String, NodeDef> = flow
147 .nodes
148 .iter()
149 .map(|n| (n.id.clone(), n.clone()))
150 .collect();
151
152 if let Err(reason) = self.start_work_item().await {
153 return FlowOutcome::Failed {
154 node: "<work>".into(),
155 reason,
156 outputs: BTreeMap::new(),
157 };
158 }
159
160 for layer in &layer_list {
161 let mut runnable = Vec::new();
162 for id in layer {
163 let status = state
164 .nodes
165 .get(id)
166 .map(|r| r.status)
167 .unwrap_or(NodeStatus::Pending);
168 match status {
169 NodeStatus::Completed => continue,
170 NodeStatus::Failed | NodeStatus::Skipped => continue,
171 NodeStatus::Pending | NodeStatus::Running => {}
172 }
173 let Some(node) = by_id.get(id).cloned() else {
174 continue;
175 };
176 let blocked_dep = node.deps.iter().any(|d| {
177 matches!(
178 state.nodes.get(d).map(|r| r.status),
179 Some(NodeStatus::Failed) | Some(NodeStatus::Skipped)
180 )
181 });
182 if blocked_dep && node.r#type != "merge" {
183 state.nodes.insert(
184 id.clone(),
185 NodeResult {
186 status: NodeStatus::Skipped,
187 output: "upstream dependency failed or was skipped".into(),
188 },
189 );
190 let _ = events
191 .send(format!("⊘ skipped {id} (upstream failure)"))
192 .await;
193 continue;
194 }
195 runnable.push(node);
196 }
197
198 let mut join = tokio::task::JoinSet::new();
199 for node in runnable {
200 let deps = self.deps.clone();
201 let cancel = cancel.clone();
202 let events = events.clone();
203 let dep_outputs: BTreeMap<String, String> = node
204 .deps
205 .iter()
206 .filter_map(|d| {
207 state
208 .nodes
209 .get(d)
210 .filter(|r| r.status == NodeStatus::Completed)
211 .map(|r| (d.clone(), r.output.clone()))
212 })
213 .collect();
214 join.spawn(async move {
215 let id = node.id.clone();
216 let mut res = execute_node(&node, &dep_outputs, &deps, &cancel, &events).await;
217 if res.is_ok() && !node.accept.is_empty() {
218 let _ = events
219 .send(format!(
220 "⍗ accept {} ({} check(s))",
221 node.id,
222 node.accept.len()
223 ))
224 .await;
225 res = verify_accept(&node, &deps, &cancel)
226 .await
227 .map(|_| res.unwrap_or_default());
228 }
229 (id, res)
230 });
231 }
232
233 while let Some(res) = join.join_next().await {
234 let Ok((id, result)) = res else {
235 continue;
236 };
237 let (status, output) = match result {
238 Ok(text) => (NodeStatus::Completed, text),
239 Err(reason) => (NodeStatus::Failed, reason),
240 };
241 let failed = status == NodeStatus::Failed;
242 state.nodes.insert(
243 id.clone(),
244 NodeResult {
245 status,
246 output: output.clone(),
247 },
248 );
249 let _ = events
250 .send(format!("{} {}", if failed { "✗" } else { "✓" }, id))
251 .await;
252 self.persist(state);
253
254 if status == NodeStatus::Completed
255 && let Err(reason) = self
256 .record_work_evidence(&id, &state.run_id, &flow.name)
257 .await
258 {
259 let _ = self
260 .finish_work_item(vak_session::types::WorkItemStatus::Failed)
261 .await;
262 return FlowOutcome::Failed {
263 node: id,
264 reason,
265 outputs: collect_outputs(state),
266 };
267 }
268
269 if failed && by_id.get(&id).is_some_and(|n| n.required) {
270 // Mark every transitive dependent skipped so the ledger
271 // reflects why they never ran.
272 let mut stack: Vec<String> = vec![id.clone()];
273 while let Some(failed_id) = stack.pop() {
274 for n in flow.nodes.iter() {
275 if n.deps.contains(&failed_id)
276 && matches!(
277 state.nodes.get(&n.id).map(|r| r.status),
278 None | Some(NodeStatus::Pending)
279 )
280 {
281 state.nodes.insert(
282 n.id.clone(),
283 NodeResult {
284 status: NodeStatus::Skipped,
285 output: format!("upstream '{failed_id}' failed"),
286 },
287 );
288 stack.push(n.id.clone());
289 }
290 }
291 }
292 self.persist(state);
293 let _ = self
294 .finish_work_item(vak_session::types::WorkItemStatus::Failed)
295 .await;
296 return FlowOutcome::Failed {
297 node: id,
298 reason: output,
299 outputs: collect_outputs(state),
300 };
301 }
302 }
303
304 if cancel.is_cancelled() {
305 let _ = self
306 .finish_work_item(vak_session::types::WorkItemStatus::Interrupted)
307 .await;
308 return FlowOutcome::Aborted;
309 }
310 }
311
312 if let Err(reason) = self
313 .record_work_evidence("__flow_completed__", &state.run_id, &flow.name)
314 .await
315 {
316 let _ = self
317 .finish_work_item(vak_session::types::WorkItemStatus::Failed)
318 .await;
319 return FlowOutcome::Failed {
320 node: "<work>".into(),
321 reason,
322 outputs: collect_outputs(state),
323 };
324 }
325 if let Err(reason) = self
326 .finish_work_item(vak_session::types::WorkItemStatus::ReadyForVerification)
327 .await
328 {
329 return FlowOutcome::Failed {
330 node: "<work>".into(),
331 reason,
332 outputs: collect_outputs(state),
333 };
334 }
335 FlowOutcome::Completed {
336 outputs: collect_outputs(state),
337 }
338 }
339
340 fn persist(&self, state: &FlowState) {
341 if let Ok(json) = serde_json::to_string_pretty(state) {
342 if let Some(parent) = self.deps.state_path.parent() {
343 let _ = std::fs::create_dir_all(parent);
344 }
345 let _ = std::fs::write(&self.deps.state_path, json);
346 }
347 }
348
349 async fn record_work_evidence(
350 &self,
351 node_id: &str,
352 run_id: &str,
353 flow_name: &str,
354 ) -> Result<(), String> {
355 let Some(work) = &self.deps.work else {
356 return Ok(());
357 };
358 let mut session = work.session.lock().await;
359 let Some(projection) = session
360 .work_projection()
361 .map_err(|error| format!("flow work projection is invalid: {error}"))?
362 else {
363 return Err("flow work context has no contract".into());
364 };
365 if projection.contract.contract_id != work.contract_id {
366 return Err("flow work context targets a different contract".into());
367 }
368 session
369 .append_work(vak_session::types::WorkEvent {
370 contract_id: work.contract_id.clone(),
371 revision: projection.contract.revision,
372 kind: vak_session::types::WorkEventKind::EvidenceAttached {
373 item_id: work.work_item_id.clone(),
374 evidence: vak_session::types::EvidenceRef::FlowNode {
375 flow: flow_name.into(),
376 run_id: run_id.into(),
377 node_id: node_id.into(),
378 },
379 },
380 })
381 .map_err(|error| format!("flow evidence write failed: {error}"))?;
382 Ok(())
383 }
384
385 async fn start_work_item(&self) -> Result<(), String> {
386 let Some(work) = &self.deps.work else {
387 return Ok(());
388 };
389 let mut session = work.session.lock().await;
390 let Some(projection) = session.work_projection().map_err(|e| e.to_string())? else {
391 return Err("flow work context has no contract".into());
392 };
393 let Some(state) = projection.items.get(&work.work_item_id) else {
394 return Err("flow work context has no work item".into());
395 };
396 if state.status == vak_session::types::WorkItemStatus::Ready {
397 session
398 .append_work(vak_session::types::WorkEvent {
399 contract_id: work.contract_id.clone(),
400 revision: projection.contract.revision,
401 kind: vak_session::types::WorkEventKind::ItemStatusChanged {
402 item_id: work.work_item_id.clone(),
403 from: vak_session::types::WorkItemStatus::Ready,
404 to: vak_session::types::WorkItemStatus::Running,
405 attempt: state.attempt.saturating_add(1),
406 reason: "flow execution started".into(),
407 },
408 })
409 .map_err(|e| e.to_string())?;
410 } else if state.status != vak_session::types::WorkItemStatus::Running {
411 return Err(format!(
412 "flow work item is {:?}, not ready or running",
413 state.status
414 ));
415 }
416 Ok(())
417 }
418
419 async fn finish_work_item(
420 &self,
421 status: vak_session::types::WorkItemStatus,
422 ) -> Result<(), String> {
423 let Some(work) = &self.deps.work else {
424 return Ok(());
425 };
426 let mut session = work.session.lock().await;
427 let Some(projection) = session.work_projection().map_err(|e| e.to_string())? else {
428 return Err("flow work context has no contract".into());
429 };
430 let Some(state) = projection.items.get(&work.work_item_id) else {
431 return Err("flow work context has no work item".into());
432 };
433 if state.status != vak_session::types::WorkItemStatus::Running {
434 return Ok(());
435 }
436 session
437 .append_work(vak_session::types::WorkEvent {
438 contract_id: work.contract_id.clone(),
439 revision: projection.contract.revision,
440 kind: vak_session::types::WorkEventKind::ItemStatusChanged {
441 item_id: work.work_item_id.clone(),
442 from: vak_session::types::WorkItemStatus::Running,
443 to: status,
444 attempt: state.attempt,
445 reason: "flow execution returned".into(),
446 },
447 })
448 .map_err(|e| e.to_string())?;
449 Ok(())
450 }
451}
452
453fn collect_outputs(state: &FlowState) -> BTreeMap<String, String> {
454 state
455 .nodes
456 .iter()
457 .filter(|(_, r)| r.status == NodeStatus::Completed)
458 .map(|(id, r)| (id.clone(), r.output.clone()))
459 .collect()
460}
461
462fn render(template: &str, dep_outputs: &BTreeMap<String, String>) -> String {
463 let mut out = template.to_string();
464 for (id, output) in dep_outputs {
465 out = out.replace(&format!("{{{{{id}}}}}"), output);
466 }
467 out
468}
469
470async fn execute_node(
471 node: &NodeDef,
472 dep_outputs: &BTreeMap<String, String>,
473 deps: &ExecutorDeps,
474 cancel: &CancellationToken,
475 events: &tokio::sync::mpsc::Sender<String>,
476) -> Result<String, String> {
477 match node.r#type.as_str() {
478 "bash" => {
479 let command = render(node.command.as_deref().unwrap_or_default(), dep_outputs);
480 let ctx = ToolContext {
481 cwd: deps.cwd.clone(),
482 cancel: cancel.child_token(),
483 sandbox: deps.sandbox.clone(),
484 sandbox_sink: None,
485 agent_id: None,
486 new_documents: Vec::new(),
487 };
488 let tool = deps
489 .tools
490 .iter()
491 .find(|tool| tool.name() == "bash")
492 .ok_or_else(|| "bash tool unavailable".to_string())?;
493 let args = match node.timeout_ms {
494 Some(t) => serde_json::json!({"command": command, "timeout_ms": t}),
495 None => serde_json::json!({"command": command}),
496 };
497 authorize_flow_tool("bash", &args, deps).await?;
498 let out = tool.execute(&args, &ctx).await;
499 // A node's output is rendered into later nodes, prompts
500 // included, and a flow keeps no evidence ledger to recall from.
501 let content = vak_tools::bounded(out.content);
502 if out.is_error {
503 Err(content)
504 } else {
505 Ok(content)
506 }
507 }
508 "agent" => {
509 let prompt = render(node.prompt.as_deref().unwrap_or_default(), dep_outputs);
510 let readonly = node.readonly;
511 let tools = if readonly {
512 deps.read_only_tools.clone()
513 } else {
514 deps.tools.clone()
515 };
516 let mode = if readonly { Mode::ReadOnly } else { deps.mode };
517
518 let session_id = format!(
519 "flow-{}-{}",
520 node.id,
521 std::time::SystemTime::now()
522 .duration_since(std::time::UNIX_EPOCH)
523 .unwrap_or_default()
524 .as_nanos()
525 );
526 let header = vak_session::types::SessionHeader {
527 agent: deps.agent_identity.clone().or_else(|| {
528 Some(vak_session::types::AgentIdentity {
529 id: "vak".into(),
530 revision: 1,
531 name: "Vakyartha".into(),
532 character: "vak".into(),
533 personality: String::new(),
534 animation: "subtle".into(),
535 voice: "default".into(),
536 behaviour: String::new(),
537 responsibilities: String::new(),
538 instructions: String::new(),
539 })
540 }),
541 session_id: session_id.clone(),
542 created_at: chrono::Utc::now(),
543 cwd: deps.cwd.clone(),
544 parent_session_id: Some(deps.parent_session_id.clone()),
545 contract_id: None,
546 work_item_id: None,
547 conversation: deps
548 .conversation_context
549 .clone()
550 .or_else(|| Some(vak_session::ConversationContext::local(&session_id, "flow"))),
551 contract: vak_session::types::FrozenContract {
552 app_version: env!("CARGO_PKG_VERSION").into(),
553 provider: deps.provider.name().into(),
554 model: deps.model.clone(),
555 route_ladder: Vec::new(),
556 route_objective: String::new(),
557 route_annotations: Vec::new(),
558 system_prompt: deps.system_prompt.clone(),
559 permission_mode: match mode {
560 Mode::ReadOnly => "read-only",
561 Mode::WorkspaceWrite => "workspace-write",
562 Mode::FullAccess => "full-access",
563 }
564 .into(),
565 capabilities: Vec::new(),
566 prompt_layers: deps.prompt_layers.clone(),
567 },
568 };
569 let path = SessionPath::new_session_file(&deps.sessions_home, &deps.cwd, &session_id);
570 let log = SessionLog::create(path, header)
571 .map_err(|e| format!("cannot create node session: {e}"))?;
572
573 let mut cfg = AgentConfig::new(deps.system_prompt.clone());
574 cfg.outcome = deps.outcome.clone();
575 cfg.max_retries = deps.max_retries;
576 cfg.retry_base_backoff_ms = deps.retry_base_backoff_ms;
577 cfg.request_timeout = deps.request_timeout;
578 cfg.circuit_breaker = deps.circuit_breaker.clone();
579 cfg.run_retry_attempts = deps.run_retry_attempts;
580 cfg.run_retry_base_backoff_ms = deps.run_retry_base_backoff_ms;
581 cfg.dispatch_ceiling = deps.dispatch_ceiling;
582 cfg.spend_gate = deps.spend_gate.clone();
583 cfg.model = deps.model.clone();
584 cfg.tools = tools;
585 cfg.max_turns = deps
586 .outcome
587 .as_ref()
588 .and_then(|outcome| outcome.max_turns)
589 .map_or(deps.max_turns, |cap| deps.max_turns.min(cap));
590 cfg.permission = deps.permission.clone();
591 cfg.mode = mode;
592 cfg.approval_mode = deps.approval_mode;
593 cfg.approver = deps.approver.clone();
594 cfg.sandbox = deps.sandbox.clone();
595
596 let mut agent = Agent::new(deps.provider.clone(), log, cfg);
597 let steering = vak_agent::SteeringQueues::new();
598 let (ev_tx, mut ev_rx) = tokio::sync::mpsc::channel::<vak_agent::AgentEvent>(256);
599 let pump = tokio::spawn(async move { while ev_rx.recv().await.is_some() {} });
600 let outcome = agent
601 .run(&prompt, &steering, cancel.child_token(), ev_tx)
602 .await;
603 let _ = pump.await;
604
605 match outcome {
606 vak_agent::TurnOutcome::Completed { response } => {
607 let text = response.text_content();
608 if text.is_empty() {
609 Ok(format!("(node '{}' completed without output)", node.id))
610 } else {
611 Ok(text)
612 }
613 }
614 vak_agent::TurnOutcome::Aborted { partial } => Err(format!(
615 "agent node aborted. Partial:\n{}",
616 partial.map(|p| p.text_content()).unwrap_or_default()
617 )),
618 vak_agent::TurnOutcome::Failed { error } => Err(error.to_string()),
619 vak_agent::TurnOutcome::MaxTurnsReached => {
620 Err("agent node hit its turn limit".into())
621 }
622 }
623 }
624 "approval" => {
625 let message = render(node.message.as_deref().unwrap_or_default(), dep_outputs);
626 let _ = events.send(format!("⏸ approval needed: {message}")).await;
627 match &deps.approver {
628 Some(a) => {
629 if a.approve("approval", "", &message).await {
630 Ok("approved".into())
631 } else {
632 Err("denied by user".into())
633 }
634 }
635 None => Err("no approver available for approval node".into()),
636 }
637 }
638 "merge" => {
639 let mut report = String::from("<merge-report>\n");
640 for dep in &node.deps {
641 match dep_outputs.get(dep) {
642 Some(out) => report.push_str(&format!("[{dep}] ok\n{out}\n")),
643 None => report.push_str(&format!(
644 "[{dep}] unavailable (failed or skipped upstream)\n"
645 )),
646 }
647 }
648 report.push_str("</merge-report>");
649 Ok(report)
650 }
651 other => Err(format!("unknown node type '{other}'")),
652 }
653}
654
655async fn authorize_flow_tool(
656 tool: &str,
657 args: &serde_json::Value,
658 deps: &ExecutorDeps,
659) -> Result<(), String> {
660 let Some(engine) = &deps.permission else {
661 return Ok(());
662 };
663 match engine.evaluate(tool, args, deps.mode, &deps.cwd) {
664 Decision::Allow => Ok(()),
665 Decision::Deny { reason } => Err(reason),
666 Decision::Ask { reason, source } => {
667 if vak_agent::auto_approve(
668 deps.approval_mode,
669 source,
670 tool,
671 args,
672 deps.mode,
673 deps.sandbox.is_some(),
674 &deps.cwd,
675 ) {
676 return Ok(());
677 }
678 match &deps.approver {
679 Some(approver) if approver.approve(tool, &args.to_string(), &reason).await => {
680 Ok(())
681 }
682 Some(_) => Err(format!("denied by user: {reason}")),
683 None => Err(format!("{reason} (no approver available)")),
684 }
685 }
686 }
687}
688

1/// Run a node's done-contract: every entry executes as brokered bash and
2/// must exit 0. First failure rejects the node with the failing check.
3async fn verify_accept(
4 node: &NodeDef,
5 deps: &ExecutorDeps,
6 cancel: &CancellationToken,
7) -> Result<(), String> {
8 if node.accept.is_empty() {
9 return Ok(());
10 }
11 let tool = deps
12 .tools
13 .iter()
14 .find(|t| t.name() == "bash")
15 .ok_or_else(|| "accept requires bash tool".to_string())?;
16 for check in &node.accept {
17 let command = check.trim_start_matches("verify:").trim();
18 let ctx = ToolContext {
19 cwd: deps.cwd.clone(),
20 cancel: cancel.child_token(),
21 sandbox: deps.sandbox.clone(),
22 sandbox_sink: None,
23 agent_id: None,
24 new_documents: Vec::new(),
25 };
26 let args = serde_json::json!({"command": command});
27 authorize_flow_tool("bash", &args, deps).await?;
28 let out = tool.execute(&args, &ctx).await;
29 if out.is_error {
30 return Err(format!(
31 "done-contract failed: `{command}`\n{}",
32 vak_tools::bounded(out.content)
33 ));
34 }
35 let _ = cancel.child_token();
36 }
37 Ok(())
38}
39
40use std::collections::BTreeMap;
41use std::path::PathBuf;
42use std::sync::Arc;
43
44use tokio_util::sync::CancellationToken;
45
46use vak_agent::{Agent, AgentConfig, ApprovalMode, Approver};
47use vak_llm::Provider;
48use vak_permission::{Decision, Mode, PermissionEngine};
49use vak_session::{SessionLog, SessionPath};
50use vak_tools::sandbox::Sandbox;
51use vak_tools::{Tool, ToolContext};
52
53use crate::parse::layers;
54use crate::types::{FlowDef, FlowState, NodeDef, NodeResult, NodeStatus};
55
56#[derive(Clone)]
57pub struct ExecutorDeps {
58 pub provider: Arc<dyn Provider>,
59 pub system_prompt: String,
60 pub prompt_layers: Vec<vak_session::types::PromptLayerDescriptor>,
61 pub model: String,
62 pub tools: Vec<Arc<dyn Tool>>,
63 pub read_only_tools: Vec<Arc<dyn Tool>>,
64 pub max_turns: usize,
65 pub outcome: Option<vak_intent::OutcomeSpec>,
66 pub max_retries: u32,
67 pub retry_base_backoff_ms: u64,
68 pub request_timeout: Option<std::time::Duration>,
69 pub circuit_breaker: Option<Arc<vak_agent::CircuitBreaker>>,
70 pub run_retry_attempts: u32,
71 pub run_retry_base_backoff_ms: u64,
72 pub dispatch_ceiling: u32,
73 pub spend_gate: Option<Arc<dyn vak_agent::SpendGate>>,
74 pub permission: Option<Arc<PermissionEngine>>,
75 pub mode: Mode,
76 pub approval_mode: ApprovalMode,
77 pub approver: Option<Arc<dyn Approver>>,
78 pub sandbox: Option<Arc<dyn Sandbox>>,
79 pub cwd: PathBuf,
80 pub sessions_home: PathBuf,
81 pub parent_session_id: String,
82 /// Where the run-state ledger is persisted.
83 pub state_path: PathBuf,
84 /// Parent Agent/conversation ownership carried into dynamic flow nodes.
85 pub agent_identity: Option<vak_session::types::AgentIdentity>,
86 pub conversation_context: Option<vak_session::ConversationContext>,
87 pub work: Option<FlowWorkContext>,
88}
89
90#[derive(Clone)]
91pub struct FlowWorkContext {
92 pub session: Arc<tokio::sync::Mutex<SessionLog>>,
93 pub contract_id: String,
94 pub work_item_id: String,
95}
96
97#[derive(Debug)]
98pub enum FlowOutcome {
99 Completed {
100 outputs: BTreeMap<String, String>,
101 },
102 Failed {
103 node: String,
104 reason: String,
105 outputs: BTreeMap<String, String>,
106 },
107 Aborted,
108}
109
110pub struct Executor {
111 deps: Arc<ExecutorDeps>,
112}
113
114impl Executor {
115 pub fn new(deps: ExecutorDeps) -> Self {
116 Executor {
117 deps: Arc::new(deps),
118 }
119 }
120
121 pub async fn run(
122 &self,
123 flow: &FlowDef,
124 state: &mut FlowState,
125 cancel: CancellationToken,
126 events: tokio::sync::mpsc::Sender<String>,
127 ) -> FlowOutcome {
128 if state.outcome.is_none() {
129 state.outcome = self.deps.outcome.clone();
130 }
131 let Ok(layer_list) = layers(flow) else {
132 return FlowOutcome::Failed {
133 node: "<flow>".into(),
134 reason: "invalid graph".into(),
135 outputs: BTreeMap::new(),
136 };
137 };
138
139 for id in &layer_list.iter().flatten().cloned().collect::<Vec<_>>() {
140 state.nodes.entry(id.clone()).or_insert(NodeResult {
141 status: NodeStatus::Pending,
142 output: String::new(),
143 });
144 }
145
146 let by_id: BTreeMap<String, NodeDef> = flow
147 .nodes
148 .iter()
149 .map(|n| (n.id.clone(), n.clone()))
150 .collect();
151
152 if let Err(reason) = self.start_work_item().await {
153 return FlowOutcome::Failed {
154 node: "<work>".into(),
155 reason,
156 outputs: BTreeMap::new(),
157 };
158 }
159
160 for layer in &layer_list {
161 let mut runnable = Vec::new();
162 for id in layer {
163 let status = state
164 .nodes
165 .get(id)
166 .map(|r| r.status)
167 .unwrap_or(NodeStatus::Pending);
168 match status {
169 NodeStatus::Completed => continue,
170 NodeStatus::Failed | NodeStatus::Skipped => continue,
171 NodeStatus::Pending | NodeStatus::Running => {}
172 }
173 let Some(node) = by_id.get(id).cloned() else {
174 continue;
175 };
176 let blocked_dep = node.deps.iter().any(|d| {
177 matches!(
178 state.nodes.get(d).map(|r| r.status),
179 Some(NodeStatus::Failed) | Some(NodeStatus::Skipped)
180 )
181 });
182 if blocked_dep && node.r#type != "merge" {
183 state.nodes.insert(
184 id.clone(),
185 NodeResult {
186 status: NodeStatus::Skipped,
187 output: "upstream dependency failed or was skipped".into(),
188 },
189 );
190 let _ = events
191 .send(format!("⊘ skipped {id} (upstream failure)"))
192 .await;
193 continue;
194 }
195 runnable.push(node);
196 }
197
198 let mut join = tokio::task::JoinSet::new();
199 for node in runnable {
200 let deps = self.deps.clone();
201 let cancel = cancel.clone();
202 let events = events.clone();
203 let dep_outputs: BTreeMap<String, String> = node
204 .deps
205 .iter()
206 .filter_map(|d| {
207 state
208 .nodes
209 .get(d)
210 .filter(|r| r.status == NodeStatus::Completed)
211 .map(|r| (d.clone(), r.output.clone()))
212 })
213 .collect();
214 join.spawn(async move {
215 let id = node.id.clone();
216 let mut res = execute_node(&node, &dep_outputs, &deps, &cancel, &events).await;
217 if res.is_ok() && !node.accept.is_empty() {
218 let _ = events
219 .send(format!(
220 "⍗ accept {} ({} check(s))",
221 node.id,
222 node.accept.len()
223 ))
224 .await;
225 res = verify_accept(&node, &deps, &cancel)
226 .await
227 .map(|_| res.unwrap_or_default());
228 }
229 (id, res)
230 });
231 }
232
233 while let Some(res) = join.join_next().await {
234 let Ok((id, result)) = res else {
235 continue;
236 };
237 let (status, output) = match result {
238 Ok(text) => (NodeStatus::Completed, text),
239 Err(reason) => (NodeStatus::Failed, reason),
240 };
241 let failed = status == NodeStatus::Failed;
242 state.nodes.insert(
243 id.clone(),
244 NodeResult {
245 status,
246 output: output.clone(),
247 },
248 );
249 let _ = events
250 .send(format!("{} {}", if failed { "✗" } else { "✓" }, id))
251 .await;
252 self.persist(state);
253
254 if status == NodeStatus::Completed
255 && let Err(reason) = self
256 .record_work_evidence(&id, &state.run_id, &flow.name)
257 .await
258 {
259 let _ = self
260 .finish_work_item(vak_session::types::WorkItemStatus::Failed)
261 .await;
262 return FlowOutcome::Failed {
263 node: id,
264 reason,
265 outputs: collect_outputs(state),
266 };
267 }
268
269 if failed && by_id.get(&id).is_some_and(|n| n.required) {
270 // Mark every transitive dependent skipped so the ledger
271 // reflects why they never ran.
272 let mut stack: Vec<String> = vec![id.clone()];
273 while let Some(failed_id) = stack.pop() {
274 for n in flow.nodes.iter() {
275 if n.deps.contains(&failed_id)
276 && matches!(
277 state.nodes.get(&n.id).map(|r| r.status),
278 None | Some(NodeStatus::Pending)
279 )
280 {
281 state.nodes.insert(
282 n.id.clone(),
283 NodeResult {
284 status: NodeStatus::Skipped,
285 output: format!("upstream '{failed_id}' failed"),
286 },
287 );
288 stack.push(n.id.clone());
289 }
290 }
291 }
292 self.persist(state);
293 let _ = self
294 .finish_work_item(vak_session::types::WorkItemStatus::Failed)
295 .await;
296 return FlowOutcome::Failed {
297 node: id,
298 reason: output,
299 outputs: collect_outputs(state),
300 };
301 }
302 }
303
304 if cancel.is_cancelled() {
305 let _ = self
306 .finish_work_item(vak_session::types::WorkItemStatus::Interrupted)
307 .await;
308 return FlowOutcome::Aborted;
309 }
310 }
311
312 if let Err(reason) = self
313 .record_work_evidence("__flow_completed__", &state.run_id, &flow.name)
314 .await
315 {
316 let _ = self
317 .finish_work_item(vak_session::types::WorkItemStatus::Failed)
318 .await;
319 return FlowOutcome::Failed {
320 node: "<work>".into(),
321 reason,
322 outputs: collect_outputs(state),
323 };
324 }
325 if let Err(reason) = self
326 .finish_work_item(vak_session::types::WorkItemStatus::ReadyForVerification)
327 .await
328 {
329 return FlowOutcome::Failed {
330 node: "<work>".into(),
331 reason,
332 outputs: collect_outputs(state),
333 };
334 }
335 FlowOutcome::Completed {
336 outputs: collect_outputs(state),
337 }
338 }
339
340 fn persist(&self, state: &FlowState) {
341 if let Ok(json) = serde_json::to_string_pretty(state) {
342 if let Some(parent) = self.deps.state_path.parent() {
343 let _ = std::fs::create_dir_all(parent);
344 }
345 let _ = std::fs::write(&self.deps.state_path, json);
346 }
347 }
348
349 async fn record_work_evidence(
350 &self,
351 node_id: &str,
352 run_id: &str,
353 flow_name: &str,
354 ) -> Result<(), String> {
355 let Some(work) = &self.deps.work else {
356 return Ok(());
357 };
358 let mut session = work.session.lock().await;
359 let Some(projection) = session
360 .work_projection()
361 .map_err(|error| format!("flow work projection is invalid: {error}"))?
362 else {
363 return Err("flow work context has no contract".into());
364 };
365 if projection.contract.contract_id != work.contract_id {
366 return Err("flow work context targets a different contract".into());
367 }
368 session
369 .append_work(vak_session::types::WorkEvent {
370 contract_id: work.contract_id.clone(),
371 revision: projection.contract.revision,
372 kind: vak_session::types::WorkEventKind::EvidenceAttached {
373 item_id: work.work_item_id.clone(),
374 evidence: vak_session::types::EvidenceRef::FlowNode {
375 flow: flow_name.into(),
376 run_id: run_id.into(),
377 node_id: node_id.into(),
378 },
379 },
380 })
381 .map_err(|error| format!("flow evidence write failed: {error}"))?;
382 Ok(())
383 }
384
385 async fn start_work_item(&self) -> Result<(), String> {
386 let Some(work) = &self.deps.work else {
387 return Ok(());
388 };
389 let mut session = work.session.lock().await;
390 let Some(projection) = session.work_projection().map_err(|e| e.to_string())? else {
391 return Err("flow work context has no contract".into());
392 };
393 let Some(state) = projection.items.get(&work.work_item_id) else {
394 return Err("flow work context has no work item".into());
395 };
396 if state.status == vak_session::types::WorkItemStatus::Ready {
397 session
398 .append_work(vak_session::types::WorkEvent {
399 contract_id: work.contract_id.clone(),
400 revision: projection.contract.revision,
401 kind: vak_session::types::WorkEventKind::ItemStatusChanged {
402 item_id: work.work_item_id.clone(),
403 from: vak_session::types::WorkItemStatus::Ready,
404 to: vak_session::types::WorkItemStatus::Running,
405 attempt: state.attempt.saturating_add(1),
406 reason: "flow execution started".into(),
407 },
408 })
409 .map_err(|e| e.to_string())?;
410 } else if state.status != vak_session::types::WorkItemStatus::Running {
411 return Err(format!(
412 "flow work item is {:?}, not ready or running",
413 state.status
414 ));
415 }
416 Ok(())
417 }
418
419 async fn finish_work_item(
420 &self,
421 status: vak_session::types::WorkItemStatus,
422 ) -> Result<(), String> {
423 let Some(work) = &self.deps.work else {
424 return Ok(());
425 };
426 let mut session = work.session.lock().await;
427 let Some(projection) = session.work_projection().map_err(|e| e.to_string())? else {
428 return Err("flow work context has no contract".into());
429 };
430 let Some(state) = projection.items.get(&work.work_item_id) else {
431 return Err("flow work context has no work item".into());
432 };
433 if state.status != vak_session::types::WorkItemStatus::Running {
434 return Ok(());
435 }
436 session
437 .append_work(vak_session::types::WorkEvent {
438 contract_id: work.contract_id.clone(),
439 revision: projection.contract.revision,
440 kind: vak_session::types::WorkEventKind::ItemStatusChanged {
441 item_id: work.work_item_id.clone(),
442 from: vak_session::types::WorkItemStatus::Running,
443 to: status,
444 attempt: state.attempt,
445 reason: "flow execution returned".into(),
446 },
447 })
448 .map_err(|e| e.to_string())?;
449 Ok(())
450 }
451}
452
453fn collect_outputs(state: &FlowState) -> BTreeMap<String, String> {
454 state
455 .nodes
456 .iter()
457 .filter(|(_, r)| r.status == NodeStatus::Completed)
458 .map(|(id, r)| (id.clone(), r.output.clone()))
459 .collect()
460}
461
462fn render(template: &str, dep_outputs: &BTreeMap<String, String>) -> String {
463 let mut out = template.to_string();
464 for (id, output) in dep_outputs {
465 out = out.replace(&format!("{{{{{id}}}}}"), output);
466 }
467 out
468}
469
470async fn execute_node(
471 node: &NodeDef,
472 dep_outputs: &BTreeMap<String, String>,
473 deps: &ExecutorDeps,
474 cancel: &CancellationToken,
475 events: &tokio::sync::mpsc::Sender<String>,
476) -> Result<String, String> {
477 match node.r#type.as_str() {
478 "bash" => {
479 let command = render(node.command.as_deref().unwrap_or_default(), dep_outputs);
480 let ctx = ToolContext {
481 cwd: deps.cwd.clone(),
482 cancel: cancel.child_token(),
483 sandbox: deps.sandbox.clone(),
484 sandbox_sink: None,
485 agent_id: None,
486 new_documents: Vec::new(),
487 };
488 let tool = deps
489 .tools
490 .iter()
491 .find(|tool| tool.name() == "bash")
492 .ok_or_else(|| "bash tool unavailable".to_string())?;
493 let args = match node.timeout_ms {
494 Some(t) => serde_json::json!({"command": command, "timeout_ms": t}),
495 None => serde_json::json!({"command": command}),
496 };
497 authorize_flow_tool("bash", &args, deps).await?;
498 let out = tool.execute(&args, &ctx).await;
499 // A node's output is rendered into later nodes, prompts
500 // included, and a flow keeps no evidence ledger to recall from.
501 let content = vak_tools::bounded(out.content);
502 if out.is_error {
503 Err(content)
504 } else {
505 Ok(content)
506 }
507 }
508 "agent" => {
509 let prompt = render(node.prompt.as_deref().unwrap_or_default(), dep_outputs);
510 let readonly = node.readonly;
511 let tools = if readonly {
512 deps.read_only_tools.clone()
513 } else {
514 deps.tools.clone()
515 };
516 let mode = if readonly { Mode::ReadOnly } else { deps.mode };
517
518 let session_id = format!(
519 "flow-{}-{}",
520 node.id,
521 std::time::SystemTime::now()
522 .duration_since(std::time::UNIX_EPOCH)
523 .unwrap_or_default()
524 .as_nanos()
525 );
526 let header = vak_session::types::SessionHeader {
527 agent: deps.agent_identity.clone().or_else(|| {
528 Some(vak_session::types::AgentIdentity {
529 id: "vak".into(),
530 revision: 1,
531 name: "Vakyartha".into(),
532 character: "vak".into(),
533 personality: String::new(),
534 animation: "subtle".into(),
535 voice: "default".into(),
536 behaviour: String::new(),
537 responsibilities: String::new(),
538 instructions: String::new(),
539 })
540 }),
541 session_id: session_id.clone(),
542 created_at: chrono::Utc::now(),
543 cwd: deps.cwd.clone(),
544 parent_session_id: Some(deps.parent_session_id.clone()),
545 contract_id: None,
546 work_item_id: None,
547 conversation: deps
548 .conversation_context
549 .clone()
550 .or_else(|| Some(vak_session::ConversationContext::local(&session_id, "flow"))),
551 contract: vak_session::types::FrozenContract {
552 app_version: env!("CARGO_PKG_VERSION").into(),
553 provider: deps.provider.name().into(),
554 model: deps.model.clone(),
555 route_ladder: Vec::new(),
556 route_objective: String::new(),
557 route_annotations: Vec::new(),
558 system_prompt: deps.system_prompt.clone(),
559 permission_mode: match mode {
560 Mode::ReadOnly => "read-only",
561 Mode::WorkspaceWrite => "workspace-write",
562 Mode::FullAccess => "full-access",
563 }
564 .into(),
565 capabilities: Vec::new(),
566 prompt_layers: deps.prompt_layers.clone(),
567 },
568 };
569 let path = SessionPath::new_session_file(&deps.sessions_home, &deps.cwd, &session_id);
570 let log = SessionLog::create(path, header)
571 .map_err(|e| format!("cannot create node session: {e}"))?;
572
573 let mut cfg = AgentConfig::new(deps.system_prompt.clone());
574 cfg.outcome = deps.outcome.clone();
575 cfg.max_retries = deps.max_retries;
576 cfg.retry_base_backoff_ms = deps.retry_base_backoff_ms;
577 cfg.request_timeout = deps.request_timeout;
578 cfg.circuit_breaker = deps.circuit_breaker.clone();
579 cfg.run_retry_attempts = deps.run_retry_attempts;
580 cfg.run_retry_base_backoff_ms = deps.run_retry_base_backoff_ms;
581 cfg.dispatch_ceiling = deps.dispatch_ceiling;
582 cfg.spend_gate = deps.spend_gate.clone();
583 cfg.model = deps.model.clone();
584 cfg.tools = tools;
585 cfg.max_turns = deps
586 .outcome
587 .as_ref()
588 .and_then(|outcome| outcome.max_turns)
589 .map_or(deps.max_turns, |cap| deps.max_turns.min(cap));
590 cfg.permission = deps.permission.clone();
591 cfg.mode = mode;
592 cfg.approval_mode = deps.approval_mode;
593 cfg.approver = deps.approver.clone();
594 cfg.sandbox = deps.sandbox.clone();
595
596 let mut agent = Agent::new(deps.provider.clone(), log, cfg);
597 let steering = vak_agent::SteeringQueues::new();
598 let (ev_tx, mut ev_rx) = tokio::sync::mpsc::channel::<vak_agent::AgentEvent>(256);
599 let pump = tokio::spawn(async move { while ev_rx.recv().await.is_some() {} });
600 let outcome = agent
601 .run(&prompt, &steering, cancel.child_token(), ev_tx)
602 .await;
603 let _ = pump.await;
604
605 match outcome {
606 vak_agent::TurnOutcome::Completed { response } => {
607 let text = response.text_content();
608 if text.is_empty() {
609 Ok(format!("(node '{}' completed without output)", node.id))
610 } else {
611 Ok(text)
612 }
613 }
614 vak_agent::TurnOutcome::Aborted { partial } => Err(format!(
615 "agent node aborted. Partial:\n{}",
616 partial.map(|p| p.text_content()).unwrap_or_default()
617 )),
618 vak_agent::TurnOutcome::Failed { error } => Err(error.to_string()),
619 vak_agent::TurnOutcome::MaxTurnsReached => {
620 Err("agent node hit its turn limit".into())
621 }
622 }
623 }
624 "approval" => {
625 let message = render(node.message.as_deref().unwrap_or_default(), dep_outputs);
626 let _ = events.send(format!("⏸ approval needed: {message}")).await;
627 match &deps.approver {
628 Some(a) => {
629 if a.approve("approval", "", &message).await {
630 Ok("approved".into())
631 } else {
632 Err("denied by user".into())
633 }
634 }
635 None => Err("no approver available for approval node".into()),
636 }
637 }
638 "merge" => {
639 let mut report = String::from("<merge-report>\n");
640 for dep in &node.deps {
641 match dep_outputs.get(dep) {
642 Some(out) => report.push_str(&format!("[{dep}] ok\n{out}\n")),
643 None => report.push_str(&format!(
644 "[{dep}] unavailable (failed or skipped upstream)\n"
645 )),
646 }
647 }
648 report.push_str("</merge-report>");
649 Ok(report)
650 }
651 other => Err(format!("unknown node type '{other}'")),
652 }
653}
654
655async fn authorize_flow_tool(
656 tool: &str,
657 args: &serde_json::Value,
658 deps: &ExecutorDeps,
659) -> Result<(), String> {
660 let Some(engine) = &deps.permission else {
661 return Ok(());
662 };
663 match engine.evaluate(tool, args, deps.mode, &deps.cwd) {
664 Decision::Allow => Ok(()),
665 Decision::Deny { reason } => Err(reason),
666 Decision::Ask { reason, source } => {
667 if vak_agent::auto_approve(
668 deps.approval_mode,
669 source,
670 tool,
671 args,
672 deps.mode,
673 deps.sandbox.is_some(),
674 &deps.cwd,
675 ) {
676 return Ok(());
677 }
678 match &deps.approver {
679 Some(approver) if approver.approve(tool, &args.to_string(), &reason).await => {
680 Ok(())
681 }
682 Some(_) => Err(format!("denied by user: {reason}")),
683 None => Err(format!("{reason} (no approver available)")),
684 }
685 }
686 }
687}
688

Indexing the workspace…