behavior_contracts/

plan.rs

//! plan — execution-plan.md reference implementation.
//!
//! run_plan(plan, ops, exec): execute stage groups sequentially, interpret
//! null-binding skip, Skip propagation, and Error Policy Kind (fail/retry/continue)
//! to produce a deterministic result tree. Node execution is delegated to `exec`.

use crate::value::Value;

#[derive(Debug, Clone)]
pub struct ExecutionPlanSpec {
    pub groups: Vec<Vec<usize>>,
    pub concurrency: i64,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PolicyKind {
    Fail,
    Retry,
    Continue,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RelationKind {
    Single,
    Connection,
}

#[derive(Debug, Clone)]
pub struct OpSpec {
    pub id: String,
    pub parent: Option<usize>,
    pub bind_field: Option<String>,
    pub relation_kind: Option<RelationKind>,
    /// Raw policy string from the vector (validated at run time so unknown
    /// policies fail closed with UNKNOWN_POLICY rather than at parse time).
    pub policy: Option<String>,
}

/// consumer-supplied node execution outcome (mock).
#[derive(Debug, Clone)]
pub enum ExecOutcome {
    Ok(Value),
    Error(String),
}

#[derive(Debug, Clone)]
enum OpState {
    Ok(Value),
    Skipped,
    // Failed is never stored: fail policy raises immediately.
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PlanFailureCode {
    OpFailed,
    UnknownPolicy,
    InvalidPlan,
}

impl PlanFailureCode {
    pub fn as_str(self) -> &'static str {
        match self {
            PlanFailureCode::OpFailed => "OP_FAILED",
            PlanFailureCode::UnknownPolicy => "UNKNOWN_POLICY",
            PlanFailureCode::InvalidPlan => "INVALID_PLAN",
        }
    }
}

#[derive(Debug, Clone)]
pub struct PlanFailure {
    pub code: PlanFailureCode,
    pub message: String,
}

impl std::fmt::Display for PlanFailure {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}: {}", self.code.as_str(), self.message)
    }
}
impl std::error::Error for PlanFailure {}

fn fail<T>(code: PlanFailureCode, message: impl Into<String>) -> Result<T, PlanFailure> {
    Err(PlanFailure {
        code,
        message: message.into(),
    })
}

#[derive(Debug)]
pub struct RunResult {
    states: Vec<Option<OpState>>,
    pub executed: Vec<String>,
    pub skipped: Vec<String>,
    ops_ref: Vec<(String, Option<RelationKind>)>,
}

impl RunResult {
    /// Build the observed result tree (opId → value; Skip → unproduced form).
    pub fn final_tree(&self) -> Vec<(String, Value)> {
        let mut tree = Vec::new();
        for (i, st) in self.states.iter().enumerate() {
            match st {
                Some(OpState::Ok(v)) => tree.push((self.ops_ref[i].0.clone(), v.clone())),
                Some(OpState::Skipped) => tree.push((
                    self.ops_ref[i].0.clone(),
                    unproduced_value(self.ops_ref[i].1),
                )),
                None => {}
            }
        }
        tree
    }
}

fn unproduced_value(kind: Option<RelationKind>) -> Value {
    if kind == Some(RelationKind::Connection) {
        Value::Obj(vec![
            ("items".into(), Value::Arr(vec![])),
            ("cursor".into(), Value::Null),
        ])
    } else {
        Value::Null
    }
}

fn parse_policy(raw: &Option<String>) -> Result<PolicyKind, PlanFailure> {
    match raw.as_deref() {
        None | Some("fail") => Ok(PolicyKind::Fail),
        Some("retry") => Ok(PolicyKind::Retry),
        Some("continue") => Ok(PolicyKind::Continue),
        Some(other) => fail(
            PlanFailureCode::UnknownPolicy,
            format!("unknown policy kind: {other}"),
        ),
    }
}

