cuttlefish_core/

graph.rs

//! The graph AST: `nodes = {...}` and `branches = {...}`.
//!
//! A node's `in` is an expression over other nodes' outputs, not just a bare
//! name — `Record`/`List` are what make fan-in possible. See
//! `docs/superpowers/specs/2026-08-03-dag-core-design.md` for the full
//! rationale; this module is purely the parsed shape, with no typechecking
//! or execution logic (those live in `cuttlefish-host`).

use crate::lex::{Tok, Token};
use crate::spec::SpecError;
use std::collections::BTreeMap;
use std::path::PathBuf;

/// What feeds a node's input.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum InputExpr {
    /// `other_node.out` — this node's whole output.
    FromNode(String),
    /// `{ field = expr; ... }` — build a record from several nodes.
    Record(BTreeMap<String, InputExpr>),
    /// `[ expr, expr, ... ]` — build a list from several nodes, order significant.
    List(Vec<InputExpr>),
}

/// One check a node's output must pass before it counts as done.
///
/// Evaluated in declaration order, short-circuiting on the first failure —
/// see [`Node::accept`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AcceptCheck {
    /// Validate against the JSON Schema at this path. Deterministic, and
    /// costs no inference.
    Schema(PathBuf),
    /// Ask a model whether the output is acceptable.
    Judge {
        /// Which model grades. `None` means the spec's own `model`.
        model: Option<crate::spec::ModelRef>,
        /// The grading prompt. The host appends the node's input and the
        /// output under judgement, since "does this cite numbers *from the
        /// input*" is unanswerable without both.
        prompt: String,
    },
}

/// One rung of a node's recovery ladder, climbed in order until a rung
/// succeeds or the rungs run out.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Rung {
    /// Up to N further attempts, unchanged.
    Retry(u32),
    /// One attempt against a different model.
    Reroute(crate::spec::ModelRef),
    /// Terminal: stop, record why, and surface it to `cuttlefish
    /// escalations`. Must be the last rung — see `GraphParser::ladder`.
    Escalate,
}

/// One node in the graph.
#[derive(Debug, Clone, PartialEq)]
pub struct Node {
    /// The block/bundle this node runs — same string shape as today's
    /// `pipeline` entries (a path, or a bare `name@version`).
    pub block: PathBuf,
    /// What feeds this node. `None` only for a node with no inbound edge —
    /// the graph's entry point(s).
    pub input: Option<InputExpr>,
    /// Bounded-loop marker: the `Ty::Text` output field compared against
    /// `"done"`. Requires `max_iterations`.
    pub repeat_until: Option<String>,
    /// Mandatory alongside `repeat_until` — enforced at parse time, not left
    /// as a typecheck-time gap, since a missing bound is a spec-authoring
    /// mistake regardless of what the graph shape turns out to be.
    pub max_iterations: Option<u32>,
    /// Fan-out marker: a JSONL manifest, one JSON value per line, each the
    /// complete input for one run of this node's block. `None` for an
    /// ordinary node, which runs exactly once.
    ///
    /// Mutually exclusive with [`Node::repeat_until`] — both describe
    /// iteration, but over different things (manifest items vs. this node's
    /// own prior output), and there is no coherent combined meaning.
    pub over: Option<PathBuf>,
    /// Checks this node's output must pass beyond its declared type.
    ///
    /// Ordered and short-circuiting. Empty means the declared type is the
    /// only contract — today's behaviour, unchanged.
    pub accept: Vec<AcceptCheck>,
    /// What to try when an attempt fails, in order.
    ///
    /// Empty (or omitted) means one attempt and no recovery — again,
    /// today's behaviour. A ladder that runs out without [`Rung::Escalate`]
    /// is an ordinary failure.
    pub on_fail: Vec<Rung>,
}

/// `nodes = { name = { ... }; ... }`
#[derive(Debug, Clone, PartialEq, Default)]
pub struct NodeGraph {
    /// Insertion order preserved (`BTreeMap` would reorder alphabetically,
    /// which is fine for lookup but wrong for any diagnostic that lists
    /// nodes "in the order the author wrote them").
    pub nodes: Vec<(String, Node)>,
}

