use crate::analyzer::{CompiledProgram, ExternalInput, compile};
use crate::{
BinaryOp, Block, CanonicalValue as V, Diagnostic, Expr, ExprKind, Graph, GraphChange,
GraphEvent, NodeKind, NodeState, Schema, Span, ToolRegistry, UnaryOp,
};
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, HashMap};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use thiserror::Error;
#[derive(Debug, Clone)]
pub struct ToolContext {
pub node_id: String,
pub operation_id: String,
pub dispatch_id: String,
pub attempt: u32,
pub schema_version: String,
}
#[derive(Debug, Clone, Error)]
#[error("{code}: {message}")]
pub struct ToolError {
pub code: String,
pub message: String,
pub retryable: bool,
pub uncertain: bool,
pub details: BTreeMap<String, V>,
pub span: Option<Span>,
pub retry_after: Option<std::time::Duration>,
}
impl ToolError {
pub fn new(code: impl Into<String>, message: impl Into<String>) -> Self {
Self {
code: code.into(),
message: message.into(),
retryable: false,
uncertain: false,
details: BTreeMap::new(),
span: None,
retry_after: None,
}
}
pub fn retryable(mut self, yes: bool) -> Self {
self.retryable = yes;
self
}
pub fn with_retry_after(mut self, delay: std::time::Duration) -> Self {
self.retry_after = Some(delay);
self
}
}
type Handler = Arc<dyn Fn(&[V], &ToolContext) -> Result<V, ToolError> + Send + Sync>;
#[derive(Clone)]
pub struct Runtime {
registry: ToolRegistry,
handlers: BTreeMap<String, Handler>,
inputs: BTreeMap<String, (Schema, V)>,
default_loop_limit: u32,
max_loop_limit: u32,
max_graph_nodes: usize,
max_eval_depth: usize,
max_active_dispatches: usize,
max_worker_threads: usize,
retry_backoff: Option<RetryBackoff>,
}
#[derive(Debug, Clone, Copy)]
struct RetryBackoff {
base: std::time::Duration,
factor: f64,
cap: std::time::Duration,
}
struct DispatchSemaphore {
permits: Mutex<usize>,
released: Condvar,
}
impl DispatchSemaphore {
fn new(permits: usize) -> Self {
Self {
permits: Mutex::new(permits.max(1)),
released: Condvar::new(),
}
}
fn acquire(&self) -> DispatchPermit<'_> {
let mut permits = self.permits.lock().unwrap();
while *permits == 0 {
permits = self.released.wait(permits).unwrap();
}
*permits -= 1;
DispatchPermit(self)
}
}
struct DispatchPermit<'a>(&'a DispatchSemaphore);
impl Drop for DispatchPermit<'_> {
fn drop(&mut self) {
*self.0.permits.lock().unwrap() += 1;
self.0.released.notify_one();
}
}
pub struct RuntimeBuilder {
runtime: Runtime,
}
impl Runtime {
pub fn builder() -> RuntimeBuilder {
RuntimeBuilder {
runtime: Runtime {
registry: ToolRegistry::new(),
handlers: BTreeMap::new(),
inputs: BTreeMap::new(),
default_loop_limit: 16,
max_loop_limit: 1024,
max_graph_nodes: 250_000,
max_eval_depth: 512,
max_active_dispatches: 64,
max_worker_threads: 64,
retry_backoff: None,
},
}
}
pub fn compile(&self, source: &str) -> Result<CompiledProgram, Vec<Diagnostic>> {
let inputs = self
.inputs
.iter()
.map(|(name, (schema, _))| ExternalInput {
name: name.clone(),
schema: schema.clone(),
})
.collect::<Vec<_>>();
compile(source, &self.registry, &inputs)
}
pub fn run(&self, program: &CompiledProgram) -> Result<Execution, ToolError> {
self.run_observed(program, |_| {})
}
pub fn run_observed<F>(
&self,
program: &CompiledProgram,
observer: F,
) -> Result<Execution, ToolError>
where
F: Fn(&GraphEvent) + Send + Sync + 'static,
{
if program.registry_digest != self.registry.digest() {
return Err(ToolError::new(
"RL7102",
"compiled program registry digest does not match this runtime",
));
}
let graph = Arc::new(LiveGraph::new(Arc::new(observer)));
let mut ev = Evaluator {
runtime: self,
graph: graph.clone(),
scopes: vec![],
attempt: 0,
owners: vec![],
workflow: program.source_digest.clone(),
dynamic: vec![],
successful_operations: Arc::new(Mutex::new(HashMap::new())),
dispatch_generations: Arc::new(Mutex::new(HashMap::new())),
operation_nodes: Arc::new(Mutex::new(HashMap::new())),
last_output: None,
binding_caches: vec![],
depth: 0,
dispatches: Arc::new(DispatchSemaphore::new(self.max_active_dispatches)),
worker_budget: Arc::new(AtomicUsize::new(self.max_worker_threads)),
};
let mut root = HashMap::new();
for (n, (_, v)) in &self.inputs {
root.insert(n.clone(), Binding::Value(v.clone(), None));
}
for s in &program.program.statements {
if let crate::StmtKind::Binding { name, value } = &s.kind {
root.insert(name.clone(), Binding::Expr(value.clone()));
}
}
ev.scopes.push(root);
let node = ev
.graph
.begin(NodeKind::Root, "return", program.program.result.span, 0);
ev.graph.running(node);
let outcome = ev
.guards_and_effects(&program.program.statements)
.and_then(|_| ev.eval(&program.program.result));
match outcome {
Ok(v) => {
if let Some(producer) = ev.last_output {
ev.graph.data(producer, node, "result", "return");
}
ev.graph.success(node, v.clone());
Ok(Execution {
value: v,
graph: graph.snapshot(),
})
}
Err(e) => {
ev.graph.fail(node, e.to_string());
Err(e)
}
}
}
fn retry_delay(
&self,
reattempt: u32,
retry_after: Option<std::time::Duration>,
) -> std::time::Duration {
match (self.retry_backoff, retry_after) {
(Some(backoff), Some(after)) => after.min(backoff.cap),
(Some(backoff), None) => {
let exponent = reattempt.saturating_sub(1).min(i32::MAX as u32) as i32;
let scaled = backoff.base.as_secs_f64() * backoff.factor.powi(exponent);
std::time::Duration::try_from_secs_f64(scaled)
.unwrap_or(backoff.cap)
.min(backoff.cap)
}
(None, Some(after)) => after,
(None, None) => std::time::Duration::ZERO,
}
}
}
impl RuntimeBuilder {
pub fn registry(mut self, registry: ToolRegistry) -> Self {
self.runtime.registry = registry;
self
}
pub fn input(mut self, name: impl Into<String>, schema: Schema, value: V) -> Self {
self.runtime.inputs.insert(name.into(), (schema, value));
self
}
pub fn tool<F>(mut self, name: impl Into<String>, handler: F) -> Self
where
F: Fn(&[V], &ToolContext) -> Result<V, ToolError> + Send + Sync + 'static,
{
self.runtime.handlers.insert(name.into(), Arc::new(handler));
self
}
pub fn loop_limits(mut self, default: u32, max: u32) -> Self {
self.runtime.default_loop_limit = default;
self.runtime.max_loop_limit = max;
self
}
pub fn graph_node_limit(mut self, limit: usize) -> Self {
self.runtime.max_graph_nodes = limit;
self
}
pub fn eval_depth_limit(mut self, limit: usize) -> Self {
self.runtime.max_eval_depth = limit.max(1);
self
}
pub fn dispatch_limit(mut self, limit: usize) -> Self {
self.runtime.max_active_dispatches = limit.max(1);
self
}
pub fn worker_thread_limit(mut self, limit: usize) -> Self {
self.runtime.max_worker_threads = limit;
self
}
pub fn retry_backoff(
mut self,
base: std::time::Duration,
factor: f64,
cap: std::time::Duration,
) -> Self {
self.runtime.retry_backoff = Some(RetryBackoff { base, factor, cap });
self
}
pub fn with_prelude(mut self) -> Self {
for (descriptor, handler) in crate::prelude::entries() {
if self.runtime.registry.get(&descriptor.name).is_some() {
continue;
}
let name = descriptor.name.clone();
self.runtime
.registry
.register(descriptor)
.expect("prelude descriptors are valid and deduplicated");
self.runtime.handlers.insert(name, Arc::new(handler));
}
self
}
pub fn build(self) -> Result<Runtime, String> {
for n in self.runtime.registry.names() {
if !self.runtime.handlers.contains_key(n) {
return Err(format!("registered tool `{n}` has no implementation"));
}
}
for n in self.runtime.handlers.keys() {
if self.runtime.registry.get(n).is_none() {
return Err(format!("implementation `{n}` has no descriptor"));
}
}
for (n, (s, v)) in &self.runtime.inputs {
if !s.accepts(v) {
return Err(format!("input `{n}` does not match its schema"));
}
}
Ok(self.runtime)
}
}
#[derive(Debug, Clone)]
pub struct Execution {
pub value: V,
pub graph: Graph,
}
#[derive(Clone)]
enum Binding {
Expr(Expr),
Value(V, Option<usize>),
Evaluating,
}
type BindingCache = Arc<Mutex<HashMap<(usize, String), (V, Option<usize>)>>>;
struct Evaluator<'a> {
runtime: &'a Runtime,
graph: Arc<LiveGraph>,
scopes: Vec<HashMap<String, Binding>>,
attempt: u32,
owners: Vec<usize>,
workflow: String,
dynamic: Vec<String>,
successful_operations: Arc<Mutex<HashMap<String, V>>>,
dispatch_generations: Arc<Mutex<HashMap<String, u32>>>,
operation_nodes: Arc<Mutex<HashMap<String, usize>>>,
last_output: Option<usize>,
binding_caches: Vec<(usize, BindingCache)>,
depth: usize,
dispatches: Arc<DispatchSemaphore>,
worker_budget: Arc<AtomicUsize>,
}
struct LiveGraph {
state: Mutex<LiveGraphState>,
observer: Arc<dyn Fn(&GraphEvent) + Send + Sync>,
}
#[derive(Default)]
struct LiveGraphState {
graph: Graph,
sequence: u64,
}
impl LiveGraph {
fn new(observer: Arc<dyn Fn(&GraphEvent) + Send + Sync>) -> Self {
Self {
state: Mutex::new(LiveGraphState::default()),
observer,
}
}
fn publish(&self, state: &mut LiveGraphState, change: GraphChange) {
state.sequence += 1;
(self.observer)(&GraphEvent {
sequence: state.sequence,
change,
});
}
fn begin(&self, kind: NodeKind, label: impl Into<String>, span: Span, attempt: u32) -> usize {
let mut state = self.state.lock().unwrap();
let index = state.graph.begin(kind, label, span, attempt);
let node = state.graph.nodes[index].clone();
self.publish(&mut state, GraphChange::NodeAdded(node));
index
}
fn update(&self, index: usize, apply: impl FnOnce(&mut Graph, usize)) {
let mut state = self.state.lock().unwrap();
apply(&mut state.graph, index);
let node = state.graph.nodes[index].clone();
self.publish(&mut state, GraphChange::NodeUpdated(node));
}
fn running(&self, index: usize) {
self.update(index, |graph, index| graph.running(index));
}
fn dispatching(&self, index: usize) {
self.update(index, |graph, index| {
graph.nodes[index].state = NodeState::Dispatching
});
}
fn label(&self, index: usize, label: String) {
self.update(index, |graph, index| graph.nodes[index].label = label);
}
fn blocked(&self, index: usize) {
self.update(index, |graph, index| {
graph.nodes[index].state = NodeState::Blocked
});
}
fn success(&self, index: usize, value: V) {
self.update(index, |graph, index| graph.success(index, value));
}
fn fail(&self, index: usize, error: impl Into<String>) {
let error = error.into();
self.update(index, |graph, index| graph.fail(index, error));
}
fn contains(&self, parent: usize, child: usize) {
let mut state = self.state.lock().unwrap();
state.graph.contains(parent, child);
let edge = state.graph.edges.last().unwrap().clone();
self.publish(&mut state, GraphChange::EdgeAdded(edge));
}
fn data(&self, producer: usize, consumer: usize, producer_path: &str, consumer_path: &str) {
let mut state = self.state.lock().unwrap();
let edge = crate::Edge {
from: state.graph.nodes[producer].id.clone(),
to: state.graph.nodes[consumer].id.clone(),
kind: crate::EdgeKind::Data {
producer_path: producer_path.into(),
consumer_path: consumer_path.into(),
},
};
state.graph.edges.push(edge.clone());
self.publish(&mut state, GraphChange::EdgeAdded(edge));
}
fn retry_of(&self, retry: usize, previous_attempt: usize) {
let mut state = self.state.lock().unwrap();
let edge = crate::Edge {
from: state.graph.nodes[retry].id.clone(),
to: state.graph.nodes[previous_attempt].id.clone(),
kind: crate::EdgeKind::RetryOf,
};
state.graph.edges.push(edge.clone());
self.publish(&mut state, GraphChange::EdgeAdded(edge));
}
fn node_id(&self, index: usize) -> String {
self.state.lock().unwrap().graph.nodes[index].id.clone()
}
fn node_count(&self) -> usize {
self.state.lock().unwrap().graph.nodes.len()
}
fn snapshot(&self) -> Graph {
self.state.lock().unwrap().graph.clone()
}
}
impl Evaluator<'_> {
fn eval(&mut self, e: &Expr) -> Result<V, ToolError> {
if self.depth >= self.runtime.max_eval_depth {
return Err(lang(
"RL4106",
&format!(
"expression nesting exceeded the host depth limit of {}",
self.runtime.max_eval_depth
),
));
}
self.depth += 1;
let result = self.eval_inner(e);
self.depth -= 1;
result.map_err(|mut error| {
if error.span.is_none() {
error.span = Some(e.span);
}
error
})
}
fn eval_inner(&mut self, e: &Expr) -> Result<V, ToolError> {
self.last_output = None;
match &e.kind {
ExprKind::Null => Ok(V::Null),
ExprKind::Boolean(v) => Ok(V::Boolean(*v)),
ExprKind::Integer(s) => s
.parse()
.map(V::Integer)
.map_err(|_| lang("RL5101", "NUMERIC_OVERFLOW")),
ExprKind::Number(s) => {
V::number(s.parse().map_err(|_| lang("RL5103", "NON_FINITE_NUMBER"))?)
.map_err(|x| lang("RL5103", &x.to_string()))
}
ExprKind::String(s) => Ok(V::String(s.clone())),
ExprKind::Name(n) => self.name(n),
ExprKind::List(xs) => {
let node = self.node(NodeKind::Composite, "list", e.span)?;
let r = xs
.iter()
.map(|x| self.eval(x))
.collect::<Result<Vec<_>, _>>()
.map(V::List);
self.finish(node, r)
}
ExprKind::Object(xs) => {
let node = self.node(NodeKind::Composite, "object", e.span)?;
let mut o = BTreeMap::new();
for (k, x) in xs {
let key = match k {
crate::ObjectKey::Static(k) => k.clone(),
crate::ObjectKey::Computed(key_expr) => {
match self.eval(key_expr).and_then(|v| {
computed_key(v).map_err(|mut err| {
err.span = err.span.or(Some(key_expr.span));
err
})
}) {
Ok(key) => key,
Err(err) => {
self.graph.fail(node, err.to_string());
return Err(err);
}
}
}
};
o.insert(
key,
match self.eval(x) {
Ok(v) => v,
Err(err) => {
self.graph.fail(node, err.to_string());
return Err(err);
}
},
);
}
self.finish(node, Ok(V::Object(o)))
}
ExprKind::Member { target, field } => {
let node = self.node(NodeKind::Project, format!(".{field}"), e.span)?;
let r = self
.eval(target)
.and_then(|v| project(v, &V::String(field.clone())));
self.finish(node, r)
}
ExprKind::Index { target, index } => {
let node = self.node(NodeKind::Project, "index", e.span)?;
let r = (|| {
let v = self.eval(target)?;
let i = self.eval(index)?;
project(v, &i)
})();
self.finish(node, r)
}
ExprKind::Call { callee, arguments } => self.call(e.span, callee, arguments),
ExprKind::Unary { op, value } => {
let node = self.node(NodeKind::Compute, format!("{op:?}"), e.span)?;
let r = self.eval(value).and_then(|v| unary(*op, v));
self.finish(node, r)
}
ExprKind::Binary { op, left, right } => self.binary(e.span, *op, left, right),
ExprKind::Conditional {
then_expr,
condition,
else_expr,
} => {
let node = self.node(NodeKind::Branch, "if", e.span)?;
let c = self.eval(condition)?;
let chosen = match c {
V::Boolean(true) => then_expr,
V::Boolean(false) => else_expr,
other => {
let mut er = lang(
"RL5202",
&format!(
"condition evaluated to {}; conditions must be true or false — \
Runlet has no truthiness, compare explicitly (e.g. \
`value != null`, `list.count(xs) > 0`, `text != \"\"`)",
value_kind(&other)
),
);
er.span = Some(condition.span);
self.graph.fail(node, er.to_string());
return Err(er);
}
};
let r = self.eval(chosen);
self.finish(node, r)
}
ExprKind::If {
condition,
then_block,
else_block,
} => {
let node = self.node(NodeKind::Branch, "if", e.span)?;
let c = match self.eval(condition) {
Ok(v) => v,
Err(err) => {
self.graph.fail(node, err.to_string());
return Err(err);
}
};
let r = match c {
V::Boolean(true) => self.block(then_block),
V::Boolean(false) => match else_block {
Some(block) => self.block(block),
None => Ok(V::Null),
},
other => {
let mut er = lang(
"RL5202",
&format!(
"condition evaluated to {}; conditions must be true or false — \
Runlet has no truthiness, compare explicitly (e.g. \
`value != null`, `list.length(xs) > 0`, `text != \"\"`)",
value_kind(&other)
),
);
er.span = Some(condition.span);
Err(er)
}
};
self.finish(node, r)
}
ExprKind::For {
binding,
collection,
limit,
body,
} => self.loop_expr(
e.span,
binding,
collection,
limit.unwrap_or(self.runtime.default_loop_limit),
body,
),
ExprKind::Fold {
accumulator,
init,
binding,
collection,
body,
} => self.fold_expr(e.span, accumulator, init, binding, collection, body),
ExprKind::Fail { arguments } => {
let node = self.node(NodeKind::Compute, "fail", e.span)?;
let r = self.fail_error(arguments).and_then(Err);
self.finish(node, r)
}
ExprKind::Boundary {
retries,
body,
error_binding,
catch,
} => self.boundary(e.span, *retries, body, error_binding, catch),
}
}
fn fail_error(&mut self, arguments: &[Expr]) -> Result<ToolError, ToolError> {
let mut strings = Vec::new();
for (i, a) in arguments.iter().take(2).enumerate() {
match self.eval(a)? {
V::String(s) => strings.push(s),
other => {
return Err(lang(
"RL5201",
&format!(
"fail {} must be a string, got {}",
if i == 0 { "code" } else { "message" },
value_kind(&other)
),
));
}
}
}
let mut error = ToolError::new(
strings.first().cloned().unwrap_or_else(|| "FAILED".into()),
strings.get(1).cloned().unwrap_or_default(),
);
if let Some(details) = arguments.get(2) {
match self.eval(details)? {
V::Object(o) => error.details = o,
other => {
return Err(lang(
"RL5201",
&format!("fail details must be an object, got {}", value_kind(&other)),
));
}
}
}
Ok(error)
}
fn fold_expr(
&mut self,
span: Span,
accumulator: &str,
init: &Expr,
binding: &str,
collection: &Expr,
body: &Block,
) -> Result<V, ToolError> {
let node = self.node(NodeKind::Loop, "fold", span)?;
let prelude: Result<(V, V), ToolError> = (|| {
let acc = self.eval(init)?;
let c = self.eval(collection)?;
Ok((acc, c))
})();
let (mut acc, c) = match prelude {
Ok(x) => x,
Err(e) => {
self.graph.fail(node, e.to_string());
return Err(e);
}
};
let values = match c {
V::List(x) => x,
V::Object(o) => o
.into_iter()
.map(|(k, v)| {
V::Object(BTreeMap::from([
("key".into(), V::String(k)),
("value".into(), v),
]))
})
.collect(),
other => {
let e = lang(
"RL5204",
&format!(
"cannot iterate a {} value; `fold` needs a list or an object",
value_kind(&other)
),
);
self.graph.fail(node, e.to_string());
return Err(e);
}
};
for (i, value) in values.into_iter().enumerate() {
let it = self.node(
NodeKind::Iteration,
iteration_label(binding, i, &value),
body.span,
)?;
self.graph.contains(node, it);
self.owners.push(it);
self.dynamic
.push(format!("{i}:{}", value.digest_hex().unwrap_or_default()));
self.scopes.push(HashMap::from([
(accumulator.to_string(), Binding::Value(acc.clone(), None)),
(binding.to_string(), Binding::Value(value, None)),
]));
let result = self.block(body);
self.scopes.pop();
self.dynamic.pop();
self.owners.pop();
match result {
Ok(v) => {
self.graph.success(it, v.clone());
acc = v;
}
Err(e) if e.code == SKIP_SIGNAL => {
self.graph.success(it, acc.clone());
}
Err(e) => {
self.graph.fail(it, e.to_string());
self.graph.fail(node, e.to_string());
return Err(e);
}
}
}
self.finish(node, Ok(acc))
}
fn name(&mut self, n: &str) -> Result<V, ToolError> {
let found = (0..self.scopes.len())
.rev()
.find(|&i| self.scopes[i].contains_key(n))
.ok_or_else(|| lang("RL8101", &format!("unknown runtime name `{n}`")))?;
let binding = self.scopes[found].remove(n).unwrap();
match binding {
Binding::Value(v, producer) => {
self.last_output = producer;
self.scopes[found].insert(n.into(), Binding::Value(v.clone(), producer));
Ok(v)
}
Binding::Evaluating => Err(lang(
"RL4104",
&format!(
"binding `{n}` depends on itself; bindings are immutable, so accumulator \
patterns like `x = x + 1` cannot work — reduce with a fold instead: \
`total = fold acc = 0 for item in items {{ return acc + item }}`"
),
)),
Binding::Expr(e) => {
let cache = self
.binding_caches
.iter()
.find(|(base, _)| found < *base)
.map(|(_, cache)| cache.clone());
if let Some(cache) = &cache {
let cached = cache.lock().unwrap().get(&(found, n.to_string())).cloned();
if let Some((v, producer)) = cached {
self.last_output = producer;
self.scopes[found].insert(n.into(), Binding::Value(v.clone(), producer));
return Ok(v);
}
}
self.scopes[found].insert(n.into(), Binding::Evaluating);
let r = self.eval(&e);
match &r {
Ok(v) => {
self.scopes[found]
.insert(n.into(), Binding::Value(v.clone(), self.last_output));
if let Some(cache) = &cache {
cache
.lock()
.unwrap()
.insert((found, n.to_string()), (v.clone(), self.last_output));
}
}
Err(_) => {
self.scopes[found].insert(n.into(), Binding::Expr(e));
}
}
r
}
}
}
fn call(&mut self, span: Span, callee: &Expr, args: &[Expr]) -> Result<V, ToolError> {
let name = path(callee).ok_or_else(|| lang("RL2104", "value is not callable"))?;
let desc = self
.runtime
.registry
.get(&name)
.ok_or_else(|| lang("RL2102", &format!("unknown tool `{name}`")))?;
let node = self.blocked_node(NodeKind::Call, &name, span)?;
let mut values = vec![];
for (index, (a, expected)) in args.iter().zip(&desc.input.parameters).enumerate() {
match self.eval(a).and_then(|v| self.convert(a.span, v, expected)) {
Ok(v) => {
if let Some(producer) = self.last_output {
self.graph
.data(producer, node, "output", &format!("argument[{index}]"));
}
values.push(v)
}
Err(e) => {
self.graph.fail(node, e.to_string());
return Err(e);
}
}
}
if values.len() < desc.input.required_count() || values.len() > desc.input.parameters.len()
{
let e = lang("RL6102", "TOOL_INPUT_SCHEMA_MISMATCH");
self.graph.fail(node, e.to_string());
return Err(e);
}
if let Some(V::String(display)) = values.first() {
let display = display.chars().take(80).collect::<String>();
self.graph.label(node, format!("{name} · {display}"));
}
self.graph.dispatching(node);
let canonical = V::List(values.clone())
.rcve()
.map_err(|x| lang("RL5207", &x.to_string()))?;
let identity = format!(
"{}\0{}\0{}\0{}\0{}",
self.workflow,
name,
desc.schema_version,
span.start,
self.dynamic.join("/")
);
let op = hex::encode(Sha256::digest([identity.as_bytes(), &canonical].concat()));
if let Some(previous_attempt) = self
.operation_nodes
.lock()
.unwrap()
.insert(op.clone(), node)
{
self.graph.retry_of(node, previous_attempt);
}
let previous = self.successful_operations.lock().unwrap().get(&op).cloned();
if let Some(value) = previous {
return self.finish(node, Ok(value));
}
let dispatch_id = {
let mut generations = self.dispatch_generations.lock().unwrap();
let generation = generations.entry(op.clone()).or_default();
let dispatch_id = format!("{op}:{}", *generation);
*generation += 1;
dispatch_id
};
let ctx = ToolContext {
node_id: self.graph.node_id(node),
operation_id: op.clone(),
dispatch_id,
attempt: self.attempt,
schema_version: desc.schema_version.clone(),
};
self.graph.running(node);
let handler = self
.runtime
.handlers
.get(&name)
.ok_or_else(|| lang("RL8103", "missing tool implementation"))?;
let dispatches = self.dispatches.clone();
let permit = dispatches.acquire();
let mut r = handler(&values, &ctx).and_then(|v| {
if desc.output.accepts(&v) {
Ok(v)
} else {
Err(lang("RL6103", "TOOL_OUTPUT_SCHEMA_MISMATCH"))
}
});
drop(permit);
if let Err(error) = &mut r {
error.retryable &= matches!(
desc.execution,
crate::ExecutionPolicy::Pure
| crate::ExecutionPolicy::Idempotent
| crate::ExecutionPolicy::Recoverable
);
}
if let Ok(value) = &r {
self.successful_operations
.lock()
.unwrap()
.insert(op, value.clone());
}
self.finish(node, r)
}
fn binary(
&mut self,
span: Span,
op: BinaryOp,
left: &Expr,
right: &Expr,
) -> Result<V, ToolError> {
let node = self.node(
if matches!(op, BinaryOp::And | BinaryOp::Or) {
NodeKind::Branch
} else {
NodeKind::Compute
},
format!("{op:?}"),
span,
)?;
self.binary_inner(node, op, left, right)
}
fn binary_inner(
&mut self,
node: usize,
op: BinaryOp,
left: &Expr,
right: &Expr,
) -> Result<V, ToolError> {
let l = self.eval(left)?;
if let V::Boolean(b) = l {
if op == BinaryOp::And && !b {
return self.finish(node, Ok(V::Boolean(false)));
}
if op == BinaryOp::Or && b {
return self.finish(node, Ok(V::Boolean(true)));
}
}
let r = self.eval(right)?;
let out = binary(op, l, r);
self.finish(node, out)
}
fn convert(&mut self, span: Span, value: V, expected: &Schema) -> Result<V, ToolError> {
if expected.accepts(&value) {
return Ok(value);
}
let node = self.node(NodeKind::Convert, "convert", span)?;
let result = convert_value(value, expected);
self.finish(node, result)
}
fn loop_expr(
&mut self,
span: Span,
binding: &str,
collection: &Expr,
limit: u32,
body: &Block,
) -> Result<V, ToolError> {
if limit == 0 || limit > self.runtime.max_loop_limit {
return Err(lang("RL4102", "loop limit outside host policy"));
}
let node = self.node(NodeKind::Loop, format!("for limit {limit}"), span)?;
let c = self.eval(collection)?;
let values = match c {
V::List(x) => x,
V::Object(o) => o
.into_iter()
.map(|(k, v)| {
V::Object(BTreeMap::from([
("key".into(), V::String(k)),
("value".into(), v),
]))
})
.collect(),
other => {
let e = lang(
"RL5204",
&format!(
"cannot iterate a {} value; `for` needs a list or an object",
value_kind(&other)
),
);
self.graph.fail(node, e.to_string());
return Err(e);
}
};
let values = Arc::new(values);
let next = Arc::new(AtomicUsize::new(1));
let results = Arc::new(Mutex::new(vec![None; values.len()]));
let worker_count = (limit as usize).min(values.len().saturating_sub(1));
self.binding_caches
.push((self.scopes.len(), Arc::new(Mutex::new(HashMap::new()))));
let run_iteration = |mut child: Evaluator<'_>, i: usize, value: V| {
let it = match child.node(
NodeKind::Iteration,
iteration_label(binding, i, &value),
body.span,
) {
Ok(it) => it,
Err(error) => return Err(error),
};
child.graph.contains(node, it);
child.owners.push(it);
child
.dynamic
.push(format!("{i}:{}", value.digest_hex().unwrap_or_default()));
child.scopes.push(HashMap::from([(
binding.into(),
Binding::Value(value, None),
)]));
let result = child.block(body);
match result {
Ok(value) => {
child.graph.success(it, value.clone());
Ok(Some(value))
}
Err(error) if error.code == SKIP_SIGNAL => {
child.graph.success(it, V::Null);
Ok(None)
}
Err(error) => {
child.graph.fail(it, error.to_string());
Err(error)
}
}
};
if let Some(value) = values.first().cloned() {
let first = run_iteration(self.clone_for_branch(), 0, value);
results.lock().unwrap()[0] = Some(first);
}
let desired_extra = worker_count.saturating_sub(1);
let mut reserved = 0;
let _ =
self.worker_budget
.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |available| {
reserved = available.min(desired_extra);
Some(available - reserved)
});
let templates = (0..reserved)
.map(|_| self.clone_for_branch())
.collect::<Vec<_>>();
std::thread::scope(|scope| {
for template in templates {
let values = values.clone();
let next = next.clone();
let results = results.clone();
let run_iteration = &run_iteration;
scope.spawn(move || {
loop {
let i = next.fetch_add(1, Ordering::Relaxed);
let Some(value) = values.get(i).cloned() else {
break;
};
let result = run_iteration(template.clone_for_branch(), i, value);
results.lock().unwrap()[i] = Some(result);
}
});
}
loop {
let i = next.fetch_add(1, Ordering::Relaxed);
let Some(value) = values.get(i).cloned() else {
break;
};
let result = run_iteration(self.clone_for_branch(), i, value);
results.lock().unwrap()[i] = Some(result);
}
});
self.worker_budget.fetch_add(reserved, Ordering::Relaxed);
self.binding_caches.pop();
let mut out = Vec::with_capacity(values.len());
for result in results.lock().unwrap().iter_mut() {
match result.take().expect("every iteration produced a result") {
Ok(Some(value)) => out.push(value),
Ok(None) => {}
Err(error) => {
self.graph.fail(node, error.to_string());
return Err(error);
}
}
}
self.finish(node, Ok(V::List(out)))
}
fn boundary(
&mut self,
span: Span,
retries: u32,
body: &Block,
error_binding: &str,
catch: &Block,
) -> Result<V, ToolError> {
let node = self.node(
NodeKind::Boundary,
format!("boundary retry {retries}"),
span,
)?;
let mut last = None;
for attempt in 0..=retries {
self.attempt = attempt;
self.owners.push(node);
let r = self.block(body);
self.owners.pop();
match r {
Ok(v) => {
self.attempt = 0;
return self.finish(node, Ok(v));
}
Err(e) => {
let retry = e.retryable && attempt < retries;
let retry_after = e.retry_after;
last = Some(e);
if !retry {
break;
}
let delay = self.runtime.retry_delay(attempt + 1, retry_after);
if !delay.is_zero() {
std::thread::sleep(delay);
}
}
}
}
let e = last.unwrap();
let error = error_value(&e, self.attempt);
self.scopes.push(HashMap::from([(
error_binding.into(),
Binding::Value(error, None),
)]));
self.owners.push(node);
let r = self.block(catch);
self.owners.pop();
self.scopes.pop();
self.attempt = 0;
self.finish(node, r)
}
fn block(&mut self, b: &Block) -> Result<V, ToolError> {
let mut scope = HashMap::new();
for s in &b.statements {
if let crate::StmtKind::Binding { name, value } = &s.kind {
scope.insert(name.clone(), Binding::Expr(value.clone()));
}
}
self.scopes.push(scope);
let r = self
.guards_and_effects(&b.statements)
.and_then(|_| self.eval(&b.result));
self.scopes.pop();
r
}
fn guards_and_effects(&mut self, stmts: &[crate::Stmt]) -> Result<(), ToolError> {
for s in stmts {
match &s.kind {
crate::StmtKind::Skip { condition } => {
let taken = match condition {
None => true,
Some(c) => match self.eval(c)? {
V::Boolean(b) => b,
other => {
let mut e = lang(
"RL5202",
&format!(
"skip condition evaluated to {}; conditions must be \
true or false — Runlet has no truthiness, compare \
explicitly (e.g. `value != null`)",
value_kind(&other)
),
);
e.span = Some(c.span);
return Err(e);
}
},
};
if taken {
let mut e = lang(SKIP_SIGNAL, "skip");
e.span = Some(s.span);
return Err(e);
}
}
crate::StmtKind::Binding { name, value } => {
if crate::analyzer::contains_effectful_call(value, &self.runtime.registry) {
self.name(name)?;
}
}
}
}
Ok(())
}
fn budget(&self) -> Result<(), ToolError> {
if self.graph.node_count() >= self.runtime.max_graph_nodes {
return Err(lang(
"RL4105",
&format!(
"execution graph exceeded the host limit of {} nodes",
self.runtime.max_graph_nodes
),
));
}
Ok(())
}
fn node(&mut self, k: NodeKind, l: impl Into<String>, s: Span) -> Result<usize, ToolError> {
self.budget()?;
let i = self.graph.begin(k, l, s, self.attempt);
self.graph.running(i);
if let Some(&p) = self.owners.last() {
self.graph.contains(p, i)
}
Ok(i)
}
fn blocked_node(
&mut self,
k: NodeKind,
l: impl Into<String>,
s: Span,
) -> Result<usize, ToolError> {
self.budget()?;
let i = self.graph.begin(k, l, s, self.attempt);
self.graph.blocked(i);
if let Some(&parent) = self.owners.last() {
self.graph.contains(parent, i)
}
Ok(i)
}
fn clone_for_branch(&self) -> Evaluator<'_> {
Evaluator {
runtime: self.runtime,
graph: self.graph.clone(),
scopes: self.scopes.clone(),
attempt: self.attempt,
owners: self.owners.clone(),
workflow: self.workflow.clone(),
dynamic: self.dynamic.clone(),
successful_operations: self.successful_operations.clone(),
dispatch_generations: self.dispatch_generations.clone(),
operation_nodes: self.operation_nodes.clone(),
last_output: self.last_output,
binding_caches: self.binding_caches.clone(),
depth: self.depth,
dispatches: self.dispatches.clone(),
worker_budget: self.worker_budget.clone(),
}
}
fn finish(&mut self, node: usize, r: Result<V, ToolError>) -> Result<V, ToolError> {
match &r {
Ok(v) => {
self.graph.success(node, v.clone());
self.last_output = Some(node);
}
Err(e) => self.graph.fail(node, e.to_string()),
}
r
}
}
fn iteration_label(binding: &str, index: usize, value: &V) -> String {
let key = match value {
V::String(value) => Some(value.as_str()),
V::Object(fields) => fields.get("name").and_then(|value| match value {
V::String(value) => Some(value.as_str()),
_ => None,
}),
_ => None,
};
key.map_or_else(
|| format!("{binding}[{index}]"),
|key| format!("{binding}[{index}] {key}"),
)
}
fn path(e: &Expr) -> Option<String> {
match &e.kind {
ExprKind::Name(n) => Some(n.clone()),
ExprKind::Member { target, field } => Some(format!("{}.{}", path(target)?, field)),
_ => None,
}
}
fn lang(code: &str, msg: &str) -> ToolError {
ToolError::new(code, msg)
}
pub(crate) const SKIP_SIGNAL: &str = "RL4107";
fn project(v: V, i: &V) -> Result<V, ToolError> {
match (v, i) {
(V::List(x), V::Integer(i)) => idx(x.len(), *i)
.and_then(|n| x.get(n).cloned())
.ok_or_else(|| lang("RL5205", "INDEX_OUT_OF_BOUNDS")),
(V::String(s), V::Integer(i)) => {
let x = s.chars().collect::<Vec<_>>();
idx(x.len(), *i)
.and_then(|n| x.get(n))
.map(|c| V::String(c.to_string()))
.ok_or_else(|| lang("RL5205", "INDEX_OUT_OF_BOUNDS"))
}
(V::Bytes(x), V::Integer(i)) => idx(x.len(), *i)
.and_then(|n| x.get(n))
.map(|x| V::Integer(*x as i64))
.ok_or_else(|| lang("RL5205", "INDEX_OUT_OF_BOUNDS")),
(V::Object(mut o), V::String(k)) => o
.remove(k)
.ok_or_else(|| lang("RL5206", &format!("KEY_NOT_FOUND: `{k}`"))),
_ => Err(lang("RL5203", "NOT_INDEXABLE")),
}
}
fn idx(len: usize, i: i64) -> Option<usize> {
let n = if i < 0 { len as i64 + i } else { i };
(n >= 0 && n < len as i64).then_some(n as usize)
}
fn unary(op: UnaryOp, v: V) -> Result<V, ToolError> {
match (op, v) {
(UnaryOp::Not, V::Boolean(x)) => Ok(V::Boolean(!x)),
(UnaryOp::Negate, V::Integer(x)) => x
.checked_neg()
.map(V::Integer)
.ok_or_else(|| lang("RL5101", "NUMERIC_OVERFLOW")),
(UnaryOp::Negate, V::Number(x)) => {
V::number(-x).map_err(|_| lang("RL5103", "NON_FINITE_NUMBER"))
}
(UnaryOp::Not, v) => Err(lang(
"RL5202",
&format!(
"cannot apply `not` to a {} value; Runlet has no truthiness — compare \
explicitly first (e.g. `not (value != null)`)",
value_kind(&v)
),
)),
(UnaryOp::Negate, v) => Err(lang(
"RL5201",
&format!("cannot negate a {} value", value_kind(&v)),
)),
}
}
fn binary(op: BinaryOp, l: V, r: V) -> Result<V, ToolError> {
use BinaryOp::*;
match (op, l, r) {
(Add, V::Integer(a), V::Integer(b)) => a
.checked_add(b)
.map(V::Integer)
.ok_or_else(|| lang("RL5101", "NUMERIC_OVERFLOW")),
(Subtract, V::Integer(a), V::Integer(b)) => a
.checked_sub(b)
.map(V::Integer)
.ok_or_else(|| lang("RL5101", "NUMERIC_OVERFLOW")),
(Multiply, V::Integer(a), V::Integer(b)) => a
.checked_mul(b)
.map(V::Integer)
.ok_or_else(|| lang("RL5101", "NUMERIC_OVERFLOW")),
(Divide, _, V::Integer(0))
| (Divide, _, V::Number(0.0))
| (Remainder, _, V::Integer(0)) => Err(lang("RL5102", "DIVISION_BY_ZERO")),
(Remainder, V::Integer(a), V::Integer(b)) => a
.checked_rem(b)
.map(V::Integer)
.ok_or_else(|| lang("RL5101", "NUMERIC_OVERFLOW")),
(Add, V::String(a), V::String(b)) => Ok(V::String(a + &b)),
(Add, V::String(a), b) => Ok(V::String(a + &format_value(&b)?)),
(Add, a, V::String(b)) => Ok(V::String(format_value(&a)? + &b)),
(Add, V::List(mut a), V::List(b)) => {
a.extend(b);
Ok(V::List(a))
}
(Add, V::Object(mut a), V::Object(b)) => {
a.extend(b);
Ok(V::Object(a))
}
(Equal, V::Integer(a), V::Number(b)) | (Equal, V::Number(b), V::Integer(a)) => {
Ok(V::Boolean(exact(a)? == b))
}
(NotEqual, V::Integer(a), V::Number(b)) | (NotEqual, V::Number(b), V::Integer(a)) => {
Ok(V::Boolean(exact(a)? != b))
}
(Equal, a, b) => Ok(V::Boolean(a == b)),
(NotEqual, a, b) => Ok(V::Boolean(a != b)),
(And, V::Boolean(a), V::Boolean(b)) => Ok(V::Boolean(a && b)),
(Or, V::Boolean(a), V::Boolean(b)) => Ok(V::Boolean(a || b)),
(And | Or, a, b) => Err(lang(
"RL5202",
&format!(
"`and`/`or` need boolean operands, got {} and {}; Runlet has no truthiness — \
compare explicitly (e.g. `value != null`, `text != \"\"`)",
value_kind(&a),
value_kind(&b)
),
)),
(In, a, V::List(b)) => Ok(V::Boolean(b.contains(&a))),
(In, V::String(a), V::String(b)) => Ok(V::Boolean(b.contains(&a))),
(In, V::String(a), V::Object(b)) => Ok(V::Boolean(b.contains_key(&a))),
(op, a, b)
if matches!(
op,
Add | Subtract | Multiply | Divide | Less | LessEqual | Greater | GreaterEqual
) =>
{
numeric_binary(op, a, b)
}
(op, a, b) => Err(lang(
"RL5201",
&format!(
"operator `{op:?}` cannot combine {} and {} values",
value_kind(&a),
value_kind(&b)
),
)),
}
}
fn numeric_binary(op: BinaryOp, a: V, b: V) -> Result<V, ToolError> {
let (a, b) = match (a, b) {
(V::Integer(a), V::Integer(b)) => (a as f64, b as f64),
(V::Integer(a), V::Number(b)) => (exact(a)?, b),
(V::Number(a), V::Integer(b)) => (a, exact(b)?),
(V::Number(a), V::Number(b)) => (a, b),
(V::String(a), V::String(b)) => {
return Ok(V::Boolean(match op {
BinaryOp::Less => a < b,
BinaryOp::LessEqual => a <= b,
BinaryOp::Greater => a > b,
BinaryOp::GreaterEqual => a >= b,
_ => false,
}));
}
_ => return Err(lang("RL5201", "INVALID_NUMERIC_OPERANDS")),
};
use BinaryOp::*;
match op {
Add => V::number(a + b),
Subtract => V::number(a - b),
Multiply => V::number(a * b),
Divide => V::number(a / b),
Less => return Ok(V::Boolean(a < b)),
LessEqual => return Ok(V::Boolean(a <= b)),
Greater => return Ok(V::Boolean(a > b)),
GreaterEqual => return Ok(V::Boolean(a >= b)),
_ => unreachable!(),
}
.map_err(|_| lang("RL5103", "NON_FINITE_NUMBER"))
}
fn exact(x: i64) -> Result<f64, ToolError> {
let f = x as f64;
if f as i64 == x {
Ok(f)
} else {
Err(lang("RL5208", "LOSSY_NUMERIC_WIDENING"))
}
}
fn format_value(v: &V) -> Result<String, ToolError> {
match v {
V::Null | V::Bytes(_) => Err(lang("RL5207", "NOT_JSON_REPRESENTABLE")),
V::String(s) => Ok(s.clone()),
V::Integer(x) => Ok(x.to_string()),
V::Boolean(x) => Ok(x.to_string()),
V::Number(_) | V::List(_) | V::Object(_) => v
.presentation_json()
.map_err(|e| lang("RL5207", &e.to_string())),
}
}
fn computed_key(v: V) -> Result<String, ToolError> {
match v {
V::String(s) => Ok(s),
V::Integer(_) | V::Number(_) | V::Boolean(_) => format_value(&v),
other => Err(lang(
"RL5209",
&format!(
"computed property keys must be strings or scalar values, got {}",
value_kind(&other)
),
)),
}
}
fn convert_value(value: V, expected: &Schema) -> Result<V, ToolError> {
if expected.accepts(&value) {
return Ok(value);
}
match (value, expected) {
(V::Integer(x), Schema::Number { .. }) => {
V::number(exact(x)?).map_err(|_| lang("RL5103", "NON_FINITE_NUMBER"))
}
(
v @ (V::Integer(_) | V::Number(_) | V::Boolean(_) | V::List(_) | V::Object(_)),
Schema::String { .. },
) => Ok(V::String(format_value(&v)?)),
(V::List(values), Schema::List { items, .. }) => values
.into_iter()
.map(|v| convert_value(v, items))
.collect::<Result<Vec<_>, _>>()
.map(V::List),
(
V::Object(values),
Schema::Object {
properties,
required,
additional,
},
) => {
let missing = required
.iter()
.filter(|key| !values.contains_key(*key))
.cloned()
.collect::<Vec<_>>();
if !missing.is_empty() {
return Err(lang(
"RL5208",
&format!(
"missing required object propert{}: {}",
if missing.len() == 1 { "y" } else { "ies" },
missing.join(", ")
),
));
}
let mut out = BTreeMap::new();
for (key, value) in values {
match properties.get(&key) {
Some(property) => {
if matches!(value, V::Null)
&& !required.contains(&key)
&& !property.schema.accepts(&V::Null)
{
continue;
}
out.insert(key, convert_value(value, &property.schema)?);
}
None if *additional => {
out.insert(key, value);
}
None => {
return Err(lang(
"RL5208",
&format!("unexpected object property `{key}`"),
));
}
}
}
Ok(V::Object(out))
}
(value, Schema::Union { variants, .. }) => {
let converted = variants
.iter()
.filter_map(|schema| convert_value(value.clone(), schema).ok())
.collect::<Vec<_>>();
if converted.len() == 1 {
Ok(converted.into_iter().next().unwrap())
} else {
Err(lang("RL5208", "ambiguous or invalid union conversion"))
}
}
(value, expected) => Err(lang(
"RL5208",
&format!(
"{} value cannot be safely converted to the expected {} schema",
value_kind(&value),
schema_kind(expected)
),
)),
}
}
fn value_kind(v: &V) -> &'static str {
match v {
V::Null => "null",
V::Boolean(_) => "boolean",
V::Integer(_) => "integer",
V::Number(_) => "number",
V::String(_) => "string",
V::Bytes(_) => "bytes",
V::List(_) => "list",
V::Object(_) => "object",
}
}
fn schema_kind(s: &Schema) -> &'static str {
match s {
Schema::Any => "any",
Schema::Null => "null",
Schema::Boolean => "boolean",
Schema::Integer { .. } => "integer",
Schema::Number { .. } => "number",
Schema::String { .. } => "string",
Schema::Bytes => "bytes",
Schema::List { .. } => "list",
Schema::Map { .. } => "map",
Schema::Object { .. } => "object",
Schema::Union { .. } => "union",
Schema::Never => "never",
}
}
fn error_value(e: &ToolError, attempt: u32) -> V {
let mut o = e.details.clone();
for (k, v) in [
("code", V::String(e.code.clone())),
("message", V::String(e.message.clone())),
("retryable", V::Boolean(e.retryable)),
("node_id", V::String(String::new())),
("attempt", V::Integer(attempt as i64)),
("uncertain", V::Boolean(e.uncertain)),
(
"span",
match e.span {
Some(span) => V::Object(BTreeMap::from([
("start".to_string(), V::Integer(span.start as i64)),
("end".to_string(), V::Integer(span.end as i64)),
])),
None => V::Null,
},
),
] {
o.insert(k.into(), v);
}
V::Object(o)
}