1use std::collections::BTreeMap;
16
17use ingot_ir::{AgentIr, Decision, Node, NodeKind, RefScope, TemplatePart, Value as IrValue};
18use serde_json::{json, Value};
19
20use crate::events::{Artifact, EventSink, RunEvent, VerifyOutcome};
21use crate::price::{parse_micros, Pricing, Spend};
22use crate::provider::{CompletionRequest, ModelProvider, ModelSelection, ProviderError, Usage};
23use crate::schema;
24use crate::snapshot::{self, Resumption};
25use crate::tools::{
26 ApprovalRequest, ConsultError, ConsultRequest, HumanChannel, ToolError, ToolHost,
27 ToolInvocation,
28};
29use crate::{RunError, RunReport};
30
31const NON_STREAMING_CEILING: u32 = 16_000;
38
39const STREAMING_CEILING: u32 = 64_000;
46
47pub struct RunOptions {
48 pub inputs: BTreeMap<String, Value>,
50 pub approval: HumanChannel,
51 pub max_steps: u32,
54 pub memory: BTreeMap<String, Value>,
58 pub stop_at: Option<String>,
64 pub resume: Option<Resumption>,
69 pub pricing: Pricing,
76}
77
78impl Default for RunOptions {
79 fn default() -> Self {
80 RunOptions {
81 inputs: BTreeMap::new(),
82 approval: HumanChannel::Deny,
83 max_steps: 1_000,
84 memory: BTreeMap::new(),
85 stop_at: None,
86 resume: None,
87 pricing: Pricing::default(),
88 }
89 }
90}
91
92pub type AgentRegistry = BTreeMap<String, AgentIr>;
94
95struct Interp<'a> {
96 ir: &'a AgentIr,
97 registry: &'a AgentRegistry,
98 provider: &'a mut dyn ModelProvider,
99 tools: &'a mut dyn ToolHost,
100 sink: &'a mut dyn EventSink,
101 approval: &'a mut HumanChannel,
102
103 bindings: BTreeMap<String, Value>,
104 state: BTreeMap<String, Value>,
105 memory: BTreeMap<String, Value>,
108 outputs: BTreeMap<String, Artifact>,
109
110 steps: u32,
111 max_steps: u32,
112 usage: Usage,
113 pricing: Pricing,
114 spend: Spend,
115
116 stop_at: Option<String>,
118 stopped: Option<(String, String)>,
124 run_inputs: BTreeMap<String, Value>,
126 model_calls: u32,
128 consultations: u32,
129 tool_calls: u32,
130}
131
132pub fn run(
134 ir: &AgentIr,
135 registry: &AgentRegistry,
136 provider: &mut dyn ModelProvider,
137 tools: &mut dyn ToolHost,
138 sink: &mut dyn EventSink,
139 options: RunOptions,
140) -> Result<RunReport, RunError> {
141 let RunOptions {
142 inputs,
143 mut approval,
144 max_steps,
145 memory,
146 stop_at,
147 resume,
148 pricing,
149 } = options;
150 run_nested(
151 ir,
152 registry,
153 provider,
154 tools,
155 sink,
156 inputs,
157 &mut approval,
158 max_steps,
159 memory,
160 Interruption { stop_at, resume },
161 &pricing,
162 )
163}
164
165#[allow(clippy::too_many_arguments)]
173fn run_nested(
174 ir: &AgentIr,
175 registry: &AgentRegistry,
176 provider: &mut dyn ModelProvider,
177 tools: &mut dyn ToolHost,
178 sink: &mut dyn EventSink,
179 inputs: BTreeMap<String, Value>,
180 approval: &mut HumanChannel,
181 step_ceiling: u32,
182 stored: BTreeMap<String, Value>,
186 interruption: Interruption,
189 pricing: &Pricing,
192) -> Result<RunReport, RunError> {
193 check_ir_version(ir)?;
194
195 let inputs = match &interruption.resume {
200 Some(snapshot) => {
201 if !inputs.is_empty() && inputs != snapshot.inputs {
202 return Err(RunError::InputsAfterResume);
203 }
204 snapshot.inputs.clone()
205 }
206 None => inputs,
207 };
208
209 let mut bindings = BTreeMap::new();
210 for (name, declared_type) in &ir.inputs {
211 let Some(value) = inputs.get(name) else {
212 return Err(RunError::MissingInput {
213 name: name.clone(),
214 ty: declared_type.clone(),
215 });
216 };
217 schema::validate(value, declared_type, &ir.types).map_err(|reason| {
218 RunError::InvalidInput {
219 name: name.clone(),
220 reason,
221 }
222 })?;
223 bindings.insert(name.clone(), value.clone());
224 }
225 for name in inputs.keys() {
226 if !ir.inputs.contains_key(name) {
227 return Err(RunError::UnknownInput {
228 name: name.clone(),
229 expected: ir.inputs.keys().cloned().collect(),
230 });
231 }
232 }
233
234 let mut memory = BTreeMap::new();
240 for (name, field) in &ir.persistent {
241 let value = stored
242 .get(name)
243 .cloned()
244 .unwrap_or_else(|| field.initial.clone());
245 schema::validate(&value, &field.ty, &ir.types).map_err(|reason| {
246 RunError::InvalidMemory {
247 field: name.clone(),
248 reason,
249 }
250 })?;
251 memory.insert(name.clone(), value);
252 }
253 for name in stored.keys() {
254 if !ir.persistent.contains_key(name) {
255 return Err(RunError::UnknownMemoryField {
256 field: name.clone(),
257 expected: ir.persistent.keys().cloned().collect(),
258 });
259 }
260 }
261
262 let Interruption { stop_at, resume } = interruption;
266 if let Some(label) = &stop_at {
267 check_stop_label(ir, label)?;
268 }
269 if let Some(snapshot) = &resume {
270 snapshot.check(ir).map_err(RunError::Snapshot)?;
271 }
272
273 let entry = match &resume {
277 Some(snapshot) => {
278 bindings = snapshot.bindings.clone();
279 Some(snapshot.resume_at.clone())
280 }
281 None => ir.entry.clone(),
282 };
283
284 sink.emit(RunEvent::RunStarted {
285 agent: ir.agent.clone(),
286 provider: provider.name().to_string(),
287 });
288
289 let max_steps = match ir.budget.steps {
290 Some(limit) if limit >= 0 => (limit as u32).min(step_ceiling),
291 _ => step_ceiling,
292 };
293
294 let mut interp = Interp {
295 ir,
296 registry,
297 provider,
298 tools,
299 sink,
300 approval,
301 bindings,
302 state: resume
306 .as_ref()
307 .map(|snapshot| snapshot.state.clone())
308 .unwrap_or_default(),
309 memory,
310 outputs: resume
311 .as_ref()
312 .map(|snapshot| snapshot.outputs.clone())
313 .unwrap_or_default(),
314 steps: resume.as_ref().map(|snapshot| snapshot.steps).unwrap_or(0),
315 max_steps,
316 usage: resume
317 .as_ref()
318 .map(|snapshot| snapshot.usage)
319 .unwrap_or_default(),
320 pricing: pricing.clone(),
321 spend: resume
322 .as_ref()
323 .map(|snapshot| snapshot.spend.clone())
324 .unwrap_or_default(),
325 stop_at,
326 stopped: None,
327 run_inputs: match &resume {
328 Some(snapshot) => snapshot.inputs.clone(),
329 None => inputs,
330 },
331 consultations: resume
332 .as_ref()
333 .map(|snapshot| snapshot.consultations)
334 .unwrap_or(0),
335 model_calls: resume
336 .as_ref()
337 .map(|snapshot| snapshot.model_calls)
338 .unwrap_or(0),
339 tool_calls: resume
340 .as_ref()
341 .map(|snapshot| snapshot.tool_calls)
342 .unwrap_or(0),
343 };
344
345 let result = interp.run_region(entry.as_deref());
346
347 match result {
348 Ok(()) => {
349 let stopped = interp.take_snapshot();
350 let report = RunReport {
351 agent: ir.agent.clone(),
352 outputs: interp.outputs,
353 memory: interp.memory,
354 stopped,
355 usage: interp.usage,
356 steps: interp.steps,
357 spend: interp.spend.clone(),
358 };
359 if let Some(snapshot) = &report.stopped {
360 sink.emit(RunEvent::RunStopped {
361 node: snapshot.stopped_at.clone(),
362 label: snapshot.label.clone(),
363 });
364 return Ok(report);
369 }
370 sink.emit(RunEvent::RunFinished {
371 steps: report.steps,
372 usage: report.usage,
373 });
374 for name in ir.outputs.keys() {
375 if !report.outputs.contains_key(name) {
376 return Err(RunError::OutputNotProduced { name: name.clone() });
377 }
378 }
379 Ok(report)
380 }
381 Err(error) => {
382 sink.emit(RunEvent::RunFailed {
383 reason: error.to_string(),
384 });
385 Err(error)
386 }
387 }
388}
389
390#[derive(Default)]
393struct Interruption {
394 stop_at: Option<String>,
395 resume: Option<Resumption>,
396}
397
398fn check_stop_label(ir: &AgentIr, label: &str) -> Result<(), RunError> {
403 if snapshot::resumable_labels(ir).iter().any(|it| it == label) {
404 return Ok(());
405 }
406 let nested = snapshot::all_checkpoint_labels(ir)
407 .iter()
408 .any(|it| it == label);
409 Err(RunError::NotResumable {
410 label: label.to_string(),
411 nested,
412 available: snapshot::resumable_labels(ir),
413 })
414}
415
416fn check_ir_version(ir: &AgentIr) -> Result<(), RunError> {
417 let major = ir.ir_version.split('.').next().unwrap_or_default();
418 if major != "0" {
419 return Err(RunError::UnsupportedIrVersion {
420 found: ir.ir_version.clone(),
421 supported: ingot_ir::IR_VERSION.to_string(),
422 });
423 }
424 Ok(())
425}
426
427impl Interp<'_> {
428 fn node(&self, id: &str) -> Result<&Node, RunError> {
429 self.ir
430 .node(id)
431 .ok_or_else(|| RunError::MalformedIr(format!("node `{id}` does not exist")))
432 }
433
434 fn take_snapshot(&self) -> Option<Resumption> {
436 let (node, label) = self.stopped.clone()?;
437 let resume_at = self.ir.node(&node).and_then(|node| node.next.clone())?;
441 Some(Resumption {
442 ingot_snapshot: snapshot::SNAPSHOT_VERSION.to_string(),
443 kind: snapshot::KIND.to_string(),
444 agent: self.ir.agent.clone(),
445 artifact: snapshot::artifact_digest(self.ir),
446 label,
447 stopped_at: node,
448 resume_at,
449 inputs: self.run_inputs.clone(),
450 bindings: self.bindings.clone(),
451 state: self.state.clone(),
452 outputs: self.outputs.clone(),
453 steps: self.steps,
454 usage: self.usage,
455 spend: self.spend.clone(),
456 model_calls: self.model_calls,
457 consultations: self.consultations,
458 tool_calls: self.tool_calls,
459 })
460 }
461
462 fn run_region(&mut self, entry: Option<&str>) -> Result<(), RunError> {
464 let mut current = entry.map(str::to_string);
465 while let Some(id) = current {
466 if self.stopped.is_some() {
467 return Ok(());
468 }
469 let node = self.node(&id)?.clone();
470 self.sink.emit(RunEvent::NodeStarted {
471 node: node.id.clone(),
472 kind: node.kind.as_str().to_string(),
473 });
474 self.run_node(&node)?;
475 current = node.next.clone();
476 }
477 Ok(())
478 }
479
480 fn run_node(&mut self, node: &Node) -> Result<(), RunError> {
481 match node.kind {
482 NodeKind::LlmCall => self.run_llm_call(node),
483 NodeKind::ToolCall => self.run_tool_call(node),
484 NodeKind::AgentCall => self.run_agent_call(node),
485 NodeKind::Branch => self.run_branch(node),
486 NodeKind::Parallel => self.run_parallel(node),
487 NodeKind::Loop => self.run_loop(node),
488 NodeKind::Approval => self.run_approval(node),
489 NodeKind::Consult => self.run_consult(node),
490 NodeKind::Verify => self.run_verify(node),
491 NodeKind::StateRead => self.run_state_read(node),
492 NodeKind::StateWrite => self.run_state_write(node),
493 NodeKind::ArtifactEmit => self.run_emit(node),
494 NodeKind::Checkpoint => {
495 let label = node.label.clone().unwrap_or_default();
496 self.sink.emit(RunEvent::Checkpoint {
497 node: node.id.clone(),
498 label: label.clone(),
499 });
500 if node.resumable && self.stop_at.as_deref() == Some(label.as_str()) {
504 self.stopped = Some((node.id.clone(), label));
505 }
506 Ok(())
507 }
508 }
509 }
510
511 fn charge_step(&mut self, node: &Node) -> Result<(), RunError> {
514 self.steps += 1;
515 if self.steps > self.max_steps {
516 return Err(RunError::BudgetExceeded {
517 budget: "steps".to_string(),
518 limit: self.max_steps.to_string(),
519 node: node.id.clone(),
520 });
521 }
522 Ok(())
523 }
524
525 fn charge_cost(&mut self, node: &Node, model: &str, usage: Usage) -> Result<(), RunError> {
532 let Some(budget) = self.ir.budget.cost.clone() else {
533 return Ok(());
536 };
537 self.spend
538 .add(model, self.pricing.charge(model, usage, &budget.currency));
539
540 let Some(limit) = parse_micros(&budget.amount) else {
541 return Ok(());
542 };
543 if self.spend.is_complete() && self.spend.micros() > limit {
544 return Err(RunError::BudgetExceeded {
545 budget: "cost".to_string(),
546 limit: format!("{} {}", budget.amount, budget.currency.to_ascii_uppercase()),
547 node: node.id.clone(),
548 });
549 }
550 Ok(())
551 }
552
553 fn charge_tokens(&mut self, node: &Node, usage: Usage) -> Result<(), RunError> {
554 self.usage.add(usage);
555 if let Some(limit) = self.ir.budget.tokens {
556 if limit >= 0 && self.usage.total() > limit as u64 {
557 return Err(RunError::BudgetExceeded {
558 budget: "tokens".to_string(),
559 limit: limit.to_string(),
560 node: node.id.clone(),
561 });
562 }
563 }
564 Ok(())
565 }
566
567 fn check_effects(&mut self, node: &Node, effects: &[String]) -> Result<bool, RunError> {
574 let mut needs_approval = Vec::new();
575 for effect in effects {
576 if effect == "model_access" {
577 continue;
578 }
579 let subject = subject_for_effect(effect);
580 match self.ir.policy.get(subject) {
581 Some(rule) => match rule.decision {
582 Decision::Allow => {}
583 Decision::RequireApproval => needs_approval.push(effect.clone()),
584 Decision::Deny => {
585 return Err(RunError::CapabilityDenied {
586 node: node.id.clone(),
587 effect: effect.clone(),
588 explicit: true,
589 })
590 }
591 },
592 None => {
593 return Err(RunError::CapabilityDenied {
594 node: node.id.clone(),
595 effect: effect.clone(),
596 explicit: false,
597 })
598 }
599 }
600 }
601 Ok(!needs_approval.is_empty())
602 }
603
604 fn run_llm_call(&mut self, node: &Node) -> Result<(), RunError> {
607 self.charge_step(node)?;
608 self.model_calls += 1;
612
613 let response_type = node
614 .response_type
615 .clone()
616 .ok_or_else(|| RunError::MalformedIr(format!("`{}` has no responseType", node.id)))?;
617 let shape = schema::response_shape(&response_type, &self.ir.types).map_err(|error| {
618 RunError::UnsupportedResponseType {
619 node: node.id.clone(),
620 ty: error.ty,
621 reason: error.reason,
622 }
623 })?;
624
625 let prompt_value = node
626 .prompt
627 .as_ref()
628 .ok_or_else(|| RunError::MalformedIr(format!("`{}` has no prompt", node.id)))?;
629 let prompt = self.render_prompt(prompt_value)?;
630
631 let mut system = None;
632 let mut context = Vec::new();
633 for argument in &node.args {
634 let value = self.eval(&argument.value)?;
635 match argument.name.as_str() {
636 "system" => system = value.as_str().map(str::to_string),
637 "temperature" | "max_tokens" => {}
642 name => context.push((name.to_string(), value)),
643 }
644 }
645
646 let request = CompletionRequest {
647 node: node.id.clone(),
648 model: self.model_selection(),
649 system,
650 prompt,
651 context,
652 response_type: response_type.clone(),
653 shape,
654 max_tokens: self.max_output_tokens(),
655 };
656
657 let node_id = node.id.clone();
662 let mut shown = false;
663 let attempt = {
664 let sink = &mut *self.sink;
665 self.provider.complete_streaming(&request, &mut |text| {
666 shown = true;
667 sink.delta(&node_id, text);
668 })
669 };
670
671 let response = match attempt {
676 Ok(response) => response,
677 Err(error) => {
678 if shown {
679 self.sink.settled(&node_id, false);
680 }
681 return Err(RunError::Provider {
682 node: node.id.clone(),
683 source: error,
684 });
685 }
686 };
687
688 if let Err(reason) = schema::validate(&response.value, &response_type, &self.ir.types) {
689 if shown {
690 self.sink.settled(&node_id, false);
691 }
692 return Err(RunError::Provider {
693 node: node.id.clone(),
694 source: ProviderError::InvalidResponse(reason),
695 });
696 }
697 if shown {
698 self.sink.settled(&node_id, true);
699 }
700
701 self.sink.emit(RunEvent::ModelCall {
702 node: node.id.clone(),
703 model: response.model.clone(),
704 response_type,
705 usage: response.usage,
706 });
707 self.charge_tokens(node, response.usage)?;
708 self.charge_cost(node, &response.model, response.usage)?;
709 self.bind(node, response.value);
710 Ok(())
711 }
712
713 fn run_tool_call(&mut self, node: &Node) -> Result<(), RunError> {
714 self.tool_calls += 1;
715 let reference = node
716 .tool
717 .clone()
718 .ok_or_else(|| RunError::MalformedIr(format!("`{}` names no tool", node.id)))?;
719 let binding = self
720 .ir
721 .tools
722 .iter()
723 .find(|tool| tool.reference == reference)
724 .ok_or_else(|| {
725 RunError::MalformedIr(format!(
726 "`{}` calls `{reference}`, which the artifact does not grant",
727 node.id
728 ))
729 })?
730 .clone();
731
732 self.check_effects(node, &node.effects)?;
733 self.charge_step(node)?;
734
735 if !self.tools.provides(&binding.name) {
736 return Err(RunError::Tool {
737 node: node.id.clone(),
738 source: ToolError::NotAvailable(binding.name.clone()),
739 });
740 }
741
742 let mut arguments = BTreeMap::new();
743 for argument in &node.args {
744 arguments.insert(argument.name.clone(), self.eval(&argument.value)?);
745 }
746
747 let invocation = ToolInvocation {
748 node: node.id.clone(),
749 agent: self.ir.agent.clone(),
750 reference: binding.reference.clone(),
751 name: binding.name.clone(),
752 transport: binding.transport.clone(),
753 arguments,
754 effects: node.effects.clone(),
755 result_type: binding.signature.result.clone(),
756 };
757
758 let result = self
759 .tools
760 .call(&invocation)
761 .map_err(|source| RunError::Tool {
762 node: node.id.clone(),
763 source,
764 })?;
765
766 schema::validate(&result, &binding.signature.result, &self.ir.types).map_err(|reason| {
767 RunError::Tool {
768 node: node.id.clone(),
769 source: ToolError::InvalidResult(reason),
770 }
771 })?;
772
773 self.sink.emit(RunEvent::ToolCall {
774 node: node.id.clone(),
775 tool: binding.reference,
776 effects: node.effects.clone(),
777 });
778 self.bind(node, result);
779 Ok(())
780 }
781
782 fn run_agent_call(&mut self, node: &Node) -> Result<(), RunError> {
783 let name = node
784 .agent
785 .clone()
786 .ok_or_else(|| RunError::MalformedIr(format!("`{}` names no agent", node.id)))?;
787 let sub = self
788 .registry
789 .get(&name)
790 .ok_or_else(|| RunError::AgentNotAvailable {
791 node: node.id.clone(),
792 agent: name.clone(),
793 })?;
794
795 self.check_effects(node, &node.effects)?;
796 self.charge_step(node)?;
797 self.sink.emit(RunEvent::AgentCall {
798 node: node.id.clone(),
799 agent: name.clone(),
800 });
801
802 let mut inputs = BTreeMap::new();
803 for argument in &node.args {
804 inputs.insert(argument.name.clone(), self.eval(&argument.value)?);
805 }
806
807 let sub = sub.clone();
815 let report = run_nested(
816 &sub,
817 self.registry,
818 self.provider,
819 self.tools,
820 self.sink,
821 inputs,
822 &mut *self.approval,
823 self.max_steps.saturating_sub(self.steps).max(1),
824 BTreeMap::new(),
828 Interruption::default(),
832 &self.pricing,
833 )
834 .map_err(|error| RunError::SubAgent {
835 node: node.id.clone(),
836 agent: name.clone(),
837 source: Box::new(error),
838 })?;
839
840 self.steps += report.steps;
841 self.charge_tokens(node, report.usage)?;
842
843 let value = report
844 .outputs
845 .values()
846 .next()
847 .map(|artifact| artifact.value.clone())
848 .unwrap_or(Value::Null);
849 self.bind(node, value);
850 Ok(())
851 }
852
853 fn run_branch(&mut self, node: &Node) -> Result<(), RunError> {
854 let condition = node
855 .condition
856 .as_ref()
857 .ok_or_else(|| RunError::MalformedIr(format!("`{}` has no condition", node.id)))?;
858 let taken = self.eval(condition)?.as_bool().ok_or_else(|| {
859 RunError::MalformedIr(format!(
860 "`{}` condition did not evaluate to a boolean",
861 node.id
862 ))
863 })?;
864
865 self.sink.emit(RunEvent::BranchTaken {
866 node: node.id.clone(),
867 arm: if taken { "then".into() } else { "else".into() },
868 });
869 let arm = if taken {
870 node.then.as_deref()
871 } else {
872 node.otherwise.as_deref()
873 };
874 self.run_region(arm)
875 }
876
877 fn run_parallel(&mut self, node: &Node) -> Result<(), RunError> {
882 let source = node
883 .source
884 .as_ref()
885 .ok_or_else(|| RunError::MalformedIr(format!("`{}` has no source", node.id)))?;
886 let binder = node
887 .binder
888 .clone()
889 .ok_or_else(|| RunError::MalformedIr(format!("`{}` has no binder", node.id)))?;
890 let items = self.eval(source)?;
891 let items = items.as_array().cloned().ok_or_else(|| {
892 RunError::MalformedIr(format!("`{}` source did not evaluate to a list", node.id))
893 })?;
894
895 let last_body_node = node
896 .body
897 .as_deref()
898 .map(|entry| self.last_of_region(entry))
899 .transpose()?
900 .flatten();
901
902 let total = items.len();
903 let mut collected = Vec::with_capacity(total);
904 let shadowed = self.bindings.remove(&binder);
905
906 for (index, item) in items.into_iter().enumerate() {
907 self.sink.emit(RunEvent::MapIteration {
908 node: node.id.clone(),
909 index,
910 total,
911 });
912 self.bindings.insert(binder.clone(), item);
913 self.run_region(node.body.as_deref())?;
914 if self.stopped.is_some() {
915 break;
916 }
917
918 let value = match &last_body_node {
927 Some(id) => {
928 let last = self.node(id)?.clone();
929 match &last.binding {
930 Some(name) => self.bindings.get(name).cloned().unwrap_or(Value::Null),
931 None => {
932 return Err(RunError::MalformedIr(format!(
933 "`{}` ends its body at `{id}`, which binds nothing, so an iteration has no value to collect",
934 node.id
935 )))
936 }
937 }
938 }
939 None => Value::Null,
940 };
941 collected.push(value);
942 }
943
944 self.bindings.remove(&binder);
945 if let Some(shadowed) = shadowed {
946 self.bindings.insert(binder, shadowed);
947 }
948 self.bind(node, Value::Array(collected));
949 Ok(())
950 }
951
952 fn run_loop(&mut self, node: &Node) -> Result<(), RunError> {
953 let max = node.max_iterations.unwrap_or(0).max(0) as u32;
956 for iteration in 0..max {
957 if let Some(guard) = &node.guard {
958 let keep_going = self.eval(guard)?.as_bool().ok_or_else(|| {
959 RunError::MalformedIr(format!(
960 "`{}` guard did not evaluate to a boolean",
961 node.id
962 ))
963 })?;
964 if !keep_going {
965 break;
966 }
967 }
968 self.sink.emit(RunEvent::LoopIteration {
969 node: node.id.clone(),
970 iteration: iteration + 1,
971 });
972 self.run_region(node.body.as_deref())?;
973 if self.stopped.is_some() {
977 break;
978 }
979 }
980 Ok(())
981 }
982
983 fn run_approval(&mut self, node: &Node) -> Result<(), RunError> {
984 let reason = node
985 .label
986 .clone()
987 .unwrap_or_else(|| "approval required".to_string());
988 self.sink.emit(RunEvent::ApprovalRequested {
989 node: node.id.clone(),
990 effects: node.effects.clone(),
991 reason: reason.clone(),
992 });
993
994 let allowed = match self.approval {
995 HumanChannel::AssumeYes => true,
996 HumanChannel::Deny => false,
997 HumanChannel::Ask(handler) => handler.approve(&ApprovalRequest {
998 node: node.id.clone(),
999 effects: node.effects.clone(),
1000 reason: reason.clone(),
1001 }),
1002 };
1003
1004 self.sink.emit(RunEvent::ApprovalDecided {
1005 node: node.id.clone(),
1006 allowed,
1007 });
1008 if allowed {
1009 Ok(())
1010 } else {
1011 Err(RunError::ApprovalDenied {
1012 node: node.id.clone(),
1013 reason,
1014 })
1015 }
1016 }
1017
1018 fn run_consult(&mut self, node: &Node) -> Result<(), RunError> {
1026 self.charge_step(node)?;
1027 let index = self.consultations as usize;
1028 self.consultations += 1;
1032
1033 let question_value = node
1034 .prompt
1035 .as_ref()
1036 .ok_or_else(|| RunError::MalformedIr(format!("`{}` has no question", node.id)))?;
1037 let question = self.render_prompt(question_value)?;
1038
1039 let mut choices = Vec::new();
1040 let mut context = Vec::new();
1041 for argument in &node.args {
1042 let value = self.eval(&argument.value)?;
1043 match argument.name.as_str() {
1044 "choices" => {
1045 choices = value
1046 .as_array()
1047 .map(|items| {
1048 items
1049 .iter()
1050 .filter_map(|item| item.as_str().map(str::to_string))
1051 .collect()
1052 })
1053 .unwrap_or_default();
1054 }
1055 name => context.push((name.to_string(), value)),
1056 }
1057 }
1058
1059 self.sink.emit(RunEvent::ConsultationAsked {
1060 node: node.id.clone(),
1061 index,
1062 question: question.clone(),
1063 choices: choices.clone(),
1064 });
1065
1066 let request = ConsultRequest {
1067 node: node.id.clone(),
1068 index,
1069 question: question.clone(),
1070 choices: choices.clone(),
1071 context,
1072 };
1073
1074 let answered = match self.approval {
1075 HumanChannel::Ask(interlocutor) => interlocutor.consult(&request),
1076 HumanChannel::AssumeYes => Err(ConsultError::NoChannel(
1080 "`--yes` approves a gate and cannot answer a question; there is no safe side to guess"
1081 .to_string(),
1082 )),
1083 HumanChannel::Deny => Err(ConsultError::NoChannel(
1084 "this run has no channel to a person".to_string(),
1085 )),
1086 };
1087
1088 let answer = match answered {
1089 Ok(answer) if choices.is_empty() || choices.iter().any(|choice| choice == &answer) => {
1090 answer
1091 }
1092 Ok(answer) => {
1096 return Err(RunError::ConsultFailed {
1097 node: node.id.clone(),
1098 question,
1099 reason: ConsultError::NotAChoice { answer, choices }.to_string(),
1100 })
1101 }
1102 Err(error) => {
1103 return Err(RunError::ConsultFailed {
1104 node: node.id.clone(),
1105 question,
1106 reason: error.to_string(),
1107 })
1108 }
1109 };
1110
1111 self.sink.emit(RunEvent::ConsultationAnswered {
1112 node: node.id.clone(),
1113 index,
1114 answer: answer.clone(),
1115 });
1116 self.bind(node, Value::String(answer));
1117 Ok(())
1118 }
1119
1120 fn run_verify(&mut self, node: &Node) -> Result<(), RunError> {
1121 let verifier = node
1122 .verifier
1123 .clone()
1124 .ok_or_else(|| RunError::MalformedIr(format!("`{}` names no verifier", node.id)))?;
1125 for argument in &node.args {
1126 self.eval(&argument.value)?;
1127 }
1128
1129 let Some(condition) = &node.condition else {
1134 self.sink.emit(RunEvent::Verified {
1135 node: node.id.clone(),
1136 verifier: verifier.clone(),
1137 outcome: VerifyOutcome::NotPerformed,
1138 });
1139 return Ok(());
1140 };
1141
1142 let held = self.eval(condition)?.as_bool().ok_or_else(|| {
1143 RunError::MalformedIr(format!(
1144 "`{}` condition did not evaluate to a boolean",
1145 node.id
1146 ))
1147 })?;
1148
1149 self.sink.emit(RunEvent::Verified {
1150 node: node.id.clone(),
1151 verifier: verifier.clone(),
1152 outcome: if held {
1153 VerifyOutcome::Passed
1154 } else {
1155 VerifyOutcome::Failed
1156 },
1157 });
1158
1159 if held {
1163 Ok(())
1164 } else {
1165 Err(RunError::VerificationFailed {
1166 node: node.id.clone(),
1167 verifier: verifier.clone(),
1168 })
1169 }
1170 }
1171
1172 fn run_state_read(&mut self, node: &Node) -> Result<(), RunError> {
1173 let field = node
1174 .field
1175 .clone()
1176 .ok_or_else(|| RunError::MalformedIr(format!("`{}` names no state field", node.id)))?;
1177 let value = match node.scope {
1178 Some(RefScope::Memory) => self.memory.get(&field).cloned().ok_or_else(|| {
1179 RunError::MalformedIr(format!(
1180 "`{}` reads `memory.{field}`, which is not declared",
1181 node.id
1182 ))
1183 })?,
1184 _ => self
1185 .state
1186 .get(&field)
1187 .cloned()
1188 .ok_or_else(|| RunError::StateNotSet {
1189 node: node.id.clone(),
1190 field: field.clone(),
1191 })?,
1192 };
1193 self.bind(node, value);
1194 Ok(())
1195 }
1196
1197 fn run_state_write(&mut self, node: &Node) -> Result<(), RunError> {
1198 let field = node
1199 .field
1200 .clone()
1201 .ok_or_else(|| RunError::MalformedIr(format!("`{}` names no state field", node.id)))?;
1202 let value = node
1203 .value
1204 .as_ref()
1205 .ok_or_else(|| RunError::MalformedIr(format!("`{}` has no value", node.id)))?;
1206 let value = self.eval(value)?;
1207 let persistent = matches!(node.scope, Some(RefScope::Memory));
1208 let declared = if persistent {
1209 self.ir.persistent.get(&field).map(|field| field.ty.clone())
1210 } else {
1211 self.ir.state.get(&field).cloned()
1212 };
1213 if let Some(declared) = declared {
1214 let root = if persistent { "memory" } else { "state" };
1215 schema::validate(&value, &declared, &self.ir.types).map_err(|reason| {
1216 RunError::TypeMismatch {
1217 node: node.id.clone(),
1218 what: format!("{root}.{field}"),
1219 reason,
1220 }
1221 })?;
1222 }
1223 if persistent {
1224 self.memory.insert(field.clone(), value);
1225 } else {
1226 self.state.insert(field.clone(), value);
1227 }
1228 self.sink.emit(RunEvent::StateWritten {
1229 node: node.id.clone(),
1230 field,
1231 });
1232 Ok(())
1233 }
1234
1235 fn run_emit(&mut self, node: &Node) -> Result<(), RunError> {
1236 let output = node
1237 .output
1238 .clone()
1239 .ok_or_else(|| RunError::MalformedIr(format!("`{}` names no output", node.id)))?;
1240 let value = node
1241 .value
1242 .as_ref()
1243 .ok_or_else(|| RunError::MalformedIr(format!("`{}` has no value", node.id)))?;
1244 let value = self.eval(value)?;
1245
1246 let declared = self.ir.outputs.get(&output).cloned().ok_or_else(|| {
1247 RunError::MalformedIr(format!(
1248 "`{}` emits `{output}`, which is not declared",
1249 node.id
1250 ))
1251 })?;
1252 let content_type = declared
1253 .strip_prefix("artifact<")
1254 .and_then(|rest| rest.strip_suffix('>'))
1255 .unwrap_or(&declared)
1256 .to_string();
1257 schema::validate(&value, &content_type, &self.ir.types).map_err(|reason| {
1258 RunError::TypeMismatch {
1259 node: node.id.clone(),
1260 what: format!("output `{output}`"),
1261 reason,
1262 }
1263 })?;
1264
1265 self.outputs.insert(
1266 output.clone(),
1267 Artifact {
1268 name: output.clone(),
1269 content_type,
1270 value,
1271 },
1272 );
1273 self.sink.emit(RunEvent::Emitted {
1274 node: node.id.clone(),
1275 output,
1276 });
1277 Ok(())
1278 }
1279
1280 fn bind(&mut self, node: &Node, value: Value) {
1283 if let Some(name) = &node.binding {
1284 self.bindings.insert(name.clone(), value);
1285 }
1286 }
1287
1288 fn last_of_region(&self, entry: &str) -> Result<Option<String>, RunError> {
1290 let mut current = Some(entry.to_string());
1291 let mut last = None;
1292 let mut visited = 0usize;
1293 while let Some(id) = current {
1294 let node = self.node(&id)?;
1295 last = Some(node.id.clone());
1296 current = node.next.clone();
1297 visited += 1;
1298 if visited > self.ir.nodes.len() {
1299 return Err(RunError::MalformedIr(
1300 "the node graph contains a cycle".to_string(),
1301 ));
1302 }
1303 }
1304 Ok(last)
1305 }
1306
1307 fn model_selection(&self) -> ModelSelection {
1308 match &self.ir.requirements.model {
1309 ingot_ir::ModelRequirement::Exact { reference } => {
1310 ModelSelection::Exact(reference.clone())
1311 }
1312 ingot_ir::ModelRequirement::Capabilities {
1313 capabilities,
1314 context_tokens,
1315 } => ModelSelection::Capabilities {
1316 capabilities: capabilities.clone(),
1317 min_context_tokens: context_tokens.as_ref().map(|tokens| tokens.min),
1318 },
1319 ingot_ir::ModelRequirement::Unspecified => ModelSelection::Default,
1320 }
1321 }
1322
1323 fn max_output_tokens(&self) -> u32 {
1331 let ceiling = if self.provider.streams() {
1332 STREAMING_CEILING
1333 } else {
1334 NON_STREAMING_CEILING
1335 };
1336 let remaining = match self.ir.budget.tokens {
1337 Some(limit) if limit >= 0 => (limit as u64)
1338 .saturating_sub(self.usage.total())
1339 .min(u32::MAX as u64) as u32,
1340 _ => ceiling,
1341 };
1342 remaining.clamp(1, ceiling)
1343 }
1344
1345 fn render_prompt(&mut self, value: &IrValue) -> Result<String, RunError> {
1346 match value {
1347 IrValue::Literal {
1348 value: Value::String(text),
1349 ..
1350 } => Ok(text.clone()),
1351 IrValue::Template { parts } => {
1352 let mut out = String::new();
1353 for part in parts {
1354 match part {
1355 TemplatePart::Text { value } => out.push_str(value),
1356 TemplatePart::Value { value, ty } => {
1357 let resolved = self.eval(value)?;
1358 out.push_str(&render_value(&resolved, ty));
1359 }
1360 }
1361 }
1362 Ok(out)
1363 }
1364 other => {
1365 let resolved = self.eval(other)?;
1366 Ok(render_value(&resolved, "string"))
1367 }
1368 }
1369 }
1370
1371 fn eval(&mut self, value: &IrValue) -> Result<Value, RunError> {
1373 match value {
1374 IrValue::Literal { value, .. } => Ok(value.clone()),
1375 IrValue::Ref { scope, path } => self.eval_ref(*scope, path),
1376 IrValue::List { items } => {
1377 let mut out = Vec::with_capacity(items.len());
1378 for item in items {
1379 out.push(self.eval(item)?);
1380 }
1381 Ok(Value::Array(out))
1382 }
1383 IrValue::Template { .. } => Ok(Value::String(self.render_prompt(value)?)),
1384 IrValue::Unary { op, operand } => {
1385 let operand = self.eval(operand)?;
1386 match op.as_str() {
1387 "!" => Ok(json!(!truthy(&operand))),
1388 "-" => match operand.as_i64() {
1389 Some(int) => Ok(json!(-int)),
1390 None => Ok(json!(-operand.as_f64().unwrap_or_default())),
1391 },
1392 other => Err(RunError::MalformedIr(format!(
1393 "unknown unary operator `{other}`"
1394 ))),
1395 }
1396 }
1397 IrValue::Binary { op, lhs, rhs } => {
1398 let lhs = self.eval(lhs)?;
1399 let rhs = self.eval(rhs)?;
1400 eval_binary(op, &lhs, &rhs)
1401 }
1402 IrValue::Builtin { name, args } => {
1403 let mut evaluated = Vec::with_capacity(args.len());
1404 for arg in args {
1405 evaluated.push(self.eval(arg)?);
1406 }
1407 eval_builtin(name, &evaluated)
1408 }
1409 IrValue::Unknown => Err(RunError::MalformedIr(
1410 "the artifact contains an unresolved value; it was not built from a clean compile"
1411 .to_string(),
1412 )),
1413 }
1414 }
1415
1416 fn eval_ref(&mut self, scope: RefScope, path: &[String]) -> Result<Value, RunError> {
1417 let Some((root, fields)) = path.split_first() else {
1418 return Err(RunError::MalformedIr(
1419 "a reference has an empty path".to_string(),
1420 ));
1421 };
1422 let mut current = match scope {
1423 RefScope::Input | RefScope::Binding => {
1424 self.bindings.get(root).cloned().ok_or_else(|| {
1425 RunError::MalformedIr(format!("`{root}` is not bound at this point"))
1426 })?
1427 }
1428 RefScope::State => {
1429 self.state
1430 .get(root)
1431 .cloned()
1432 .ok_or_else(|| RunError::StateNotSet {
1433 node: String::new(),
1434 field: root.clone(),
1435 })?
1436 }
1437 RefScope::Memory => self.memory.get(root).cloned().ok_or_else(|| {
1441 RunError::MalformedIr(format!("`memory.{root}` is not a declared field"))
1442 })?,
1443 };
1444 for field in fields {
1445 current = current
1446 .get(field)
1447 .cloned()
1448 .ok_or_else(|| RunError::MalformedIr(format!("no field `{field}` on the value")))?;
1449 }
1450 Ok(current)
1451 }
1452}
1453
1454fn render_value(value: &Value, ty: &str) -> String {
1455 match value {
1456 Value::String(text) => text.clone(),
1459 Value::Null => String::new(),
1460 other => match ty {
1461 "json" => serde_json::to_string_pretty(other).unwrap_or_else(|_| other.to_string()),
1462 _ => other.to_string(),
1463 },
1464 }
1465}
1466
1467fn truthy(value: &Value) -> bool {
1468 value.as_bool().unwrap_or(false)
1469}
1470
1471fn eval_binary(op: &str, lhs: &Value, rhs: &Value) -> Result<Value, RunError> {
1472 let result = match op {
1473 "==" => json!(lhs == rhs),
1474 "!=" => json!(lhs != rhs),
1475 "&&" => json!(truthy(lhs) && truthy(rhs)),
1476 "||" => json!(truthy(lhs) || truthy(rhs)),
1477 "<" | "<=" | ">" | ">=" => {
1478 let (a, b) = numeric_pair(lhs, rhs, op)?;
1479 json!(match op {
1480 "<" => a < b,
1481 "<=" => a <= b,
1482 ">" => a > b,
1483 _ => a >= b,
1484 })
1485 }
1486 "+" | "-" => {
1487 if let (Some(a), Some(b)) = (lhs.as_i64(), rhs.as_i64()) {
1488 json!(if op == "+" { a + b } else { a - b })
1489 } else {
1490 let (a, b) = numeric_pair(lhs, rhs, op)?;
1491 json!(if op == "+" { a + b } else { a - b })
1492 }
1493 }
1494 other => {
1495 return Err(RunError::MalformedIr(format!(
1496 "unknown binary operator `{other}`"
1497 )));
1498 }
1499 };
1500 Ok(result)
1501}
1502
1503fn numeric_pair(lhs: &Value, rhs: &Value, op: &str) -> Result<(f64, f64), RunError> {
1504 match (lhs.as_f64(), rhs.as_f64()) {
1505 (Some(a), Some(b)) => Ok((a, b)),
1506 _ => Err(RunError::MalformedIr(format!(
1507 "`{op}` needs two numbers but was given {lhs} and {rhs}"
1508 ))),
1509 }
1510}
1511
1512fn eval_builtin(name: &str, args: &[Value]) -> Result<Value, RunError> {
1513 match name {
1514 "len" => {
1515 let Some(value) = args.first() else {
1516 return Err(RunError::MalformedIr(
1517 "`len` takes one argument".to_string(),
1518 ));
1519 };
1520 let length = match value {
1521 Value::Array(items) => items.len(),
1522 Value::String(text) => text.chars().count(),
1523 Value::Object(map) => map.len(),
1524 other => {
1525 return Err(RunError::MalformedIr(format!(
1526 "`len` cannot measure {other}"
1527 )))
1528 }
1529 };
1530 Ok(json!(length))
1531 }
1532 other => Err(RunError::MalformedIr(format!("unknown builtin `{other}`"))),
1533 }
1534}
1535
1536fn subject_for_effect(effect: &str) -> &str {
1541 match effect {
1542 "secret_access" => "secrets",
1543 other => other,
1544 }
1545}
1546
1547#[cfg(test)]
1548mod tests {
1549 use super::*;
1550
1551 #[test]
1552 fn secret_access_maps_to_the_secrets_subject() {
1553 assert_eq!(subject_for_effect("secret_access"), "secrets");
1554 assert_eq!(subject_for_effect("network"), "network");
1555 }
1556
1557 #[test]
1558 fn len_measures_lists_strings_and_objects() {
1559 assert_eq!(eval_builtin("len", &[json!([1, 2, 3])]).unwrap(), json!(3));
1560 assert_eq!(eval_builtin("len", &[json!("abc")]).unwrap(), json!(3));
1561 assert_eq!(eval_builtin("len", &[json!({"a": 1})]).unwrap(), json!(1));
1562 }
1563
1564 #[test]
1565 fn comparison_operators_work_on_numbers() {
1566 assert_eq!(eval_binary(">", &json!(3), &json!(1)).unwrap(), json!(true));
1567 assert_eq!(
1568 eval_binary("<=", &json!(3), &json!(3)).unwrap(),
1569 json!(true)
1570 );
1571 }
1572
1573 #[test]
1574 fn equality_works_on_any_value() {
1575 assert_eq!(
1576 eval_binary("==", &json!("a"), &json!("a")).unwrap(),
1577 json!(true)
1578 );
1579 assert_eq!(
1580 eval_binary("!=", &json!([1]), &json!([2])).unwrap(),
1581 json!(true)
1582 );
1583 }
1584
1585 #[test]
1586 fn integer_arithmetic_stays_integral() {
1587 assert_eq!(eval_binary("+", &json!(2), &json!(3)).unwrap(), json!(5));
1588 }
1589
1590 #[test]
1591 fn strings_render_without_json_quoting() {
1592 assert_eq!(render_value(&json!("hello"), "string"), "hello");
1593 assert_eq!(render_value(&json!(3), "int"), "3");
1594 }
1595}