impl NodeGraph {
    /// The one-node graph `block = "...";` desugars to.
    pub fn single(block: PathBuf) -> Self {
        Self {
            nodes: vec![(
                "block".to_string(),
                Node {
                    block,
                    input: None,
                    repeat_until: None,
                    max_iterations: None,
                    over: None,
                    accept: Vec::new(),
                    on_fail: Vec::new(),
                },
            )],
        }
    }

    /// Look up a node by name.
    pub fn get(&self, name: &str) -> Option<&Node> {
        self.nodes
            .iter()
            .find(|(n, _)| n == name)
            .map(|(_, node)| node)
    }
}

/// Whether a graph is a strict linear chain — a single strand where node
/// `i`'s sole input (if any) is `FromNode` of node `i-1`, nothing more:
///
/// - `branches` must be empty (no conditional dispatch to encode).
/// - No node declares `repeat_until` (no loop to encode).
/// - No node's `input` is `Record`/`List` fan-in.
/// - **Beyond per-node checks:** the *whole graph* must be one strand, not
///   just individually-simple nodes that still fan out or fan in as a
///   group. Concretely: the first declared node has no input; every
///   subsequent declared node's input must be exactly `FromNode` of the
///   node declared immediately before it; and no node may be referenced by
///   more than one other node's `FromNode` (that would be fan-out — two
///   nodes both reading node `k`'s output — which individually satisfies
///   every per-node check above while still not being a chain).
///
/// `cuttlefish build`'s bundle format only knows how to encode this exact
/// shape, walked in `spec.nodes`' declaration order.
pub fn is_simple_chain(graph: &NodeGraph, branches: &Branches) -> bool {
    if !branches.decisions.is_empty() {
        return false;
    }
    for (i, (_, node)) in graph.nodes.iter().enumerate() {
        if node.repeat_until.is_some() {
            return false;
        }
        // The bundle manifest carries a node's name, kind, resolution and
        // signature — nothing about *how* it executes. `over` would
        // therefore be dropped at bundle time and silently absent at run
        // time, so a bundled fan-out node runs once against the job input
        // instead of once per manifest line, and returns something that
        // looks entirely reasonable. Refusing to bundle is the only honest
        // option until the format carries this.
        if node.over.is_some() {
            return false;
        }
        match (i, &node.input) {
            (0, None) => {}
            (0, Some(_)) => return false, // the entry node must have no input
            (_, Some(InputExpr::FromNode(referenced))) => {
                let (previous_name, _) = &graph.nodes[i - 1];
                if referenced != previous_name {
                    return false; // not chained to the immediately-preceding node
                }
            }
            _ => return false, // missing input, or Record/List fan-in
        }
    }
    // Fan-out check: provably unreachable given the position check above
    // already passed (if every node's input is exactly its immediate
    // predecessor, no target can have two referrers) — kept as an explicit,
    // cheap assertion rather than an implicit invariant, so a future change
    // to the loop above that weakens it trips this instead of silently
    // regressing.
    let mut referenced_counts = std::collections::HashMap::new();
    for (_, node) in &graph.nodes {
        if let Some(InputExpr::FromNode(referenced)) = &node.input {
            *referenced_counts.entry(referenced.clone()).or_insert(0) += 1;
        }
    }
    referenced_counts.values().all(|&count| count <= 1)
}

/// `branches = { node_name = { "label" -> target; ... }; ... }`
#[derive(Debug, Clone, PartialEq, Default)]
pub struct Branches {
    /// (branching node name) -> (label -> target node name), insertion order.
    pub decisions: Vec<(String, Vec<(String, String)>)>,
}

/// A self-contained recursive-descent parser for the `nodes = {...}` and
/// `branches = {...}` bodies, operating directly on a token slice.
pub struct GraphParser<'a> {
    /// The full token stream being parsed.
    pub tokens: &'a [Token],
    /// Current cursor position into `tokens`.
    pub at: usize,
}

