//! Materialised view of the funnel: pure in-memory state built by
//! replaying [`super::event::Event`]s.
//!
//! Loading a fresh funnel:
//! 1. `Funnel::default()` — empty state.
//! 2. For each row in the Iceberg `funnel_events` table, call [`Funnel::apply`].
//! 3. Query.
//!
//! Writing:
//! 1. Build the [`Event`] with `chrono::Utc::now()`.
//! 2. `Funnel::apply` it (validates + mutates in-memory state).
//! 3. Append to the log (caller's job — see [`super::store`]).
//!
//! Validation lives in `apply`: we reject events that reference
//! unknown ids, would create a duplicate edge, a self-loop, or a
//! dependency **cycle** (an edge whose `to` already reaches its `from`) —
//! a cycle would otherwise permanently stall the plan in `topo_ready` /
//! `simulate_flow`. Bad events are rejected with `anyhow::Error`; the log
//! stays a clean DAG because the store calls `apply` before `append`.
//! (Node *status* transitions are deliberately not lifecycle-gated — see
//! `.nornir/design.md`.)
use std::collections::{BTreeMap, BTreeSet};
use anyhow::{anyhow, bail, Result};
use chrono::{DateTime, Utc};
use super::event::{CommitRef, Event, NodeStatus, PlanStatus, RunOutcome, TriageDecision};
use super::ids::{IdeaId, NodeId, PlanId, RunId};
#[derive(Debug, Clone)]
pub struct Idea {
pub id: IdeaId,
pub source: String,
pub text: String,
pub refs: Vec<String>,
/// `idea` (default) or `error` — a pasted error report is a first-class
/// funnel item (FI3), triageable to a plan node like any idea.
pub item_kind: super::event::ItemKind,
pub submitted_at: DateTime<Utc>,
pub triage: Option<(TriageDecision, Option<String>, DateTime<Utc>)>,
/// The funnel-prompt-planner **mode** (`"auto"` | `"manual"`) this idea's
/// prompt-driven decomposition ran under, set by an [`Event::IdeaModeSet`].
/// `None` for ideas never routed through `nornir funnel prompt` (including
/// every pre-keystone log row — additive/migration-safe).
///
/// Kept free-form on purpose (see [`crate::mode`]); read it as a typed
/// [`FunnelMode`](crate::mode::FunnelMode) via [`Idea::mode_kind`].
pub mode: Option<String>,
}
impl Idea {
/// The idea's [`FunnelMode`](crate::mode::FunnelMode), parsed from the
/// free-form [`mode`](Idea::mode) token. `None` when the mode was never set
/// (a pre-keystone idea) **or** the stored token isn't a recognised mode —
/// the forgiving contract that keeps the free-form field future-proof.
#[must_use]
pub fn mode_kind(&self) -> Option<super::mode::FunnelMode> {
self.mode
.as_deref()
.and_then(super::mode::FunnelMode::parse)
}
}
#[derive(Debug, Clone)]
pub struct Plan {
pub id: PlanId,
pub idea_id: IdeaId,
pub summary: String,
pub planner: String,
pub created_at: DateTime<Utc>,
pub status: PlanStatus,
/// Nodes keyed by id.
pub nodes: BTreeMap<NodeId, PlanNode>,
/// Edges as (from, to). Stored as a Set so duplicates can't slip in.
pub edges: BTreeSet<(NodeId, NodeId)>,
}
#[derive(Debug, Clone)]
pub struct PlanNode {
pub id: NodeId,
pub kind: String,
pub params: serde_json::Map<String, serde_json::Value>,
pub targets: Vec<String>,
pub prompt_excerpt: Option<String>,
pub created_at: DateTime<Utc>,
pub status: NodeStatus,
pub last_status_change: DateTime<Utc>,
}
/// The latest [`Event::CardAssigned`] recorded for a card (an idea or a
/// decomposed task node) — who owns it, the role they were picked for, and why.
#[derive(Debug, Clone)]
pub struct Assignment {
pub member: String,
pub role: Option<String>,
pub why: Option<String>,
pub ts: DateTime<Utc>,
}
/// A single execution attempt against a plan node, folded from
/// [`Event::RunRecorded`] — the funnel's **run history**.
///
/// The event carries the full record (who ran it, the outcome, the log
/// pointer, what it produced); before this existed `apply` validated the
/// references, bumped `next_run` and dropped every payload field, so run
/// history was not recoverable from a replayed `Funnel` and each consumer
/// had to re-read the raw log itself. Retaining it here keeps the state
/// engine the ONE source of run history (no parallel per-consumer store).
#[derive(Debug, Clone)]
pub struct Run {
pub run_id: RunId,
pub plan_id: PlanId,
pub node_id: NodeId,
pub ran_by: String,
pub outcome: RunOutcome,
pub log_ref: Option<String>,
pub produced_commits: Vec<CommitRef>,
pub produced_test_runs: Vec<String>,
pub ts: DateTime<Utc>,
}
#[derive(Debug, Default, Clone)]
pub struct Funnel {
pub ideas: BTreeMap<IdeaId, Idea>,
pub plans: BTreeMap<PlanId, Plan>,
/// Every [`Event::RunRecorded`] folded in **log order** (oldest first) —
/// the append-only run history. Replay-deterministic: the same log always
/// yields the same vector.
pub runs: Vec<Run>,
/// High-water marks so new ids stay monotonic across sessions.
pub next_idea: u64,
pub next_plan: u64,
pub next_node: u64,
pub next_run: u64,
/// Latest assignment per card id (an [`IdeaId`] or [`NodeId`] string),
/// derived by folding every [`Event::CardAssigned`] (last-write-wins).
pub assignments: BTreeMap<String, Assignment>,
/// Latest kanban lane (wire token) per card id, derived by folding every
/// [`Event::CardStateChanged`] (last-write-wins). Idea/error/test cards
/// normally derive their lane from [`super::history::HistoryStatus`]
/// instead; this map exists for cards — chiefly decomposed task nodes —
/// whose lane isn't a function of triage/plan-existence (the `Proposed`
/// staging lane, EPIC funnel-prompt-planner §2).
pub card_lanes: BTreeMap<String, String>,
/// The **Plan-DAG ↔ Component-DAG wiring layer** — one [`Wire`] per
/// `(plan_node, component)` pair, folded last-write-wins from every
/// [`Event::PlanComponentWired`]. Keyed `(plan_node, component)` so the map's
/// natural order IS the deterministic projection order
/// [`Wiring::from_funnel`](crate::wiring::Wiring::from_funnel) hands out.
///
/// Rejected wires stay in the map (the log never deletes); they simply do not
/// [`bind`](crate::wiring::Wire::binds). See [`crate::wiring`].
pub wires: BTreeMap<(String, String), super::wiring::Wire>,
}
impl Funnel {
/// Apply a single event to the in-memory state. Returns Err if the
/// event would violate an invariant (unknown id, duplicate edge,
/// illegal status transition). The caller MUST NOT append a
/// rejected event to the log.
pub fn apply(&mut self, event: &Event) -> Result<()> {
match event {
Event::IdeaSubmitted {
id,
source,
text,
refs,
item_kind,
ts,
} => {
if self.ideas.contains_key(id) {
bail!("duplicate IdeaSubmitted for `{id}`");
}
self.ideas.insert(
id.clone(),
Idea {
id: id.clone(),
source: source.clone(),
text: text.clone(),
refs: refs.clone(),
item_kind: *item_kind,
submitted_at: *ts,
triage: None,
mode: None,
},
);
self.next_idea = self.next_idea.max(parse_seq(id.as_str()) + 1);
}
Event::IdeaTriaged {
idea_id,
decision,
why,
ts,
} => {
let idea = self
.ideas
.get_mut(idea_id)
.ok_or_else(|| anyhow!("IdeaTriaged for unknown idea `{idea_id}`"))?;
idea.triage = Some((*decision, why.clone(), *ts));
}
Event::PlanCreated {
id,
idea_id,
summary,
planner,
ts,
} => {
if !self.ideas.contains_key(idea_id) {
bail!("PlanCreated for unknown idea `{idea_id}`");
}
if self.plans.contains_key(id) {
bail!("duplicate PlanCreated for `{id}`");
}
self.plans.insert(
id.clone(),
Plan {
id: id.clone(),
idea_id: idea_id.clone(),
summary: summary.clone(),
planner: planner.clone(),
created_at: *ts,
status: PlanStatus::Draft,
nodes: BTreeMap::new(),
edges: BTreeSet::new(),
},
);
self.next_plan = self.next_plan.max(parse_seq(id.as_str()) + 1);
}
Event::NodeAdded {
plan_id,
node_id,
kind,
params,
targets,
prompt_excerpt,
ts,
} => {
if !self.plans.contains_key(plan_id) {
bail!("NodeAdded for unknown plan `{plan_id}`");
}
// Node ids must be unique **funnel-wide**, not merely within the
// plan. Cards are addressed by a bare string id (`CardAssigned` /
// `CardStateChanged` → `card_exists`, which scans every plan), so a
// node id repeated across two plans makes an assignment or a lane
// move aimed at one plan's node silently apply to both, last-write-
// wins, with no way to disambiguate. Uniqueness used to rest purely
// on producers allocating via `NodeId::seq(funnel.next_node)`; that
// is a caller-side convention, and `parse_seq` falls back to 0 for
// uuid-style ids, so a spliced/hand-rolled log could collide. Make
// it an invariant of the engine instead.
if let Some(owner) = self.plan_owning_node(node_id) {
if owner == *plan_id {
bail!("duplicate NodeAdded `{node_id}` in plan `{plan_id}`");
}
bail!(
"NodeAdded `{node_id}` in plan `{plan_id}` collides with the \
same node id already in plan `{owner}` (node ids must be \
unique across the whole funnel)"
);
}
let plan = self
.plans
.get_mut(plan_id)
.expect("plan presence checked above");
plan.nodes.insert(
node_id.clone(),
PlanNode {
id: node_id.clone(),
kind: kind.clone(),
params: params.clone(),
targets: targets.clone(),
prompt_excerpt: prompt_excerpt.clone(),
created_at: *ts,
status: NodeStatus::Pending,
last_status_change: *ts,
},
);
self.next_node = self.next_node.max(parse_seq(node_id.as_str()) + 1);
}
Event::EdgeAdded {
plan_id,
from_node,
to_node,
ts: _,
} => {
let plan = self
.plans
.get_mut(plan_id)
.ok_or_else(|| anyhow!("EdgeAdded for unknown plan `{plan_id}`"))?;
if !plan.nodes.contains_key(from_node) {
bail!("EdgeAdded references unknown from_node `{from_node}`");
}
if !plan.nodes.contains_key(to_node) {
bail!("EdgeAdded references unknown to_node `{to_node}`");
}
if from_node == to_node {
bail!("EdgeAdded would create self-loop on `{from_node}`");
}
// A cycle (a→b, b→c, c→a) can't be caught by the self-loop /
// duplicate-edge guards — each edge is individually valid. Left
// unchecked it slips into the log and then `topo_ready` /
// `simulate_flow` skip the whole plan forever ("cycle permanently
// stalls a plan"). Reject the closing edge here so the log stays a
// DAG and every plan remains drainable. Adding `from → to` closes a
// cycle iff `from` is already reachable from `to`.
if edge_reaches(&plan.edges, to_node, from_node) {
bail!(
"EdgeAdded {from_node} -> {to_node} would create a cycle in plan `{plan_id}` \
({to_node} already reaches {from_node})"
);
}
let inserted = plan.edges.insert((from_node.clone(), to_node.clone()));
if !inserted {
bail!("duplicate edge {from_node} -> {to_node} in plan `{plan_id}`");
}
}
Event::NodeStatusChanged {
plan_id,
node_id,
status,
why: _,
ts,
} => {
let plan = self
.plans
.get_mut(plan_id)
.ok_or_else(|| anyhow!("NodeStatusChanged for unknown plan `{plan_id}`"))?;
let node = plan
.nodes
.get_mut(node_id)
.ok_or_else(|| anyhow!("NodeStatusChanged for unknown node `{node_id}`"))?;
node.status = *status;
node.last_status_change = *ts;
}
Event::RunRecorded {
plan_id,
node_id,
run_id,
ran_by,
outcome,
log_ref,
produced_commits,
produced_test_runs,
ts,
} => {
let plan = self
.plans
.get(plan_id)
.ok_or_else(|| anyhow!("RunRecorded for unknown plan `{plan_id}`"))?;
if !plan.nodes.contains_key(node_id) {
bail!("RunRecorded for unknown node `{node_id}`");
}
if self.runs.iter().any(|r| &r.run_id == run_id) {
bail!("duplicate RunRecorded for run `{run_id}`");
}
self.runs.push(Run {
run_id: run_id.clone(),
plan_id: plan_id.clone(),
node_id: node_id.clone(),
ran_by: ran_by.clone(),
outcome: *outcome,
log_ref: log_ref.clone(),
produced_commits: produced_commits.clone(),
produced_test_runs: produced_test_runs.clone(),
ts: *ts,
});
self.next_run = self.next_run.max(parse_seq(run_id.as_str()) + 1);
}
Event::PlanStatusChanged {
plan_id,
status,
why: _,
ts: _,
} => {
let plan = self
.plans
.get_mut(plan_id)
.ok_or_else(|| anyhow!("PlanStatusChanged for unknown plan `{plan_id}`"))?;
plan.status = *status;
}
Event::IdeaModeSet {
idea_id,
mode,
ts: _,
} => {
let idea = self
.ideas
.get_mut(idea_id)
.ok_or_else(|| anyhow!("IdeaModeSet for unknown idea `{idea_id}`"))?;
idea.mode = Some(mode.clone());
}
Event::CardAssigned {
node,
member,
role,
why,
ts,
} => {
if !self.card_exists(node) {
bail!("CardAssigned for unknown card `{node}` (not an idea or task-node id)");
}
self.assignments.insert(
node.clone(),
Assignment {
member: member.clone(),
role: role.clone(),
why: why.clone(),
ts: *ts,
},
);
}
Event::CardStateChanged {
node,
from_lane: _,
to_lane,
why: _,
ts: _,
} => {
if !self.card_exists(node) {
bail!(
"CardStateChanged for unknown card `{node}` (not an idea or task-node id)"
);
}
self.card_lanes.insert(node.clone(), to_lane.clone());
}
Event::PlanComponentWired {
plan_node,
component,
origin,
status,
confidence,
why,
ts,
} => {
// The plan side of the wire MUST name a real card — a wire to a
// node that doesn't exist would silently inflate the wiring
// coverage of a plan that never had that step, which is exactly
// the dishonesty the wiring layer exists to prevent.
if !self.card_exists(plan_node) {
bail!(
"PlanComponentWired for unknown plan node `{plan_node}` \
(not an idea or task-node id)"
);
}
if component.is_empty() {
bail!("PlanComponentWired with an empty component id for `{plan_node}`");
}
// The component side is deliberately NOT validated: the component
// namespace belongs to Mímir, not the funnel, and the funnel holds
// no component graph. An id the graph doesn't hold is reported as a
// dangling wire by `WiringReport`, not rejected at write time —
// otherwise a plan could never be wired ahead of the code.
self.wires.insert(
(plan_node.clone(), component.clone()),
super::wiring::Wire {
plan_node: plan_node.clone(),
component: super::wiring::ComponentId::new(component.clone()),
origin: origin.clone(),
status: status.clone(),
confidence: *confidence,
why: why.clone(),
ts: *ts,
},
);
}
}
Ok(())
}
/// Does `id` name an existing card — either an idea/error/test item
/// (`i-NNN`) or a task node inside any plan (`n-NNN`)? Shared validity
/// check for [`Event::CardAssigned`]/[`Event::CardStateChanged`], which
/// address cards by a generic string id rather than a typed one.
fn card_exists(&self, id: &str) -> bool {
self.ideas.contains_key(&IdeaId::new(id))
|| self
.plans
.values()
.any(|p| p.nodes.contains_key(&NodeId::new(id)))
}
/// The plan that already owns `node_id`, if any — the funnel-wide
/// node-id lookup backing the uniqueness invariant enforced in
/// [`Funnel::apply`]'s `NodeAdded` arm.
pub fn plan_owning_node(&self, node_id: &NodeId) -> Option<PlanId> {
self.plans
.iter()
.find(|(_, p)| p.nodes.contains_key(node_id))
.map(|(id, _)| id.clone())
}
/// Every recorded run, oldest first — the funnel's run history as folded
/// from [`Event::RunRecorded`].
pub fn runs(&self) -> &[Run] {
&self.runs
}
/// The runs recorded against one node, oldest first.
pub fn runs_for_node(&self, node_id: &NodeId) -> Vec<&Run> {
self.runs.iter().filter(|r| &r.node_id == node_id).collect()
}
/// The runs recorded against one plan, oldest first.
pub fn runs_for_plan(&self, plan_id: &PlanId) -> Vec<&Run> {
self.runs.iter().filter(|r| &r.plan_id == plan_id).collect()
}
/// The run history as a stable JSON array (newest first) — the wire shape
/// UI panes render. Mirrors [`super::history::history_to_json`]'s
/// discipline: pure projection, no clock, no I/O.
pub fn runs_to_json(&self) -> serde_json::Value {
let rows: Vec<serde_json::Value> = self
.runs
.iter()
.rev()
.map(|r| {
serde_json::json!({
"run_id": r.run_id.as_str(),
"plan_id": r.plan_id.as_str(),
"node_id": r.node_id.as_str(),
"ran_by": r.ran_by,
"outcome": r.outcome,
"log_ref": r.log_ref,
"produced_commits": r.produced_commits.iter()
.map(|c| serde_json::json!({"repo": c.repo, "sha": c.sha}))
.collect::<Vec<_>>(),
"produced_test_runs": r.produced_test_runs,
"ts": r.ts.to_rfc3339(),
})
})
.collect();
serde_json::Value::Array(rows)
}
}
// (legacy seq helpers removed — bumping is inlined in each apply arm)
/// Does `target` reach `start` by following existing directed `edges`
/// (`from → to`)? A BFS over the plan's current edge set — used by
/// [`Funnel::apply`] to reject a `EdgeAdded from → to` that would close a cycle
/// (which happens exactly when `to` already reaches `from`). Excludes the edge
/// being added (it isn't in `edges` yet), and never revisits a node, so it
/// terminates on any graph in O(V + E).
fn edge_reaches(edges: &BTreeSet<(NodeId, NodeId)>, start: &NodeId, target: &NodeId) -> bool {
let mut seen: BTreeSet<&NodeId> = BTreeSet::new();
let mut stack: Vec<&NodeId> = vec![start];
while let Some(cur) = stack.pop() {
if cur == target {
return true;
}
if !seen.insert(cur) {
continue;
}
for (from, to) in edges {
if from == cur {
stack.push(to);
}
}
}
false
}
/// Parse the trailing numeric suffix of an id like `"p-007"` → `7`.
/// Non-numeric or unsuffixed ids fall back to 0 (so seq-allocation
/// remains correct even if external tools generate uuid-style ids).
fn parse_seq(s: &str) -> u64 {
s.rsplit_once('-')
.and_then(|(_, tail)| tail.parse::<u64>().ok())
.unwrap_or(0)
}
// The next_*_from helpers above return *what next_* should become*;
// they don't mutate. We do the actual assignments in `apply` for
// clarity / borrow-checker friendliness:
impl Funnel {
/// Re-evaluate every node's `Pending` → `Ready` transition based
/// on the current edge set. Called by [`super::topo::topo_ready`]
/// just before returning a list, so callers see the latest
/// readiness without needing to emit explicit
/// `NodeStatusChanged` events for the trivial transition.
pub fn promote_ready(&mut self) {
for plan in self.plans.values_mut() {
// Snapshot status map so we can borrow nodes mutably below.
let status: BTreeMap<NodeId, NodeStatus> = plan
.nodes
.iter()
.map(|(id, n)| (id.clone(), n.status))
.collect();
for node in plan.nodes.values_mut() {
if node.status != NodeStatus::Pending {
continue;
}
let deps_ok = plan
.edges
.iter()
.filter(|(_, to)| to == &node.id)
.all(|(from, _)| matches!(status.get(from), Some(NodeStatus::Done)));
if deps_ok {
node.status = NodeStatus::Ready;
}
}
}
}
}
#[cfg(test)]
mod cycle_tests {
use super::*;
use crate::event::{Event, ItemKind, PlanStatus};
use crate::topo::simulate_flow;
use chrono::Utc;
/// Build one Active plan whose 3 nodes n-001/2/3 form a chain a→b (only the
/// first edge added), ready for the caller to try closing a cycle.
fn chain_plan() -> (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: "cycle guard".into(),
refs: vec![],
item_kind: ItemKind::Idea,
ts: now,
})
.unwrap();
f.apply(&Event::PlanCreated {
id: plan.clone(),
idea_id: idea,
summary: "chain".into(),
planner: "test".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: None,
ts: now,
})
.unwrap();
}
f.apply(&Event::PlanStatusChanged {
plan_id: plan.clone(),
status: PlanStatus::Active,
why: None,
ts: now,
})
.unwrap();
// a → b (n-001 → n-002).
f.apply(&Event::EdgeAdded {
plan_id: plan.clone(),
from_node: NodeId::seq(1),
to_node: NodeId::seq(2),
ts: now,
})
.unwrap();
(f, plan)
}
fn add_edge(f: &mut Funnel, plan: &PlanId, from: u64, to: u64) -> Result<()> {
f.apply(&Event::EdgeAdded {
plan_id: plan.clone(),
from_node: NodeId::seq(from),
to_node: NodeId::seq(to),
ts: Utc::now(),
})
}
/// RED-when-broken: an `EdgeAdded` that closes a cycle (a→b, b→c, then c→a)
/// is REJECTED, and the acyclic edges before it still applied. Without the
/// cycle guard the c→a edge is accepted and the plan stalls forever.
#[test]
fn edge_closing_a_cycle_is_rejected() {
let (mut f, plan) = chain_plan(); // has a→b
add_edge(&mut f, &plan, 2, 3).unwrap(); // b→c: still a DAG
// c→a would close the a→b→c→a cycle — must be rejected.
let err = add_edge(&mut f, &plan, 3, 1).unwrap_err();
let msg = format!("{err:#}");
assert!(msg.contains("cycle"), "error names the cycle: {msg}");
// The rejected edge did NOT slip into the log — the plan is still a DAG.
let p = &f.plans[&plan];
assert_eq!(p.edges.len(), 2, "only the two acyclic edges are stored");
assert!(!p.edges.contains(&(NodeId::seq(3), NodeId::seq(1))));
}
/// A longer back-edge is caught too: a→b→c, then c→a (transitive reach), and
/// a direct back-edge b→a. Reachability, not just adjacency, is checked.
#[test]
fn transitive_and_direct_back_edges_are_both_rejected() {
let (mut f, plan) = chain_plan(); // a→b
add_edge(&mut f, &plan, 2, 3).unwrap(); // b→c
// Direct back-edge b→a (b reaches a? no — a reaches b; but a→b means
// adding b→a closes a 2-cycle since a already reaches... check: adding
// from=2 to=1 cycles iff to(1) reaches from(2): 1→2 yes) → rejected.
assert!(add_edge(&mut f, &plan, 2, 1).is_err(), "b→a closes a→b");
// Transitive back-edge c→a: 1 reaches 3 (1→2→3), so 3→1 cycles.
assert!(add_edge(&mut f, &plan, 3, 1).is_err(), "c→a closes a→b→c");
// A genuinely new forward edge (a→c, a "shortcut") is still allowed — it
// keeps the graph acyclic, so the guard must NOT over-reject.
assert!(
add_edge(&mut f, &plan, 1, 3).is_ok(),
"a→c is a DAG shortcut"
);
}
/// The whole point: after the cycle edge is rejected, the plan still drains
/// to all-Done (it never permanently stalls). A stalled plan would leave
/// nodes `remaining`.
#[test]
fn plan_still_drains_after_a_cycle_edge_is_refused() {
let (mut f, plan) = chain_plan(); // a→b
add_edge(&mut f, &plan, 2, 3).unwrap(); // b→c → a→b→c chain
let _ = add_edge(&mut f, &plan, 3, 1); // refused cycle edge (ignored)
let sim = simulate_flow(&mut f, 10);
assert_eq!(sim.total, 3);
assert_eq!(sim.done, 3, "every node reaches Done — no stall");
assert_eq!(sim.remaining, 0);
assert_eq!(sim.waves, 3, "the 3-node chain drains in exactly 3 waves");
}
}
#[cfg(test)]
mod run_recorded_tests {
//! Coverage for the [`Event::RunRecorded`] `apply` arm. It **validates**
//! (the plan+node must exist, the run id must be fresh), **retains the run**
//! in [`Funnel::runs`] — the funnel's run history, so a replayed `Funnel`
//! carries it and no consumer needs a parallel store — and **bumps**
//! `next_run`. It does NOT touch node lifecycle (status flips ride a separate
//! `NodeStatusChanged`). These tests pin exactly that contract so a regression
//! that dropped a validation, mis-bumped the high-water mark, stopped
//! retaining the run, or started mutating the node is caught.
use super::*;
use crate::event::{Event, ItemKind, NodeStatus, PlanStatus, RunOutcome};
use crate::ids::{IdeaId, NodeId, PlanId, RunId};
use chrono::Utc;
/// One Active plan with a single Pending node n-001, ready to record runs against.
fn plan_with_node() -> (Funnel, PlanId, NodeId) {
let mut f = Funnel::default();
let now = Utc::now();
let idea = IdeaId::seq(0);
let plan = PlanId::seq(0);
let node = NodeId::seq(1);
f.apply(&Event::IdeaSubmitted {
id: idea.clone(),
source: "test".into(),
text: "run recorder".into(),
refs: vec![],
item_kind: ItemKind::Idea,
ts: now,
})
.unwrap();
f.apply(&Event::PlanCreated {
id: plan.clone(),
idea_id: idea,
summary: "runs".into(),
planner: "test".into(),
ts: now,
})
.unwrap();
f.apply(&Event::NodeAdded {
plan_id: plan.clone(),
node_id: node.clone(),
kind: "code:write".into(),
params: Default::default(),
targets: vec![],
prompt_excerpt: None,
ts: now,
})
.unwrap();
f.apply(&Event::PlanStatusChanged {
plan_id: plan.clone(),
status: PlanStatus::Active,
why: None,
ts: now,
})
.unwrap();
(f, plan, node)
}
fn run(plan: &PlanId, node: &NodeId, run_id: RunId, outcome: RunOutcome) -> Event {
Event::RunRecorded {
plan_id: plan.clone(),
node_id: node.clone(),
run_id,
ran_by: "agent:test".into(),
outcome,
log_ref: None,
produced_commits: vec![],
produced_test_runs: vec![],
ts: Utc::now(),
}
}
/// A valid `RunRecorded` is accepted, bumps `next_run` to `seq+1`, and leaves the
/// node's lifecycle untouched — recording a run is NOT a status transition.
/// RED-when-broken: an off-by-one in the high-water bump, or a stray status
/// mutation, fails the exact-value asserts below.
#[test]
fn valid_run_bumps_high_water_and_never_touches_node_status() {
let (mut f, plan, node) = plan_with_node();
assert_eq!(f.next_run, 0, "no run recorded yet");
let status_before = f.plans[&plan].nodes[&node].status;
assert_eq!(status_before, NodeStatus::Pending);
// r-005 (numeric suffix) → next_run high-water = 5 + 1.
f.apply(&run(&plan, &node, RunId::seq(5), RunOutcome::Ok))
.unwrap();
assert_eq!(f.next_run, 6, "next_run is the recorded seq + 1");
// The run is RETAINED as history, but must not add a node or flip its status.
assert_eq!(f.runs().len(), 1, "the run is retained as history");
assert_eq!(f.runs()[0].run_id, RunId::seq(5));
assert_eq!(f.runs()[0].outcome, RunOutcome::Ok);
assert_eq!(f.runs()[0].ran_by, "agent:test");
assert_eq!(
f.plans[&plan].nodes[&node].status,
NodeStatus::Pending,
"status is untouched"
);
assert_eq!(f.plans[&plan].nodes.len(), 1, "no node added by a run");
// A Failed-outcome run likewise leaves the node status alone (the caller emits a
// separate NodeStatusChanged to move it to Failed).
f.apply(&run(&plan, &node, RunId::seq(6), RunOutcome::Failed))
.unwrap();
assert_eq!(
f.plans[&plan].nodes[&node].status,
NodeStatus::Pending,
"a failed run is not a status flip"
);
assert_eq!(f.next_run, 7);
}
/// The high-water mark is monotone: a later run whose id has a smaller seq never
/// lowers `next_run`. RED-when-broken: a plain assignment (instead of `.max`)
/// would regress it to 3 here.
#[test]
fn high_water_is_monotone_not_last_write() {
let (mut f, plan, node) = plan_with_node();
f.apply(&run(&plan, &node, RunId::seq(9), RunOutcome::Ok))
.unwrap();
assert_eq!(f.next_run, 10);
f.apply(&run(&plan, &node, RunId::seq(2), RunOutcome::Ok))
.unwrap();
assert_eq!(
f.next_run, 10,
"a lower seq never lowers the high-water mark"
);
}
/// `RunRecorded` validates its references: an unknown plan or an unknown node is
/// rejected (so the store never appends a dangling run), and a rejected event
/// leaves `next_run` unchanged. RED-when-broken: dropping either guard makes these
/// `unwrap_err()` calls panic.
#[test]
fn unknown_plan_or_node_is_rejected_and_high_water_unchanged() {
let (mut f, plan, node) = plan_with_node();
let e = f
.apply(&run(&PlanId::seq(99), &node, RunId::seq(1), RunOutcome::Ok))
.unwrap_err();
assert!(
format!("{e:#}").contains("unknown plan"),
"names the unknown plan: {e:#}"
);
let e = f
.apply(&run(&plan, &NodeId::seq(42), RunId::seq(1), RunOutcome::Ok))
.unwrap_err();
assert!(
format!("{e:#}").contains("unknown node"),
"names the unknown node: {e:#}"
);
assert_eq!(
f.next_run, 0,
"a rejected run must not bump the high-water mark"
);
assert!(
f.runs().is_empty(),
"a rejected run must not enter the history"
);
}
/// **The run history is recoverable from a replayed funnel** — the whole point of
/// retaining runs. Every `RunRecorded` payload field survives the fold, the order
/// is log order (oldest first), and the per-node / per-plan / JSON projections
/// agree with it. RED-when-broken: revert the arm to the old validate-and-drop and
/// `runs()` is empty ⇒ every assert below fails.
#[test]
fn runs_are_retained_in_log_order_with_every_payload_field() {
let (mut f, plan, node) = plan_with_node();
f.apply(&Event::RunRecorded {
plan_id: plan.clone(),
node_id: node.clone(),
run_id: RunId::seq(1),
ran_by: "agent:alpha".into(),
outcome: RunOutcome::Failed,
log_ref: Some("s3://logs/r-001".into()),
produced_commits: vec![crate::event::CommitRef {
repo: "dwarves".into(),
sha: "deadbeef".into(),
}],
produced_test_runs: vec!["tr-1".into()],
ts: Utc::now(),
})
.unwrap();
f.apply(&run(&plan, &node, RunId::seq(2), RunOutcome::Ok))
.unwrap();
assert_eq!(f.runs().len(), 2, "both runs retained");
// Log order, oldest first.
assert_eq!(f.runs()[0].run_id, RunId::seq(1));
assert_eq!(f.runs()[1].run_id, RunId::seq(2));
// Every payload field survived the fold — nothing dropped on the floor.
let r0 = &f.runs()[0];
assert_eq!(r0.ran_by, "agent:alpha");
assert_eq!(r0.outcome, RunOutcome::Failed);
assert_eq!(r0.log_ref.as_deref(), Some("s3://logs/r-001"));
assert_eq!(r0.produced_commits.len(), 1);
assert_eq!(r0.produced_commits[0].sha, "deadbeef");
assert_eq!(r0.produced_test_runs, vec!["tr-1".to_string()]);
assert_eq!(r0.plan_id, plan);
assert_eq!(r0.node_id, node);
// Projections agree with the ledger.
assert_eq!(f.runs_for_node(&node).len(), 2);
assert_eq!(f.runs_for_plan(&plan).len(), 2);
assert!(
f.runs_for_node(&NodeId::seq(42)).is_empty(),
"an unrelated node has no runs"
);
// The JSON wire shape is NEWEST first (what a history pane renders).
let json = f.runs_to_json();
let rows = json.as_array().unwrap();
assert_eq!(rows.len(), 2);
assert_eq!(rows[0]["run_id"], "r-002", "newest first");
assert_eq!(rows[1]["run_id"], "r-001");
assert_eq!(rows[1]["outcome"], "failed");
assert_eq!(rows[1]["ran_by"], "agent:alpha");
assert_eq!(rows[1]["produced_commits"][0]["repo"], "dwarves");
}
/// A replayed duplicate `run_id` is rejected — the ledger holds each run once, so
/// a log replayed twice (or a spliced log) can't double-count history.
/// RED-when-broken: drop the dup guard and `runs()` grows to 2.
#[test]
fn duplicate_run_id_is_rejected() {
let (mut f, plan, node) = plan_with_node();
f.apply(&run(&plan, &node, RunId::seq(1), RunOutcome::Ok))
.unwrap();
let e = f
.apply(&run(&plan, &node, RunId::seq(1), RunOutcome::Failed))
.unwrap_err();
assert!(
format!("{e:#}").contains("duplicate RunRecorded"),
"names the dup: {e:#}"
);
assert_eq!(f.runs().len(), 1, "the ledger holds each run exactly once");
}
}
#[cfg(test)]
mod node_id_uniqueness_tests {
//! **Cross-plan node-id uniqueness is an invariant of `apply`, not a caller
//! convention.** Node ids used to be unique only *within* a plan: the `NodeAdded`
//! arm checked `plan.nodes.contains_key`, scoped to the one plan it had borrowed.
//! Funnel-wide uniqueness rested entirely on producers allocating via
//! `NodeId::seq(funnel.next_node)` against the global high-water counter — and
//! `parse_seq` falls back to 0 for uuid-style ids, so a hand-rolled or spliced log
//! could collide silently.
//!
//! The collision is load-bearing: `CardAssigned` / `CardStateChanged` address cards
//! by a bare string id resolved through `card_exists`, which scans EVERY plan. With
//! `n-001` in two plans, an assignment aimed at one plan's node lands in the flat
//! `assignments` / `card_lanes` map for both, last-write-wins, undisambiguatable.
use super::*;
use crate::event::{Event, ItemKind};
use crate::ids::{IdeaId, NodeId, PlanId};
use chrono::Utc;
/// Two plans (`p-000`, `p-001`), each hanging off its own idea — the setup a
/// cross-plan id collision needs.
fn two_plans() -> (Funnel, PlanId, PlanId) {
let mut f = Funnel::default();
let now = Utc::now();
let mut make = |n: u64| -> PlanId {
let idea = IdeaId::seq(n);
let plan = PlanId::seq(n);
f.apply(&Event::IdeaSubmitted {
id: idea.clone(),
source: "test".into(),
text: format!("idea {n}"),
refs: vec![],
item_kind: ItemKind::Idea,
ts: now,
})
.unwrap();
f.apply(&Event::PlanCreated {
id: plan.clone(),
idea_id: idea,
summary: format!("plan {n}"),
planner: "test".into(),
ts: now,
})
.unwrap();
plan
};
let a = make(0);
let b = make(1);
(f, a, b)
}
fn add_node(f: &mut Funnel, plan: &PlanId, node: &NodeId) -> Result<()> {
f.apply(&Event::NodeAdded {
plan_id: plan.clone(),
node_id: node.clone(),
kind: "code:write".into(),
params: Default::default(),
targets: vec![],
prompt_excerpt: None,
ts: Utc::now(),
})
}
/// **RED-when-broken (the flagged bug).** The SAME node id in a second plan is
/// rejected. Revert the `NodeAdded` arm to the plan-scoped
/// `plan.nodes.contains_key` check and this `unwrap_err()` panics — the collision
/// lands silently, exactly as the audit described.
#[test]
fn the_same_node_id_in_a_second_plan_is_rejected() {
let (mut f, a, b) = two_plans();
let n = NodeId::seq(1);
add_node(&mut f, &a, &n).unwrap();
let e = add_node(&mut f, &b, &n).unwrap_err();
let msg = format!("{e:#}");
assert!(msg.contains("collides"), "explains the collision: {msg}");
assert!(
msg.contains(a.as_str()),
"names the plan that already owns it: {msg}"
);
// The rejected event changed nothing: the second plan stays empty.
assert_eq!(
f.plans[&b].nodes.len(),
0,
"a rejected NodeAdded must not mutate"
);
assert_eq!(f.plans[&a].nodes.len(), 1);
assert_eq!(
f.plan_owning_node(&n),
Some(a),
"the id resolves to exactly one plan"
);
}
/// The uniqueness guard does not break the ordinary path: distinct ids land in
/// different plans, and a same-plan duplicate keeps its original (more specific)
/// error message.
#[test]
fn distinct_ids_still_land_and_same_plan_dup_keeps_its_message() {
let (mut f, a, b) = two_plans();
add_node(&mut f, &a, &NodeId::seq(1)).unwrap();
add_node(&mut f, &b, &NodeId::seq(2)).unwrap();
assert_eq!(f.plans[&a].nodes.len(), 1);
assert_eq!(f.plans[&b].nodes.len(), 1);
assert_eq!(f.next_node, 3, "the high-water mark tracks the largest seq");
let e = add_node(&mut f, &a, &NodeId::seq(1)).unwrap_err();
let msg = format!("{e:#}");
assert!(
msg.contains("duplicate NodeAdded"),
"same-plan dup keeps its message: {msg}"
);
}
/// Uniqueness holds for ids the seq counter cannot protect. `parse_seq` cannot read
/// a uuid-style suffix and falls back to 0, so the counter lands on a constant 1 no
/// matter which non-numeric id was inserted — it does NOT track those ids, and the
/// caller-side "allocate from `next_node`" convention therefore offers them no
/// protection at all. Only the `apply` invariant does.
/// RED-when-broken: without the funnel-wide check both inserts succeed.
#[test]
fn non_seq_ids_are_still_unique_funnel_wide() {
let (mut f, a, b) = two_plans();
let uuid = NodeId::new("3f2a9c1e-node");
add_node(&mut f, &a, &uuid).unwrap();
// The fallback pins the counter to 1 regardless of the id — proof it is not
// tracking this id space, so the counter cannot be what keeps ids unique.
assert_eq!(f.next_node, 1, "parse_seq's 0-fallback ignores the real id");
assert!(
add_node(&mut f, &b, &uuid).is_err(),
"still rejected funnel-wide"
);
assert_eq!(f.plans[&b].nodes.len(), 0);
// A second, DIFFERENT uuid-style id also maps to the same counter value —
// the counter genuinely carries no uniqueness information here.
let other = NodeId::new("7c1b4d2f-node");
add_node(&mut f, &b, &other).unwrap();
assert_eq!(
f.next_node, 1,
"still 1 — the counter never distinguished them"
);
}
/// The collision the invariant actually prevents: a card assignment aimed at one
/// plan's node can no longer be ambiguous, because the id names exactly one node
/// funnel-wide. With the duplicate rejected up front, `card_exists` resolves
/// unambiguously and `assignments` has one meaning.
#[test]
fn card_assignment_stays_unambiguous() {
let (mut f, a, b) = two_plans();
let n = NodeId::seq(1);
add_node(&mut f, &a, &n).unwrap();
assert!(
add_node(&mut f, &b, &n).is_err(),
"the ambiguity is refused at the source"
);
f.apply(&Event::CardAssigned {
node: n.as_str().to_string(),
member: "alice".into(),
role: Some("author".into()),
why: None,
ts: Utc::now(),
})
.unwrap();
assert_eq!(f.assignments[n.as_str()].member, "alice");
assert_eq!(
f.plan_owning_node(&n),
Some(a),
"the assigned card belongs to exactly one plan"
);
}
}
#[cfg(test)]
mod replay_determinism_tests {
//! The core event-replay property: folding the SAME log into two independently
//! constructed fresh funnels yields byte-identical observable state. `apply` reads
//! no wall clock, no environment and no unordered collection, so replay is a pure
//! function of the log — this guard fails the moment that stops being true (e.g. a
//! `HashMap` fold or a `now()` read sneaks into `apply`).
use super::*;
use crate::card::{board, board_to_json, cards_from_funnel};
use crate::event::{Event, ItemKind, NodeStatus, PlanStatus, RunOutcome, TriageDecision};
use crate::history::{history, history_to_json};
use crate::ids::{IdeaId, NodeId, PlanId, RunId};
use crate::plan_dag::PlanDag;
use chrono::{Duration, Utc};
/// A log spanning EVERY event variant, in a fixed write order with fixed
/// timestamps, so replaying it is a deterministic function of the slice alone.
fn full_log() -> 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 n1 = NodeId::seq(1);
let n2 = NodeId::seq(2);
vec![
Event::IdeaSubmitted {
id: idea.clone(),
source: "human:alice".into(),
text: "build the funnel #core".into(),
refs: vec!["https://example.com/n/1".into()],
item_kind: ItemKind::Idea,
ts: at(0),
},
Event::IdeaTriaged {
idea_id: idea.clone(),
decision: TriageDecision::Accept,
why: Some("worth it".into()),
ts: at(1),
},
Event::IdeaModeSet {
idea_id: idea.clone(),
mode: "manual".into(),
ts: at(2),
},
Event::PlanCreated {
id: plan.clone(),
idea_id: idea,
summary: "decompose".into(),
planner: "decompose".into(),
ts: at(3),
},
Event::PlanStatusChanged {
plan_id: plan.clone(),
status: PlanStatus::Active,
why: None,
ts: at(4),
},
Event::NodeAdded {
plan_id: plan.clone(),
node_id: n1.clone(),
kind: "code:write".into(),
params: Default::default(),
targets: vec![],
prompt_excerpt: Some("impl foo".into()),
ts: at(5),
},
Event::NodeAdded {
plan_id: plan.clone(),
node_id: n2.clone(),
kind: "test:write".into(),
params: Default::default(),
targets: vec![],
prompt_excerpt: Some("test foo".into()),
ts: at(6),
},
Event::EdgeAdded {
plan_id: plan.clone(),
from_node: n1.clone(),
to_node: n2.clone(),
ts: at(7),
},
Event::CardStateChanged {
node: n1.as_str().into(),
from_lane: None,
to_lane: "proposed".into(),
why: None,
ts: at(8),
},
Event::CardAssigned {
node: n1.as_str().into(),
member: "ada".into(),
role: Some("coder".into()),
why: None,
ts: at(9),
},
Event::RunRecorded {
plan_id: plan.clone(),
node_id: n1.clone(),
run_id: RunId::seq(3),
ran_by: "agent:autonom".into(),
outcome: RunOutcome::Ok,
log_ref: Some("s3://runs/3".into()),
produced_commits: vec![],
produced_test_runs: vec![],
ts: at(10),
},
Event::NodeStatusChanged {
plan_id: plan.clone(),
node_id: n1.clone(),
status: NodeStatus::Done,
why: None,
ts: at(11),
},
]
}
fn fold(log: &[Event]) -> Funnel {
let mut f = Funnel::default();
for e in log {
f.apply(e).unwrap();
}
f
}
/// Two fresh folds of the same log agree on the high-water marks AND on every
/// downstream projection (history, board, and the Plan-DAG blob). A single
/// determinism-breaking change in `apply` diverges at least one of these.
#[test]
fn folding_the_same_log_twice_yields_identical_state() {
let log = full_log();
let a = fold(&log);
let b = fold(&log);
// High-water marks (the mutable scalar state `apply` maintains).
assert_eq!(
(a.next_idea, a.next_plan, a.next_node, a.next_run),
(b.next_idea, b.next_plan, b.next_node, b.next_run),
"high-water marks are a pure function of the log"
);
// The RunRecorded (r-003) and NodeAdded (n-002) bumps actually landed.
assert_eq!(a.next_run, 4, "r-003 bumped next_run to 4");
assert_eq!(a.next_node, 3, "n-002 bumped next_node to 3");
// Every projection is byte-identical across the two independent folds.
assert_eq!(
history_to_json(&history(&a, None, None)),
history_to_json(&history(&b, None, None)),
"history diverged"
);
assert_eq!(
board_to_json(&board(&cards_from_funnel(&a))).to_string(),
board_to_json(&board(&cards_from_funnel(&b))).to_string(),
"board diverged"
);
assert_eq!(
PlanDag::from_funnel(&a, None).state_json().to_string(),
PlanDag::from_funnel(&b, None).state_json().to_string(),
"Plan-DAG diverged"
);
}
/// Replay is order-faithful for last-write-wins fields: the fold reflects the LAST
/// write in log order (not, say, an unordered pick). Re-asserts the property on the
/// two card-lane / assignment folds specifically.
#[test]
fn last_write_wins_fields_follow_log_order() {
let mut log = full_log();
let plan = PlanId::seq(1);
let n1 = NodeId::seq(1);
let base = Utc::now() + Duration::seconds(100);
// A later reassignment + lane move must win over the earlier ones in the log.
log.push(Event::CardAssigned {
node: n1.as_str().into(),
member: "bengt".into(),
role: None,
why: None,
ts: base,
});
log.push(Event::CardStateChanged {
node: n1.as_str().into(),
from_lane: Some("proposed".into()),
to_lane: "planned".into(),
why: None,
ts: base,
});
let _ = plan;
let f = fold(&log);
assert_eq!(
f.assignments["n-001"].member, "bengt",
"the later assignment wins"
);
assert_eq!(f.card_lanes["n-001"], "planned", "the later lane move wins");
// And it is still deterministic across a second fold.
let g = fold(&log);
assert_eq!(f.assignments["n-001"].member, g.assignments["n-001"].member);
assert_eq!(f.card_lanes["n-001"], g.card_lanes["n-001"]);
}
}