open-standards/jcp/jcp-rs/crates/jcp-orchestrate/examples/fleet.rs

//! Orchestrate a heterogeneous fleet: route work across harnesses, fail over on
//! budget exhaustion, retry a flaky one, deny the ungranted — every dispatch
//! capability-gated, joule-budgeted, and sealed.
//!
//! ```text
//! cargo run -p jcp-orchestrate --example fleet
//! ```

use std::cell::Cell;

use ed25519_dalek::SigningKey;
use jcp_orchestrate::{Harness, HarnessOutcome, Orchestrator, Task};

/// A harness that completes at a fixed quoted cost (stands in for any backend —
/// a different agent framework, a local model, a remote worker).
struct Worker {
    id: String,
    kinds: Vec<String>,
    cost: u64,
}
impl Harness for Worker {
    fn id(&self) -> &str {
        &self.id
    }
    fn handles(&self, kind: &str) -> bool {
        self.kinds.iter().any(|k| k == kind)
    }
    fn estimate(&self, _task: &Task) -> u64 {
        self.cost
    }
    fn run(&self, _task: &Task) -> HarnessOutcome {
        HarnessOutcome::done(self.cost)
    }
}

/// A harness that fails transiently `n` times before completing.
struct Flaky {
    id: String,
    cost: u64,
    fails: Cell<u32>,
}
impl Harness for Flaky {
    fn id(&self) -> &str {
        &self.id
    }
    fn handles(&self, kind: &str) -> bool {
        kind == "scrape"
    }
    fn estimate(&self, _task: &Task) -> u64 {
        self.cost
    }
    fn run(&self, _task: &Task) -> HarnessOutcome {
        let left = self.fails.get();
        if left > 0 {
            self.fails.set(left - 1);
            HarnessOutcome::retry(self.cost, "endpoint timeout")
        } else {
            HarnessOutcome::done(self.cost)
        }
    }
}

fn main() {
    let mut fleet = Orchestrator::new(
        SigningKey::from_bytes(&[1u8; 32]), // the principal authorising the fleet
        SigningKey::from_bytes(&[7u8; 32]), // the sealing steward
        "did:example:fleet",
        3, // max attempts per item
    );

    // Two harnesses can summarize; the primary runs out of budget first and the
    // backup picks up the overflow. A third harness scrapes (flaky). Nothing is
    // granted "translate".
    fleet.register(Box::new(Worker { id: "summarizer-A".into(), kinds: vec!["summarize".into()], cost: 100 }), &["summarize"], 250).unwrap();
    fleet.register(Box::new(Worker { id: "summarizer-B".into(), kinds: vec!["summarize".into()], cost: 100 }), &["summarize"], 1_000).unwrap();
    fleet.register(Box::new(Flaky { id: "scraper".into(), cost: 80, fails: Cell::new(2) }), &["scrape"], 1_000).unwrap();

    // A batch of mixed work.
    for n in 0..4 {
        fleet.submit(Task::new(format!("sum-{n}"), "summarize", vec![n]));
    }
    fleet.submit(Task::new("scrape-1", "scrape", b"https://example".to_vec()));
    fleet.submit(Task::new("xlate-1", "translate", b"bonjour".to_vec())); // ungranted kind

    let report = fleet.run();

    println!("── run report ──");
    println!("  done   : {}", report.done);
    println!("  denied : {}", report.denied);
    println!("  failed : {}", report.failed);
    println!("  joules : {} µJ authorised across the fleet", report.total_uj);
    println!("  sealed : {} decision receipts", report.receipts);
    println!("\n── per-item ──");
    for id in ["sum-0", "sum-1", "sum-2", "sum-3", "scrape-1", "xlate-1"] {
        println!("  {id:10} → {:?}", fleet.state_of(id).unwrap());
    }
    println!("\n── per-harness joules ──");
    for h in ["summarizer-A", "summarizer-B", "scraper"] {
        println!("  {h:14} {} µJ", fleet.spent_uj(h));
    }

    // The whole run is one ordered, verifiable ledger.
    let steward = fleet.steward().to_string();
    let pk = fleet.steward_pubkey();
    let verified = fleet.receipts().iter().filter(|r| jcp_receipt::extract(r, &steward, &pk).is_ok()).count();
    println!("\naudit: {verified}/{} sealed decisions verify under the steward key", fleet.receipts().len());
}