impl<'a> GraphParser<'a> {
    fn peek(&self) -> Option<&'a Tok> {
        self.tokens.get(self.at).map(|t| &t.tok)
    }
    fn here(&self) -> String {
        match self.tokens.get(self.at) {
            Some(t) => format!("{} at {}", t.tok.describe(), t.span),
            None => "end of input".into(),
        }
    }
    fn expect(&mut self, want: &Tok) -> Result<(), SpecError> {
        match self.peek() {
            Some(got) if got == want => {
                self.at += 1;
                Ok(())
            }
            _ => Err(SpecError::Malformed(format!(
                "expected {}, found {}",
                want.describe(),
                self.here()
            ))),
        }
    }
    fn ident(&mut self) -> Result<String, SpecError> {
        match self.tokens.get(self.at).map(|t| &t.tok) {
            Some(Tok::Ident(name)) => {
                self.at += 1;
                Ok(name.clone())
            }
            _ => Err(SpecError::Malformed(format!(
                "expected a name, found {}",
                self.here()
            ))),
        }
    }
    fn string(&mut self) -> Result<String, SpecError> {
        match self.tokens.get(self.at).map(|t| &t.tok) {
            Some(Tok::Str(s)) => {
                self.at += 1;
                Ok(s.clone())
            }
            _ => Err(SpecError::Malformed(format!(
                "expected a quoted string, found {}",
                self.here()
            ))),
        }
    }
    fn skip_semi(&mut self) {
        if self.peek() == Some(&Tok::Semicolon) {
            self.at += 1;
        }
    }

    /// `{ name = { field* }; ... }` — the whole `nodes = {...}` body.
    ///
    /// Returns the parsed graph together with the token position just past
    /// its closing `}` — `spec.rs`'s `Parser` (a *separate* struct walking
    /// the same token slice, Task 3) needs that position to resume parsing
    /// the rest of the `spec {...}` body afterward, since it can't see
    /// `GraphParser`'s internal cursor otherwise.
    pub fn node_graph(&mut self) -> Result<(NodeGraph, usize), SpecError> {
        self.expect(&Tok::OpenBrace)?;
        let mut nodes = Vec::new();
        while self.peek().is_some() && self.peek() != Some(&Tok::CloseBrace) {
            let name = self.ident()?;
            self.expect(&Tok::Equals)?;
            let node = self.node_body()?;
            nodes.push((name, node));
            self.skip_semi();
        }
        self.expect(&Tok::CloseBrace)?;
        if nodes.is_empty() {
            return Err(SpecError::Malformed("nodes needs at least one node".into()));
        }
        Ok((NodeGraph { nodes }, self.at))
    }

    /// `{ block = "..."; in = expr; over = "..."; repeat_until = "..."; max_iterations = N; }`
    fn node_body(&mut self) -> Result<Node, SpecError> {
        self.expect(&Tok::OpenBrace)?;
        let (mut block, mut input, mut repeat_until, mut max_iterations, mut over) =
            (None, None, None, None, None);
        let (mut accept, mut on_fail) = (Vec::new(), Vec::new());
        while self.peek().is_some() && self.peek() != Some(&Tok::CloseBrace) {
            let key = self.ident()?;
            self.expect(&Tok::Equals)?;
            match key.as_str() {
                "block" => block = Some(PathBuf::from(self.string()?)),
                "in" => input = Some(self.input_expr()?),
                "over" => over = Some(PathBuf::from(self.string()?)),
                "accept" => accept = self.accept_checks()?,
                "on_fail" => on_fail = self.ladder()?,
                "repeat_until" => repeat_until = Some(self.string_or_field()?),
                "max_iterations" => max_iterations = Some(self.number()?),
                other => return Err(SpecError::UnknownField(other.to_string())),
            }
            self.skip_semi();
        }
        self.expect(&Tok::CloseBrace)?;
        if let (Some(_), None) = (&repeat_until, &max_iterations) {
            return Err(SpecError::Malformed(
                "repeat_until requires max_iterations".into(),
            ));
        }
        if over.is_some() && repeat_until.is_some() {
            return Err(SpecError::Malformed(
                "a node cannot declare both `over` (run once per manifest item) and \
                 `repeat_until` (re-run on its own output) — they are two different \
                 iteration semantics with no combined meaning"
                    .into(),
            ));
        }
        Ok(Node {
            block: block.ok_or(SpecError::MissingField("block"))?,
            input,
            repeat_until,
            max_iterations,
            over,
            accept,
            on_fail,
        })
    }

    /// `[ Schema "p.json", Judge "prompt", Judge { model = M "t"; prompt = "..."; } ]`
    fn accept_checks(&mut self) -> Result<Vec<AcceptCheck>, SpecError> {
        let mut checks = Vec::new();
        self.expect(&Tok::OpenBracket)?;
        while self.peek() != Some(&Tok::CloseBracket) {
            let kind = self.ident()?;
            match kind.as_str() {
                "Schema" => checks.push(AcceptCheck::Schema(PathBuf::from(self.string()?))),
                "Judge" => checks.push(self.judge()?),
                other => {
                    return Err(SpecError::Malformed(format!(
                        "unknown accept check `{other}` — expected `Schema` or `Judge`"
                    )))
                }
            }
            if self.peek() == Some(&Tok::Comma) {
                self.at += 1;
            } else {
                break;
            }
        }
        self.expect(&Tok::CloseBracket)?;
        Ok(checks)
    }

    /// Either `Judge "prompt"` or `Judge { model = M "t"; prompt = "..."; }`.
    ///
    /// Two spellings because the bare one costs nothing to declare and suits
    /// a cheap sanity check, while naming a model is what lets a strong,
    /// slow model grade a fast one's bulk output.
    fn judge(&mut self) -> Result<AcceptCheck, SpecError> {
        if self.peek() != Some(&Tok::OpenBrace) {
            return Ok(AcceptCheck::Judge {
                model: None,
                prompt: self.string()?,
            });
        }
        self.at += 1;
        let (mut model, mut prompt) = (None, None);
        while self.peek().is_some() && self.peek() != Some(&Tok::CloseBrace) {
            let key = self.ident()?;
            self.expect(&Tok::Equals)?;
            match key.as_str() {
                "model" => {
                    let provider = self.ident()?;
                    model = Some(crate::spec::ModelRef::new(provider, self.string()?));
                }
                "prompt" => prompt = Some(self.string()?),
                other => return Err(SpecError::UnknownField(other.to_string())),
            }
            self.skip_semi();
        }
        self.expect(&Tok::CloseBrace)?;
        Ok(AcceptCheck::Judge {
            model,
            prompt: prompt.ok_or(SpecError::MissingField("prompt"))?,
        })
    }

    /// `[ retry 2, reroute Ollama "m", escalate ]`
    fn ladder(&mut self) -> Result<Vec<Rung>, SpecError> {
        let mut rungs: Vec<Rung> = Vec::new();
        self.expect(&Tok::OpenBracket)?;
        while self.peek() != Some(&Tok::CloseBracket) {
            // A terminal rung with anything after it means the author
            // expected the tail to run. It never would — so say so rather
            // than accepting a ladder whose second half is decorative.
            if rungs.last() == Some(&Rung::Escalate) {
                return Err(SpecError::Malformed(
                    "`escalate` must be the last rung of an on_fail ladder — nothing after it \
                     can ever run"
                        .into(),
                ));
            }
            let kind = self.ident()?;
            match kind.as_str() {
                "retry" => {
                    let n = self.number()?;
                    if n == 0 {
                        return Err(SpecError::Malformed(
                            "`retry 0` expresses nothing — write `retry 1`, or omit the rung"
                                .into(),
                        ));
                    }
                    rungs.push(Rung::Retry(n));
                }
                "reroute" => {
                    let provider = self.ident()?;
                    rungs.push(Rung::Reroute(crate::spec::ModelRef::new(
                        provider,
                        self.string()?,
                    )));
                }
                "escalate" => rungs.push(Rung::Escalate),
                other => {
                    return Err(SpecError::Malformed(format!(
                        "unknown on_fail rung `{other}` — expected `retry`, `reroute`, or \
                         `escalate`"
                    )))
                }
            }
            if self.peek() == Some(&Tok::Comma) {
                self.at += 1;
            } else {
                break;
            }
        }
        self.expect(&Tok::CloseBracket)?;
        Ok(rungs)
    }

    /// `repeat_until = "done"` — a bare field-name string, not a node
    /// reference, so this reuses `string()` (kept as its own method name at
    /// the call site above for readability, not because parsing differs).
    fn string_or_field(&mut self) -> Result<String, SpecError> {
        self.string()
    }

    fn number(&mut self) -> Result<u32, SpecError> {
        // Numbers aren't tokenized separately today (see lex.rs) — an
        // integer like `5` lexes as `Ident("5")` since digits satisfy
        // `is_alphanumeric()`. Parsing it here, rather than adding a
        // dedicated numeric token, keeps this the only place that cares.
        let s = self.ident()?;
        s.parse::<u32>()
            .map_err(|_| SpecError::Malformed(format!("`{s}` is not a valid max_iterations")))
    }

    /// `node.out` | `{ field = expr; ... }` | `[ expr, ... ]`
    fn input_expr(&mut self) -> Result<InputExpr, SpecError> {
        match self.peek() {
            Some(Tok::OpenBrace) => {
                self.at += 1;
                let mut fields = BTreeMap::new();
                while self.peek() != Some(&Tok::CloseBrace) {
                    let field = self.ident()?;
                    self.expect(&Tok::Equals)?;
                    fields.insert(field, self.input_expr()?);
                    self.skip_semi();
                }
                self.expect(&Tok::CloseBrace)?;
                Ok(InputExpr::Record(fields))
            }
            Some(Tok::OpenBracket) => {
                self.at += 1;
                let mut items = Vec::new();
                while self.peek() != Some(&Tok::CloseBracket) {
                    items.push(self.input_expr()?);
                    if self.peek() == Some(&Tok::Comma) {
                        self.at += 1;
                    } else {
                        break;
                    }
                }
                self.expect(&Tok::CloseBracket)?;
                Ok(InputExpr::List(items))
            }
            Some(Tok::Ident(reference)) => {
                let reference = reference.clone();
                self.at += 1;
                reference
                    .strip_suffix(".out")
                    .map(|node| InputExpr::FromNode(node.to_string()))
                    .ok_or_else(|| {
                        SpecError::Malformed(format!(
                            "`{reference}` is not a node reference — expected `<node>.out`"
                        ))
                    })
            }
            _ => Err(SpecError::Malformed(format!(
                "expected a node reference, `{{...}}`, or `[...]`, found {}",
                self.here()
            ))),
        }
    }

    /// `{ node_name = { "label" -> target; ... }; ... }` — the whole
    /// `branches = {...}` body. Returns `(Branches, new_at)`, same handoff
    /// convention as [`Self::node_graph`].
    pub fn branches(&mut self) -> Result<(Branches, usize), SpecError> {
        self.expect(&Tok::OpenBrace)?;
        let mut decisions = Vec::new();
        while self.peek().is_some() && self.peek() != Some(&Tok::CloseBrace) {
            let node_name = self.ident()?;
            self.expect(&Tok::Equals)?;
            self.expect(&Tok::OpenBrace)?;
            let mut labels = Vec::new();
            while self.peek() != Some(&Tok::CloseBrace) {
                let label = self.string()?;
                self.expect(&Tok::Arrow)?;
                let target = self.ident()?;
                labels.push((label, target));
                self.skip_semi();
            }
            self.expect(&Tok::CloseBrace)?;
            decisions.push((node_name, labels));
            self.skip_semi();
        }
        self.expect(&Tok::CloseBrace)?;
        Ok((Branches { decisions }, self.at))
    }
}

