//! Topological-sort dispatcher: "what should I work on next?"
//!
//! The single function the rest of the world (CLI, MCP, agents) calls
//! is [`topo_ready`]. It returns every node that:
//! - belongs to a plan whose status is `Active` (or `Draft` —
//! drafts are work-in-progress, not archived),
//! - has status `Pending` or `Ready`,
//! - has all incoming-edge dependencies satisfied (every `from`
//! node has status `Done`),
//!
//! …in topological order so dependents always come after dependencies.
//! The returned [`NextStep`] carries everything an actor needs to
//! decide "should I take this?".
use std::collections::BTreeMap;
use chrono::Utc;
use petgraph::algo::toposort;
use petgraph::graph::{DiGraph, NodeIndex};
use super::event::{Event, NodeStatus, PlanStatus};
use super::ids::{NodeId, PlanId};
use super::state::Funnel;
#[derive(Debug, Clone, serde::Serialize)]
pub struct NextStep {
pub plan_id: PlanId,
pub node_id: NodeId,
pub kind: String,
pub targets: Vec<String>,
pub summary: String,
pub prompt_excerpt: Option<String>,
}
/// Return every actionable node across every active plan, in
/// topological order. Side-effect: promotes eligible `Pending` nodes
/// to `Ready` in-memory (no event emitted — the readiness derivation
/// is deterministic from edges + statuses, so emitting events would
/// just bloat the log).
pub fn topo_ready(funnel: &mut Funnel) -> Vec<NextStep> {
funnel.promote_ready();
let mut out: Vec<NextStep> = Vec::new();
for plan in funnel.plans.values() {
if !matches!(plan.status, PlanStatus::Draft | PlanStatus::Active) {
continue;
}
let mut graph: DiGraph<NodeId, ()> = DiGraph::new();
let mut idx: BTreeMap<NodeId, NodeIndex> = BTreeMap::new();
for id in plan.nodes.keys() {
idx.insert(id.clone(), graph.add_node(id.clone()));
}
for (from, to) in &plan.edges {
if let (Some(&a), Some(&b)) = (idx.get(from), idx.get(to)) {
graph.add_edge(a, b, ());
}
}
let order = match toposort(&graph, None) {
Ok(o) => o,
Err(cyc) => {
eprintln!(
"funnel: plan `{}` has a cycle around node `{}` — skipping",
plan.id,
graph[cyc.node_id()]
);
continue;
}
};
for ni in order {
let nid = &graph[ni];
let node = &plan.nodes[nid];
if matches!(node.status, NodeStatus::Ready | NodeStatus::Pending) {
// Only Ready is genuinely actionable; surface
// Pending too so callers can see *why* nothing
// ran (deps not done).
if node.status == NodeStatus::Ready {
out.push(NextStep {
plan_id: plan.id.clone(),
node_id: nid.clone(),
kind: node.kind.clone(),
targets: node.targets.clone(),
summary: plan.summary.clone(),
prompt_excerpt: node.prompt_excerpt.clone(),
});
}
}
}
}
out
}
/// Like [`topo_ready`] but scoped to a specific plan.
pub fn topo_ready_for_plan(funnel: &mut Funnel, plan_id: &PlanId) -> Vec<NextStep> {
topo_ready(funnel)
.into_iter()
.filter(|s| &s.plan_id == plan_id)
.collect()
}
/// One recorded stage transition produced by [`simulate_flow`].
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
pub struct FlowTransition {
pub wave: usize,
pub plan_id: String,
pub node_id: String,
/// The target status the node moved to on this wave (`"done"` / `"failed"`).
pub status: String,
}
/// The rolled-up result of a [`simulate_flow`] run: how many waves it took and
/// the final Done/Failed/remaining tally over every node in every plan.
#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize)]
pub struct FlowSim {
/// Number of waves actually advanced (≤ the `steps` cap).
pub waves: usize,
pub done: usize,
pub failed: usize,
/// Nodes still Pending/Ready/Blocked/InProgress at the end (e.g. dependents
/// of a Failed node that can never become Ready).
pub remaining: usize,
pub total: usize,
pub transitions: Vec<FlowTransition>,
}
fn node_status_str(s: NodeStatus) -> &'static str {
match s {
NodeStatus::Pending => "pending",
NodeStatus::Ready => "ready",
NodeStatus::InProgress => "in_progress",
NodeStatus::Done => "done",
NodeStatus::Blocked => "blocked",
NodeStatus::Failed => "failed",
}
}
/// Time-step the whole project DAG to completion (or `steps` waves, whichever
/// comes first): repeatedly take [`topo_ready`] and advance every ready node to
/// `Done`, recording each stage transition. This is the project-flow simulator
/// (distinct from the LLM task-decomposition `decompose::simulate`): it ADVANCES
/// nodes through statuses over time, which `topo_ready` alone never does.
///
/// A DAG whose nodes all start `Pending` reaches all-`Done` in N waves, where N
/// is the length of the longest dependency chain.
pub fn simulate_flow(funnel: &mut Funnel, steps: usize) -> FlowSim {
simulate_flow_with(funnel, steps, |_| NodeStatus::Done)
}
/// [`simulate_flow`] with a caller policy deciding each ready node's fate. The
/// policy returns the status to move the node to; only `Failed` is honoured as a
/// failure, anything else is treated as `Done` (a Failed node stalls its
/// dependents, exactly as a real blocked stage would).
pub fn simulate_flow_with(
funnel: &mut Funnel,
steps: usize,
mut policy: impl FnMut(&NextStep) -> NodeStatus,
) -> FlowSim {
let mut sim = FlowSim::default();
for wave in 0..steps {
let ready = topo_ready(funnel);
if ready.is_empty() {
break;
}
sim.waves = wave + 1;
for step in &ready {
let status = match policy(step) {
NodeStatus::Failed => NodeStatus::Failed,
_ => NodeStatus::Done,
};
let ev = Event::NodeStatusChanged {
plan_id: step.plan_id.clone(),
node_id: step.node_id.clone(),
status,
why: None,
ts: Utc::now(),
};
if funnel.apply(&ev).is_ok() {
sim.transitions.push(FlowTransition {
wave: wave + 1,
plan_id: step.plan_id.to_string(),
node_id: step.node_id.to_string(),
status: node_status_str(status).to_string(),
});
}
}
}
// Final tally (promote so a last wave's dependents are reflected as ready).
funnel.promote_ready();
for plan in funnel.plans.values() {
for node in plan.nodes.values() {
sim.total += 1;
match node.status {
NodeStatus::Done => sim.done += 1,
NodeStatus::Failed => sim.failed += 1,
_ => sim.remaining += 1,
}
}
}
sim
}
#[cfg(test)]
mod sim_tests {
use super::*;
use crate::event::{ItemKind, PlanStatus};
use crate::ids::{IdeaId, NodeId, PlanId};
/// Build a 3-node chain a→b→c (all Pending) in one Active plan.
fn chain_funnel() -> (Funnel, PlanId) {
let mut f = Funnel::default();
let now = Utc::now();
let idea = IdeaId::seq(0);
let plan = PlanId::seq(0);
f.apply(&Event::IdeaSubmitted {
id: idea.clone(),
source: "test".into(),
text: "chain".into(),
refs: vec![],
item_kind: ItemKind::Idea,
ts: now,
})
.unwrap();
f.apply(&Event::PlanCreated {
id: plan.clone(),
idea_id: idea,
summary: "chain plan".into(),
planner: "test".into(),
ts: now,
})
.unwrap();
for i in 0..3u64 {
f.apply(&Event::NodeAdded {
plan_id: plan.clone(),
node_id: NodeId::seq(i),
kind: "step".into(),
params: Default::default(),
targets: vec![],
prompt_excerpt: None,
ts: now,
})
.unwrap();
}
f.apply(&Event::EdgeAdded {
plan_id: plan.clone(),
from_node: NodeId::seq(0),
to_node: NodeId::seq(1),
ts: now,
})
.unwrap();
f.apply(&Event::EdgeAdded {
plan_id: plan.clone(),
from_node: NodeId::seq(1),
to_node: NodeId::seq(2),
ts: now,
})
.unwrap();
f.apply(&Event::PlanStatusChanged {
plan_id: plan.clone(),
status: PlanStatus::Active,
why: None,
ts: now,
})
.unwrap();
(f, plan)
}
#[test]
fn all_pending_reaches_all_done_in_chain_length_waves() {
let (mut f, _plan) = chain_funnel();
let sim = simulate_flow(&mut f, 10);
assert_eq!(sim.total, 3);
assert_eq!(sim.done, 3, "every node reaches Done");
assert_eq!(sim.failed, 0);
assert_eq!(sim.remaining, 0);
assert_eq!(sim.waves, 3, "a 3-node chain drains in exactly 3 waves");
assert_eq!(sim.transitions.len(), 3);
assert!(sim.transitions.iter().all(|t| t.status == "done"));
}
#[test]
fn a_failed_first_node_stalls_its_dependents() {
let (mut f, _plan) = chain_funnel();
// Fail the very first node → nothing downstream can ever become Ready.
let sim = simulate_flow_with(&mut f, 10, |step| {
if step.node_id == NodeId::seq(0) {
NodeStatus::Failed
} else {
NodeStatus::Done
}
});
assert_eq!(sim.failed, 1);
assert_eq!(sim.done, 0);
assert_eq!(sim.remaining, 2, "b and c never become ready");
assert_eq!(sim.waves, 1, "only the first wave had a ready node");
}
}