areev-run 1.8.5

//! The run manifest (§5.3/§6.1 V7): everything a resume must know, written
//! ONCE at run start — the plan hash, the frozen tool resolutions, the
//! budgets, the principal, the redaction posture. Persisted through the
//! facade's `record_run_manifest` shape (a value-level State at epoch 0 +
//! a `run:<id> mg:harness <hash>` Fact), so the manifest replicates with
//! the file and `runs_touching` can join runs to their configuration.
//!
//! **Resolution freeze**: Named/Abstract resolution happens here, once; the
//! outcome is pinned. Resume reads the manifest and NEVER re-resolves — a
//! newer tool definition appearing between pause and resume must not
//! silently change the run.
//!
//! v1 ownership note: the F7 owner-nonce check (resume-of-a-copy detection
//! via the op-log head) needs an op-cursor read API the store does not yet
//! expose; v1 ships taint detection (forked journal tips → `RUN-E016`) and
//! the explicit `--fork` path, and records this as the documented gap.

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;

/// §5.4: what a fork descends from. Journal grains of the inherited prefix
/// carry the PARENT's run id; the fork's verifiability is this recorded
/// base-checkpoint hash (the seed checkpoint `derived_from`s it).
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct ForkBase {
    pub base_run: String,
    pub base_checkpoint: String,
    pub base_superstep: u64,
}

/// One frozen resolution.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct PinnedTool {
    pub node: String,
    pub tool_hash: String,
    pub tool_name: String,
    /// "host" | "client" (the definition's `executor_kind`, default host).
    pub executor: String,
    /// The definition's `executor_uri`, when it names a `cas://sha256:` code
    /// blob. Pinned here so a run executes the address it resolved at start
    /// even if the Definition is superseded mid-run — the same resolution
    /// freeze the bindings get. Absent for every ordinary tool, so existing
    /// manifests serialize byte-identically.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub executor_uri: Option<String>,
    /// The Definition's declared runtime ("wasm32-areev"), pinned with the
    /// address so a mid-run supersession cannot re-route a blob from the
    /// sandbox to native exec. Absent = native, and existing manifests
    /// serialize byte-identically.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub runtime: Option<String>,
    /// The declared sandbox limits (`{"fuel", "max_pages"}`), pinned with
    /// the runtime.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub runtime_limits: Option<serde_json::Value>,
    /// The Definition's declared `capabilities` (#101), frozen at start
    /// beside the pinned runtime so the effective set — `declared ∩
    /// host-granted` — replays identically and a mid-run supersession cannot
    /// widen what a module may reach. Absent for every non-capability tool,
    /// so existing manifests serialize byte-identically.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub capabilities: Option<serde_json::Value>,
}

/// The manifest, as serialized into the run-config State grain.
#[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>,
    /// D9: the redaction posture, explicit in every manifest — "none" or a
    /// redactor id — so every audit sees what the journal did or did not
    /// scrub.
    pub redaction: String,
    /// The run's input — §5's function is `(plan, manifest, input, journal)`
    /// and the manifest is where the input is journaled; verify replays
    /// seed from here, never from a checkpoint's post-reducer context.
    pub input: serde_json::Value,
    /// Per-LLM-call `max_tokens` — both the request ceiling handed to the
    /// model and the §6.7 per-dispatch reservation the scheduler checks
    /// before emitting an LLM effect. This is an OUTPUT ceiling; nothing here
    /// bounds how large a transcript may grow.
    #[serde(default)]
    pub llm_max_tokens: Option<u32>,
    /// Effects one node attempt may spend (an abstract node's turns, tool
    /// calls and re-prompts share the counter). Frozen here so a resume
    /// bounds the loop exactly as the start did, and absent on every manifest
    /// written before the knob existed — which is why it is `Option` with
    /// `skip_serializing_if`: those manifests must serialize byte-identically.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub max_effects_per_attempt: Option<u32>,
    /// Bound on ONE tool result's size in an abstract node's TRANSCRIPT, in
    /// characters. The journal keeps every result in full regardless — this
    /// bounds only what the model is shown. `None` = unbounded, the behaviour
    /// of every run before the knob existed.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub llm_tool_result_chars: Option<usize>,
    /// Ceiling on an abstract node's WHOLE transcript, in the provider's own
    /// reported prompt tokens. Reaching it emits one journaled summarizer turn
    /// and splices its result over the folded range. `None` = no ceiling, the
    /// behaviour of every run before the fold existed.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub llm_context_tokens: Option<u64>,
    /// Typed reducers (§6.5): state_key → builtin reducer name, read off
    /// the Workflow grain's `reducers` field and FROZEN here — a resume
    /// must merge exactly as the original run did. Undeclared keys are LWW.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub reducers: BTreeMap<String, String>,
    /// Present exactly when this run is a §5.4 fork.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub fork_of: Option<ForkBase>,
}

