use crate::err_run;
use areev_core::error::{Hash, Result};
use areev_core::types::Grain;
use areev_run_core::{Budgets, NodeExecutor, PlanGraph, RunError};
use areev_store::Areev;
use serde_json::json;
use std::collections::BTreeMap;
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct ForkBase {
pub base_run: String,
pub base_checkpoint: String,
pub base_superstep: u64,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct PinnedTool {
pub node: String,
pub tool_hash: String,
pub tool_name: String,
pub executor: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub executor_uri: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub runtime: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub runtime_limits: Option<serde_json::Value>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub capabilities: Option<serde_json::Value>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub read: Option<serde_json::Value>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub ask_kind: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub decide: Option<serde_json::Value>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub locked_params: Option<serde_json::Value>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub llm: Option<areev_run_core::LlmNodeConfig>,
}
pub const DECIDE_EXECUTOR: &str = "decide";
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct DeciderPin {
pub describe: String,
pub calibrated: bool,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct RunManifest {
pub run_id: String,
pub plan_hash: String,
pub principal: String,
pub pinned: Vec<PinnedTool>,
pub budgets: BudgetsSpec,
pub ask_ttl_sec: Option<i64>,
pub redaction: String,
pub input: serde_json::Value,
#[serde(default)]
pub llm_max_tokens: Option<u32>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub max_effects_per_attempt: Option<u32>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub llm_tool_result_chars: Option<usize>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub llm_context_tokens: Option<u64>,
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
pub reducers: BTreeMap<String, String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub fork_of: Option<ForkBase>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub initiator: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub llm: Option<LlmPin>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub engine: Option<EnginePin>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub input_ref: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub harness_ns: Option<String>,
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
pub allow_confirmation_asks: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub decider: Option<DeciderPin>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub dispatch_pins: Vec<PinnedTool>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub warnings: Vec<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct LlmPin {
pub provider: String,
pub model: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub region: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub tag: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub profile: Option<String>,
}
impl LlmPin {
pub fn matches(&self, other: &LlmPin) -> bool {
self == other
}
pub fn describe(&self) -> String {
let mut s = format!("{}:{}", self.provider, self.model);
if let Some(r) = &self.region {
s.push_str(&format!(" region={r}"));
}
if let Some(t) = &self.tag {
s.push_str(&format!(" tag={t}"));
}
if let Some(p) = &self.profile {
s.push_str(&format!(" profile={}", &p[..p.len().min(12)]));
}
s
}
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct EnginePin {
pub version: String,
pub scheduler_epoch: u32,
}
impl EnginePin {
pub fn current() -> Self {
EnginePin {
version: env!("CARGO_PKG_VERSION").to_string(),
scheduler_epoch: areev_run_core::SCHEDULER_EPOCH,
}
}
}
#[derive(Debug, Clone, Copy, Default, serde::Serialize, serde::Deserialize)]
pub struct BudgetsSpec {
pub max_supersteps: Option<u64>,
pub max_tokens: Option<u64>,
pub max_usd_micros: Option<u64>,
pub max_wall_ms: Option<u64>,
pub max_storage_bytes: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub max_effects: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub max_tool_calls: Option<u64>,
}
impl BudgetsSpec {
pub fn to_budgets(self) -> Budgets {
Budgets {
max_supersteps: self.max_supersteps.unwrap_or(Budgets::default().max_supersteps),
max_tokens: self.max_tokens,
max_usd_micros: self.max_usd_micros,
max_wall_ms: self.max_wall_ms,
max_storage_bytes: self.max_storage_bytes,
max_effects: self.max_effects,
max_tool_calls: self.max_tool_calls,
}
}
}
const RESULT_SHARE_OF_CEILING: u64 = 4;
const CHARS_PER_TOKEN: u64 = 4;
impl RunManifest {
#[allow(clippy::too_many_arguments)]
pub fn resolve(
m: &mut Areev,
ns: &str,
run_id: &str,
plan_hash: &Hash,
plan: &PlanGraph,
principal: &str,
budgets: BudgetsSpec,
ask_ttl_sec: Option<i64>,
input: serde_json::Value,
llm_available: bool,
) -> std::result::Result<RunManifest, RunError> {
let fields: Option<serde_json::Map<String, serde_json::Value>> =
m.get(plan_hash).ok().map(|g| g.fields.into_iter().collect());
Self::resolve_with_fields(
m,
ns,
run_id,
plan_hash,
fields.as_ref(),
plan,
principal,
budgets,
ask_ttl_sec,
input,
llm_available,
)
}
#[allow(clippy::too_many_arguments)]
pub fn resolve_with_fields(
m: &mut Areev,
ns: &str,
run_id: &str,
plan_hash: &Hash,
fields: Option<&serde_json::Map<String, serde_json::Value>>,
plan: &PlanGraph,
principal: &str,
budgets: BudgetsSpec,
ask_ttl_sec: Option<i64>,
input: serde_json::Value,
llm_available: bool,
) -> std::result::Result<RunManifest, RunError> {
let plan_ns = fields
.and_then(|f| f.get("namespace"))
.and_then(|n| n.as_str())
.map(str::to_string);
let mut reads = crate::memread::parse_reads(
fields.and_then(|f| f.get(crate::memread::READS_FIELD)),
plan,
ns,
)?;
crate::memread::pin_queries(&mut reads, m)?;
let mut pinned = Vec::with_capacity(plan.nodes.len());
let mut warnings: Vec<String> = Vec::new();
for (i, node) in plan.nodes.iter().enumerate() {
if let Some(spec) = reads.remove(node) {
let op = spec.get("op").and_then(|o| o.as_str()).unwrap_or_default();
pinned.push(PinnedTool {
node: node.clone(),
tool_hash: String::new(),
tool_name: format!("mg:{op}"),
executor: crate::memread::MEMORY_EXECUTOR.into(),
executor_uri: None,
runtime: None,
runtime_limits: None,
capabilities: None,
read: Some(spec),
ask_kind: None,
decide: None,
locked_params: None,
llm: None,
});
continue;
}
let resolved = match &plan.bindings[i] {
Some(hash_str) => {
let h = Hash::from_hex(hash_str).map_err(|_| RunError::UnresolvedRef {
what: format!("binding for node '{node}' is not a content address"),
})?;
let g = m.get_stored(&h).map_err(|e| RunError::UnresolvedRef {
what: format!("binding for node '{node}': {e}"),
})?;
if g.grain_type == areev_core::types::GrainType::Workflow {
PinnedTool {
node: node.clone(),
tool_hash: h.to_hex(),
tool_name: node.clone(),
executor: "subgraph".into(),
executor_uri: None,
runtime: None,
runtime_limits: None,
capabilities: None,
read: None,
ask_kind: None,
decide: None,
locked_params: None,
llm: None,
}
} else {
let pin = pin_from_definition(node, &h, &g)?;
warnings.extend(locked_params_warning(&pin, &g));
pin
}
}
None => {
match find_definition_by_name(m, ns, node) {
Some((h, g)) => {
let pin = pin_from_definition(node, &h, &g)?;
warnings.extend(locked_params_warning(&pin, &g));
pin
}
None if misrouted(m, ns, plan_ns.as_deref(), node) => {
let pns = plan_ns.as_deref().unwrap_or_default();
return Err(RunError::UnresolvedRef {
what: format!(
"node '{node}' has no binding, and namespace '{ns}' — where \
this run reads and journals — holds no Tool Definition \
named '{node}'. The plan grain itself lives in namespace \
'{pns}', which does. Start the run there (`--ns {pns}`), \
or bind the node to a Definition by hash so it resolves \
from any namespace"
),
});
}
None if llm_available => PinnedTool {
node: node.clone(),
tool_hash: String::new(),
tool_name: node.clone(),
executor: "abstract".into(),
executor_uri: None,
runtime: None,
runtime_limits: None,
capabilities: None,
read: None,
ask_kind: None,
decide: None,
locked_params: None,
llm: None,
},
None => return Err(RunError::NoToolLlm { node: node.clone() }),
}
}
};
pinned.push(resolved);
}
let mut resolve_dispatch = |node: &str, entry: &str| -> std::result::Result<PinnedTool, RunError> {
let by_hash = entry.len() == 64 && entry.chars().all(|c| c.is_ascii_hexdigit());
let found = if by_hash {
Hash::from_hex(entry).ok().and_then(|h| m.get_stored(&h).ok().map(|g| (h, g)))
} else {
find_definition_by_name(m, ns, entry).or_else(|| match plan_ns.as_deref() {
Some(p) if p != ns => find_definition_by_name(m, p, entry),
_ => None,
})
};
let Some((h, g)) = found else {
return Err(RunError::UnresolvedRef {
what: format!(
"node '{node}' llm.dispatch names '{entry}', which resolves to no Tool \
Definition — not in namespace '{ns}', where this run reads, nor in the \
plan's own namespace{}",
plan_ns.as_deref().map(|p| format!(" '{p}'")).unwrap_or_default()
),
});
};
let mut pin = pin_from_definition(node, &h, &g)?;
if pin.executor != "host" {
return Err(RunError::InvalidPlan {
why: format!(
"node '{node}' llm.dispatch names '{entry}', a {} tool — a dynamic tool \
dispatches through a host tool only",
pin.executor
),
});
}
pin.node = node.to_string();
Ok(pin)
};
let (blocks, dispatch_pins) = crate::llmcfg::parse_llm(
fields.and_then(|f| f.get(crate::llmcfg::LLM_FIELD)),
plan,
&pinned,
&input,
&mut resolve_dispatch,
)?;
for (n, d) in dispatch_pins.iter().enumerate() {
if let Some(o) = dispatch_pins[..n]
.iter()
.find(|o| o.tool_name == d.tool_name && o.tool_hash != d.tool_hash)
{
return Err(RunError::InvalidPlan {
why: format!(
"llm.dispatch pins two Definitions named {}: {} (node '{}') and {} \
(node '{}') — one name, one version per run",
d.tool_name, o.tool_hash, o.node, d.tool_hash, d.node
),
});
}
}
for d in &dispatch_pins {
if let Some(b) = pinned
.iter()
.find(|p| p.executor == "host" && p.tool_name == d.tool_name && p.tool_hash != d.tool_hash)
{
return Err(RunError::InvalidPlan {
why: format!(
"node '{}' llm.dispatch pins {} ({}), but node '{}' binds a different \
Definition of that name ({}) — dispatch through the bound one, or give \
the other version its own tool_name",
d.node, d.tool_name, d.tool_hash, b.node, b.tool_hash
),
});
}
}
for (node, mut block) in blocks {
for d in &mut block.dynamic_tools {
let h = Hash::from_hex(&d.dispatch.hash).map_err(|_| RunError::UnresolvedRef {
what: format!("dispatch hash {} is not a content address", d.dispatch.hash),
})?;
d.dispatch.definition_schema = m
.get_stored(&h)
.ok()
.and_then(|g| g.to_tool().ok())
.and_then(|t| t.input_schema);
}
if let Some(p) = pinned.iter_mut().find(|p| p.node == node) {
p.llm = Some(block);
}
}
let mut reducers: BTreeMap<String, String> = BTreeMap::new();
if let Some(fields) = fields {
if let Some(table) = fields.get("reducers").and_then(|v| v.as_object()) {
for (key, name) in table {
let Some(name) = name.as_str() else {
return Err(RunError::InvalidPlan {
why: format!("reducer for state key '{key}' is not a string"),
});
};
if !crate::reducers::is_builtin(name) {
return Err(RunError::InvalidPlan {
why: format!(
"unknown reducer '{name}' for state key '{key}' \
(builtins: {:?})",
crate::reducers::BUILTIN_REDUCERS
),
});
}
reducers.insert(key.clone(), name.to_string());
}
}
}
Ok(RunManifest {
run_id: run_id.to_string(),
plan_hash: plan_hash.to_hex(),
principal: principal.to_string(),
pinned,
budgets,
ask_ttl_sec,
redaction: "none".into(),
input,
llm_max_tokens: None,
max_effects_per_attempt: None,
llm_tool_result_chars: None,
llm_context_tokens: None,
reducers,
fork_of: None,
initiator: None,
llm: None,
engine: None,
input_ref: None,
harness_ns: None,
allow_confirmation_asks: false,
decider: None,
dispatch_pins,
warnings,
})
}
pub fn with_llm_pin(mut self, pin: Option<LlmPin>) -> Self {
self.llm = pin;
self
}
pub fn with_decider_pin(mut self, pin: Option<DeciderPin>) -> std::result::Result<Self, RunError> {
if pin.is_none() {
if let Some(p) = self.pinned.iter().find(|p| p.executor == DECIDE_EXECUTOR) {
return Err(RunError::NoDecider { node: p.node.clone() });
}
}
self.decider = pin;
Ok(self)
}
pub fn with_engine_pin(mut self) -> Self {
self.engine = Some(EnginePin::current());
self
}
pub fn with_attribution(mut self, opts: &crate::RunOptions) -> Self {
self.initiator = opts.initiator.clone();
self.harness_ns = opts.harness_ns.clone();
self.allow_confirmation_asks = opts.allow_confirmation_asks;
self
}
pub fn with_input_ref(mut self, hash: String) -> Self {
self.input_ref = Some(hash);
self.input = serde_json::Value::Null;
self
}
pub fn with_limits(mut self, opts: &crate::RunOptions, model_window: Option<u64>) -> Self {
self.llm_max_tokens = opts.llm_max_tokens;
self.max_effects_per_attempt = opts.max_effects_per_attempt;
self.llm_context_tokens = opts.llm_context_tokens.or_else(|| {
model_window.map(|w| w.saturating_sub(u64::from(self.llm_max_tokens.unwrap_or(
crate::runner::DEFAULT_LLM_MAX_TOKENS,
))))
});
self.llm_tool_result_chars = opts.llm_tool_result_chars.or_else(|| {
self.llm_context_tokens
.map(|c| ((c / RESULT_SHARE_OF_CEILING) * CHARS_PER_TOKEN) as usize)
});
self
}
pub fn llm_reserve_tokens(&self) -> u64 {
u64::from(self.llm_max_tokens.unwrap_or(crate::runner::DEFAULT_LLM_MAX_TOKENS))
}
pub fn max_effects_per_attempt(&self) -> u32 {
self.max_effects_per_attempt
.unwrap_or(areev_run_core::DEFAULT_MAX_EFFECTS_PER_ATTEMPT)
}
pub fn host_pin_by_hash(&self, hash: &str) -> Option<&PinnedTool> {
self.pinned
.iter()
.chain(&self.dispatch_pins)
.find(|p| p.executor == "host" && p.tool_hash == hash)
}
pub fn code_pins(&self) -> impl Iterator<Item = &PinnedTool> {
self.pinned.iter().chain(&self.dispatch_pins)
}
pub fn executors(&self) -> Vec<NodeExecutor> {
let mut seen = std::collections::HashSet::new();
let offered: Vec<areev_run_core::OfferedTool> = self
.pinned
.iter()
.filter(|p| p.executor == "host")
.filter(|p| seen.insert(p.tool_name.clone()))
.map(|p| {
let mut t = areev_run_core::OfferedTool::new(p.tool_name.clone(), p.tool_hash.clone());
t.locked = p.locked_params.clone();
t
})
.collect();
let offer_for = |p: &PinnedTool| -> Vec<areev_run_core::OfferedTool> {
let Some(cfg) = &p.llm else { return offered.clone() };
let mut tools: Vec<areev_run_core::OfferedTool> = match &cfg.tools {
Some(names) => offered.iter().filter(|t| names.contains(&t.tool_name)).cloned().collect(),
None => offered.clone(),
};
for d in &cfg.dynamic_tools {
let def_lock = self
.host_pin_by_hash(&d.dispatch.hash)
.and_then(|q| q.locked_params.as_ref());
tools.push(areev_run_core::OfferedTool {
tool_name: d.name.clone(),
tool_hash: d.dispatch.hash.clone(),
locked: crate::llmcfg::merge_locks(d.dispatch.locked.as_ref(), def_lock),
dynamic: Some(Box::new(d.clone())),
});
}
tools
};
self.pinned
.iter()
.map(|p| match p.executor.as_str() {
"client" => NodeExecutor::Client {
tool_hash: p.tool_hash.clone(),
tool_name: p.tool_name.clone(),
approval: p.ask_kind.as_deref() != Some(ASK_KIND_CONFIRMATION),
},
"subgraph" => NodeExecutor::Subgraph { workflow_hash: p.tool_hash.clone() },
crate::memread::MEMORY_EXECUTOR => {
let spec = p.read.clone().unwrap_or(serde_json::Value::Null);
NodeExecutor::MemoryRead {
op: spec.get("op").and_then(|o| o.as_str()).unwrap_or_default().to_string(),
spec,
}
}
"abstract" => NodeExecutor::Abstract {
tools: offer_for(p),
llm: p.llm.clone().map(Box::new),
},
DECIDE_EXECUTOR => NodeExecutor::Decide {
tool_hash: p.tool_hash.clone(),
tool_name: p.tool_name.clone(),
},
_ => NodeExecutor::Host {
tool_hash: p.tool_hash.clone(),
tool_name: p.tool_name.clone(),
},
})
.collect()
}
pub fn persist_in_namespace(&self, m: &mut Areev, ns: &str) -> Result<(Hash, Hash)> {
self.persist_with_input(m, ns, None)
}
pub fn persist_with_input(
&self,
m: &mut Areev,
ns: &str,
input_ns: Option<&str>,
) -> Result<(Hash, Hash)> {
let mut manifest = self.clone();
if let Some(input_ns) = input_ns {
let mut grain = areev_core::types::State::new(self.input.clone())
.namespace(input_ns)
.created_at(0);
grain.common.author_did = Some(self.principal.clone());
grain
.common
.extra_fields
.insert("run_id".into(), serde_json::json!(self.run_id));
let h = m.add(&grain)?;
manifest = manifest.with_input_ref(h.to_hex());
}
let this = &manifest;
let config = serde_json::to_value(json!({ "areev_run": this }))
.expect("manifest serializes");
let mut state = areev_core::types::State::new(config)
.namespace(areev_core::authz::HARNESS_NS)
.created_at(0);
state.common.author_did = Some(self.principal.clone());
let config_hash = m.add(&state)?;
let mut link = areev_core::types::Fact::new(
&format!("run:{}", self.run_id),
"mg:harness",
&config_hash.to_hex(),
)
.namespace(areev_core::authz::HARNESS_NS);
link.common.author_did = Some(self.principal.clone());
link.common
.extra_fields
.insert("run_id".into(), serde_json::json!(self.run_id));
if !ns.is_empty() {
link.common.extra_fields.insert("run_ns".into(), serde_json::json!(ns));
}
let link_hash = m.add(&link)?;
Ok((config_hash, link_hash))
}
#[deprecated(
note = "use persist_in_namespace(m, ns) so the run index can be scoped by namespace"
)]
pub fn persist(&self, m: &mut Areev) -> Result<(Hash, Hash)> {
self.persist_in_namespace(m, "")
}
pub fn load(m: &mut Areev, run_id: &str) -> std::result::Result<RunManifest, RunError> {
let link = m
.latest(areev_core::authz::HARNESS_NS, &format!("run:{run_id}"), "mg:harness")
.map_err(err_run)?
.ok_or_else(|| RunError::ManifestMismatch {
why: format!("no manifest link for run '{run_id}'"),
})?;
let config_hash = link
.get_str("object")
.and_then(|h| Hash::from_hex(h).ok())
.ok_or_else(|| RunError::ManifestMismatch {
why: "manifest link carries no config hash".into(),
})?;
let config = m.get_stored(&config_hash).map_err(|e| RunError::ManifestMismatch {
why: format!("manifest config unreadable: {e}"),
})?;
let manifest = config
.fields
.get("context")
.or_else(|| config.fields.get("context_data"))
.and_then(|c| c.get("areev_run"))
.cloned()
.ok_or_else(|| RunError::ManifestMismatch {
why: "config State carries no areev_run manifest".into(),
})?;
let mut manifest: RunManifest =
serde_json::from_value(manifest).map_err(|e| RunError::ManifestMismatch {
why: format!("manifest does not parse: {e}"),
})?;
manifest.resolve_input(m)?;
Ok(manifest)
}
fn resolve_input(&mut self, m: &mut Areev) -> std::result::Result<(), RunError> {
let Some(hex) = self.input_ref.clone() else {
return Ok(());
};
let h = Hash::from_hex(&hex).map_err(|_| RunError::ManifestMismatch {
why: format!("input_ref {hex:?} is not a content address"),
})?;
let grain = m.get_stored(&h).map_err(|_| RunError::UnresolvedRef {
what: format!(
"run '{}' stores its input as grain {hex}, which is no longer readable \
— the input a run replays from cannot be reconstructed, so verify and \
resume refuse rather than replay against a different run",
self.run_id
),
})?;
self.input = grain
.fields
.get("context")
.or_else(|| grain.fields.get("context_data"))
.cloned()
.unwrap_or(serde_json::Value::Null);
Ok(())
}
}
pub fn pin_from_definition(
label: &str,
h: &Hash,
g: &areev_core::format::deserialize::DeserializedGrain,
) -> std::result::Result<PinnedTool, RunError> {
let node = label;
if g.get_str("kind") != Some("definition") {
return Err(RunError::UnresolvedRef {
what: format!(
"node '{node}' binds {} which is not a Tool Definition \
(kind = {:?})",
h.to_hex(),
g.get_str("kind").unwrap_or("execution")
),
});
}
let tool_name = g
.get_str("tool_name")
.unwrap_or(node)
.to_string();
if !areev_core::types::json_schema_subset::is_valid_tool_name(&tool_name) {
return Err(RunError::UnresolvedRef {
what: format!(
"node '{node}' binds {} whose tool_name {tool_name:?} is not a valid \
tool name (1-64 chars of [A-Za-z0-9_.-])",
h.to_hex()
),
});
}
let executor = match g.get_str("executor_kind") {
Some("client") => "client",
_ => "host",
};
let decide_node =
executor == "host" && g.get_str("executor_uri") == Some(areev_run_core::DECIDE_URI);
let executor_uri = match g.get_str("executor_uri") {
None => None,
Some(uri) if executor == "client" => {
return Err(RunError::CodeExecRefused {
condition: format!(
"node '{node}' binds a client tool carrying executor_uri {uri:?} — a \
client tool is answered by a person through `run respond`, so it has \
no executor to name"
),
})
}
Some(_) if decide_node => None,
Some(uri) => {
let hex = crate::executor::strip_cas(uri);
if hex.len() != 64 || !hex.chars().all(|c| c.is_ascii_hexdigit()) {
return Err(RunError::CodeExecRefused {
condition: format!(
"node '{node}' names executor_uri {uri:?}, which this build cannot \
dispatch — the only executable form is a content address, \
cas://sha256:<64 hex>"
),
});
}
Some(format!("cas://sha256:{}", hex.to_ascii_lowercase()))
}
};
let runtime = match g.get_str("runtime") {
None | Some("native") => None,
Some(rt) if decide_node => {
return Err(RunError::CodeExecRefused {
condition: format!(
"node '{node}' is a decision node ({}) but declares runtime {rt:?} — \
the host's decision backend answers it, and there is no code to run",
areev_run_core::DECIDE_URI
),
})
}
Some(rt) if executor_uri.is_none() => {
return Err(RunError::CodeExecRefused {
condition: format!(
"node '{node}' declares runtime {rt:?} but names no executor_uri — a runtime routes a code blob, and there is none"
),
})
}
Some(rt) if crate::executor::is_sandbox_runtime(rt) => Some(rt.to_string()),
Some(rt) => {
return Err(RunError::CodeExecRefused {
condition: format!(
"node '{node}' declares runtime {rt:?}, which this build cannot dispatch — accepted: native, wasm32-areev, wasm32-areev-io"
),
})
}
};
let runtime_limits = if runtime.is_some() { g.fields.get("runtime_limits").cloned() } else { None };
areev_core::types::capability::validate_transfer_limits(runtime_limits.as_ref())
.map_err(|detail| RunError::TransferLimitInvalid { node: node.to_string(), detail })?;
let capabilities = match g.fields.get("capabilities") {
None => None,
Some(v) if v.is_null() => None,
Some(v) => {
if !crate::executor::runtime_allows_capabilities(runtime.as_deref()) {
return Err(RunError::CodeExecRefused {
condition: format!(
"node '{node}' declares capabilities but its runtime is {:?} — \
capabilities require runtime \"wasm32-areev-io\"",
runtime.as_deref().unwrap_or("native")
),
});
}
areev_core::types::capability::Declaration::parse(v).map_err(|e| RunError::CodeExecRefused {
condition: format!("node '{node}' declares malformed capabilities: {e}"),
})?;
Some(v.clone())
}
};
if crate::executor::runtime_allows_capabilities(runtime.as_deref()) && capabilities.is_none() {
return Err(RunError::CodeExecRefused {
condition: format!(
"node '{node}' declares runtime \"wasm32-areev-io\" but no capabilities — \
a capability runtime with an empty declaration can reach nothing; \
use \"wasm32-areev\" for a pure module"
),
});
}
let ask_kind = match g.get_str("ask_kind") {
None | Some(ASK_KIND_APPROVAL) => None,
Some(_) if executor != "client" => {
return Err(RunError::InvalidPlan {
why: format!(
"node '{node}' declares ask_kind on a {executor} tool — only a \
client tool is answered by a person, so only a client tool has \
an ask kind"
),
})
}
Some(ASK_KIND_CONFIRMATION) => Some(ASK_KIND_CONFIRMATION.to_string()),
Some(other) => {
return Err(RunError::InvalidPlan {
why: format!(
"node '{node}' declares ask_kind {other:?}, which this build does \
not understand — accepted: {ASK_KIND_APPROVAL}, \
{ASK_KIND_CONFIRMATION}"
),
})
}
};
let decide = if decide_node { Some(parse_decide(node, g.fields.get("decide"))?) } else { None };
let locked_params = match g.fields.get("locked_params") {
None | Some(serde_json::Value::Null) => None,
Some(serde_json::Value::Object(o)) if o.is_empty() => None,
Some(v @ serde_json::Value::Object(_)) if executor == "host" && !decide_node => Some(v.clone()),
Some(serde_json::Value::Object(_)) => None,
Some(other) => {
return Err(RunError::InvalidPlan {
why: format!("node '{node}' binds a Definition whose locked_params is not an object: {other}"),
})
}
};
Ok(PinnedTool {
node: node.to_string(),
tool_hash: h.to_hex(),
tool_name,
executor: if decide_node { DECIDE_EXECUTOR.into() } else { executor.into() },
executor_uri,
runtime,
runtime_limits,
capabilities,
read: None,
ask_kind,
decide,
locked_params,
llm: None,
})
}
pub fn locked_params_warning(
pin: &PinnedTool,
g: &areev_core::format::deserialize::DeserializedGrain,
) -> Option<String> {
let declared = g
.fields
.get("locked_params")
.and_then(|v| v.as_object())
.is_some_and(|o| !o.is_empty());
(declared && pin.locked_params.is_none() && pin.executor != "host").then(|| {
format!(
"RUN-W001: node '{}' binds {} ({}), a {} tool, which declares locked_params — no \
model calls it, so they are ignored",
pin.node, pin.tool_name, pin.tool_hash, pin.executor
)
})
}
fn parse_decide(
node: &str,
decl: Option<&serde_json::Value>,
) -> std::result::Result<serde_json::Value, RunError> {
let bad = |why: String| RunError::InvalidPlan { why: format!("decision node '{node}': {why}") };
let decl = match decl {
None | Some(serde_json::Value::Null) => serde_json::Map::new(),
Some(serde_json::Value::Object(o)) => o.clone(),
Some(_) => return Err(bad("`decide` must be an object".into())),
};
if let Some(k) = decl.keys().find(|k| !matches!(k.as_str(), "questions" | "into")) {
return Err(bad(format!("unknown `decide` key {k:?} (accepted: questions, into)")));
}
let into = match decl.get("into") {
None => node.to_string(),
Some(serde_json::Value::String(s)) if !s.is_empty() && !s.starts_with('$') => s.clone(),
Some(other) => {
return Err(bad(format!(
"`into` must be a non-empty state key not starting with '$', got {other}"
)))
}
};
let mut out = json!({ "into": into });
if let Some(q) = decl.get("questions") {
let parsed = areev_core::decide::questions_from_wire(q).map_err(|e| bad(e.to_string()))?;
if parsed.is_empty() {
return Err(bad("`questions` is empty".into()));
}
out["questions"] = areev_core::decide::questions_to_wire(&parsed);
}
Ok(out)
}
pub const ASK_KIND_APPROVAL: &str = "approval";
pub const ASK_KIND_CONFIRMATION: &str = "confirmation";
fn find_definition_by_name(
m: &mut Areev,
ns: &str,
name: &str,
) -> Option<(Hash, areev_core::format::deserialize::DeserializedGrain)> {
const CATALOG_SCAN: usize = 200_000;
let hits = m
.recent(ns, Some(areev_core::types::GrainType::Tool), CATALOG_SCAN)
.ok()?;
let mut best: Option<(i64, String)> = None;
let mut candidates: BTreeMap<String, areev_core::format::deserialize::DeserializedGrain> =
BTreeMap::new();
for g in hits {
if g.get_str("kind") == Some("definition") && g.get_str("tool_name") == Some(name) {
let at = g.get_i64("created_at").unwrap_or(0);
let key = (at, g.hash.to_hex());
if best.as_ref().is_none_or(|b| key > *b) {
best = Some(key);
}
candidates.insert(g.hash.to_hex(), g);
}
}
let (_, hex) = best?;
let g = candidates.remove(&hex)?;
let h = Hash::from_hex(&hex).ok()?;
Some((h, g))
}
fn misrouted(m: &mut Areev, ns: &str, plan_ns: Option<&str>, node: &str) -> bool {
match plan_ns {
Some(p) if p != ns => find_definition_by_name(m, p, node).is_some(),
_ => false,
}
}
pub fn abstract_nodes(
m: &mut Areev,
ns: &str,
plan_hash: &Hash,
) -> std::result::Result<Vec<String>, String> {
let grain = m.get(plan_hash).map_err(|e| format!("{e}"))?;
let wf = grain.to_workflow().map_err(|e| format!("{e}"))?;
let plan = PlanGraph::build(&wf).map_err(|e| e.to_string())?;
let reads = grain.fields.get(crate::memread::READS_FIELD).and_then(|r| r.as_object());
let mut out = Vec::new();
for (i, node) in plan.nodes.iter().enumerate() {
if reads.is_some_and(|r| r.contains_key(node)) {
continue;
}
if plan.bindings[i].is_none() && find_definition_by_name(m, ns, node).is_none() {
out.push(node.clone());
}
}
Ok(out)
}
#[cfg(test)]
mod tests {
use super::*;
fn bare() -> RunManifest {
RunManifest {
run_id: "r1".into(),
plan_hash: "deadbeef".into(),
principal: "user:t".into(),
pinned: vec![],
budgets: BudgetsSpec::default(),
ask_ttl_sec: None,
redaction: "none".into(),
input: json!({}),
llm_max_tokens: None,
max_effects_per_attempt: None,
llm_tool_result_chars: None,
llm_context_tokens: None,
reducers: BTreeMap::new(),
fork_of: None,
initiator: None,
llm: None,
engine: None,
input_ref: None,
harness_ns: None,
allow_confirmation_asks: false,
decider: None,
dispatch_pins: Vec::new(),
warnings: Vec::new(),
}
}
fn host_pin(node: &str, tool: &str) -> PinnedTool {
PinnedTool {
node: node.into(),
tool_hash: format!("hash-{tool}"),
tool_name: tool.into(),
executor: "host".into(),
executor_uri: None,
runtime: None,
runtime_limits: None,
capabilities: None,
read: None,
ask_kind: None,
decide: None,
locked_params: None,
llm: None,
}
}
#[test]
fn a_definition_bound_to_two_nodes_is_offered_once() {
let mut m = bare();
m.pinned = vec![
host_pin("parse", "parse_attachments"),
PinnedTool {
node: "extract".into(),
tool_hash: String::new(),
tool_name: "extract".into(),
executor: "abstract".into(),
executor_uri: None,
runtime: None,
runtime_limits: None,
capabilities: None,
read: None,
ask_kind: None,
decide: None,
locked_params: None,
llm: None,
},
host_pin("reply_done", "reply_email"),
host_pin("reply_rejected", "reply_email"),
];
let executors = m.executors();
assert_eq!(executors.len(), 4);
match &executors[1] {
NodeExecutor::Abstract { tools, .. } => {
let names: Vec<&str> = tools.iter().map(|t| t.tool_name.as_str()).collect();
assert_eq!(names, vec!["parse_attachments", "reply_email"]);
assert_eq!(tools[1].tool_hash, "hash-reply_email");
}
other => panic!("expected the abstract node's offer, got {other:?}"),
}
for (i, node) in [(2usize, "reply_done"), (3, "reply_rejected")] {
match &executors[i] {
NodeExecutor::Host { tool_name, tool_hash } => {
assert_eq!(tool_name, "reply_email", "{node} runs its own binding");
assert_eq!(tool_hash, "hash-reply_email");
}
other => panic!("{node}: expected a host executor, got {other:?}"),
}
}
}
#[test]
fn a_memory_read_is_never_offered_and_never_host() {
let spec = json!({"op": "entity_at", "ns": "ops", "into": "cover", "subject": "s",
"relation": "p", "at": 1, "axis": "world"});
let read_pin = |read: Option<serde_json::Value>| PinnedTool {
node: "cover".into(),
tool_hash: String::new(),
tool_name: "mg:entity_at".into(),
executor: "memory".into(),
executor_uri: None,
runtime: None,
runtime_limits: None,
capabilities: None,
read,
ask_kind: None,
decide: None,
locked_params: None,
llm: None,
};
let mut m = bare();
m.pinned = vec![
host_pin("parse", "parse_attachments"),
read_pin(Some(spec.clone())),
PinnedTool { executor: "abstract".into(), ..host_pin("decide", "decide") },
];
let executors = m.executors();
assert_eq!(executors[1], NodeExecutor::MemoryRead { op: "entity_at".into(), spec });
let NodeExecutor::Abstract { tools, .. } = &executors[2] else { panic!("{executors:?}") };
let offered: Vec<&str> = tools.iter().map(|t| t.tool_name.as_str()).collect();
assert_eq!(offered, vec!["parse_attachments"], "a read is never offered to a model");
m.pinned = vec![read_pin(None)];
assert!(
matches!(m.executors()[0], NodeExecutor::MemoryRead { .. }),
"a memory pin without its declaration still never becomes Host"
);
let json = serde_json::to_string(&m).unwrap();
assert!(!json.contains("\"read\""), "an absent read does not reach the wire: {json}");
}
#[test]
fn an_unset_effect_cap_serializes_as_it_did_before_the_field_existed() {
let json = serde_json::to_string(&bare()).unwrap();
assert_eq!(
json,
r#"{"run_id":"r1","plan_hash":"deadbeef","principal":"user:t","pinned":[],"budgets":{"max_supersteps":null,"max_tokens":null,"max_usd_micros":null,"max_wall_ms":null,"max_storage_bytes":null},"ask_ttl_sec":null,"redaction":"none","input":{},"llm_max_tokens":null}"#
);
}
#[test]
fn a_pre_existing_manifest_reads_back_at_the_default_cap() {
let stored = r#"{"run_id":"r1","plan_hash":"deadbeef","principal":"user:t",
"pinned":[],"budgets":{"max_supersteps":null,"max_tokens":null,
"max_usd_micros":null,"max_wall_ms":null,"max_storage_bytes":null},
"ask_ttl_sec":null,"redaction":"none","input":{},"llm_max_tokens":null}"#;
let m: RunManifest = serde_json::from_str(stored).unwrap();
assert_eq!(m.max_effects_per_attempt, None);
assert_eq!(m.max_effects_per_attempt(), 16);
assert_eq!(
m.llm_tool_result_chars, None,
"unbounded tool results: what every pre-1.9 run did"
);
assert_eq!(areev_run_core::DEFAULT_MAX_EFFECTS_PER_ATTEMPT, 16);
}
#[test]
fn a_pinned_cap_wins_over_the_default() {
let m = RunManifest { max_effects_per_attempt: Some(40), ..bare() };
assert_eq!(m.max_effects_per_attempt(), 40);
let json = serde_json::to_string(&m).unwrap();
assert!(json.contains(r#""max_effects_per_attempt":40"#), "{json}");
}
#[test]
fn a_pin_without_a_lock_or_a_block_keeps_its_bytes() {
let json = serde_json::to_string(&host_pin("parse", "parse")).unwrap();
assert!(!json.contains("locked_params") && !json.contains("\"llm\""), "{json}");
}
#[test]
fn a_nodes_offer_is_its_subset_then_its_dynamic_tools() {
let mut m = bare();
let mut mcp = host_pin("mcp", "mcp.call");
mcp.locked_params = Some(json!({"credential": "gateway"}));
let cfg = areev_run_core::LlmNodeConfig {
tools: Some(vec!["lookup".into()]),
dynamic_tools: vec![areev_run_core::DynamicTool {
name: "paid_to".into(),
description: String::new(),
input_schema: json!({"type": "object"}),
dispatch: areev_run_core::DynamicDispatch {
definition: "mcp.call".into(),
hash: "hash-mcp.call".into(),
locked: Some(json!({"url": "u", "credential": "loses"})),
args_into: Some("arguments".into()),
definition_schema: None,
},
}],
..Default::default()
};
m.pinned = vec![
host_pin("lookup", "lookup"),
mcp,
PinnedTool { executor: "abstract".into(), llm: Some(cfg), ..host_pin("agent", "agent") },
PinnedTool { executor: "abstract".into(), ..host_pin("other", "other") },
];
let ex = m.executors();
let NodeExecutor::Abstract { tools, llm } = &ex[2] else { panic!("{ex:?}") };
let names: Vec<&str> = tools.iter().map(|t| t.tool_name.as_str()).collect();
assert_eq!(names, vec!["lookup", "paid_to"]);
assert_eq!(tools[1].tool_hash, "hash-mcp.call");
assert_eq!(tools[1].locked, Some(json!({"url": "u", "credential": "gateway"})));
assert!(llm.is_some());
let NodeExecutor::Abstract { tools, llm } = &ex[3] else { panic!("{ex:?}") };
let names: Vec<&str> = tools.iter().map(|t| t.tool_name.as_str()).collect();
assert_eq!(names, vec!["lookup", "mcp.call"]);
assert_eq!(tools[1].locked, Some(json!({"credential": "gateway"})));
assert!(llm.is_none());
}
}