#[cfg(test)]
mod is_simple_chain_tests {
    use super::*;

    fn node(block: &str, input: Option<InputExpr>) -> Node {
        Node {
            block: PathBuf::from(block),
            input,
            repeat_until: None,
            max_iterations: None,
            over: None,
            accept: Vec::new(),
            on_fail: Vec::new(),
        }
    }

    fn from_node(name: &str) -> InputExpr {
        InputExpr::FromNode(name.to_string())
    }

    #[test]
    fn a_genuine_three_node_chain_is_simple() {
        let graph = NodeGraph {
            nodes: vec![
                ("a".into(), node("blocks/a", None)),
                ("b".into(), node("blocks/b", Some(from_node("a")))),
                ("c".into(), node("blocks/c", Some(from_node("b")))),
            ],
        };
        assert!(is_simple_chain(&graph, &Branches::default()));
    }

    #[test]
    fn record_fan_in_is_not_simple() {
        let mut fields = BTreeMap::new();
        fields.insert("x".to_string(), from_node("a"));
        fields.insert("y".to_string(), from_node("b"));
        let graph = NodeGraph {
            nodes: vec![
                ("a".into(), node("blocks/a", None)),
                ("b".into(), node("blocks/b", None)),
                (
                    "c".into(),
                    node("blocks/c", Some(InputExpr::Record(fields))),
                ),
            ],
        };
        assert!(!is_simple_chain(&graph, &Branches::default()));
    }