/// Serializable twin of the core's `Budgets` (kept separate so the manifest
/// wire shape is owned here, not by the pure crate).
#[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>,
}

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,
        }
    }
}

/// A single tool result may occupy at most this fraction of the context
/// ceiling. A quarter is a heuristic and named as one: large enough that an
/// ordinary file read arrives whole, small enough that one dump cannot carry
/// the next request past the window on its own.
const RESULT_SHARE_OF_CEILING: u64 = 4;

/// The runtime's characters-per-token conversion, used ONLY to turn a token
/// ceiling into a character bound — never to measure a transcript, which is
/// always the provider's own reported count. Same ratio the store's estimator
/// uses, and the same honesty applies: it converts a limit, it does not
/// pretend to count tokens.
const CHARS_PER_TOKEN: u64 = 4;

impl RunManifest {
    /// Freeze resolutions for every node (V3 + V7):
    /// - **Bound** (a `bindings` hash): must resolve to a Tool Definition.
    /// - **Named** (no binding): a Definition head whose `tool_name` equals
    ///   the node id, found via the definition catalogue.
    /// - **Abstract** (neither): Wave 2 (`RUN-E006` until then).
    // Mirrored to the CLI/MCP surfaces like `record_tool_call`: grouped
    // params would make the scalar-in mirrors diverge.
    #[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> {
        // Where the plan itself lives. Read once, used only when a node fails
        // to resolve: a plan run in a namespace that is not its own is the
        // misroute #230 reported, and the plan grain is the one thing that
        // knows where its Definitions were authored.
        let plan_ns =
            m.get(plan_hash).ok().and_then(|g| g.get_str("namespace").map(str::to_string));
        let mut pinned = Vec::with_capacity(plan.nodes.len());
        for (i, node) in plan.nodes.iter().enumerate() {
            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(&h).map_err(|e| RunError::UnresolvedRef {
                        what: format!("binding for node '{node}': {e}"),
                    })?;
                    if g.grain_type == areev_core::types::GrainType::Workflow {
                        // A binding to another plan = a subgraph node.
                        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,
                        }
                    } else {
                        pin_from_definition(node, &h, &g)?
                    }
                }
                None => {
                    // Named: the newest Definition with tool_name == node;
                    // otherwise ABSTRACT (§6.2) — the node label is an
                    // instruction the model interprets — which needs the
                    // tool-calling seam configured.
                    match find_definition_by_name(m, ns, node) {
                        Some((h, g)) => pin_from_definition(node, &h, &g)?,
                        // The misroute FIRST, before either fallback (#230).
                        // A node whose name matches a real Definition sitting
                        // in the plan's own namespace is a run started in the
                        // wrong place, not an instruction for a model: its
                        // `executor_uri`, runtime and capabilities would all
                        // be silently dropped, and with an LLM configured the
                        // node would quietly become an abstract one and the
                        // run would report Completed having called nothing.
                        // Costs one extra catalogue scan, and only on a miss.
                        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,
                        },
                        None => return Err(RunError::NoToolLlm { node: node.clone() }),
                    }
                }
            };
            pinned.push(resolved);
        }
        // Reducer table (§6.5): declared on the Workflow grain, validated
        // here — an unknown name fails at run start, not at first merge.
        let mut reducers: BTreeMap<String, String> = BTreeMap::new();
        if let Ok(wf_grain) = m.get(plan_hash) {
            if let Some(table) = wf_grain.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,
        })
    }

    /// Freeze the caller's run-level LLM limits into the manifest.
    ///
    /// Deliberately NOT arguments to [`RunManifest::resolve`]: not one of them
    /// affects resolution — they are ceilings the scheduler reads back on
    /// every resume and verify — and threading each through an
    /// already-ten-argument signature would make the next one an eleventh,
    /// with two call sites free to disagree about which knobs they passed.
    /// One place to add a limit, one place that can forget it.
    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;
        // The ceiling the operator set — or, failing that, one derived from
        // the model's own window. This is what makes folding automatic: a long
        // agent should not need a flag to survive, and a host that knows the
        // window can answer the question the operator would have had to.
        //
        // Reserved output comes off the top, because a prompt that fits and a
        // reply that does not is the same rejection. A backend that reports no
        // window leaves this None and relies on the reactive path instead.
        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,
            ))))
        });
        // And the per-result bound follows the ceiling, because a ceiling on
        // its own is not safe: the fold measures the PREVIOUS turn, so one
        // oversized result appended since can carry the next request past the
        // window before anything checks. Deriving it means an operator who
        // sets one bound gets the other, rather than discovering the gap.
        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
    }

    /// The per-dispatch token reserve (§6.7) — the scheduler refuses to
    /// emit an LLM effect that could not fit under the token ceiling.
    pub fn llm_reserve_tokens(&self) -> u64 {
        u64::from(self.llm_max_tokens.unwrap_or(crate::runner::DEFAULT_LLM_MAX_TOKENS))
    }

    /// Effects one node attempt may spend, as the scheduler's `StepEnv` reads
    /// it. Every `StepEnv` the driver builds — drive, verify — takes it from
    /// here, so a resumed or replayed run is bounded exactly as the start was.
    pub fn max_effects_per_attempt(&self) -> u32 {
        self.max_effects_per_attempt
            .unwrap_or(areev_run_core::DEFAULT_MAX_EFFECTS_PER_ATTEMPT)
    }

    /// The executor vector the scheduler env consumes, in node order. An
    /// abstract node's offered tools are the manifest's HOST-pinned tools —
    /// derived, not stored, so the manifest stays minimal and the offer is
    /// deterministic from it.
    pub fn executors(&self) -> Vec<NodeExecutor> {
        // ONE OFFER PER DEFINITION, however many nodes bind it. `pinned` holds one
        // entry per plan node, so a Definition bound to two nodes — the ordinary
        // shape for a terminal step reached from two branches — appeared twice in
        // the offer, and every provider refuses a tools list with a repeated name
        // ("tools: Tool names must be unique", HTTP 400 on the whole request; #270).
        // First occurrence wins, so the offer's order stays that of the plan.
        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| areev_run_core::OfferedTool {
                tool_name: p.tool_name.clone(),
                tool_hash: p.tool_hash.clone(),
            })
            .collect();
        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(),
                },
                "subgraph" => NodeExecutor::Subgraph { workflow_hash: p.tool_hash.clone() },
                "abstract" => NodeExecutor::Abstract { tools: offered.clone() },
                _ => NodeExecutor::Host {
                    tool_hash: p.tool_hash.clone(),
                    tool_name: p.tool_name.clone(),
                },
            })
            .collect()
    }

    /// Persist via the established run-manifest shape (a value-level State
    /// at epoch 0 + the `run:<id> mg:harness <hash>` link Fact in
    /// `agent:harness`). Store-direct like every other runtime control
    /// write: the driver's `run.execute` check is THE authorization decision
    /// for starting a run — requiring executor principals to additionally
    /// hold `write ON agent:harness` would make every governed deployment
    /// hand out grants on a governance namespace. Attribution still lands on
    /// the grains (`author_did`). The link also carries the run's session
    /// namespace (`run_ns`), which is where its journal lives — the run index
    /// is what `areev run list` and the console page through, and without
    /// the namespace on it a per-tenant listing could only be filtered
    /// after the fact, over an already-truncated page (#165). Returns
    /// (config hash, link hash).
    pub fn persist_in_namespace(&self, m: &mut Areev, ns: &str) -> Result<(Hash, Hash)> {
        let config = serde_json::to_value(json!({ "areev_run": self }))
            .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))
    }

    /// The pre-#165 signature, kept so a host that persists manifests itself
    /// keeps compiling. It stamps no session namespace, so the run lists as
    /// unattributed — visible unscoped, excluded and counted under any
    /// scope. Prefer [`persist_in_namespace`](Self::persist_in_namespace).
    #[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, "")
    }

    /// Load a run's manifest back from the file (resume/verify path).
    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(&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(),
            })?;
        serde_json::from_value(manifest).map_err(|e| RunError::ManifestMismatch {
            why: format!("manifest does not parse: {e}"),
        })
    }
}

