async-nodes 0.1.0

//! # An agentic deployment gate (mastery)
//!
//! One graph, every flagship feature on stage at once:
//!
//! - **Failure recovery as topology**: a flaky preflight check retries with exponential
//!   backoff (`RecoveryPolicy::Retry`), and a notification node that *always* fails
//!   routes its failure onto a visible `OnError` edge instead of aborting the run.
//! - **Human-in-the-loop**: the approval gate *suspends the entire run* mid-graph; the
//!   host injects the verdict later with `engine.resume(...)`.
//! - **Serializable workflows**: the whole graph round-trips through versioned JSON
//!   (`GraphDef`) and re-instantiates from a registry — then runs *again*, taking the
//!   other approval branch. A GUI could load that same JSON.
//! - **A live event stream**: a `BspInterceptor` prints the run as it happens.
//!
//! Run it: `cargo run --example agentic_deploy`

use std::sync::Arc;

use async_nodes::{
    BspContext, BspEngine, BspError, BspGraph, BspInterceptor, BspNode, Condition,
    EngineResult, InputPin, NodeId, NodeOutcome, NodeRegistry, Recovery, RecoveryPolicy,
    RetryPolicy, RunContext, async_trait::async_trait,
};
use indexmap::IndexMap;

// =========================================================================
// STATE — yours to define; the engine moves it and never looks inside.
// =========================================================================

#[derive(Clone, Default)]
struct DeployState {
    plan: Vec<String>,
    log: Vec<String>,
    approved: bool,
}

impl DeployState {
    fn say(&mut self, line: impl Into<String>) {
        let line = line.into();
        println!("    [state] {line}");
        self.log.push(line);
    }
}

// =========================================================================
// NODES — plain `BspNode<S>` impls. Nodes never name their successors;
// the edges decide what happens next.
// =========================================================================

/// Drafts the rollout plan into the state.
struct Plan;

#[async_trait]
impl BspNode<DeployState> for Plan {
    fn name(&self) -> &str {
        "plan"
    }

    async fn execute(&self, run: &mut RunContext<'_, DeployState>) -> Result<NodeOutcome, BspError> {
        let s = &mut *run.state;
        s.plan = vec!["migrate db".into(), "deploy api".into(), "deploy web".into()];
        s.say(format!("planned {} steps", s.plan.len()));
        Ok(NodeOutcome::Success(Default::default()))
    }
}

/// A preflight check that flakes twice before it passes.
///
/// The retry policy lives on the *node*, not in your business logic: the engine
/// re-invokes `execute` with backoff (and jitter), honoring server hints when present.
/// `execution_no` is the retry-aware attempt counter — here we use it to flake
/// exactly twice; a real node would use it to undo partial writes on re-entry.
struct Preflight;

#[async_trait]
impl BspNode<DeployState> for Preflight {
    fn name(&self) -> &str {
        "preflight"
    }

    fn recovery(&self) -> RecoveryPolicy {
        RecoveryPolicy::Retry {
            policy: RetryPolicy::default()
                .with_max_retries(5)
                .with_base_delay(std::time::Duration::from_millis(10)),
            // The classification hook: which failures deserve another attempt.
            // (llm-net plugs its HTTP/quota/structural classifier in right here.)
            classify: Arc::new(|_| Recovery::Retry { after: None }),
        }
    }

    async fn execute(&self, run: &mut RunContext<'_, DeployState>) -> Result<NodeOutcome, BspError> {
        if run.execution_no < 2 {
            return Err(BspError::NodeExecutionFailed {
                node_id: run.node_id,
                node_name: "preflight".to_string(),
                message: format!("simulated flake (attempt {})", run.execution_no + 1),
            });
        }
        run.state.say("preflight green".to_string());
        Ok(NodeOutcome::Success(Default::default()))
    }
}

/// The human gate. First entry: suspend the whole run, handing the host a payload and
/// the pin it must fill. On resume, the injected verdict lands in the input gather.
struct ApprovalGate;

#[async_trait]
impl BspNode<DeployState> for ApprovalGate {
    fn name(&self) -> &str {
        "approval_gate"
    }

    fn input_ports(&self) -> Vec<usize> {
        vec![0] // port 0: the human's verdict, injected on resume
    }