    #[test]
    fn list_fan_in_is_not_simple() {
        let graph = NodeGraph {
            nodes: vec![
                ("a".into(), node("blocks/a", None)),
                ("b".into(), node("blocks/b", None)),
                (
                    "c".into(),
                    node(
                        "blocks/c",
                        Some(InputExpr::List(vec![from_node("a"), from_node("b")])),
                    ),
                ),
            ],
        };
        assert!(!is_simple_chain(&graph, &Branches::default()));
    }

    #[test]
    fn a_repeat_until_node_is_not_simple() {
        let mut looped = node("blocks/b", Some(from_node("a")));
        looped.repeat_until = Some("done".to_string());
        looped.max_iterations = Some(5);
        let graph = NodeGraph {
            nodes: vec![("a".into(), node("blocks/a", None)), ("b".into(), looped)],
        };
        assert!(!is_simple_chain(&graph, &Branches::default()));
    }

    /// A `.cfbundle` manifest records each node's name, kind, resolution and
    /// signature — nothing about *how* it executes. A fan-out node bundled
    /// anyway would lose `over` on the way in and be silently ignored on the
    /// way out, running once against the job input instead of N times over
    /// the manifest, and producing a plausible-looking result. Refusing to
    /// bundle is the only honest option until the format carries it.
    #[test]
    fn a_fan_out_node_is_not_simple_because_a_bundle_cannot_carry_over() {
        let mut fanned = node("blocks/a", None);
        fanned.over = Some(PathBuf::from("corpus/manifest.jsonl"));
        let graph = NodeGraph {
            nodes: vec![
                ("a".into(), fanned),
                ("b".into(), node("blocks/b", Some(from_node("a")))),
            ],
        };
        assert!(!is_simple_chain(&graph, &Branches::default()));
    }