/// Run the plan. `exec(op, bound_value)` returns the node outcome (mock).
pub fn run_plan<F>(
    plan: Option<&ExecutionPlanSpec>,
    ops: &[OpSpec],
    mut exec: F,
) -> Result<RunResult, PlanFailure>
where
    F: FnMut(&OpSpec, Option<&Value>) -> ExecOutcome,
{
    let n = ops.len();
    let mut states: Vec<Option<OpState>> = vec![None; n];
    let mut executed_idx: Vec<usize> = Vec::new();
    let mut skipped_idx: Vec<usize> = Vec::new();

    let stages: Vec<Vec<usize>> = match plan {
        Some(p) => p.groups.clone(),
        None => (0..n).map(|i| vec![i]).collect(),
    };

    validate_coverage(&stages, n)?;

    for stage in &stages {
        let mut ordered = stage.clone();
        ordered.sort_unstable(); // determinism: concurrency does not change results
        for &i in &ordered {
            let st = run_op(
                i,
                &ops[i],
                &states,
                &mut exec,
                &mut executed_idx,
                &mut skipped_idx,
            )?;
            states[i] = Some(st);
        }
    }

    executed_idx.sort_unstable();
    skipped_idx.sort_unstable();

    Ok(RunResult {
        executed: executed_idx.iter().map(|&i| ops[i].id.clone()).collect(),
        skipped: skipped_idx.iter().map(|&i| ops[i].id.clone()).collect(),
        ops_ref: ops
            .iter()
            .map(|o| (o.id.clone(), o.relation_kind))
            .collect(),
        states,
    })
}

#[allow(clippy::too_many_arguments)]
fn run_op<F>(
    i: usize,
    op: &OpSpec,
    states: &[Option<OpState>],
    exec: &mut F,
    executed_idx: &mut Vec<usize>,
    skipped_idx: &mut Vec<usize>,
) -> Result<OpState, PlanFailure>
where
    F: FnMut(&OpSpec, Option<&Value>) -> ExecOutcome,
{
    let policy = parse_policy(&op.policy)?; // fail-closed on unknown

    let mut bound_value: Option<Value> = None;
    if let Some(pidx) = op.parent {
        match states.get(pidx).and_then(|s| s.as_ref()) {
            Some(OpState::Ok(pv)) => {
                if let Some(field) = &op.bind_field {
                    let bound = pv.obj_get(field);
                    match bound {
                        None | Some(Value::Null) => {
                            skipped_idx.push(i);
                            return Ok(OpState::Skipped);
                        }
                        Some(b) => bound_value = Some(b.clone()),
                    }
                }
            }
            _ => {
                // parent skipped / failed / not produced → skip (chain)
                skipped_idx.push(i);
                return Ok(OpState::Skipped);
            }
        }
    }

    let outcome = exec(op, bound_value.as_ref());
    executed_idx.push(i);

    match outcome {
        ExecOutcome::Ok(v) => Ok(OpState::Ok(v)),
        ExecOutcome::Error(e) => match policy {
            PolicyKind::Fail => fail(
                PlanFailureCode::OpFailed,
                format!("operation '{}' failed under 'fail' policy: {e}", op.id),
            ),
            PolicyKind::Retry => fail(
                PlanFailureCode::OpFailed,
                format!(
                    "operation '{}' failed under 'retry' policy (exhausted): {e}",
                    op.id
                ),
            ),
            PolicyKind::Continue => {
                // failed op produces no Port → treat as unproduced/skip downstream.
                skipped_idx.push(i);
                executed_idx.pop(); // ran but produced no Port
                Ok(OpState::Skipped)
            }
        },
    }
}

fn validate_coverage(stages: &[Vec<usize>], n: usize) -> Result<(), PlanFailure> {
    let mut seen = std::collections::HashSet::new();
    for stage in stages {
        for &i in stage {
            if i >= n {
                return fail(
                    PlanFailureCode::InvalidPlan,
                    format!("stage index {i} out of range [0,{n})"),
                );
            }
            if !seen.insert(i) {
                return fail(
                    PlanFailureCode::InvalidPlan,
                    format!("operation {i} appears in more than one stage"),
                );
            }
        }
    }
    if seen.len() != n {
        return fail(
            PlanFailureCode::InvalidPlan,
            format!(
                "plan does not cover all {n} operations (covered {})",
                seen.len()
            ),
        );
    }
    Ok(())
}