    async fn execute(&self, run: &mut RunContext<'_, DeployState>) -> Result<NodeOutcome, BspError> {
        match run.input::<bool>(0) {
            Some(approved) => {
                run.state.approved = approved;
                run.state.say(format!("human verdict: {approved}"));
                Ok(NodeOutcome::Success(Default::default()))
            }
            None => {
                // No verdict yet — freeze. Counters, frontier, plan: all packaged
                // into the suspension snapshot and continued, never restarted.
                run.state.say("suspending for human review…".to_string());
                Ok(NodeOutcome::Suspend {
                    payload: serde_json::json!({
                        "question": "Approve deployment?",
                        "plan": run.state.plan,
                    }),
                    waiting_inputs: vec![0],
                })
            }
        }
    }
}

struct Deploy;

#[async_trait]
impl BspNode<DeployState> for Deploy {
    fn name(&self) -> &str {
        "deploy"
    }

    async fn execute(&self, run: &mut RunContext<'_, DeployState>) -> Result<NodeOutcome, BspError> {
        let steps = run.state.plan.clone();
        for step in steps {
            run.state.say(format!("deployed {step}"));
        }
        Ok(NodeOutcome::Success(Default::default()))
    }
}

struct Abort;

#[async_trait]
impl BspNode<DeployState> for Abort {
    fn name(&self) -> &str {
        "abort"
    }

    async fn execute(&self, run: &mut RunContext<'_, DeployState>) -> Result<NodeOutcome, BspError> {
        run.state.say("deployment aborted, plan archived".to_string());
        Ok(NodeOutcome::Success(Default::default()))
    }
}

/// A notification that always fails — on purpose. Its recovery policy routes the
/// failure onto the `OnError` edge instead of killing the completed deployment.
struct Notify;

#[async_trait]
impl BspNode<DeployState> for Notify {
    fn name(&self) -> &str {
        "notify"
    }

    fn recovery(&self) -> RecoveryPolicy {
        RecoveryPolicy::RouteToErrorEdge
    }

    async fn execute(&self, run: &mut RunContext<'_, DeployState>) -> Result<NodeOutcome, BspError> {
        Err(BspError::NodeExecutionFailed {
            node_id: run.node_id,
            node_name: "notify".to_string(),
            message: "SMTP unreachable (simulated)".to_string(),
        })
    }
}

struct Audit;

#[async_trait]
impl BspNode<DeployState> for Audit {
    fn name(&self) -> &str {
        "audit"
    }

    async fn execute(&self, run: &mut RunContext<'_, DeployState>) -> Result<NodeOutcome, BspError> {
        run.state
            .say("notification failed → logged to audit trail instead".to_string());
        Ok(NodeOutcome::Success(Default::default()))
    }
}

// =========================================================================
// LIVE TRACE — every interesting thing the engine does is an event.
// =========================================================================

struct Trace;

impl BspInterceptor for Trace {
    fn on_superstep_start(&self, step: u64, frontier: &[NodeId]) {
        println!("── superstep {step} ─ frontier {frontier:?}");
    }

    fn on_node_retry(&self, node_id: NodeId, attempt: u32, wait: std::time::Duration, err: &BspError) {
        println!("   ↻ node {node_id} retry #{attempt} in {wait:?} ({err})");
    }

    fn on_edge_taken(&self, source: NodeId, target: NodeId, condition_label: &str) {
        println!("   ⤷ edge {source} → {target}  [{condition_label}]");
    }

    fn on_suspended(&self, suspensions: &[async_nodes::PendingSuspension]) {
        for s in suspensions {
            println!("   ⏸ suspended at node {}: {}", s.node_id, s.payload);
        }
    }
}

// =========================================================================
// THE GRAPH — and its JSON afterlife.
// =========================================================================

fn build_graph() -> BspGraph<DeployState> {
    let mut graph = BspGraph::new();
    let plan = graph.add_node(Plan);
    let preflight = graph.add_node(Preflight);
    let gate = graph.add_node(ApprovalGate);
    let deploy = graph.add_node(Deploy);
    let abort = graph.add_node(Abort);
    let notify = graph.add_node(Notify);
    let audit = graph.add_node(Audit);

    // Named conditions keep the graph serializable: the JSON keeps the name, the
    // registry rebinds the closure at load time.
    graph.register_condition("approved", |s: &DeployState, _| s.approved);
    graph.register_condition("rejected", |s: &DeployState, _| !s.approved);

    graph.add_edge(plan.id, preflight.id);
    graph.add_edge(preflight.id, gate.id);
    graph
        .add_edge(gate.id, deploy.id)
        .when(Condition::Named("approved".into()));
    graph
        .add_edge(gate.id, abort.id)
        .when(Condition::Named("rejected".into()));
    graph.add_edge(deploy.id, notify.id);
    // The error path is topology you can see, not a catch block you can't.
    graph.add_edge_with(notify.id, audit.id, Condition::OnError);

    graph
}