/// Read a Tool Definition into the pinned shape a run (or a trigger's
/// connector, #185) executes under: name, executor kind, `executor_uri`,
/// runtime, limits and capability declaration, each either dispatchable or
/// refused by name.
///
/// `label` is what a refusal calls the caller — a node name on the run path,
/// the connector name on the trigger path. Public because the trigger
/// evaluator resolves a connector Definition the same way and must reach the
/// same verdict: two readers of one declaration would drift the first time
/// either moved, and the one that drifts quietly is the heartbeat nobody
/// watches.
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();
    // Refuse a malformed tool name at resolve time, before the run starts.
    // This name reaches a host tool as `$AREEV_TOOL_NAME`, and it arrives from
    // a plan grain — which may have been imported from a bundle whose author we
    // do not vouch for (import verifies content integrity, not authorship). It
    // is not shell-interpolated, so this is not injection; the point is that an
    // attacker-influenced string with newlines or shell metacharacters should
    // never reach a child's environment, because a perfectly ordinary tool
    // script that does something like `eval "$AREEV_TOOL_NAME"` turns it into
    // one. Failing here also fails loudly and early rather than mid-superstep.
    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",
    };
    // `executor_uri` used to be written, parsed, CAL-buildable — and read by
    // nothing, so a Definition naming `executor://crm.lookup@v3` executed
    // whatever `--tool-cmd` happened to be. That is the failure with no
    // symptom this runtime exists to refuse, so every value now either
    // dispatches or is refused by name.
    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(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()))
        }
    };
    // The declared runtime (#86). Fail closed on both axes: a runtime on a
    // tool with no code blob describes nothing, and an unknown runtime —
    // possible on a grain that arrived by sync from another implementation,
    // since our own write path refuses it — must not silently fall back to
    // native exec, which would run wasm bytes as a program.
    let runtime = match g.get_str("runtime") {
        None | Some("native") => None,
        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 };
    // The declared capability set (#101). Fail closed on both axes, the same
    // way the runtime does: a declaration on a tool whose runtime cannot honour
    // it describes nothing, and a malformed one must be refused at start rather
    // than at the first call — a grain can arrive by sync from another
    // implementation, and our own write path is not the only way in.
    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())
        }
    };
    // The inverse: a capability RUNTIME with nothing declared can reach
    // nothing, and saying so at start beats a module that instantiates and
    // then has every call refused.
    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"
            ),
        });
    }
    Ok(PinnedTool {
        node: node.to_string(),
        tool_hash: h.to_hex(),
        tool_name,
        executor: executor.into(),
        executor_uri,
        runtime,
        runtime_limits,
        capabilities,
    })
}