    #[test]
    fn a_branches_decision_is_not_simple() {
        let graph = NodeGraph {
            nodes: vec![
                ("a".into(), node("blocks/a", None)),
                ("b".into(), node("blocks/b", Some(from_node("a")))),
            ],
        };
        let branches = Branches {
            decisions: vec![("a".to_string(), vec![("done".to_string(), "b".to_string())])],
        };
        assert!(!is_simple_chain(&graph, &branches));
    }

    /// Two independent, individually-simple nodes that both declare
    /// `in = a.out` — fan-out from `a`. Neither `b` nor `c` individually has
    /// fan-in (each has exactly one `FromNode` input); what makes this not a
    /// chain is a whole-graph property (two referrers of `a`), not anything
    /// visible from a single node in isolation. In this implementation the
    /// position check (node `i`'s input must be exactly node `i-1`) already
    /// rejects this case before the dedicated fan-out tally ever runs — `c`'s
    /// immediate predecessor is `b`, not `a` — which is exactly why that
    /// tally is documented as unreachable-but-kept-explicit. This test
    /// pins the required *behavior* (fan-out is rejected), independent of
    /// which internal check catches it.
    #[test]
    fn fan_out_from_a_shared_predecessor_is_not_simple() {
        let graph = NodeGraph {
            nodes: vec![
                ("a".into(), node("blocks/a", None)),
                ("b".into(), node("blocks/b", Some(from_node("a")))),
                ("c".into(), node("blocks/c", Some(from_node("a")))),
            ],
        };
        assert!(!is_simple_chain(&graph, &Branches::default()));
    }
}