/// The registry is how a JSON document becomes a running graph again: factories for
/// node types, predicates for named conditions. A GUI's palette is this same list.
fn registry() -> NodeRegistry<DeployState> {
    let mut r = NodeRegistry::new();
    r.register_node("plan", |_, _| Ok(Box::new(Plan)));
    r.register_node("preflight", |_, _| Ok(Box::new(Preflight)));
    r.register_node("approval_gate", |_, _| Ok(Box::new(ApprovalGate)));
    r.register_node("deploy", |_, _| Ok(Box::new(Deploy)));
    r.register_node("abort", |_, _| Ok(Box::new(Abort)));
    r.register_node("notify", |_, _| Ok(Box::new(Notify)));
    r.register_node("audit", |_, _| Ok(Box::new(Audit)));
    r.register_condition("approved", |s: &DeployState, _| s.approved);
    r.register_condition("rejected", |s: &DeployState, _| !s.approved);
    r
}

/// One full run up to the approval gate; returns the engine and the suspension.
async fn run_until_gate(
    graph: BspGraph<DeployState>,
    gate: NodeId,
) -> (BspEngine<DeployState>, Box<async_nodes::SuspendedExecution<DeployState>>) {
    let engine = BspEngine::new(graph)
        .with_interceptor(Arc::new(Trace))
        .with_max_steps(1_000);

    match engine.run(DeployState::default()).await {
        Ok(EngineResult::Suspended(suspended)) => {
            // Sanity: the gate is the only thing we're waiting on.
            assert_eq!(suspended.suspensions[0].node_id, gate);
            (engine, suspended)
        }
        Ok(other) => panic!("expected the run to suspend at the gate, got {}", verdict(&other)),
        Err(e) => panic!("expected the run to suspend at the gate, got error: {e}"),
    }
}

fn verdict(result: &EngineResult<DeployState>) -> &'static str {
    match result {
        EngineResult::Complete(_) => "complete",
        EngineResult::Suspended(_) => "suspended",
        EngineResult::StepLimitReached(_) => "step-limit",
    }
}

#[tokio::main]
async fn main() {
    // ---------------------------------------------------------------------
    // RUN 1 — the programmatic graph, APPROVED by the "human".
    // ---------------------------------------------------------------------
    println!("━━━ RUN 1: programmatic graph, human approves ━━━");
    let graph = build_graph();
    let gate_id = graph.node_ids().nth(2).expect("gate is the 3rd node");
    let (engine, suspended) = run_until_gate(graph, gate_id).await;

    // The host reads the payload, asks its human, and injects the verdict into the
    // gate's waiting input pin. Resume continues the exact superstep accounting.
    let mut injections = IndexMap::new();
    injections.insert(
        InputPin {
            node_id: gate_id,
            port: 0,
        },
        async_nodes::BspValue::new(true),
    );
    let result = engine.resume(*suspended, injections).await.unwrap();
    let EngineResult::Complete(state) = result else {
        panic!("expected completion, got {}", verdict(&result));
    };
    println!("━━━ run 1 complete: {} state log lines ━━━\n", state.user.log.len());

    // ---------------------------------------------------------------------
    // RUN 2 — the SAME graph, but loaded from JSON, and the human REJECTS.
    // ---------------------------------------------------------------------
    println!("━━━ RUN 2: graph re-instantiated from GraphDef JSON, human rejects ━━━");
    let def = build_graph().to_def();
    let json = def.to_json().expect("serialize graph definition");
    println!("the workflow as versioned data (excerpt):");
    for line in json.lines().take(14) {
        println!("  {line}");
    }
    println!("  …\n");

    let def = async_nodes::GraphDef::from_json(&json).expect("parse graph definition");
    let graph2 = BspGraph::from_def(&def, &registry(), &BspContext::new())
        .expect("registry completeness");
    let gate2 = graph2.node_ids().nth(2).expect("gate is the 3rd node");
    let (engine2, suspended2) = run_until_gate(graph2, gate2).await;

    let mut injections = IndexMap::new();
    injections.insert(
        InputPin {
            node_id: gate2,
            port: 0,
        },
        async_nodes::BspValue::new(false), // rejected this time
    );
    let result2 = engine2.resume(*suspended2, injections).await.unwrap();
    let EngineResult::Complete(state2) = result2 else {
        panic!("expected completion, got {}", verdict(&result2));
    };

    println!("━━━ run 2 complete ━━━");
    println!("\nfinal state log (run 2, from JSON):");
    for line in &state2.user.log {
        println!("  • {line}");
    }
}