/// Newest Definition with this tool_name, or None. Definitions are few; a
/// bounded scan of the namespace's Tool grains is the v1 catalogue (a
/// dedicated definition index is a later optimization, never a semantics
/// change). Deterministic tiebreak: newest `created_at`, then highest hash
/// — the provisional-head rule's shape, so every host resolves identically.
fn find_definition_by_name(
    m: &mut Areev,
    ns: &str,
    name: &str,
) -> Option<(Hash, areev_core::format::deserialize::DeserializedGrain)> {
    // Journal intents/results are Tool grains IN THE SAME NAMESPACE (two
    // per effect), so a small window would let an active file crowd its own
    // definitions out of the catalogue — and with an LLM configured the
    // miss silently becomes an abstract node. Scan wide until a dedicated
    // definition index exists.
    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))
}

/// Is this unresolved node a MISROUTED run rather than an abstract one
/// (#230)? True when the plan grain lives in another namespace and a Tool
/// Definition named `node` is there — the run was started somewhere the
/// plan's own tools cannot be seen.
///
/// Deliberately narrow: it asks the plan's namespace and no other. A sweep
/// over every namespace in the memory would be both unbounded (journal Tool
/// grains crowd the catalogue) and less certain — the plan grain is the one
/// place that records where its nodes were authored to resolve. A run whose
/// namespace IS the plan's, or whose plan names a namespace with no such
/// Definition, falls through to the existing verdicts unchanged: abstract
/// with a model configured, `RUN-E006` without.
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,
    }
}

