src/plan_dag.rs
//! **Plan-DAG projection** — the funnel's cards + their ordering/dependency edges
//! as a neutral, egui-free **graph model** (nodes + edges + per-node metadata +
//! a topological order).
//!
//! This is the *deep* half of the funnel Plan-DAG UI (`nornir viz`): it assembles
//! the DAG from real funnel data (the [`cards_from_funnel`](crate::card::cards_from_funnel)
//! projection + each [`Plan`](crate::state::Plan)'s decomposition lineage and
//! dependency edges), and a *thin* renderer (`facett-plandag`'s `PlanDagView` in
//! nornir) maps it onto pixels. Keeping the graph assembly here — pure, BTree-
//! ordered, clock-free — makes it deterministic (FC-7) and golden-testable with no
//! warehouse, no GPU and no facett dep.
//!
//! ## Shape
//! ```text
//! PlanDag { nodes: Vec<PlanDagNode>, edges: Vec<PlanDagEdge>, sel_plan }
//! PlanDagNode { id, title, group?, status, is_bug, staleness_hours, weight_bytes, fanout }
//! PlanDagEdge { from, to, kind } // kind = Mother (idea→task) | Dep (task→task)
//! ```
//! The three numeric metadata scalars are the neutral cost bag a metric-bearing
//! renderer fills nodes by: **staleness** (age, oldest = hottest), **weight** (the
//! card body size) and **fan-out** (a plan's breadth / a node's out-degree).
use std::collections::{BTreeMap, BTreeSet};
use petgraph::algo::toposort;
use petgraph::graph::{DiGraph, NodeIndex};
use serde::{Deserialize, Serialize};
use crate::card::{cards_from_funnel, Card, CardType, Lane};
use crate::event::{Event, NodeStatus};
use crate::state::Funnel;
/// Neutral node run-status (a host renderer maps this onto its own palette). Idea
/// cards derive theirs from the card [`Lane`]; decomposed task-node cards from the
/// real [`NodeStatus`] in the plan.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum DagStatus {
/// Untriaged / pending — no work started.
Idle,
/// In flight — accepted/proposed idea, or a ready/in-progress task.
Running,
/// Completed — planned idea, or a Done task.
Done,
/// Dropped idea, or a Failed task.
Failed,
}
impl DagStatus {
/// A short, stable JSON/wire token.
#[must_use]
pub fn as_str(&self) -> &'static str {
match self {
DagStatus::Idle => "idle",
DagStatus::Running => "running",
DagStatus::Done => "done",
DagStatus::Failed => "failed",
}
}
/// The lane→status map for idea/error/test cards.
fn from_lane(l: Lane) -> Self {
match l {
Lane::Planned => DagStatus::Done,
Lane::Accepted | Lane::Proposed => DagStatus::Running,
Lane::Inbox => DagStatus::Idle,
Lane::Dropped => DagStatus::Failed,
}
}
/// The node-status→status map for decomposed task-node cards.
fn from_node_status(s: NodeStatus) -> Self {
match s {
NodeStatus::Done => DagStatus::Done,
NodeStatus::InProgress | NodeStatus::Ready => DagStatus::Running,
NodeStatus::Pending | NodeStatus::Blocked => DagStatus::Idle,
NodeStatus::Failed => DagStatus::Failed,
}
}
}
/// Which lineage an edge represents.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum EdgeKind {
/// Mother idea → one of its decomposed task nodes (the decomposition lineage).
Mother,
/// Task node → dependent task node (a `Plan.edges` dependency edge).
Dep,
}
impl EdgeKind {
#[must_use]
pub fn as_str(&self) -> &'static str {
match self {
EdgeKind::Mother => "mother",
EdgeKind::Dep => "dep",
}
}
}
/// One graph node — a funnel card (a mother idea, or a decomposed task node).
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct PlanDagNode {
/// Stable card id (`i-NNN` idea / `n-NNN` task node) — the render + selection key.
pub id: String,
/// One-line title (the card summary).
pub title: String,
/// The plan this task node belongs to (`Some(plan_id)`); `None` for a mother
/// idea. A renderer folds a group into a collapsible super-node.
pub group: Option<String>,
/// Neutral run status (drives node border colour).
pub status: DagStatus,
/// True for a bug/error card (a renderer can highlight it).
pub is_bug: bool,
/// Age relative to the newest card in the set, in hours (0 = newest, larger =
/// older). Clock-free: computed against the set's max timestamp, never `now`.
pub staleness_hours: f64,
/// Card body size in bytes (a rough "card weight").
pub weight_bytes: u64,
/// Fan-out: a mother idea's linked-plan count; a task node's dependency
/// out-degree.
pub fanout: u64,
}
/// One directed edge `from → to` (by stable card id) + its lineage kind.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct PlanDagEdge {
pub from: String,
pub to: String,
pub kind: EdgeKind,
}
/// The assembled Plan-DAG: nodes + edges + the scope it was built for.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct PlanDag {
pub nodes: Vec<PlanDagNode>,
pub edges: Vec<PlanDagEdge>,
/// The plan this DAG was scoped to (`None` = the whole funnel).
pub sel_plan: Option<String>,
}
/// One surviving node that changed between two [`PlanDag`] revisions (see
/// [`PlanDag::diff`]). `changed` lists the differing aspects in a fixed order:
/// any of `"title"`, `"status"`, `"group"`, `"is_bug"`, `"deps"` (the node's
/// incident-edge set).
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct NodeChange {
pub id: String,
pub changed: Vec<String>,
}
/// The structural delta between two [`PlanDag`] revisions — the DAG twin of the
/// coverage / bench diff gates. Produced by [`PlanDag::diff`]; a replan step reads
/// it to show "what changed". Every list is deterministically ordered (ids/tuples
/// sorted); [`is_empty`](PlanDiff::is_empty) is `true` for an unchanged revision.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct PlanDiff {
/// Node ids in the new graph but not the baseline.
pub added_nodes: Vec<String>,
/// Node ids in the baseline but not the new graph.
pub removed_nodes: Vec<String>,
/// Edges in the new graph but not the baseline.
pub added_edges: Vec<PlanDagEdge>,
/// Edges in the baseline but not the new graph.
pub removed_edges: Vec<PlanDagEdge>,
/// Nodes present in both whose payload or wiring changed.
pub changed_nodes: Vec<NodeChange>,
}
impl PlanDiff {
/// `true` when the two revisions are structurally identical — nothing added,
/// removed, or changed. An identical (re-projected) plan diffs to empty.
#[must_use]
pub fn is_empty(&self) -> bool {
self.added_nodes.is_empty()
&& self.removed_nodes.is_empty()
&& self.added_edges.is_empty()
&& self.removed_edges.is_empty()
&& self.changed_nodes.is_empty()
}
/// Total count of differences (added + removed + changed) — the diff's "size".
#[must_use]
pub fn change_count(&self) -> usize {
self.added_nodes.len()
+ self.removed_nodes.len()
+ self.added_edges.len()
+ self.removed_edges.len()
+ self.changed_nodes.len()
}
/// The robot-observable delta blob (LAW 6): the added/removed/changed lists +
/// the roll-up flags a headless replan test asserts.
#[must_use]
pub fn state_json(&self) -> serde_json::Value {
let edges = |es: &[PlanDagEdge]| {
es.iter()
.map(|e| serde_json::json!({ "from": e.from, "to": e.to, "kind": e.kind.as_str() }))
.collect::<Vec<_>>()
};
serde_json::json!({
"is_empty": self.is_empty(),
"change_count": self.change_count(),
"added_nodes": self.added_nodes,
"removed_nodes": self.removed_nodes,
"added_edges": edges(&self.added_edges),
"removed_edges": edges(&self.removed_edges),
"changed_nodes": self.changed_nodes.iter().map(|c| serde_json::json!({
"id": c.id, "changed": c.changed,
})).collect::<Vec<_>>(),
})
}
}
/// One **bottleneck** ranking row — a node scored by how much work it constrains
/// (see [`PlanDag::bottlenecks`]). The headline metric is
/// [`downstream`](Bottleneck::downstream): how many other nodes (transitively)
/// depend on this one, i.e. how much work its completion **unblocks**. The
/// [`fan_out`](Bottleneck::fan_out) / [`fan_in`](Bottleneck::fan_in) degrees are
/// the local shape (a wide fan-out is a divergence point a human can parallelise;
/// a wide fan-in is a join everything waits on).
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct Bottleneck {
/// Stable card id (`i-NNN` / `n-NNN`) — the same key [`PlanDagNode::id`] uses.
pub id: String,
/// In-degree: how many edges point AT this node (a join width).
pub fan_in: u64,
/// Out-degree: how many edges leave this node (a divergence width — the number
/// of successors immediately unblocked when it finishes).
pub fan_out: u64,
/// **Transitive descendant count** — every node reachable by following edges
/// out of this one (its whole downstream cone, itself excluded). This is the
/// "unblocks the most work" score: the amount of work stalled behind this node.
pub downstream: u64,
}
impl Bottleneck {
/// The robot-observable ranking row (LAW 6).
#[must_use]
pub fn state_json(&self) -> serde_json::Value {
serde_json::json!({
"id": self.id,
"fan_in": self.fan_in,
"fan_out": self.fan_out,
"downstream": self.downstream,
})
}
}
/// The **birth stamp** of one Plan-DAG node — the log **step** (0-based index of
/// the event, in the replayed slice, that first brought it into the DAG) and that
/// event's RFC3339 **timestamp**. See [`DagGrowth`].
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct NodeBirth {
/// Stable card id (`i-NNN` / `n-NNN`) — the same key [`PlanDagNode::id`] uses.
pub id: String,
/// 0-based event index at which this node first appeared in the DAG.
pub step: usize,
/// RFC3339 timestamp of that birth event.
pub at: String,
}
/// The **birth stamp** of one Plan-DAG edge — the [`step`](EdgeBirth::step) +
/// timestamp at which the edge `from → to` (of the given [`kind`](EdgeBirth::kind))
/// first appeared in the DAG. See [`DagGrowth`].
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct EdgeBirth {
pub from: String,
pub to: String,
pub kind: EdgeKind,
/// 0-based event index at which this edge first appeared in the DAG.
pub step: usize,
/// RFC3339 timestamp of that birth event.
pub at: String,
}
/// A birth-stamped **growth timeline** of a Plan-DAG — for every node and edge the
/// event log ever brings into the graph (scoped exactly like
/// [`PlanDag::from_funnel`]), the **step** (0-based event index) and timestamp at
/// which it FIRST appeared, in birth order (step ascending, then id). Produced by
/// [`DagGrowth::from_events`].
///
/// This is the **temporal** seam the live funnel pane draws the DAG *growing* along:
/// the funnel event log is append-only and never removes a node or edge, so the DAG
/// only ever grows — a renderer plays [`nodes`](DagGrowth::nodes) /
/// [`edges`](DagGrowth::edges) forward to reveal each element as its birth step is
/// reached (the "DAG growing live" animation), and a history scrubber reads
/// [`node_count_at`](DagGrowth::node_count_at) /
/// [`edge_count_at`](DagGrowth::edge_count_at) (or filters the birth marks by step)
/// to show the graph *as of* any point in the log.
///
/// Where [`PlanDag::diff`] compares two arbitrary revisions, this is the running
/// growth of a *single* log: `diff` answers "what changed between A and B", this
/// answers "when did each element arrive". Built by replaying the events through the
/// same [`Funnel::apply`] fold + [`PlanDag::from_funnel`] projection every other
/// query uses (no reimplementation), so it agrees with them by construction. It runs
/// on the cold render/history path, not a projection hot loop. Deterministic (FC-7):
/// birth order is (step, then id/tuple) sorted; an invariant-violating event (which
/// the store never appends) births nothing and leaves the DAG standing.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct DagGrowth {
/// Every node's birth stamp, in birth order (step asc, then id).
pub nodes: Vec<NodeBirth>,
/// Every edge's birth stamp, in birth order (step asc, then `(from,to,kind)`).
pub edges: Vec<EdgeBirth>,
/// Total number of events replayed — the final step count (`step` values are
/// `0 ..= steps.saturating_sub(1)`).
pub steps: usize,
/// The plan the timeline was scoped to (`None` = the whole funnel), mirroring
/// [`PlanDag::sel_plan`].
pub sel_plan: Option<String>,
// ── incremental fold state (live-stream only) ────────────────────────────
// These carry the running accumulator [`extend`](DagGrowth::extend) folds one
// event at a time into, so the live pane never re-replays the whole log. They
// are an implementation detail of the streaming path: skipped on the wire (the
// serialized timeline is just the births), NOT compared by `PartialEq` (two
// logs with identical births are equal regardless of how they were built), and
// `Funnel` is neither `Serialize` nor `Eq` anyway. A DagGrowth produced by the
// batch [`from_events`](DagGrowth::from_events) leaves them at their default
// (empty) — it is a frozen snapshot; only a stream built in-memory via
// [`new`](DagGrowth::new) (or `from_events` — see the note there) can be
// `extend`ed further. A deserialized DagGrowth is likewise frozen.
#[serde(skip)]
funnel: Funnel,
#[serde(skip)]
seen_nodes: BTreeSet<String>,
#[serde(skip)]
seen_edges: BTreeSet<(String, String, EdgeKind)>,
}
// Two growth timelines are equal iff they expose the same OBSERVABLE history — the
// births, step count and scope. The private incremental fold state (the retained
// `Funnel` + seen sets) is deliberately excluded: the batch `from_events` and the
// streaming `extend` build byte-identical timelines but keep different (or no) fold
// state, and they must still compare equal (the equivalence guard + the determinism
// assert both rely on this). `Funnel` is not `Eq`, so this also can't be derived.
impl PartialEq for DagGrowth {
fn eq(&self, other: &Self) -> bool {
self.nodes == other.nodes
&& self.edges == other.edges
&& self.steps == other.steps
&& self.sel_plan == other.sel_plan
}
}
impl Eq for DagGrowth {}
impl DagGrowth {
/// Replay `events` (in log order) and record, for every Plan-DAG node and edge,
/// the **first** step + timestamp at which it entered the DAG scoped by
/// `sel_plan` (`None` = whole funnel, `Some(pid)` = one plan — same scoping as
/// [`PlanDag::from_funnel`]).
///
/// One [`Funnel::apply`] + [`PlanDag::from_funnel`] per event; each element is
/// stamped on first sight and never overwritten, so the stamp is its true birth.
/// An event that `apply` rejects (an invariant violation the store would never
/// have appended) changes nothing and is skipped — it births no element.
#[must_use]
pub fn from_events(events: &[Event], sel_plan: Option<&str>) -> Self {
let mut funnel = Funnel::default();
let mut seen_nodes: BTreeSet<String> = BTreeSet::new();
let mut seen_edges: BTreeSet<(String, String, EdgeKind)> = BTreeSet::new();
let mut nodes: Vec<NodeBirth> = Vec::new();
let mut edges: Vec<EdgeBirth> = Vec::new();
for (step, ev) in events.iter().enumerate() {
// A rejected event never reaches the log (the store `apply`s before it
// appends), so it births nothing and the DAG stands still.
if funnel.apply(ev).is_err() {
continue;
}
let at = ev.ts().to_rfc3339();
let dag = PlanDag::from_funnel(&funnel, sel_plan);
for n in &dag.nodes {
if seen_nodes.insert(n.id.clone()) {
nodes.push(NodeBirth {
id: n.id.clone(),
step,
at: at.clone(),
});
}
}
for e in &dag.edges {
if seen_edges.insert((e.from.clone(), e.to.clone(), e.kind)) {
edges.push(EdgeBirth {
from: e.from.clone(),
to: e.to.clone(),
kind: e.kind,
step,
at: at.clone(),
});
}
}
}
// Birth order: step asc, then id / (from,to,kind) — deterministic (FC-7).
nodes.sort_by(|a, b| a.step.cmp(&b.step).then_with(|| a.id.cmp(&b.id)));
edges.sort_by(|a, b| {
a.step
.cmp(&b.step)
.then_with(|| (&a.from, &a.to, a.kind).cmp(&(&b.from, &b.to, b.kind)))
});
Self {
nodes,
edges,
steps: events.len(),
sel_plan: sel_plan.map(str::to_string),
// Retain the accumulated fold state so a batch-loaded timeline can be
// continued with `extend` (hydrate the live pane from the log so far,
// then stream new ticks). Excluded from `PartialEq`, skipped on the wire.
funnel,
seen_nodes,
seen_edges,
}
}
/// A fresh, empty growth **stream** scoped by `sel_plan`, ready to be fed one
/// event at a time via [`extend`](DagGrowth::extend).
///
/// This is the entry point for the **live** funnel pane: build a stream once at
/// pane open (or hydrate it with a batch [`from_events`](DagGrowth::from_events)
/// over the log so far — a `from_events` result carries its accumulated fold
/// state and can be `extend`ed further), then push each new event as it lands to
/// grow the DAG per-tick without re-replaying the whole log.
#[must_use]
pub fn new(sel_plan: Option<&str>) -> Self {
Self {
sel_plan: sel_plan.map(str::to_string),
..Self::default()
}
}
/// Fold **one** event into the stream, stamping the births it introduces — the
/// **incremental** twin of [`from_events`](DagGrowth::from_events) for the live
/// pane that pushes a single event per tick.
///
/// `from_events` re-projects the whole DAG after every event and, when a live
/// pane calls it each tick, re-replays the ENTIRE log every time (O(steps²) over
/// a session). `extend` instead keeps the running [`Funnel`] accumulator inside
/// the stream, applies just this one event to it, projects the new frame once,
/// and appends only the nodes/edges seen for the first time — the per-tick
/// "growing live" animation without a full re-fold. Cost per call is a single
/// `apply` + one [`PlanDag::from_funnel`] frame diff against the seen set.
///
/// Growth stays monotone (the log never removes an element, so `node_count_at` /
/// `edge_count_at` are non-decreasing) and each element is stamped at the step it
/// FIRST appears and never overwritten — its true birth. An event that `apply`
/// rejects (an invariant the store would never have appended) births nothing but
/// still counts as a step, exactly as in the batch path.
///
/// Byte-for-byte equivalence with the batch path is a tested guard: feeding a log
/// event-by-event through `extend` yields the SAME `nodes`/`edges` (same order,
/// same steps) as `from_events` on that log.
pub fn extend(&mut self, ev: &Event) {
// This event's 0-based step index, then the step count advances — a rejected
// event still consumes a step (matches `from_events`' `steps = events.len()`).
let step = self.steps;
self.steps += 1;
// A rejected event never reaches the log; it births nothing, DAG stands still.
if self.funnel.apply(ev).is_err() {
return;
}
let at = ev.ts().to_rfc3339();
let dag = PlanDag::from_funnel(&self.funnel, self.sel_plan.as_deref());
// Collect only the first-seen births from this single frame, then sort THIS
// batch (nodes by id, edges by (from,to,kind)) before appending. Steps arrive
// strictly ascending, so an append of an intra-step-sorted batch keeps the
// whole vector in (step asc, then id / tuple) order — byte-identical to the
// global sort `from_events` does at the end.
let mut new_nodes: Vec<NodeBirth> = Vec::new();
for n in &dag.nodes {
if self.seen_nodes.insert(n.id.clone()) {
new_nodes.push(NodeBirth {
id: n.id.clone(),
step,
at: at.clone(),
});
}
}
new_nodes.sort_by(|a, b| a.id.cmp(&b.id));
self.nodes.extend(new_nodes);
let mut new_edges: Vec<EdgeBirth> = Vec::new();
for e in &dag.edges {
if self
.seen_edges
.insert((e.from.clone(), e.to.clone(), e.kind))
{
new_edges.push(EdgeBirth {
from: e.from.clone(),
to: e.to.clone(),
kind: e.kind,
step,
at: at.clone(),
});
}
}
new_edges.sort_by(|a, b| (&a.from, &a.to, a.kind).cmp(&(&b.from, &b.to, b.kind)));
self.edges.extend(new_edges);
}
/// How many nodes the DAG holds **as of** `step` (every node born at or before
/// it). The node-count growth curve a progress bar / sparkline plots; monotone
/// non-decreasing in `step` because the log never removes a node.
#[must_use]
pub fn node_count_at(&self, step: usize) -> usize {
self.nodes.iter().filter(|n| n.step <= step).count()
}
/// How many edges the DAG holds **as of** `step` — the edge-count growth curve.
/// Monotone non-decreasing (the log never removes an edge).
#[must_use]
pub fn edge_count_at(&self, step: usize) -> usize {
self.edges.iter().filter(|e| e.step <= step).count()
}
/// The robot-observable growth blob (LAW 6): the scope, step count, per-element
/// birth stamps, and the final node/edge totals a headless test asserts.
#[must_use]
pub fn state_json(&self) -> serde_json::Value {
serde_json::json!({
"sel_plan": self.sel_plan,
"steps": self.steps,
"node_count": self.nodes.len(),
"edge_count": self.edges.len(),
"nodes": self.nodes.iter().map(|n| serde_json::json!({
"id": n.id, "step": n.step, "at": n.at,
})).collect::<Vec<_>>(),
"edges": self.edges.iter().map(|e| serde_json::json!({
"from": e.from, "to": e.to, "kind": e.kind.as_str(), "step": e.step, "at": e.at,
})).collect::<Vec<_>>(),
})
}
}
impl PlanDag {
/// Assemble the Plan-DAG from a replayed [`Funnel`].
///
/// * `sel_plan = Some(pid)` scopes the graph to plan `pid`: its mother idea card
/// plus every task-node card in that plan, and only edges among them.
/// * `sel_plan = None` projects the WHOLE funnel: every idea/error/test card and
/// every plan's decomposition + dependency edges.
///
/// Deterministic (FC-7): nodes follow [`cards_from_funnel`]'s id order, edges are
/// sorted `(from, to, kind)`, and staleness is measured against the set's newest
/// card (no wall-clock read).
#[must_use]
pub fn from_funnel(funnel: &Funnel, sel_plan: Option<&str>) -> Self {
let cards = cards_from_funnel(funnel);
// Real per-task-node status (the card only carries a lane; the accurate
// node lifecycle lives on the plan node).
let mut node_status: BTreeMap<String, NodeStatus> = BTreeMap::new();
for plan in funnel.plans.values() {
for node in plan.nodes.values() {
node_status.insert(node.id.as_str().to_string(), node.status);
}
}
// Which mother idea id owns the selected plan (for scoping to it).
let sel_mother: Option<String> = sel_plan.and_then(|pid| {
funnel
.plans
.values()
.find(|p| p.id.as_str() == pid)
.map(|p| p.idea_id.as_str().to_string())
});
// The card subset in scope.
let in_scope = |c: &Card| -> bool {
match sel_plan {
None => true,
Some(pid) => {
c.plan_ids.iter().any(|p| p == pid)
|| sel_mother.as_deref() == Some(c.id.as_str())
}
}
};
let picked: Vec<&Card> = cards.iter().filter(|c| in_scope(c)).collect();
let included: BTreeSet<&str> = picked.iter().map(|c| c.id.as_str()).collect();
// Newest timestamp among the picked cards — the staleness reference.
let max_ts = picked
.iter()
.filter_map(|c| parse_ts(&c.created))
.max()
.unwrap_or(0);
// ── edges: mother→child + intra-plan dependency ──────────────────────
let mut edge_set: BTreeSet<(String, String, EdgeKind)> = BTreeSet::new();
for plan in funnel.plans.values() {
if let Some(pid) = sel_plan {
if plan.id.as_str() != pid {
continue;
}
}
let mother = plan.idea_id.as_str();
for node in plan.nodes.values() {
let nid = node.id.as_str();
if included.contains(mother) && included.contains(nid) {
edge_set.insert((mother.to_string(), nid.to_string(), EdgeKind::Mother));
}
}
for (from, to) in &plan.edges {
let (f, t) = (from.as_str(), to.as_str());
if included.contains(f) && included.contains(t) {
edge_set.insert((f.to_string(), t.to_string(), EdgeKind::Dep));
}
}
}
// Dependency out-degree per task node (for its fan-out metric).
let mut out_degree: BTreeMap<String, u64> = BTreeMap::new();
for (f, _t, k) in &edge_set {
if *k == EdgeKind::Dep {
*out_degree.entry(f.clone()).or_default() += 1;
}
}
let nodes: Vec<PlanDagNode> = picked
.iter()
.map(|c| {
let is_task = node_status.contains_key(c.id.as_str());
let status = if let Some(ns) = node_status.get(c.id.as_str()) {
DagStatus::from_node_status(*ns)
} else {
DagStatus::from_lane(c.lane)
};
let staleness_hours = parse_ts(&c.created)
.map(|t| ((max_ts - t) as f64 / 3600.0).max(0.0))
.unwrap_or(0.0);
let fanout = if is_task {
out_degree.get(c.id.as_str()).copied().unwrap_or(0)
} else {
c.plan_ids.len() as u64
};
PlanDagNode {
id: c.id.clone(),
title: c.title.clone(),
group: if is_task {
c.plan_ids.first().cloned()
} else {
None
},
status,
is_bug: c.card_type == CardType::Bug,
staleness_hours,
weight_bytes: c.body.len() as u64,
fanout,
}
})
.collect();
let mut edges: Vec<PlanDagEdge> = edge_set
.into_iter()
.map(|(from, to, kind)| PlanDagEdge { from, to, kind })
.collect();
edges.sort_by(|a, b| (&a.from, &a.to, a.kind).cmp(&(&b.from, &b.to, b.kind)));
Self {
nodes,
edges,
sel_plan: sel_plan.map(str::to_string),
}
}
/// Node count.
#[must_use]
pub fn node_count(&self) -> usize {
self.nodes.len()
}
/// Edge count.
#[must_use]
pub fn edge_count(&self) -> usize {
self.edges.len()
}
/// How many edges are decomposition (mother→child) edges.
#[must_use]
pub fn mother_edges(&self) -> usize {
self.edges
.iter()
.filter(|e| e.kind == EdgeKind::Mother)
.count()
}
/// How many edges are task→task dependency edges.
#[must_use]
pub fn dep_edges(&self) -> usize {
self.edges
.iter()
.filter(|e| e.kind == EdgeKind::Dep)
.count()
}
/// A topological order of the node ids (dependencies before dependents:
/// `from` before `to` for every edge). `Err(cycle_node_ids)` if the graph has a
/// cycle — a mother idea's decomposition is always acyclic, so this only trips
/// on a malformed dependency edge set.
pub fn topo_order(&self) -> Result<Vec<String>, Vec<String>> {
let mut g: DiGraph<String, ()> = DiGraph::new();
let mut idx: BTreeMap<&str, NodeIndex> = BTreeMap::new();
for n in &self.nodes {
let ni = g.add_node(n.id.clone());
idx.insert(n.id.as_str(), ni);
}
for e in &self.edges {
if let (Some(&a), Some(&b)) = (idx.get(e.from.as_str()), idx.get(e.to.as_str())) {
g.add_edge(a, b, ());
}
}
match toposort(&g, None) {
Ok(order) => Ok(order.into_iter().map(|ni| g[ni].clone()).collect()),
Err(cyc) => Err(vec![g[cyc.node_id()].clone()]),
}
}
/// The **longest-path depth** of every node — the length (in edges) of the
/// longest dependency chain that ends at it. A root (no incoming edge) has
/// depth `0`; a node whose deepest predecessor has depth `d` has depth
/// `d + 1`. This is the layout **layer** a renderer places the node in (depth
/// = column, so dependencies always sit left of their dependents) and the
/// scheduling **wave** the flow simulator would advance it on.
///
/// Computed in a single topological pass (dependency-before-dependent), so it
/// is O(V + E). Returns an empty map if the graph has a cycle (no valid
/// layering exists) — mirrors [`topo_order`](PlanDag::topo_order)'s `Err`.
#[must_use]
pub fn depths(&self) -> BTreeMap<String, u64> {
let mut depth: BTreeMap<String, u64> = BTreeMap::new();
let order = match self.topo_order() {
Ok(o) => o,
Err(_) => return BTreeMap::new(),
};
// Predecessors per node, so each node's depth = max(pred depth) + 1.
let mut preds: BTreeMap<&str, Vec<&str>> = BTreeMap::new();
for n in &self.nodes {
preds.entry(n.id.as_str()).or_default();
}
for e in &self.edges {
preds
.entry(e.to.as_str())
.or_default()
.push(e.from.as_str());
}
for id in &order {
let d = preds
.get(id.as_str())
.map(|ps| {
ps.iter()
.map(|p| depth.get(*p).copied().unwrap_or(0) + 1)
.max()
.unwrap_or(0)
})
.unwrap_or(0);
depth.insert(id.clone(), d);
}
depth
}
/// The **critical path** — the longest dependency chain through the whole DAG
/// (mother→child→…→leaf), as an ordered list of node ids. This is the
/// bottleneck lineage: the minimum number of sequential waves the plan can
/// finish in is `critical_path().len().saturating_sub(1)`, and a renderer can
/// highlight it as the schedule-defining path.
///
/// Ties (several chains of equal maximum length) break deterministically on
/// the endpoint's id, then on each predecessor's id, so the result is stable
/// (FC-7). Empty for an empty or cyclic graph.
#[must_use]
pub fn critical_path(&self) -> Vec<String> {
let order = match self.topo_order() {
Ok(o) => o,
Err(_) => return Vec::new(),
};
let mut preds: BTreeMap<&str, Vec<&str>> = BTreeMap::new();
for n in &self.nodes {
preds.entry(n.id.as_str()).or_default();
}
for e in &self.edges {
preds
.entry(e.to.as_str())
.or_default()
.push(e.from.as_str());
}
// best[n] = longest chain length ending at n; from[n] = the chosen pred.
let mut best: BTreeMap<&str, u64> = BTreeMap::new();
let mut from: BTreeMap<&str, Option<&str>> = BTreeMap::new();
for id in &order {
let mut chosen: Option<&str> = None;
let mut chosen_len = 0u64;
if let Some(ps) = preds.get(id.as_str()) {
for p in ps {
let cand = best.get(*p).copied().unwrap_or(0) + 1;
// Longer wins; equal length breaks on the smaller pred id
// (deterministic tie-break).
if cand > chosen_len
|| (cand == chosen_len && chosen.map(|c| *p < c).unwrap_or(false))
{
chosen_len = cand;
chosen = Some(*p);
}
}
}
best.insert(id.as_str(), chosen_len);
from.insert(id.as_str(), chosen);
}
// Pick the endpoint with the greatest chain length; tie-break on id.
let mut end: Option<&str> = None;
let mut end_len = 0u64;
for id in &order {
let l = best.get(id.as_str()).copied().unwrap_or(0);
if l > end_len || (l == end_len && end.map(|e| id.as_str() < e).unwrap_or(true)) {
end_len = l;
end = Some(id.as_str());
}
}
// Walk the predecessor chain back from the endpoint, then reverse.
let mut path: Vec<String> = Vec::new();
let mut cur = end;
while let Some(c) = cur {
path.push(c.to_string());
cur = from.get(c).copied().flatten();
}
path.reverse();
path
}
/// Number of edges on the [`critical_path`](PlanDag::critical_path) — the DAG's
/// depth, i.e. the minimum number of sequential waves the plan can finish in.
/// `0` for an empty graph or a single isolated node.
#[must_use]
pub fn critical_path_len(&self) -> usize {
self.critical_path().len().saturating_sub(1)
}
/// The **Dep-only critical path** — the longest chain that follows ONLY task→task
/// [`Dep`](EdgeKind::Dep) edges, ignoring [`Mother`](EdgeKind::Mother) decomposition
/// lineage entirely. Where [`critical_path`](PlanDag::critical_path) measures the
/// deepest idea→…→leaf lineage (which always spends its first edge on the mother→child
/// hop), this one measures the pure *dependency* bottleneck: how many task→task waves
/// the work itself imposes, decoupled from how it was decomposed. It is therefore
/// always ≤ the mixed critical path and, whenever the deepest chain runs through the
/// mother, strictly shorter.
///
/// A self-contained twin of [`critical_path`](PlanDag::critical_path): same
/// single-pass topological longest-path walk and the same deterministic tie-break
/// (endpoint id, then predecessor id), but built over the Dep-edge predecessor set
/// only. Every task node — even one with no Dep edge — is a candidate endpoint, so a
/// dependency-free plan yields a single-node path (length 0). Empty for an empty or
/// cyclic graph.
#[must_use]
pub fn critical_path_deps(&self) -> Vec<String> {
let order = match self.topo_order() {
Ok(o) => o,
Err(_) => return Vec::new(),
};
// Predecessors via Dep edges ONLY — the Mother lineage is invisible here.
let mut preds: BTreeMap<&str, Vec<&str>> = BTreeMap::new();
for n in &self.nodes {
preds.entry(n.id.as_str()).or_default();
}
for e in &self.edges {
if e.kind == EdgeKind::Dep {
preds
.entry(e.to.as_str())
.or_default()
.push(e.from.as_str());
}
}
// best[n] = longest Dep chain length ending at n; from[n] = the chosen pred.
let mut best: BTreeMap<&str, u64> = BTreeMap::new();
let mut from: BTreeMap<&str, Option<&str>> = BTreeMap::new();
for id in &order {
let mut chosen: Option<&str> = None;
let mut chosen_len = 0u64;
if let Some(ps) = preds.get(id.as_str()) {
for p in ps {
let cand = best.get(*p).copied().unwrap_or(0) + 1;
if cand > chosen_len
|| (cand == chosen_len && chosen.map(|c| *p < c).unwrap_or(false))
{
chosen_len = cand;
chosen = Some(*p);
}
}
}
best.insert(id.as_str(), chosen_len);
from.insert(id.as_str(), chosen);
}
// Pick the endpoint with the greatest Dep-chain length; tie-break on id.
let mut end: Option<&str> = None;
let mut end_len = 0u64;
for id in &order {
let l = best.get(id.as_str()).copied().unwrap_or(0);
if l > end_len || (l == end_len && end.map(|e| id.as_str() < e).unwrap_or(true)) {
end_len = l;
end = Some(id.as_str());
}
}
// Walk the Dep-predecessor chain back from the endpoint, then reverse.
let mut path: Vec<String> = Vec::new();
let mut cur = end;
while let Some(c) = cur {
path.push(c.to_string());
cur = from.get(c).copied().flatten();
}
path.reverse();
path
}
/// Number of edges on the [`critical_path_deps`](PlanDag::critical_path_deps) — the
/// pure task→task dependency depth, i.e. the sequential wave count the *dependencies*
/// alone impose (mother lineage excluded). `0` for an empty graph or a plan with no
/// dependency edges. Always ≤ [`critical_path_len`](PlanDag::critical_path_len).
#[must_use]
pub fn critical_path_deps_len(&self) -> usize {
self.critical_path_deps().len().saturating_sub(1)
}
/// The **layered layout** — every node id bucketed by its longest-path
/// [`depth`](PlanDag::depths), one inner vector per layer in ascending depth
/// order (the index IS the layer = the render column, dependencies always to
/// the left of their dependents). Within a layer the ids are sorted, so the
/// layout is deterministic (FC-7).
///
/// `layers()[d]` is exactly the set of cards a renderer draws in column `d`;
/// the layer count `layers().len()` is the DAG **height** =
/// [`critical_path_len`](PlanDag::critical_path_len) `+ 1` (the number of nodes
/// on the critical path). Empty for an empty or cyclic graph — no valid layering
/// exists, mirroring [`depths`](PlanDag::depths).
#[must_use]
pub fn layers(&self) -> Vec<Vec<String>> {
let depth = self.depths();
if depth.is_empty() {
return Vec::new();
}
let height = depth.values().copied().max().map_or(0, |m| m as usize + 1);
let mut layers: Vec<Vec<String>> = vec![Vec::new(); height];
for (id, d) in &depth {
layers[*d as usize].push(id.clone());
}
for layer in &mut layers {
layer.sort();
}
layers
}
/// The **peak layout width** — the size of the widest [`layer`](PlanDag::layers),
/// i.e. the most cards the layered layout stacks in a single column. Two nodes
/// share a layer only when neither is on the other's longest path, so this is
/// the peak parallelism the layout surfaces (a lower bound on the DAG's true
/// max antichain) — the count a renderer sizes its canvas by and a scheduler
/// reads as "the most concurrent work the plan exposes at one depth". `0` for
/// an empty or cyclic graph; `1` for a pure chain (one node per layer).
#[must_use]
pub fn width(&self) -> usize {
self.layers().iter().map(Vec::len).max().unwrap_or(0)
}
/// The **slack (float)** of every node — how many layout layers it may slip
/// without extending the DAG's critical depth. In CPM terms its *earliest*
/// layer is its longest-path [`depth`](PlanDag::depths) `e(n)`, and the longest
/// chain *from* it to a sink is `r(n)`; the whole plan's depth is `D`
/// (= [`critical_path_len`](PlanDag::critical_path_len)), so a node can occupy
/// any layer in `e(n) ..= D - r(n)` and its slack is `slack(n) = D - e(n) - r(n)`.
///
/// A node with **zero slack lies on some longest path** — it is *critical*:
/// delaying it delays the whole plan. A node with slack `k` has `k` waves of
/// scheduling freedom a dispatcher can spend (start it late, or overlap it).
/// Where [`critical_path`](PlanDag::critical_path) returns one chain, this grades
/// EVERY node by tension. Computed in one forward + one reverse topological pass,
/// O(V + E). Empty for an empty or cyclic graph (mirrors [`depths`](PlanDag::depths)).
#[must_use]
pub fn slack(&self) -> BTreeMap<String, u64> {
let depth = self.depths();
if depth.is_empty() {
return BTreeMap::new();
}
let order = match self.topo_order() {
Ok(o) => o,
Err(_) => return BTreeMap::new(),
};
let d_max = depth.values().copied().max().unwrap_or(0);
// Successors per node — for the reverse longest-path (n → sink) pass.
let mut succ: BTreeMap<&str, Vec<&str>> = BTreeMap::new();
for n in &self.nodes {
succ.entry(n.id.as_str()).or_default();
}
for e in &self.edges {
succ.entry(e.from.as_str()).or_default().push(e.to.as_str());
}
// rdepth[n] = longest chain (edges) from n to a sink; fold in REVERSE topo
// order so every successor is settled before n.
let mut rdepth: BTreeMap<&str, u64> = BTreeMap::new();
for id in order.iter().rev() {
let r = succ
.get(id.as_str())
.map(|ss| {
ss.iter()
.map(|s| rdepth.get(*s).copied().unwrap_or(0) + 1)
.max()
.unwrap_or(0)
})
.unwrap_or(0);
rdepth.insert(id.as_str(), r);
}
// slack = D - earliest - rdepth. The longest path THROUGH any node is
// e(n) + r(n) ≤ D, so the subtraction never underflows.
let mut slack: BTreeMap<String, u64> = BTreeMap::new();
for n in &self.nodes {
let e = depth.get(n.id.as_str()).copied().unwrap_or(0);
let r = rdepth.get(n.id.as_str()).copied().unwrap_or(0);
slack.insert(n.id.clone(), d_max - e - r);
}
slack
}
/// The **critical set** — every node with zero [`slack`](PlanDag::slack), sorted
/// by id. These are the nodes lying on *some* longest path: a **superset** of the
/// single [`critical_path`](PlanDag::critical_path) chain whenever the DAG has
/// several equally-long paths (a diamond has no non-critical node, yet
/// `critical_path` names only one branch). None of them can slip without extending
/// the plan — a renderer highlights them, a replan protects them. Empty for an
/// empty or cyclic graph.
#[must_use]
pub fn critical_nodes(&self) -> Vec<String> {
let mut out: Vec<String> = self
.slack()
.into_iter()
.filter(|(_, s)| *s == 0)
.map(|(id, _)| id)
.collect();
out.sort();
out
}
/// The **bottleneck ranking** — every node scored by how much work it
/// constrains, most-constraining first. Where [`critical_path`](PlanDag::critical_path)
/// answers "how *long* is the schedule", this answers "*which node* unblocks the
/// most work": the constraint points a human should parallelise or clear next.
///
/// Each [`Bottleneck`] carries the node's [`fan_in`](Bottleneck::fan_in) /
/// [`fan_out`](Bottleneck::fan_out) degrees and its
/// [`downstream`](Bottleneck::downstream) — the count of **all** nodes reachable
/// downstream (its transitive-descendant cone), i.e. the work stalled behind it.
/// Rows are ranked `downstream` desc, then `fan_out` desc, then `fan_in` desc,
/// then id asc — so `bottlenecks().first()` is the single node whose completion
/// unblocks the most work (typically the mother idea's decomposition root, then
/// the deepest task join). A renderer highlights the head; a scheduler reads
/// `downstream` as the priority weight.
///
/// Every node appears (an isolated leaf ranks last with all-zero scores), so the
/// full ranking is available; filter to `group.is_some()` for task-only
/// constraints. Spans both edge kinds (Mother decomposition + Dep dependency), so
/// "downstream" is everything a node gates, matching [`critical_path`](PlanDag::critical_path).
/// Computed in one reverse-topological pass accumulating each node's descendant
/// set. Empty for an empty or cyclic graph (mirrors [`depths`](PlanDag::depths)).
#[must_use]
pub fn bottlenecks(&self) -> Vec<Bottleneck> {
let order = match self.topo_order() {
Ok(o) => o,
Err(_) => return Vec::new(),
};
// Local degrees + successor adjacency.
let mut fan_in: BTreeMap<&str, u64> = BTreeMap::new();
let mut fan_out: BTreeMap<&str, u64> = BTreeMap::new();
let mut succ: BTreeMap<&str, Vec<&str>> = BTreeMap::new();
for n in &self.nodes {
fan_in.entry(n.id.as_str()).or_default();
fan_out.entry(n.id.as_str()).or_default();
succ.entry(n.id.as_str()).or_default();
}
for e in &self.edges {
*fan_out.entry(e.from.as_str()).or_default() += 1;
*fan_in.entry(e.to.as_str()).or_default() += 1;
succ.entry(e.from.as_str()).or_default().push(e.to.as_str());
}
// Transitive descendants: fold in REVERSE topological order so every
// successor's cone is settled before its predecessor unions it in.
// desc[n] = every node reachable out of n (n itself excluded).
let mut desc: BTreeMap<String, BTreeSet<&str>> = BTreeMap::new();
for id in order.iter().rev() {
let mut cone: BTreeSet<&str> = BTreeSet::new();
if let Some(ss) = succ.get(id.as_str()) {
for s in ss {
cone.insert(*s);
if let Some(sd) = desc.get(*s) {
cone.extend(sd.iter().copied());
}
}
}
desc.insert(id.clone(), cone);
}
let mut out: Vec<Bottleneck> = self
.nodes
.iter()
.map(|n| {
let id = n.id.as_str();
Bottleneck {
id: n.id.clone(),
fan_in: fan_in.get(id).copied().unwrap_or(0),
fan_out: fan_out.get(id).copied().unwrap_or(0),
downstream: desc.get(id).map_or(0, |c| c.len() as u64),
}
})
.collect();
// Rank: most work stalled behind it first; deterministic tie-break on id.
out.sort_by(|a, b| {
b.downstream
.cmp(&a.downstream)
.then_with(|| b.fan_out.cmp(&a.fan_out))
.then_with(|| b.fan_in.cmp(&a.fan_in))
.then_with(|| a.id.cmp(&b.id))
});
out
}
/// Every node reachable by following edges **out of** `id` — its whole
/// downstream cone, `id` itself excluded — sorted.
///
/// The per-node twin of the count [`bottlenecks`](PlanDag::bottlenecks)
/// reports as [`Bottleneck::downstream`], exposed because the
/// [`wiring`](crate::wiring) layer joins it to the Component-DAG: the plan
/// cone under a node decides which components the rest of that plan touches
/// (see [`Wiring::cone_blast_radius`](crate::wiring::Wiring::cone_blast_radius)).
///
/// A plain BFS with a visited set rather than the topological fold
/// `bottlenecks` uses, so it still answers on a malformed **cyclic** edge set
/// (where `topo_order` fails and `bottlenecks` returns nothing) instead of
/// silently reporting an empty cone. An unknown `id` has an empty cone.
#[must_use]
pub fn descendants(&self, id: &str) -> Vec<String> {
let mut succ: BTreeMap<&str, Vec<&str>> = BTreeMap::new();
for e in &self.edges {
succ.entry(e.from.as_str()).or_default().push(e.to.as_str());
}
let mut seen: BTreeSet<&str> = BTreeSet::new();
let mut stack: Vec<&str> = succ.get(id).cloned().unwrap_or_default();
while let Some(cur) = stack.pop() {
if cur == id || !seen.insert(cur) {
continue;
}
if let Some(ss) = succ.get(cur) {
stack.extend(ss.iter().copied());
}
}
seen.into_iter().map(str::to_string).collect()
}
/// **Diff two Plan-DAG revisions** — `self` is the new graph, `baseline` the old
/// one — into a [`PlanDiff`]: nodes/edges added (in `self`, not `baseline`),
/// removed (in `baseline`, not `self`), and the nodes present in **both** whose
/// payload or wiring changed. This is the DAG twin of the coverage / bench diff
/// gates: a replan step reads it to show "what changed" between plan revisions.
///
/// A surviving node counts as **changed** when any of its content fields
/// (`title`, `status`, `group`, `is_bug`) differs, or its **deps** — the set of
/// edges incident to it (either direction, with kind) — differs. The derived
/// layout metrics ([`staleness_hours`](PlanDagNode::staleness_hours),
/// [`weight_bytes`](PlanDagNode::weight_bytes), [`fanout`](PlanDagNode::fanout))
/// are deliberately NOT diffed: staleness is a clock artefact and fanout/deps are
/// already fully reported by the edge diff + the `"deps"` aspect, so including
/// them would only add noise.
///
/// Deterministic (FC-7): every list is id/tuple sorted, and each
/// [`NodeChange::changed`] is in the fixed order `title, status, group, is_bug,
/// deps`. An identical (re-projected) revision diffs to
/// [`PlanDiff::is_empty`]-`true`.
#[must_use]
pub fn diff(&self, baseline: &PlanDag) -> PlanDiff {
let cur: BTreeMap<&str, &PlanDagNode> =
self.nodes.iter().map(|n| (n.id.as_str(), n)).collect();
let old: BTreeMap<&str, &PlanDagNode> =
baseline.nodes.iter().map(|n| (n.id.as_str(), n)).collect();
let added_nodes: Vec<String> = cur
.keys()
.filter(|k| !old.contains_key(*k))
.map(|k| (*k).to_string())
.collect();
let removed_nodes: Vec<String> = old
.keys()
.filter(|k| !cur.contains_key(*k))
.map(|k| (*k).to_string())
.collect();
let cur_edges: BTreeSet<(&str, &str, EdgeKind)> = self
.edges
.iter()
.map(|e| (e.from.as_str(), e.to.as_str(), e.kind))
.collect();
let old_edges: BTreeSet<(&str, &str, EdgeKind)> = baseline
.edges
.iter()
.map(|e| (e.from.as_str(), e.to.as_str(), e.kind))
.collect();
let to_edge = |(f, t, k): &(&str, &str, EdgeKind)| PlanDagEdge {
from: f.to_string(),
to: t.to_string(),
kind: *k,
};
let added_edges: Vec<PlanDagEdge> = cur_edges.difference(&old_edges).map(to_edge).collect();
let removed_edges: Vec<PlanDagEdge> =
old_edges.difference(&cur_edges).map(to_edge).collect();
// The set of edges incident to `id` (either direction) — a node's wiring.
let incident = |edges: &BTreeSet<(&str, &str, EdgeKind)>,
id: &str|
-> BTreeSet<(String, String, EdgeKind)> {
edges
.iter()
.filter(|(f, t, _)| *f == id || *t == id)
.map(|(f, t, k)| (f.to_string(), t.to_string(), *k))
.collect()
};
// Nodes present in both, iterated in sorted id order (BTreeMap).
let mut changed_nodes: Vec<NodeChange> = Vec::new();
for (id, n) in &cur {
let Some(o) = old.get(id) else { continue };
let mut changed: Vec<String> = Vec::new();
if n.title != o.title {
changed.push("title".to_string());
}
if n.status != o.status {
changed.push("status".to_string());
}
if n.group != o.group {
changed.push("group".to_string());
}
if n.is_bug != o.is_bug {
changed.push("is_bug".to_string());
}
if incident(&cur_edges, id) != incident(&old_edges, id) {
changed.push("deps".to_string());
}
if !changed.is_empty() {
changed_nodes.push(NodeChange {
id: (*id).to_string(),
changed,
});
}
}
PlanDiff {
added_nodes,
removed_nodes,
added_edges,
removed_edges,
changed_nodes,
}
}
/// The robot-observable introspection blob (LAW 6): the scope, counts, per-node
/// metadata, and the topological order — everything a headless test asserts.
#[must_use]
pub fn state_json(&self) -> serde_json::Value {
serde_json::json!({
"sel_plan": self.sel_plan,
"node_count": self.node_count(),
"edge_count": self.edge_count(),
"mother_edges": self.mother_edges(),
"dep_edges": self.dep_edges(),
"critical_path": self.critical_path(),
"critical_path_len": self.critical_path_len(),
"critical_path_deps": self.critical_path_deps(),
"critical_path_deps_len": self.critical_path_deps_len(),
"depths": self.depths(),
"layers": self.layers(),
"width": self.width(),
"slack": self.slack(),
"critical_nodes": self.critical_nodes(),
"bottlenecks": self.bottlenecks().iter().map(Bottleneck::state_json).collect::<Vec<_>>(),
"topo_order": self.topo_order().ok(),
"nodes": self.nodes.iter().map(|n| serde_json::json!({
"id": n.id,
"title": n.title,
"group": n.group,
"status": n.status.as_str(),
"is_bug": n.is_bug,
"staleness_hours": n.staleness_hours,
"weight_bytes": n.weight_bytes,
"fanout": n.fanout,
})).collect::<Vec<_>>(),
"edges": self.edges.iter().map(|e| serde_json::json!({
"from": e.from, "to": e.to, "kind": e.kind.as_str(),
})).collect::<Vec<_>>(),
})
}
}
/// Parse an RFC3339 timestamp to epoch seconds (for the clock-free staleness diff).
/// Unparseable → `None` (that card contributes 0 staleness).
fn parse_ts(s: &str) -> Option<i64> {
chrono::DateTime::parse_from_rfc3339(s)
.ok()
.map(|d| d.timestamp())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::event::{Event, ItemKind, PlanStatus};
use crate::ids::{IdeaId, NodeId, PlanId};
use chrono::{Duration, Utc};
/// One idea decomposed into a 3-node chain a→b→c in an Active plan, with `a`
/// Done and `b` Ready — the shape `nornir funnel prompt` produces.
fn funnel_with_chain() -> Funnel {
let mut f = Funnel::default();
let idea = IdeaId::seq(1);
let plan = PlanId::seq(1);
let base = Utc::now() - Duration::hours(10);
f.apply(&Event::IdeaSubmitted {
id: idea.clone(),
source: "human:alice".into(),
text: "build the planner".into(),
refs: vec![],
item_kind: ItemKind::Idea,
ts: base,
})
.unwrap();
f.apply(&Event::PlanCreated {
id: plan.clone(),
idea_id: idea.clone(),
summary: "decompose → 3 subtasks".into(),
planner: "decompose".into(),
ts: base,
})
.unwrap();
f.apply(&Event::PlanStatusChanged {
plan_id: plan.clone(),
status: PlanStatus::Active,
why: None,
ts: base,
})
.unwrap();
for (i, (title, kind)) in [
("implement foo", "code:write"),
("implement bar", "code:fix"),
("test baz", "test:write"),
]
.iter()
.enumerate()
{
f.apply(&Event::NodeAdded {
plan_id: plan.clone(),
node_id: NodeId::seq(i as u64 + 1),
kind: (*kind).into(),
params: Default::default(),
targets: vec![],
prompt_excerpt: Some((*title).into()),
ts: base + Duration::hours(i as i64),
})
.unwrap();
}
// a → b → c dependency chain.
f.apply(&Event::EdgeAdded {
plan_id: plan.clone(),
from_node: NodeId::seq(1),
to_node: NodeId::seq(2),
ts: base,
})
.unwrap();
f.apply(&Event::EdgeAdded {
plan_id: plan.clone(),
from_node: NodeId::seq(2),
to_node: NodeId::seq(3),
ts: base,
})
.unwrap();
// a Done, b InProgress.
f.apply(&Event::NodeStatusChanged {
plan_id: plan.clone(),
node_id: NodeId::seq(1),
status: NodeStatus::Done,
why: None,
ts: base,
})
.unwrap();
f.apply(&Event::NodeStatusChanged {
plan_id: plan.clone(),
node_id: NodeId::seq(2),
status: NodeStatus::InProgress,
why: None,
ts: base,
})
.unwrap();
f
}
#[test]
fn assembles_mother_and_dependency_edges() {
let f = funnel_with_chain();
let dag = PlanDag::from_funnel(&f, None);
// mother idea i-001 + 3 task nodes n-001/2/3.
assert_eq!(dag.node_count(), 4, "mother + 3 task cards");
// 3 mother→child + 2 dep (a→b, b→c).
assert_eq!(
dag.mother_edges(),
3,
"one decomposition edge per task node"
);
assert_eq!(dag.dep_edges(), 2, "a→b, b→c");
assert_eq!(dag.edge_count(), 5);
// status mapping came from the real NodeStatus (not the lane).
let n1 = dag.nodes.iter().find(|n| n.id == "n-001").unwrap();
assert_eq!(n1.status, DagStatus::Done, "n-001 is Done");
let n2 = dag.nodes.iter().find(|n| n.id == "n-002").unwrap();
assert_eq!(n2.status, DagStatus::Running, "n-002 InProgress → running");
assert!(n2.is_bug, "code:fix → Bug card");
// grouping: task nodes carry their plan id, the mother does not.
assert_eq!(n1.group.as_deref(), Some("p-001"));
let mother = dag.nodes.iter().find(|n| n.id == "i-001").unwrap();
assert_eq!(mother.group, None);
assert_eq!(mother.fanout, 1, "the idea links one plan");
}
#[test]
fn topo_order_puts_dependencies_before_dependents() {
let f = funnel_with_chain();
let dag = PlanDag::from_funnel(&f, None);
let order = dag.topo_order().expect("acyclic");
let pos = |id: &str| order.iter().position(|x| x == id).unwrap();
// mother before its children; a before b before c.
assert!(pos("i-001") < pos("n-001"));
assert!(pos("n-001") < pos("n-002"), "a before b");
assert!(pos("n-002") < pos("n-003"), "b before c");
}
#[test]
fn scoping_to_a_plan_keeps_its_mother_and_children() {
let f = funnel_with_chain();
let dag = PlanDag::from_funnel(&f, Some("p-001"));
assert_eq!(dag.sel_plan.as_deref(), Some("p-001"));
assert_eq!(dag.node_count(), 4);
// Scoping to a non-existent plan yields an empty graph.
let empty = PlanDag::from_funnel(&f, Some("p-999"));
assert_eq!(empty.node_count(), 0);
assert_eq!(empty.edge_count(), 0);
}
#[test]
fn state_json_is_deterministic_and_shaped() {
let f = funnel_with_chain();
let j1 = PlanDag::from_funnel(&f, None).state_json();
let j2 = PlanDag::from_funnel(&f, None).state_json();
assert_eq!(
j1.to_string(),
j2.to_string(),
"bit-identical on re-projection"
);
assert_eq!(j1["node_count"], 4);
assert_eq!(j1["edge_count"], 5);
assert!(
j1["topo_order"].is_array(),
"acyclic → a topo order is present"
);
// staleness: the oldest card (the mother, submitted first) is the hottest.
let nodes = j1["nodes"].as_array().unwrap();
let mother = nodes.iter().find(|n| n["id"] == "i-001").unwrap();
assert!(
mother["staleness_hours"].as_f64().unwrap() >= 2.0,
"the mother is the oldest"
);
}
#[test]
fn empty_funnel_is_an_empty_graph() {
let dag = PlanDag::from_funnel(&Funnel::default(), None);
assert_eq!(dag.node_count(), 0);
assert_eq!(dag.edge_count(), 0);
assert!(dag.topo_order().unwrap().is_empty());
// The new DAG vocabulary degrades cleanly on an empty graph.
assert!(dag.depths().is_empty());
assert!(dag.critical_path().is_empty());
assert_eq!(dag.critical_path_len(), 0);
assert!(dag.critical_path_deps().is_empty());
assert_eq!(dag.critical_path_deps_len(), 0);
// The layered layout is likewise empty (no valid layering).
assert!(dag.layers().is_empty());
assert_eq!(dag.width(), 0);
// Slack / critical-set degrade to empty too.
assert!(dag.slack().is_empty());
assert!(dag.critical_nodes().is_empty());
// No nodes → no bottleneck ranking.
assert!(dag.bottlenecks().is_empty());
}
#[test]
fn slack_grades_tension_and_critical_nodes_are_zero_slack() {
// Mother → all of n-001..n-004; the only Dep chain is n-002 → n-003 → n-004;
// n-001 is a lone short branch. D = 3.
// depth: i-001=0 n-001=1 n-002=1 n-003=2 n-004=3
// rdepth: i-001=3 n-001=0 n-002=2 n-003=1 n-004=0
// slack = 3 - depth - rdepth → i-001=0 n-001=2 n-002=0 n-003=0 n-004=0
let mut f = Funnel::default();
let now = Utc::now();
let idea = IdeaId::seq(1);
let plan = PlanId::seq(1);
f.apply(&Event::IdeaSubmitted {
id: idea.clone(),
source: "x".into(),
text: "branchy".into(),
refs: vec![],
item_kind: ItemKind::Idea,
ts: now,
})
.unwrap();
f.apply(&Event::PlanCreated {
id: plan.clone(),
idea_id: idea,
summary: "branchy".into(),
planner: "x".into(),
ts: now,
})
.unwrap();
for i in 1..=4u64 {
f.apply(&Event::NodeAdded {
plan_id: plan.clone(),
node_id: NodeId::seq(i),
kind: "code:write".into(),
params: Default::default(),
targets: vec![],
prompt_excerpt: Some(format!("t{i}")),
ts: now,
})
.unwrap();
}
f.apply(&Event::EdgeAdded {
plan_id: plan.clone(),
from_node: NodeId::seq(2),
to_node: NodeId::seq(3),
ts: now,
})
.unwrap();
f.apply(&Event::EdgeAdded {
plan_id: plan.clone(),
from_node: NodeId::seq(3),
to_node: NodeId::seq(4),
ts: now,
})
.unwrap();
let dag = PlanDag::from_funnel(&f, None);
let s = dag.slack();
// The lone short branch can slip 2 layers — the slack quantification.
assert_eq!(s["n-001"], 2, "n-001 has 2 waves of scheduling freedom");
// RED-when-broken: the INTERIOR dep nodes are critical (zero slack). Only a
// correct reverse (n → sink) longest-path pass yields this — a naive
// `D - depth` (rdepth ignored) would score n-002=2, n-003=1, and only the
// leaf as critical, failing these three assertions.
assert_eq!(s["i-001"], 0, "the mother is on the longest lineage");
assert_eq!(s["n-002"], 0, "interior dep node is critical");
assert_eq!(s["n-003"], 0, "interior dep node is critical");
assert_eq!(s["n-004"], 0, "the leaf is critical");
// critical_nodes = every zero-slack node, sorted; excludes the slack branch.
assert_eq!(
dag.critical_nodes(),
vec!["i-001", "n-002", "n-003", "n-004"]
);
assert!(!dag.critical_nodes().contains(&"n-001".to_string()));
// Invariant: every node on the single critical_path() has zero slack.
for id in dag.critical_path() {
assert_eq!(
s[&id], 0,
"{id} lies on the critical path, so its slack is 0"
);
}
// Determinism (FC-7).
assert_eq!(PlanDag::from_funnel(&f, None).slack(), s);
}
#[test]
fn critical_nodes_is_a_superset_of_the_single_critical_path() {
// A diamond: mother → n-001, mother → n-002, n-001 → n-003, n-002 → n-003.
// Both branches are equally long, so NO node has slack — every node is
// critical — yet critical_path() can name only one of the two middle nodes.
let mut f = Funnel::default();
let now = Utc::now();
let idea = IdeaId::seq(1);
let plan = PlanId::seq(1);
f.apply(&Event::IdeaSubmitted {
id: idea.clone(),
source: "x".into(),
text: "diamond".into(),
refs: vec![],
item_kind: ItemKind::Idea,
ts: now,
})
.unwrap();
f.apply(&Event::PlanCreated {
id: plan.clone(),
idea_id: idea,
summary: "diamond".into(),
planner: "x".into(),
ts: now,
})
.unwrap();
for i in 1..=3u64 {
f.apply(&Event::NodeAdded {
plan_id: plan.clone(),
node_id: NodeId::seq(i),
kind: "code:write".into(),
params: Default::default(),
targets: vec![],
prompt_excerpt: Some(format!("t{i}")),
ts: now,
})
.unwrap();
}
f.apply(&Event::EdgeAdded {
plan_id: plan.clone(),
from_node: NodeId::seq(1),
to_node: NodeId::seq(3),
ts: now,
})
.unwrap();
f.apply(&Event::EdgeAdded {
plan_id: plan.clone(),
from_node: NodeId::seq(2),
to_node: NodeId::seq(3),
ts: now,
})
.unwrap();
let dag = PlanDag::from_funnel(&f, None);
// Every node is zero-slack (both branches are longest paths).
assert!(
dag.slack().values().all(|s| *s == 0),
"a balanced diamond has no float"
);
assert_eq!(
dag.critical_nodes(),
vec!["i-001", "n-001", "n-002", "n-003"]
);
// The single critical_path() names just ONE of the two middle nodes, so the
// critical SET is a strict superset here.
let cp = dag.critical_path();
assert_eq!(cp.len(), 3, "one root→mid→sink chain");
assert!(
dag.critical_nodes().len() > cp.len(),
"critical set ⊋ a single critical path"
);
for id in &cp {
assert!(
dag.critical_nodes().contains(id),
"{id} on the path is in the critical set"
);
}
}
/// Build a funnel whose critical structure separates *fan-out* from
/// *downstream reach*: a WIDE-but-shallow node (`n-001` → three leaves) versus a
/// NARROW-but-deep node (`n-005` heading a 4-long chain). When `deep_chain` is
/// false the deep chain's first edge is dropped, collapsing `n-005`'s cone — the
/// mutation the ranking must react to.
/// mother i-001 → n-001, n-005
/// n-001 → n-002, n-003, n-004 (fan_out 3, downstream 3)
/// n-005 → n-006 → n-007 → n-008 → n-009 (fan_out 1, downstream 4 when deep)
fn funnel_wide_vs_deep(deep_chain: bool) -> Funnel {
let mut f = Funnel::default();
let now = Utc::now();
let idea = IdeaId::seq(1);
let plan = PlanId::seq(1);
f.apply(&Event::IdeaSubmitted {
id: idea.clone(),
source: "x".into(),
text: "wide vs deep".into(),
refs: vec![],
item_kind: ItemKind::Idea,
ts: now,
})
.unwrap();
f.apply(&Event::PlanCreated {
id: plan.clone(),
idea_id: idea,
summary: "b".into(),
planner: "x".into(),
ts: now,
})
.unwrap();
for i in 1..=9u64 {
f.apply(&Event::NodeAdded {
plan_id: plan.clone(),
node_id: NodeId::seq(i),
kind: "code:write".into(),
params: Default::default(),
targets: vec![],
prompt_excerpt: Some(format!("t{i}")),
ts: now,
})
.unwrap();
}
let edge = |f: &mut Funnel, a: u64, b: u64| {
f.apply(&Event::EdgeAdded {
plan_id: plan.clone(),
from_node: NodeId::seq(a),
to_node: NodeId::seq(b),
ts: now,
})
.unwrap();
};
// Wide, shallow fan-out from n-001.
edge(&mut f, 1, 2);
edge(&mut f, 1, 3);
edge(&mut f, 1, 4);
// Narrow, deep chain from n-005 (its head edge optional → the mutation).
if deep_chain {
edge(&mut f, 5, 6);
}
edge(&mut f, 6, 7);
edge(&mut f, 7, 8);
edge(&mut f, 8, 9);
f
}
#[test]
fn bottlenecks_rank_by_downstream_reach_not_local_fanout() {
let dag = PlanDag::from_funnel(&funnel_wide_vs_deep(true), None);
let bn = dag.bottlenecks();
let row = |id: &str| {
bn.iter()
.find(|b| b.id == id)
.unwrap_or_else(|| panic!("{id} ranked"))
};
// Exact scores. The wide node has the bigger LOCAL fan-out …
assert_eq!(
(row("n-001").fan_out, row("n-001").downstream),
(3, 3),
"n-001: wide + shallow"
);
// … but the deep node reaches MORE work downstream despite fan_out 1.
assert_eq!(
(row("n-005").fan_out, row("n-005").downstream),
(1, 4),
"n-005: narrow + deep"
);
// The mother gates everything: all 9 tasks are in its cone.
assert_eq!(
row("i-001").downstream,
9,
"the mother idea blocks the whole plan"
);
assert_eq!(
bn[0].id, "i-001",
"the decomposition root is the top constraint"
);
// Ranks are sorted by downstream descending overall.
let pos = |id: &str| bn.iter().position(|b| b.id == id).unwrap();
for w in bn.windows(2) {
assert!(
w[0].downstream >= w[1].downstream,
"ranked by downstream desc"
);
}
// RED-when-broken: among the two task heads the DEEP node outranks the WIDE
// one — downstream reach beats local fan-out. A ranking keyed on fan_out
// first would put n-001 (fan_out 3) above n-005 (fan_out 1) and fail here.
assert!(
pos("n-005") < pos("n-001"),
"deep-reach n-005 outranks wide-fanout n-001"
);
assert!(
row("n-005").downstream > row("n-001").downstream,
"n-005 unblocks more work"
);
assert!(
row("n-005").fan_out < row("n-001").fan_out,
"yet has the smaller local fan-out"
);
// Mutation-verified: drop n-005's head edge → its downstream cone collapses to
// 0 and the wide node now outranks it. The ranking tracks the graph.
let mutated = PlanDag::from_funnel(&funnel_wide_vs_deep(false), None);
let mbn = mutated.bottlenecks();
let mrow = |id: &str| {
mbn.iter()
.find(|b| b.id == id)
.unwrap_or_else(|| panic!("{id} ranked"))
};
assert_eq!(
mrow("n-005").downstream,
0,
"the severed edge empties n-005's cone"
);
let mpos = |id: &str| mbn.iter().position(|b| b.id == id).unwrap();
assert!(
mpos("n-001") < mpos("n-005"),
"after the cut, wide n-001 outranks severed n-005"
);
// Determinism (FC-7).
assert_eq!(
PlanDag::from_funnel(&funnel_wide_vs_deep(true), None).bottlenecks(),
bn
);
}
#[test]
fn layers_group_nodes_by_depth_and_width_is_the_peak() {
// Mother → all of n-001..n-004; the only Dep chain is n-002 → n-003 → n-004;
// n-001 is a lone task. Depths: i-001=0, n-001=1, n-002=1, n-003=2, n-004=3.
let mut f = Funnel::default();
let now = Utc::now();
let idea = IdeaId::seq(1);
let plan = PlanId::seq(1);
f.apply(&Event::IdeaSubmitted {
id: idea.clone(),
source: "x".into(),
text: "branchy".into(),
refs: vec![],
item_kind: ItemKind::Idea,
ts: now,
})
.unwrap();
f.apply(&Event::PlanCreated {
id: plan.clone(),
idea_id: idea,
summary: "branchy".into(),
planner: "x".into(),
ts: now,
})
.unwrap();
for i in 1..=4u64 {
f.apply(&Event::NodeAdded {
plan_id: plan.clone(),
node_id: NodeId::seq(i),
kind: "code:write".into(),
params: Default::default(),
targets: vec![],
prompt_excerpt: Some(format!("t{i}")),
ts: now,
})
.unwrap();
}
f.apply(&Event::EdgeAdded {
plan_id: plan.clone(),
from_node: NodeId::seq(2),
to_node: NodeId::seq(3),
ts: now,
})
.unwrap();
f.apply(&Event::EdgeAdded {
plan_id: plan.clone(),
from_node: NodeId::seq(3),
to_node: NodeId::seq(4),
ts: now,
})
.unwrap();
let dag = PlanDag::from_funnel(&f, None);
let layers = dag.layers();
// One column per depth; ids sorted within a column (n-001 before n-002).
assert_eq!(
layers,
vec![
vec!["i-001".to_string()],
vec!["n-001".to_string(), "n-002".to_string()],
vec!["n-003".to_string()],
vec!["n-004".to_string()],
],
"layers bucket nodes by longest-path depth"
);
// Peak width = the widest LAYER (layer 1 stacks n-001 + n-002), NOT the
// number of layers. RED-when-broken: a 4-layer-tall DAG must still report
// width 2, so a bug returning the height (4) or the critical-path length is
// caught right here.
assert_eq!(dag.width(), 2, "two independent tasks share depth 1");
assert!(
dag.width() < dag.layers().len(),
"peak width 2 is strictly under the height 4"
);
// layers() is the inverse of depths(): every node sits in its depth's column.
let d = dag.depths();
for (col, ids) in layers.iter().enumerate() {
for id in ids {
assert_eq!(d[id] as usize, col, "{id} is drawn in its own depth column");
}
}
// Height = critical-path node count = critical_path_len + 1.
assert_eq!(
dag.layers().len(),
dag.critical_path().len(),
"height = nodes on the critical path"
);
assert_eq!(dag.layers().len(), dag.critical_path_len() + 1);
// Every edge climbs at least one layer — a valid layering.
let layer_of = |id: &str| {
layers
.iter()
.position(|l| l.iter().any(|x| x == id))
.unwrap()
};
for e in &dag.edges {
assert!(
layer_of(&e.from) < layer_of(&e.to),
"edge {}→{} must climb a layer",
e.from,
e.to
);
}
// Deterministic across re-projection (FC-7).
assert_eq!(PlanDag::from_funnel(&f, None).layers(), layers);
}
#[test]
fn critical_path_deps_is_strictly_shorter_than_the_mother_lineage() {
// The mixed critical path spends its first edge on the mother→child hop, so
// the mother lineage is one layer DEEPER than the pure task→task Dep chain.
// The Dep-only variant must drop that mother edge and return the shorter chain.
let f = funnel_with_chain();
let dag = PlanDag::from_funnel(&f, None);
// Mixed: mother → n-001 → n-002 → n-003 (3 edges, mother-rooted).
assert_eq!(
dag.critical_path(),
vec!["i-001", "n-001", "n-002", "n-003"]
);
assert_eq!(dag.critical_path_len(), 3);
// Dep-only: n-001 → n-002 → n-003 (2 edges) — the mother is invisible.
let deps = dag.critical_path_deps();
assert_eq!(
deps,
vec!["n-001", "n-002", "n-003"],
"Dep-only chain drops the mother"
);
assert!(
!deps.iter().any(|id| id.starts_with("i-")),
"no mother/idea node on a Dep-only path"
);
assert_eq!(dag.critical_path_deps_len(), 2);
// RED-when-broken: the Dep-only depth is STRICTLY less than the mixed one
// (exactly the mother→child hop that the Dep variant must ignore).
assert!(
dag.critical_path_deps_len() < dag.critical_path_len(),
"dep depth {} must be strictly under mixed depth {}",
dag.critical_path_deps_len(),
dag.critical_path_len()
);
// Deterministic across re-projection (FC-7).
assert_eq!(PlanDag::from_funnel(&f, None).critical_path_deps(), deps);
}
#[test]
fn critical_path_deps_picks_the_longer_dep_branch_ignoring_mother() {
// Mother points at all four tasks; the only Dep chain is n-002→n-003→n-004.
// The Dep-only path must follow that chain and never touch n-001 or the mother.
let mut f = Funnel::default();
let now = Utc::now();
let idea = IdeaId::seq(1);
let plan = PlanId::seq(1);
f.apply(&Event::IdeaSubmitted {
id: idea.clone(),
source: "x".into(),
text: "branchy".into(),
refs: vec![],
item_kind: ItemKind::Idea,
ts: now,
})
.unwrap();
f.apply(&Event::PlanCreated {
id: plan.clone(),
idea_id: idea,
summary: "branchy".into(),
planner: "x".into(),
ts: now,
})
.unwrap();
for i in 1..=4u64 {
f.apply(&Event::NodeAdded {
plan_id: plan.clone(),
node_id: NodeId::seq(i),
kind: "code:write".into(),
params: Default::default(),
targets: vec![],
prompt_excerpt: Some(format!("t{i}")),
ts: now,
})
.unwrap();
}
f.apply(&Event::EdgeAdded {
plan_id: plan.clone(),
from_node: NodeId::seq(2),
to_node: NodeId::seq(3),
ts: now,
})
.unwrap();
f.apply(&Event::EdgeAdded {
plan_id: plan.clone(),
from_node: NodeId::seq(3),
to_node: NodeId::seq(4),
ts: now,
})
.unwrap();
let dag = PlanDag::from_funnel(&f, None);
assert_eq!(dag.critical_path_deps(), vec!["n-002", "n-003", "n-004"]);
assert_eq!(dag.critical_path_deps_len(), 2);
// Mixed path is mother-rooted and one deeper.
assert_eq!(
dag.critical_path(),
vec!["i-001", "n-002", "n-003", "n-004"]
);
assert!(dag.critical_path_deps_len() < dag.critical_path_len());
}
#[test]
fn depths_layer_dependencies_before_dependents() {
let f = funnel_with_chain();
let dag = PlanDag::from_funnel(&f, None);
let d = dag.depths();
// mother i-001 is a root (0); each chain step is one layer deeper.
assert_eq!(d["i-001"], 0, "the mother idea is a layout root");
assert_eq!(d["n-001"], 1, "mother → n-001");
assert_eq!(d["n-002"], 2, "n-001 → n-002");
assert_eq!(d["n-003"], 3, "n-002 → n-003 — the deepest layer");
// Depth is a valid layering: every edge goes strictly downhill.
for e in &dag.edges {
assert!(
d[&e.from] < d[&e.to],
"edge {}→{} must increase depth",
e.from,
e.to
);
}
}
#[test]
fn critical_path_is_the_longest_chain() {
let f = funnel_with_chain();
let dag = PlanDag::from_funnel(&f, None);
// The single longest lineage: mother → a → b → c (3 edges, 4 nodes).
assert_eq!(
dag.critical_path(),
vec!["i-001", "n-001", "n-002", "n-003"]
);
assert_eq!(
dag.critical_path_len(),
3,
"3 sequential waves define the schedule"
);
// Deterministic across re-projection (FC-7).
let a = PlanDag::from_funnel(&f, None).critical_path();
let b = PlanDag::from_funnel(&f, None).critical_path();
assert_eq!(a, b);
}
#[test]
fn critical_path_picks_the_longer_of_two_branches() {
// A mother with a short branch (n-001) and a long branch
// (n-002 → n-003 → n-004): the critical path must follow the long one.
let mut f = Funnel::default();
let now = Utc::now();
let idea = IdeaId::seq(1);
let plan = PlanId::seq(1);
f.apply(&Event::IdeaSubmitted {
id: idea.clone(),
source: "x".into(),
text: "branchy".into(),
refs: vec![],
item_kind: ItemKind::Idea,
ts: now,
})
.unwrap();
f.apply(&Event::PlanCreated {
id: plan.clone(),
idea_id: idea,
summary: "branchy".into(),
planner: "x".into(),
ts: now,
})
.unwrap();
for i in 1..=4u64 {
f.apply(&Event::NodeAdded {
plan_id: plan.clone(),
node_id: NodeId::seq(i),
kind: "code:write".into(),
params: Default::default(),
targets: vec![],
prompt_excerpt: Some(format!("t{i}")),
ts: now,
})
.unwrap();
}
// long branch n-002 → n-003 → n-004; n-001 is a lone short branch.
f.apply(&Event::EdgeAdded {
plan_id: plan.clone(),
from_node: NodeId::seq(2),
to_node: NodeId::seq(3),
ts: now,
})
.unwrap();
f.apply(&Event::EdgeAdded {
plan_id: plan.clone(),
from_node: NodeId::seq(3),
to_node: NodeId::seq(4),
ts: now,
})
.unwrap();
let dag = PlanDag::from_funnel(&f, None);
let cp = dag.critical_path();
// mother → n-002 → n-003 → n-004 (4 nodes) beats mother → n-001 (2 nodes).
assert_eq!(cp, vec!["i-001", "n-002", "n-003", "n-004"]);
assert_eq!(dag.critical_path_len(), 3);
assert!(
!cp.contains(&"n-001".to_string()),
"the short branch is not critical"
);
}
#[test]
fn state_json_carries_the_dag_vocabulary() {
let f = funnel_with_chain();
let j = PlanDag::from_funnel(&f, None).state_json();
assert_eq!(j["critical_path_len"], 3);
assert_eq!(
j["critical_path"]
.as_array()
.unwrap()
.iter()
.map(|v| v.as_str().unwrap())
.collect::<Vec<_>>(),
["i-001", "n-001", "n-002", "n-003"]
);
assert_eq!(
j["depths"]["n-003"], 3,
"depths surface in the introspection blob"
);
// The Dep-only variant is a distinct key, one wave shorter than the mixed path.
assert_eq!(
j["critical_path_deps_len"], 2,
"task→task depth excludes the mother hop"
);
assert_eq!(
j["critical_path_deps"]
.as_array()
.unwrap()
.iter()
.map(|v| v.as_str().unwrap())
.collect::<Vec<_>>(),
["n-001", "n-002", "n-003"]
);
// The layered layout + peak width surface in the introspection blob. This
// funnel is a pure chain (mother → a → b → c), so each layer holds one node.
assert_eq!(j["width"], 1, "a chain has no concurrency — peak width 1");
let layers = j["layers"].as_array().unwrap();
assert_eq!(layers.len(), 4, "four layers = critical_path_len + 1");
assert_eq!(layers[0][0], "i-001", "the mother is the sole layer-0 node");
assert_eq!(
layers[3][0], "n-003",
"the leaf is alone in the deepest layer"
);
// Slack + critical set surface too. This chain is a single longest path, so
// every node is critical (zero slack).
assert_eq!(j["slack"]["n-001"], 0, "a chain node has no float");
assert_eq!(
j["critical_nodes"]
.as_array()
.unwrap()
.iter()
.map(|v| v.as_str().unwrap())
.collect::<Vec<_>>(),
["i-001", "n-001", "n-002", "n-003"],
"the whole chain is critical"
);
// Bottleneck ranking surfaces too. This chain (i-001→n-001→n-002→n-003) makes
// the mother the top constraint (downstream 3 = the whole chain behind it).
let bn = j["bottlenecks"].as_array().unwrap();
assert_eq!(bn[0]["id"], "i-001", "the mother gates the whole chain");
assert_eq!(
bn[0]["downstream"], 3,
"three tasks stalled behind the mother"
);
assert_eq!(bn.last().unwrap()["id"], "n-003", "the leaf blocks nothing");
assert_eq!(bn.last().unwrap()["downstream"], 0);
}
/// Build one plan revision: mother i-001 → a 3-node plan (n-001..n-003, plus
/// n-004 when `with_n4`), a dependency chain n-001→n-002 always and n-002→n-003
/// only when `keep_edge_23`, with n-001 at `n1_status` and titled `n1_title`.
/// Two calls with different flags stand in for two plan revisions.
fn revision(
with_n4: bool,
keep_edge_23: bool,
n1_status: NodeStatus,
n1_title: &str,
) -> PlanDag {
let mut f = Funnel::default();
let now = Utc::now();
let idea = IdeaId::seq(1);
let plan = PlanId::seq(1);
f.apply(&Event::IdeaSubmitted {
id: idea.clone(),
source: "human:alice".into(),
text: "diffable plan".into(),
refs: vec![],
item_kind: ItemKind::Idea,
ts: now,
})
.unwrap();
f.apply(&Event::PlanCreated {
id: plan.clone(),
idea_id: idea,
summary: "rev".into(),
planner: "x".into(),
ts: now,
})
.unwrap();
f.apply(&Event::PlanStatusChanged {
plan_id: plan.clone(),
status: PlanStatus::Active,
why: None,
ts: now,
})
.unwrap();
let count = if with_n4 { 4 } else { 3 };
for i in 1..=count {
f.apply(&Event::NodeAdded {
plan_id: plan.clone(),
node_id: NodeId::seq(i),
kind: "code:write".into(),
params: Default::default(),
targets: vec![],
prompt_excerpt: Some(if i == 1 {
n1_title.to_string()
} else {
format!("t{i}")
}),
ts: now,
})
.unwrap();
}
f.apply(&Event::EdgeAdded {
plan_id: plan.clone(),
from_node: NodeId::seq(1),
to_node: NodeId::seq(2),
ts: now,
})
.unwrap();
if keep_edge_23 {
f.apply(&Event::EdgeAdded {
plan_id: plan.clone(),
from_node: NodeId::seq(2),
to_node: NodeId::seq(3),
ts: now,
})
.unwrap();
}
if n1_status != NodeStatus::Pending {
f.apply(&Event::NodeStatusChanged {
plan_id: plan.clone(),
node_id: NodeId::seq(1),
status: n1_status,
why: None,
ts: now,
})
.unwrap();
}
PlanDag::from_funnel(&f, None)
}
#[test]
fn diff_classifies_added_removed_and_changed() {
// baseline: n-001..n-003, chain n-001→n-002→n-003, n-001 Pending, "foo".
let baseline = revision(false, true, NodeStatus::Pending, "implement foo");
// revised: + n-004 (added node + a mother→n-004 edge), the n-002→n-003 edge
// dropped (removed edge), n-001 Done + retitled (changed payload).
let revised = revision(true, false, NodeStatus::Done, "implement foo v2");
let d = revised.diff(&baseline);
assert!(!d.is_empty(), "revisions differ");
// Added node: n-004 only.
assert_eq!(
d.added_nodes,
vec!["n-004".to_string()],
"the new task is added"
);
assert!(d.removed_nodes.is_empty(), "nothing removed");
// Added edge: the mother→n-004 decomposition edge. Removed edge: n-002→n-003.
assert_eq!(
d.added_edges,
vec![PlanDagEdge {
from: "i-001".into(),
to: "n-004".into(),
kind: EdgeKind::Mother
}]
);
assert!(
d.removed_edges.contains(&PlanDagEdge {
from: "n-002".into(),
to: "n-003".into(),
kind: EdgeKind::Dep
}),
"the dropped dependency edge is a removed edge"
);
// Changed nodes — look each up by id.
let change = |id: &str| d.changed_nodes.iter().find(|c| c.id == id);
// n-001: payload changed (title + status), wiring unchanged.
let n1 = change("n-001").expect("n-001 changed");
assert_eq!(
n1.changed,
vec!["title".to_string(), "status".to_string()],
"content edit, in fixed order"
);
assert!(
!n1.changed.contains(&"deps".to_string()),
"n-001's wiring is unchanged"
);
// RED-when-broken: n-002 is a changed-DEPS node — it lost its outgoing edge.
// A classifier that ignored incident edges would miss this entirely.
let n2 = change("n-002").expect("n-002 changed");
assert_eq!(
n2.changed,
vec!["deps".to_string()],
"n-002 lost a dependency edge"
);
// n-003 also lost its incoming edge → deps changed.
assert_eq!(
change("n-003").map(|c| c.changed.clone()),
Some(vec!["deps".to_string()])
);
// Deterministic + the state_json seam carries the same delta.
assert_eq!(revised.diff(&baseline), d, "diff is deterministic");
let j = d.state_json();
assert_eq!(j["is_empty"], false);
assert_eq!(j["added_nodes"][0], "n-004");
assert_eq!(j["change_count"], d.change_count() as i64);
}
#[test]
fn identical_revision_diffs_to_empty() {
// Two independently-built but structurally identical revisions.
let a = revision(false, true, NodeStatus::Pending, "implement foo");
let b = revision(false, true, NodeStatus::Pending, "implement foo");
let d = a.diff(&b);
assert!(d.is_empty(), "an unchanged replan has no delta");
assert_eq!(d.change_count(), 0);
assert_eq!(d.state_json()["is_empty"], true);
// A DAG is identical to itself.
assert!(a.diff(&a).is_empty());
}
/// The event log for `funnel_with_chain`, in write order, so a growth timeline
/// can be replayed over the exact same shape the other tests project a snapshot
/// of. Steps: 0 idea, 1 plan, 2 active, 3/4/5 nodes a/b/c, 6/7 dep edges a→b/b→c,
/// 8 a Done.
fn chain_events() -> Vec<Event> {
let base = Utc::now();
let at = |i: i64| base + Duration::seconds(i);
let idea = IdeaId::seq(1);
let plan = PlanId::seq(1);
let mut evs = vec![
Event::IdeaSubmitted {
id: idea.clone(),
source: "human:alice".into(),
text: "build the planner".into(),
refs: vec![],
item_kind: ItemKind::Idea,
ts: at(0),
},
Event::PlanCreated {
id: plan.clone(),
idea_id: idea,
summary: "decompose → 3 subtasks".into(),
planner: "decompose".into(),
ts: at(1),
},
Event::PlanStatusChanged {
plan_id: plan.clone(),
status: PlanStatus::Active,
why: None,
ts: at(2),
},
];
for i in 1..=3u64 {
evs.push(Event::NodeAdded {
plan_id: plan.clone(),
node_id: NodeId::seq(i),
kind: "code:write".into(),
params: Default::default(),
targets: vec![],
prompt_excerpt: Some(format!("t{i}")),
ts: at(2 + i as i64),
});
}
evs.push(Event::EdgeAdded {
plan_id: plan.clone(),
from_node: NodeId::seq(1),
to_node: NodeId::seq(2),
ts: at(6),
});
evs.push(Event::EdgeAdded {
plan_id: plan.clone(),
from_node: NodeId::seq(2),
to_node: NodeId::seq(3),
ts: at(7),
});
evs.push(Event::NodeStatusChanged {
plan_id: plan.clone(),
node_id: NodeId::seq(1),
status: NodeStatus::Done,
why: None,
ts: at(8),
});
evs
}
#[test]
fn growth_stamps_first_appearance_of_each_node_and_edge() {
let evs = chain_events();
let g = DagGrowth::from_events(&evs, None);
assert_eq!(g.steps, evs.len(), "one step per replayed event");
// ── node births: the mother appears at its IdeaSubmitted (step 0), each
// task at its own NodeAdded (steps 3, 4, 5) — birth-ordered.
let nb = |id: &str| {
g.nodes
.iter()
.find(|n| n.id == id)
.unwrap_or_else(|| panic!("{id} born"))
};
assert_eq!(
nb("i-001").step,
0,
"the mother idea is born at its submission"
);
assert_eq!(nb("n-001").step, 3);
assert_eq!(nb("n-002").step, 4);
assert_eq!(nb("n-003").step, 5);
assert_eq!(
g.nodes.iter().map(|n| n.id.as_str()).collect::<Vec<_>>(),
["i-001", "n-001", "n-002", "n-003"],
"nodes are in strict birth order"
);
// The birth timestamp is the birth event's ts, not `now` or step 0's.
assert!(
nb("n-003").at > nb("i-001").at,
"a later birth carries a later timestamp"
);
// ── edge births — the RED-when-broken core. A mother→child edge is born
// with its child (steps 3/4/5), but each task→task DEP edge is born only
// at its own EdgeAdded (steps 6 and 7), NOT when its endpoints were.
// A bug that stamped every element at step 0, at the final step, or that
// attributed a dep edge to an endpoint's birth step would fail here.
let eb = |from: &str, to: &str| {
g.edges
.iter()
.find(|e| e.from == from && e.to == to)
.unwrap_or_else(|| panic!("{from}->{to} born"))
};
assert_eq!(
eb("i-001", "n-001").step,
3,
"mother→child is born with the child"
);
assert_eq!(eb("i-001", "n-001").kind, EdgeKind::Mother);
assert_eq!(
eb("n-001", "n-002").step,
6,
"dep a→b born at ITS EdgeAdded, not at a node step"
);
assert_eq!(eb("n-001", "n-002").kind, EdgeKind::Dep);
assert_eq!(
eb("n-002", "n-003").step,
7,
"dep b→c born at ITS EdgeAdded"
);
// The dep edge is born strictly after both its endpoints — a real growth order.
assert!(eb("n-001", "n-002").step > nb("n-001").step.max(nb("n-002").step));
// ── growth curve: node/edge counts as of a step are monotone non-decreasing
// and land on the exact births above.
assert_eq!(g.node_count_at(0), 1, "only the mother so far");
assert_eq!(g.node_count_at(2), 1, "plan + activate add no DAG node");
assert_eq!(g.node_count_at(3), 2);
assert_eq!(
g.node_count_at(5),
4,
"all four nodes present after the last NodeAdded"
);
assert_eq!(
g.edge_count_at(5),
3,
"three mother→child edges, no dep edge yet"
);
assert_eq!(g.edge_count_at(7), 5, "the two dep edges have arrived");
// The final NodeStatusChanged (step 8) is a CONTENT change — it births
// nothing, so the graph as of step 8 equals the graph as of step 7.
assert_eq!(
g.node_count_at(8),
g.node_count_at(7),
"a status change adds no node"
);
assert_eq!(
g.edge_count_at(8),
g.edge_count_at(7),
"a status change adds no edge"
);
// Deterministic (FC-7) + the state_json seam carries the same births.
assert_eq!(
DagGrowth::from_events(&evs, None),
g,
"growth is deterministic"
);
let j = g.state_json();
assert_eq!(j["steps"], evs.len() as i64);
assert_eq!(j["node_count"], 4);
assert_eq!(j["edge_count"], 5);
assert_eq!(j["nodes"][0]["id"], "i-001");
assert_eq!(j["nodes"][0]["step"], 0);
}
#[test]
fn growth_skips_rejected_events_and_scopes_like_from_funnel() {
// Append a cycle-closing edge (n-003 → n-001) that `apply` rejects: it never
// reaches a real log, so it must birth nothing while still counting as a step.
let mut evs = chain_events();
let plan = PlanId::seq(1);
evs.push(Event::EdgeAdded {
plan_id: plan,
from_node: NodeId::seq(3),
to_node: NodeId::seq(1),
ts: Utc::now(),
});
let g = DagGrowth::from_events(&evs, None);
assert_eq!(
g.steps,
evs.len(),
"the rejected event is still a replayed step"
);
// No n-003 → n-001 edge was ever born.
assert!(
!g.edges.iter().any(|e| e.from == "n-003" && e.to == "n-001"),
"a rejected cycle edge births no edge"
);
assert_eq!(g.edges.len(), 5, "still exactly the 3 mother + 2 dep edges");
// Scoping mirrors PlanDag::from_funnel: under `Some(p-001)` the mother card
// only enters scope once its plan exists, so it is born at the PlanCreated
// step (1) — strictly later than the whole-funnel view's step 0.
let scoped = DagGrowth::from_events(&evs, Some("p-001"));
let mother = scoped
.nodes
.iter()
.find(|n| n.id == "i-001")
.expect("mother in scope");
assert_eq!(
mother.step, 1,
"scoped: the mother is born when its plan is created"
);
assert_eq!(scoped.sel_plan.as_deref(), Some("p-001"));
// The scoped growth still contains exactly the plan's mother + 3 tasks.
assert_eq!(scoped.nodes.len(), 4);
}
#[test]
fn extend_streams_byte_identical_births_to_the_batch_fold() {
// The RED-when-broken equivalence guard: feeding the log ONE event at a time
// through the incremental `extend` must yield the exact same timeline — same
// births, same order, same step count — as the batch `from_events`. The two
// are INDEPENDENT implementations (batch re-projects + global-sorts at the
// end; the stream keeps a running Funnel and appends per tick), so a bug in
// `extend`'s stamping or ordering diverges from the batch path and this fails.
let evs = chain_events();
let batch = DagGrowth::from_events(&evs, None);
let mut stream = DagGrowth::new(None);
for ev in &evs {
stream.extend(ev);
}
// Byte-identical births (the core guard) …
assert_eq!(
stream.nodes, batch.nodes,
"streamed node births match the batch fold"
);
assert_eq!(
stream.edges, batch.edges,
"streamed edge births match the batch fold"
);
assert_eq!(stream.steps, batch.steps, "same step count");
// … and the whole observable timeline compares equal.
assert_eq!(stream, batch, "incremental growth equals batch growth");
assert_eq!(
stream.state_json(),
batch.state_json(),
"same robot-observable blob"
);
// A NEW edge is born at its OWN step under streaming, not at an endpoint's
// birth — the dep edge n-002→n-003 arrives at step 7, strictly after both
// its endpoints (steps 4 and 5). A stream that mis-stamped it (e.g. at the
// node step, at 0, or at the final step) would break this.
let eb = stream
.edges
.iter()
.find(|e| e.from == "n-002" && e.to == "n-003")
.expect("dep edge born under streaming");
assert_eq!(
eb.step, 7,
"the new dep edge is born at its own EdgeAdded step"
);
assert_eq!(eb.kind, EdgeKind::Dep);
let birth_n002 = stream.nodes.iter().find(|n| n.id == "n-002").unwrap().step;
let birth_n003 = stream.nodes.iter().find(|n| n.id == "n-003").unwrap().step;
assert!(
eb.step > birth_n002.max(birth_n003),
"edge born after both endpoints"
);
// Monotone growth holds on the streamed timeline too (never decreasing).
let mut prev_n = 0;
let mut prev_e = 0;
for s in 0..stream.steps {
let (n, e) = (stream.node_count_at(s), stream.edge_count_at(s));
assert!(
n >= prev_n && e >= prev_e,
"counts are monotone non-decreasing"
);
prev_n = n;
prev_e = e;
}
}
#[test]
fn extend_matches_batch_scoped_and_skips_rejected_events() {
// Same equivalence guard under a plan scope AND with a rejected (cycle-closing)
// event in the log: the stream must skip the rejected event's births yet still
// advance the step, landing byte-identical to the scoped batch fold.
let mut evs = chain_events();
let plan = PlanId::seq(1);
evs.push(Event::EdgeAdded {
plan_id: plan,
from_node: NodeId::seq(3),
to_node: NodeId::seq(1),
ts: Utc::now(),
});
let batch = DagGrowth::from_events(&evs, Some("p-001"));
let mut stream = DagGrowth::new(Some("p-001"));
for ev in &evs {
stream.extend(ev);
}
assert_eq!(
stream, batch,
"scoped incremental growth equals scoped batch growth"
);
assert_eq!(
stream.steps,
evs.len(),
"rejected event still counts as a step"
);
assert!(
!stream
.edges
.iter()
.any(|e| e.from == "n-003" && e.to == "n-001"),
"the rejected cycle edge births nothing under streaming"
);
// The hydrate-then-continue property: a batch fold retains its accumulator, so
// a from_events timeline can be `extend`ed further and stays consistent with a
// stream that saw every event from the start.
let split = evs.len() - 1;
let mut hydrated = DagGrowth::from_events(&evs[..split], Some("p-001"));
for ev in &evs[split..] {
hydrated.extend(ev);
}
assert_eq!(
hydrated, batch,
"batch-load then stream-continue equals full stream"
);
}
}