examples/funnel_demo.rs
//! Funnel demo CLI — drives the idea-intake funnel from the shell.
//!
//! Subcommands:
//! submit "<text>" -> emit IdeaSubmitted, print idea_id
//! plan <idea_id> "<summary>" -> emit PlanCreated, print plan_id
//! add-node <plan_id> <kind> [--targets a,b] [--needs n1,n2]
//! link <plan_id> <from_node> <to_node>
//! sort -> print topo_ready as ASCII
//! start <plan_id> <node_id> -> NodeStatusChanged -> InProgress
//! done <plan_id> <node_id> -> NodeStatusChanged -> Done
//! show -> dump funnel summary
//! bootstrap -> seed log with the funnel's own build plan
//!
//! All state lives in `<cwd>/.nornir/todo.log`.
use std::env;
use std::path::PathBuf;
use anyhow::{Context, Result, bail};
use chrono::Utc;
use nornir::funnel::event::{Event, NodeStatus, PlanStatus};
use nornir::funnel::ids::{IdeaId, NodeId, PlanId};
use nornir::funnel::store::Store;
use nornir::funnel::topo::topo_ready;
fn main() -> Result<()> {
let args: Vec<String> = env::args().skip(1).collect();
if args.is_empty() {
print_help();
return Ok(());
}
let root = warehouse_root();
let mut store = Store::open(&root)
.with_context(|| format!("open funnel warehouse at {}", root.display()))?;
match args[0].as_str() {
"submit" => cmd_submit(&mut store, &args[1..])?,
"plan" => cmd_plan(&mut store, &args[1..])?,
"add-node" => cmd_add_node(&mut store, &args[1..])?,
"link" => cmd_link(&mut store, &args[1..])?,
"sort" => cmd_sort(&mut store)?,
"start" => cmd_status(&mut store, &args[1..], NodeStatus::InProgress)?,
"done" => cmd_status(&mut store, &args[1..], NodeStatus::Done)?,
"block" => cmd_status(&mut store, &args[1..], NodeStatus::Blocked)?,
"activate" => cmd_plan_status(&mut store, &args[1..], PlanStatus::Active)?,
"show" => cmd_show(&store),
"bootstrap" => cmd_bootstrap(&mut store)?,
"migrate-ndjson" => cmd_migrate_ndjson(&mut store, &args[1..])?,
"help" | "-h" | "--help" => print_help(),
other => bail!("unknown subcommand `{other}` — try `help`"),
}
Ok(())
}
fn warehouse_root() -> PathBuf {
if let Ok(p) = env::var("NORNIR_FUNNEL_ROOT") {
return PathBuf::from(p);
}
Store::default_root()
}
fn print_help() {
println!(
"funnel_demo — drive nornir's idea-intake funnel\n\n\
USAGE: cargo run --example funnel_demo -- <subcommand> [args]\n\n\
submit \"<text>\" emit IdeaSubmitted\n\
plan <idea_id> \"<summary>\" create a plan for an idea\n\
add-node <plan_id> <kind> [--targets a,b] add a plan node\n\
link <plan_id> <from> <to> add a dependency edge\n\
sort print topo_ready() steps\n\
start <plan_id> <node_id> mark InProgress\n\
done <plan_id> <node_id> mark Done\n\
block <plan_id> <node_id> mark Blocked\n\
activate <plan_id> flip plan Draft -> Active\n\
show print funnel summary\n\
bootstrap seed log with funnel's own build plan\n\
migrate-ndjson <path> replay an old .nornir/todo.log into iceberg\n\n\
Env: NORNIR_FUNNEL_ROOT overrides the default .nornir/warehouse path."
);
}
fn cmd_submit(store: &mut Store, args: &[String]) -> Result<()> {
if args.is_empty() {
bail!("submit needs a text arg");
}
let text = args.join(" ");
let id = IdeaId::seq(store.funnel.next_idea.max(1));
store.record(Event::IdeaSubmitted {
id: id.clone(),
source: env::var("USER").unwrap_or_else(|_| "cli".into()),
text,
refs: vec![],
ts: Utc::now(),
item_kind: nornir::funnel::ItemKind::Idea,
})?;
println!("submitted {id}");
Ok(())
}
fn cmd_plan(store: &mut Store, args: &[String]) -> Result<()> {
if args.len() < 2 {
bail!("plan <idea_id> \"<summary>\"");
}
let idea_id = IdeaId::new(&args[0]);
let summary = args[1..].join(" ");
let id = PlanId::seq(store.funnel.next_plan.max(1));
store.record(Event::PlanCreated {
id: id.clone(),
idea_id,
summary,
planner: env::var("USER").unwrap_or_else(|_| "cli".into()),
ts: Utc::now(),
})?;
// Auto-activate so it shows up in topo_ready immediately.
store.record(Event::PlanStatusChanged {
plan_id: id.clone(),
status: PlanStatus::Active,
why: Some("auto-activated by demo CLI".into()),
ts: Utc::now(),
})?;
println!("plan {id} (active)");
Ok(())
}
fn cmd_add_node(store: &mut Store, args: &[String]) -> Result<()> {
if args.len() < 2 {
bail!("add-node <plan_id> <kind> [--targets a,b] [--needs n1,n2]");
}
let plan_id = PlanId::new(&args[0]);
let kind = args[1].clone();
let mut targets: Vec<String> = vec![];
let mut needs: Vec<NodeId> = vec![];
let mut i = 2;
while i < args.len() {
match args[i].as_str() {
"--targets" => {
i += 1;
if i < args.len() {
targets = args[i].split(',').map(|s| s.trim().to_string()).collect();
}
}
"--needs" => {
i += 1;
if i < args.len() {
needs = args[i].split(',').map(|s| NodeId::new(s.trim())).collect();
}
}
other => bail!("unknown flag `{other}`"),
}
i += 1;
}
let node_id = NodeId::seq(store.funnel.next_node.max(1));
store.record(Event::NodeAdded {
plan_id: plan_id.clone(),
node_id: node_id.clone(),
kind,
params: serde_json::Map::new(),
targets,
prompt_excerpt: None,
ts: Utc::now(),
})?;
for need in needs {
store.record(Event::EdgeAdded {
plan_id: plan_id.clone(),
from_node: need,
to_node: node_id.clone(),
ts: Utc::now(),
})?;
}
println!("node {node_id} in {plan_id}");
Ok(())
}
fn cmd_link(store: &mut Store, args: &[String]) -> Result<()> {
if args.len() != 3 {
bail!("link <plan_id> <from_node> <to_node>");
}
store.record(Event::EdgeAdded {
plan_id: PlanId::new(&args[0]),
from_node: NodeId::new(&args[1]),
to_node: NodeId::new(&args[2]),
ts: Utc::now(),
})?;
println!("edge {} -> {}", args[1], args[2]);
Ok(())
}
fn cmd_sort(store: &mut Store) -> Result<()> {
let steps = topo_ready(&mut store.funnel);
if steps.is_empty() {
println!("(nothing ready — submit an idea, plan it, add nodes, then `sort`)");
return Ok(());
}
println!(
"READY ({} step{}):",
steps.len(),
if steps.len() == 1 { "" } else { "s" }
);
for (i, s) in steps.iter().enumerate() {
let tgt = if s.targets.is_empty() {
String::new()
} else {
format!(" targets=[{}]", s.targets.join(","))
};
println!(
" {}. {} / {} kind={}{}\n plan: {}",
i + 1,
s.plan_id,
s.node_id,
s.kind,
tgt,
s.summary,
);
}
Ok(())
}
fn cmd_status(store: &mut Store, args: &[String], status: NodeStatus) -> Result<()> {
if args.len() < 2 {
bail!("expected <plan_id> <node_id>");
}
store.record(Event::NodeStatusChanged {
plan_id: PlanId::new(&args[0]),
node_id: NodeId::new(&args[1]),
status,
why: if args.len() > 2 {
Some(args[2..].join(" "))
} else {
None
},
ts: Utc::now(),
})?;
println!("{} / {} -> {:?}", args[0], args[1], status);
Ok(())
}
fn cmd_plan_status(store: &mut Store, args: &[String], status: PlanStatus) -> Result<()> {
if args.is_empty() {
bail!("expected <plan_id>");
}
store.record(Event::PlanStatusChanged {
plan_id: PlanId::new(&args[0]),
status,
why: None,
ts: Utc::now(),
})?;
println!("plan {} -> {:?}", args[0], status);
Ok(())
}
fn cmd_show(store: &Store) {
println!("funnel @ {}", store.root().display());
println!(" ideas: {}", store.funnel.ideas.len());
for idea in store.funnel.ideas.values() {
println!(
" {} [{}]: {}",
idea.id,
idea.source,
truncate(&idea.text, 60)
);
}
println!(" plans: {}", store.funnel.plans.len());
for plan in store.funnel.plans.values() {
println!(
" {} ({:?}) for {} — {} nodes, {} edges",
plan.id,
plan.status,
plan.idea_id,
plan.nodes.len(),
plan.edges.len(),
);
for n in plan.nodes.values() {
println!(" {} [{:?}] kind={}", n.id, n.status, n.kind);
}
}
}
fn cmd_bootstrap(store: &mut Store) -> Result<()> {
if !store.funnel.ideas.is_empty() {
println!(
"log already has {} idea(s) — skipping bootstrap",
store.funnel.ideas.len()
);
return Ok(());
}
let idea = IdeaId::seq(1);
store.record(Event::IdeaSubmitted {
id: idea.clone(),
source: "bootstrap".into(),
text: "build the idea-intake funnel itself".into(),
refs: vec![".nornir/notes/2026-05-31-funnel-dag-handoff.md".into()],
ts: Utc::now(),
item_kind: nornir::funnel::ItemKind::Idea,
})?;
let plan = PlanId::seq(1);
store.record(Event::PlanCreated {
id: plan.clone(),
idea_id: idea,
summary: "ship the funnel MVP: events, store, topo, demo, mcp".into(),
planner: "bootstrap".into(),
ts: Utc::now(),
})?;
store.record(Event::PlanStatusChanged {
plan_id: plan.clone(),
status: PlanStatus::Active,
why: None,
ts: Utc::now(),
})?;
// Five build steps. Each later step depends on the prior.
let steps: [(&str, &str, &str); 5] = [
("event_types", "rust:design", "src/funnel/event.rs"),
("store_ndjson", "rust:impl", "src/funnel/store.rs"),
("topo_dispatch", "rust:impl", "src/funnel/topo.rs"),
("demo_cli", "rust:example", "examples/funnel_demo.rs"),
("mcp_tools", "rust:mcp", "src/mcp/funnel.rs"),
];
let mut prev: Option<NodeId> = None;
let mut start_n = store.funnel.next_node.max(1);
for (label, kind, target) in steps {
let nid = NodeId::seq(start_n);
start_n += 1;
store.record(Event::NodeAdded {
plan_id: plan.clone(),
node_id: nid.clone(),
kind: kind.to_string(),
params: {
let mut m = serde_json::Map::new();
m.insert("label".into(), serde_json::Value::String(label.into()));
m
},
targets: vec![target.to_string()],
prompt_excerpt: Some(format!("Build {label}: implement {target}")),
ts: Utc::now(),
})?;
if let Some(from) = prev {
store.record(Event::EdgeAdded {
plan_id: plan.clone(),
from_node: from,
to_node: nid.clone(),
ts: Utc::now(),
})?;
}
prev = Some(nid);
}
// The first three are already done (we built them this session).
for n in 1..=3u64 {
store.record(Event::NodeStatusChanged {
plan_id: plan.clone(),
node_id: NodeId::seq(n),
status: NodeStatus::Done,
why: Some("built in scaffolding chunk 1".into()),
ts: Utc::now(),
})?;
}
println!("bootstrapped: 1 idea, 1 plan, 5 nodes, 4 edges, 3 marked done");
println!("try: cargo run --example funnel_demo -- sort");
Ok(())
}
fn cmd_migrate_ndjson(store: &mut Store, args: &[String]) -> Result<()> {
use std::io::{BufRead, BufReader};
if args.is_empty() {
bail!("migrate-ndjson <path-to-old-todo.log>");
}
let path = std::path::PathBuf::from(&args[0]);
let f = std::fs::File::open(&path).with_context(|| format!("open {}", path.display()))?;
let mut migrated = 0usize;
let mut skipped = 0usize;
for (lineno, line) in BufReader::new(f).lines().enumerate() {
let line = line.with_context(|| format!("read line {}", lineno + 1))?;
if line.trim().is_empty() {
continue;
}
let event: nornir::funnel::event::Event =
serde_json::from_str(&line).with_context(|| format!("parse line {}", lineno + 1))?;
match store.record(event) {
Ok(()) => migrated += 1,
Err(e) => {
eprintln!(" skip line {}: {e:#}", lineno + 1);
skipped += 1;
}
}
}
println!(
"migrated {migrated} events ({skipped} skipped) from {}",
path.display()
);
Ok(())
}
fn truncate(s: &str, max: usize) -> String {
if s.chars().count() <= max {
s.to_string()
} else {
let mut out: String = s.chars().take(max - 1).collect();
out.push('…');
out
}
}