/// The nodes of `plan_hash` that would resolve as **abstract** — no binding
/// and no Definition head whose `tool_name` matches — plus whether the plan
/// itself is present at all.
///
/// A pre-flight read: it takes no lease, writes nothing, and is exactly the
/// half of [`RunManifest::resolve`] that decides `RUN-E006`. It exists so a
/// declaration that names a plan can say at AUTHORING time what the plan will
/// need at fire time (#73): `areev trigger add --workflow <WF>` used to accept
/// a plan whose nodes did not resolve, report `waiting`, and fail at the first
/// firing — on the operator's mailbox rather than at their keyboard.
///
/// Abstract is not an error here, which is why this returns node ids rather
/// than a `Result`: an abstract node is legitimate with a tool-calling model
/// configured. The caller decides whether that is a warning or nothing.
pub fn abstract_nodes(
    m: &mut Areev,
    ns: &str,
    plan_hash: &Hash,
) -> std::result::Result<Vec<String>, String> {
    let wf = m
        .get(plan_hash)
        .map_err(|e| format!("{e}"))?
        .to_workflow()
        .map_err(|e| format!("{e}"))?;
    let plan = PlanGraph::build(&wf).map_err(|e| e.to_string())?;
    let mut out = Vec::new();
    for (i, node) in plan.nodes.iter().enumerate() {
        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::*;

    /// A manifest with nothing pinned and no limits set — the shape every
    /// pre-1.9 run persisted.
    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,
        }
    }

    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,
        }
    }

    /// One Definition bound to two nodes is offered to an abstract node ONCE (#270):
    /// a repeated name in a tools list is a 400 on the whole model request at every
    /// provider. Each bound node still executes its own binding.
    #[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,
            },
            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:?}"),
            }
        }
    }

    /// The whole reason `max_effects_per_attempt` is an `Option` carrying
    /// `skip_serializing_if`: a manifest written before the knob existed must
    /// serialize to the SAME bytes it did then. The manifest is a stored
    /// grain, so a widened wire shape would re-address every one of them.
    #[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}"#
        );
    }

    /// The other half: a manifest persisted before the field existed still
    /// loads, and bounds its loops at the documented default.
    #[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"
        );
        // Named here as well as in run-core, because changing it changes the
        // behaviour of every existing plan that relies on the bound.
        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}");
    }
}