plan_basic/
plan_basic.rs

1//! End-to-end demonstration of `BasicCpmPlanner` used as a library.
2//!
3//! Run with:
4//!
5//! ```bash
6//! cargo run -p mcp-flowgate-plan --example plan_basic
7//! ```
8//!
9//! This walks through the same surface an MCP client would drive over
10//! the wire, but invokes the trait directly:
11//!
12//! 1. Submit a small deliverable graph (4 deliverables, mixed prereqs)
13//! 2. Read the critical path
14//! 3. Acquire a cohort, mark its deliverables complete, repeat
15//! 4. Print the audit events captured along the way
16//!
17//! The example demonstrates that:
18//! - Deliverables with no prerequisites start in `Ready` and are
19//!   immediately acquirable.
20//! - Dependents flip to `Ready` only when all their prerequisites are
21//!   `Complete`.
22//! - Locks are atomically acquired with the cohort and released by
23//!   `mark_status(Complete)`.
24
25use std::path::PathBuf;
26use std::sync::{Arc, Mutex};
27
28use async_trait::async_trait;
29use cpm_planner::audit::{AuditEvent, AuditSink};
30use cpm_planner::plan::{CallerId, Deliverable, DeliverableStatus, PlanGraph};
31use cpm_planner::ports::Planner;
32use cpm_planner::BasicCpmPlanner;
33
34/// A trivial audit sink that buffers every event for inspection.
35#[derive(Debug, Default)]
36struct BufferingAudit {
37    events: Mutex<Vec<AuditEvent>>,
38}
39
40#[async_trait]
41impl AuditSink for BufferingAudit {
42    async fn record(&self, event: AuditEvent) -> anyhow::Result<()> {
43        let mut events = self
44            .events
45            .lock()
46            .map_err(|e| anyhow::anyhow!("audit buffer poisoned: {e}"))?;
47        events.push(event);
48        Ok(())
49    }
50}
51
52fn deliverable(
53    id: &str,
54    owned_files: &[&str],
55    prerequisites: &[&str],
56    estimated_effort_hours: f32,
57) -> Deliverable {
58    Deliverable {
59        id: id.to_string(),
60        owned_files: owned_files.iter().map(PathBuf::from).collect(),
61        prerequisites: prerequisites.iter().map(|s| s.to_string()).collect(),
62        estimated_effort_hours: Some(estimated_effort_hours),
63        metadata: serde_json::Value::Null,
64    }
65}
66
67#[tokio::main]
68async fn main() -> anyhow::Result<()> {
69    // ── 1. Build the planner with a capturing audit sink ────────────
70    let audit = Arc::new(BufferingAudit::default());
71    let planner = BasicCpmPlanner::with_audit(audit.clone());
72
73    // ── 2. Submit a small graph ─────────────────────────────────────
74    //
75    //   d1 (1h) ─┐
76    //            ├─► d3 (2h) ─► d4 (1h)
77    //   d2 (3h) ─┘                   ▲
78    //                                │
79    //                            (critical
80    //                             path: d2→d3→d4 = 6h)
81    let graph = PlanGraph {
82        deliverables: vec![
83            deliverable("d1", &["src/a.rs"], &[], 1.0),
84            deliverable("d2", &["src/b.rs"], &[], 3.0),
85            deliverable("d3", &["src/c.rs"], &["d1", "d2"], 2.0),
86            deliverable("d4", &["src/d.rs"], &["d3"], 1.0),
87        ],
88        max_chained_dispatch: None,
89    };
90    let plan_id = planner.submit_plan(graph).await?;
91    println!("submitted plan: {plan_id}");
92
93    // ── 3. Print the critical path ──────────────────────────────────
94    let status = planner.status(&plan_id).await?;
95    println!(
96        "critical path: {:?} ({:.1}h total)",
97        status.critical_path, status.critical_path_hours
98    );
99
100    // ── 4. Drive the plan to completion, cohort by cohort ───────────
101    let caller = CallerId("demo-orchestrator".to_string());
102    let mut round = 0;
103    loop {
104        round += 1;
105        // Acquire up to 4 deliverables. The planner returns only those
106        // whose prerequisites are Complete AND whose files don't overlap
107        // with anything currently locked.
108        let cohort = planner.acquire_cohort(&plan_id, &caller, 4).await?;
109        if cohort.rows.is_empty() {
110            // Two cases for empty: terminal (everything Complete) or
111            // blocked (locks held by someone else, or no Ready work).
112            // For this single-caller example, empty means terminal.
113            println!("round {round}: no work remaining; plan is terminal.");
114            break;
115        }
116        let ids: Vec<&str> = cohort
117            .rows
118            .iter()
119            .map(|r| r.deliverable.id.as_str())
120            .collect();
121        println!("round {round}: acquired cohort {ids:?}");
122
123        for row in &cohort.rows {
124            planner
125                .mark_status(
126                    &plan_id,
127                    &row.deliverable.id,
128                    &caller,
129                    DeliverableStatus::Complete,
130                )
131                .await?;
132            println!("  marked {} complete", row.deliverable.id);
133        }
134    }
135
136    // ── 5. Final status ─────────────────────────────────────────────
137    let status = planner.status(&plan_id).await?;
138    let complete = status
139        .deliverables
140        .iter()
141        .filter(|(_, s)| matches!(s, DeliverableStatus::Complete))
142        .count();
143    println!(
144        "final state: {complete}/{total} deliverables complete; {locks} locks held",
145        total = status.deliverables.len(),
146        locks = status.locks_held.len()
147    );
148
149    // ── 6. Audit trail ──────────────────────────────────────────────
150    let events = audit
151        .events
152        .lock()
153        .map_err(|e| anyhow::anyhow!("audit buffer poisoned: {e}"))?;
154    println!("\naudit events ({} total):", events.len());
155    for event in events.iter() {
156        println!("  - {}", event.event_type);
157    }
158
159    Ok(())
160}
plan_basic/plan_basic.rs

plan_basic/
plan_basic.rs
