src/funnel/demo.rs
//! Funnel **demo** generator + nuke — the E1 "funnel demo" deliverable.
//!
//! Two halves that mirror each other:
//!
//! 1. **On-disk fake repos.** [`run_demo`] materialises `--size N` tiny but
//! *real* git repos near the `git/` root, each a `Cargo.toml` +
//! `src/lib.rs`, wired into a **layered acyclic dependency graph** via
//! path-deps on lower-indexed siblings (so `funnel_fake_3` depends on
//! `funnel_fake_1` and `funnel_fake_2`, never the reverse → guaranteed
//! acyclic). Every created path is recorded in a **manifest**
//! `git/.nornir-funnel-demo.json` so [`nuke_disk`] deletes *exactly* what
//! the demo made — never a heuristic `rm`.
//!
//! 2. **In-warehouse funnel DAG.** The same run injects a funnel idea→plan
//! whose node DAG **mirrors** the on-disk repo dep graph (one node per
//! fake repo, one edge per path-dep). The idea is tagged
//! `source = "funnel:demo"` so reruns are **idempotent** (an existing demo
//! idea is detected and the inject is skipped).
//!
//! Nuke is **disk-only**: it removes the fake repos + manifest but leaves the
//! warehouse funnel data untouched, so the plan still shows up in `funnel show`
//! and the viz Funnel tab after a nuke (the event log is append-only history).
//!
//! ### Determinism
//! [`inject_dep_graph`] builds the funnel DAG from an explicit `seed` with a
//! tiny splitmix64 PRNG, so the *expected* node/edge/ready set is computable in
//! tests (golden, data-only). Edges only ever go **lower-layer → higher-layer**,
//! making the graph acyclic by construction.
use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
use chrono::Utc;
use serde::{Deserialize, Serialize};
use super::event::{Event, ItemKind, PlanStatus};
use super::ids::{IdeaId, NodeId, PlanId};
use super::store::Store;
/// The `source` tag stamped on the demo idea. Reruns key off this so a second
/// `run_demo`/`inject_demo` is a no-op instead of a duplicate plan.
pub const DEMO_SOURCE: &str = "funnel:demo";
/// Manifest filename written into the git root listing exactly what the demo
/// created on disk, so nuke is precise (never a heuristic delete).
pub const MANIFEST_NAME: &str = ".nornir-funnel-demo.json";
// ─── deterministic PRNG ─────────────────────────────────────────────────────
/// splitmix64 — a tiny, deterministic, dependency-free PRNG. Same `seed` →
/// same stream, so the injected DAG (and therefore the expected golden state)
/// is fully reproducible.
struct SplitMix64(u64);
impl SplitMix64 {
fn new(seed: u64) -> Self {
Self(seed)
}
fn next_u64(&mut self) -> u64 {
self.0 = self.0.wrapping_add(0x9E37_79B9_7F4A_7C15);
let mut z = self.0;
z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
z ^ (z >> 31)
}
/// Uniform-ish `0..n` (n>0).
fn below(&mut self, n: usize) -> usize {
(self.next_u64() % (n as u64)) as usize
}
/// Coin flip biased by `num/den`.
fn chance(&mut self, num: u64, den: u64) -> bool {
self.next_u64() % den < num
}
}
// ─── manifest ───────────────────────────────────────────────────────────────
/// What the demo created on disk, written to `git/.nornir-funnel-demo.json`.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DemoManifest {
/// Schema marker.
pub kind: String,
/// When the demo was generated.
pub created_at: chrono::DateTime<Utc>,
/// `--size N`.
pub size: usize,
/// Absolute paths of every fake-repo dir the demo created (the only paths
/// nuke is allowed to delete).
pub repos: Vec<PathBuf>,
/// The container dir created for `size > 1` (e.g. `git/funnel3_fake_projects`),
/// removed by nuke once empty. `None` for the `size == 1` single-repo layout.
pub container: Option<PathBuf>,
}
impl DemoManifest {
fn path(git_root: &Path) -> PathBuf {
git_root.join(MANIFEST_NAME)
}
fn load(git_root: &Path) -> Option<Self> {
let p = Self::path(git_root);
let txt = std::fs::read_to_string(p).ok()?;
serde_json::from_str(&txt).ok()
}
fn save(&self, git_root: &Path) -> Result<()> {
let p = Self::path(git_root);
let txt = serde_json::to_string_pretty(self).context("serialize demo manifest")?;
std::fs::write(&p, txt).with_context(|| format!("write manifest {}", p.display()))?;
Ok(())
}
}
// ─── git-root resolution ────────────────────────────────────────────────────
/// Resolve the `git/` root the fake repos live under.
///
/// Precedence:
/// 1. the warehouse `../git/` sibling convention — the warehouse root is
/// `<ws>/.nornir/warehouse`, so its great-grandparent's `git/` is the
/// monorepo git root (matches the timeline's `../git/<member>` lookup);
/// 2. fallback: `<workspace_root>/../git/`.
///
/// The funnel root passed to [`run_demo`] is the warehouse root, so we walk up
/// from it. We accept either the `.nornir/warehouse` shape or a bare dir.
pub fn resolve_git_root(funnel_root: &Path) -> PathBuf {
// <ws>/.nornir/warehouse → walk to <ws>, then sibling git/.
// warehouse → .nornir → <ws>
let ws = funnel_root
.parent() // .nornir
.and_then(|p| p.parent()) // <ws>
.unwrap_or(funnel_root);
// Prefer <ws>/../git (the monorepo sibling), else <ws>/git.
if let Some(parent) = ws.parent() {
let sib = parent.join("git");
if sib.exists() {
return sib;
}
return sib; // create-on-demand: still the sibling convention
}
ws.join("git")
}
// ─── on-disk repo names ─────────────────────────────────────────────────────
/// Directory each fake repo lives in, for `size` and 1-based index `i`.
///
/// * `size == 1` → `<git>/funnel_fake/funnel1_fake_project/`
/// * `size > 1` → `<git>/funnel{N}_fake_projects/funnel_fake_{i}/`
fn repo_dir(git_root: &Path, size: usize, i: usize) -> PathBuf {
if size == 1 {
git_root.join("funnel_fake").join("funnel1_fake_project")
} else {
git_root
.join(format!("funnel{size}_fake_projects"))
.join(format!("funnel_fake_{i}"))
}
}
/// The container dir (parent of all repos) for nuke cleanup.
fn container_dir(git_root: &Path, size: usize) -> PathBuf {
if size == 1 {
git_root.join("funnel_fake")
} else {
git_root.join(format!("funnel{size}_fake_projects"))
}
}
/// Crate name for the fake repo at 1-based index `i`.
fn crate_name(i: usize) -> String {
format!("funnel_fake_{i}")
}
// ─── the funnel DAG injector (the data-plane half) ──────────────────────────
/// One plan node in the injected demo DAG, paired with the edges into it.
#[derive(Debug, Clone)]
pub struct DemoNodeSpec {
pub node_id: NodeId,
pub kind: String,
pub targets: Vec<String>,
/// node ids this node depends on (lower-layer → this).
pub deps: Vec<NodeId>,
pub layer: usize,
}
/// Build the **layered acyclic** demo DAG spec for `plans` plans, each with
/// `nodes_per_plan` nodes arranged into `layers` layers, deterministically from
/// `seed`. Edges only go lower-layer → higher-layer (acyclic by construction).
///
/// Returns, per plan, the ordered list of node specs. The caller records the
/// matching events through a [`Store`]; tests recompute the expected golden set
/// from the same specs.
pub fn build_dep_graph_spec(
plans: usize,
nodes_per_plan: usize,
layers: usize,
seed: u64,
) -> Vec<Vec<DemoNodeSpec>> {
let mut rng = SplitMix64::new(seed);
let layers = layers.max(1);
let mut out = Vec::with_capacity(plans);
let mut next_node = 1u64;
for _plan in 0..plans {
let n = nodes_per_plan.max(1);
// Assign each node to a layer: spread round-robin so every layer is
// populated, then nudge a few by the PRNG for variety (still monotone:
// we only ever assign a layer index, edges enforce acyclicity).
let mut node_layer = Vec::with_capacity(n);
for k in 0..n {
let base = (k * layers) / n; // monotone fill 0..layers
let jitter = if layers > 1 && rng.chance(1, 4) {
// occasionally push down one layer (never up past its base+1)
(base + 1).min(layers - 1)
} else {
base
};
node_layer.push(jitter);
}
// Ensure at least one layer-0 root.
if !node_layer.iter().any(|&l| l == 0) {
node_layer[0] = 0;
}
// Allocate node ids in declaration order.
let ids: Vec<NodeId> = (0..n)
.map(|_| {
let id = NodeId::seq(next_node);
next_node += 1;
id
})
.collect();
// Build edges: each non-root node gets 1..=2 deps chosen from STRICTLY
// lower layers (deterministic). If no lower-layer node exists, it stays
// a root.
let mut specs = Vec::with_capacity(n);
for k in 0..n {
let layer = node_layer[k];
let lower: Vec<usize> = (0..n).filter(|&j| node_layer[j] < layer).collect();
let mut deps = Vec::new();
if !lower.is_empty() {
let want = 1 + rng.below(2.min(lower.len())); // 1 or 2 deps
let mut chosen = std::collections::BTreeSet::new();
for _ in 0..want {
chosen.insert(lower[rng.below(lower.len())]);
}
deps = chosen.into_iter().map(|j| ids[j].clone()).collect();
}
specs.push(DemoNodeSpec {
node_id: ids[k].clone(),
kind: "code:write".to_string(),
targets: vec![format!("{}/src/lib.rs", crate_name(k + 1))],
deps,
layer,
});
}
out.push(specs);
}
out
}
/// Inject the layered demo DAG into `store` as one idea + `plans` plans, each
/// carrying its `nodes_per_plan`-node DAG. Idempotent: if a `source =
/// "funnel:demo"` idea already exists, returns `Ok(None)` without writing.
///
/// On success returns `Some((idea_id, plan_ids))`.
pub fn inject_dep_graph(
store: &mut Store,
plans: usize,
nodes_per_plan: usize,
layers: usize,
seed: u64,
) -> Result<Option<(IdeaId, Vec<PlanId>)>> {
// Idempotency: already injected?
if store.funnel.ideas.values().any(|i| i.source == DEMO_SOURCE) {
return Ok(None);
}
// Strictly increasing timestamps so replay order is deterministic (the
// Iceberg scan order is not stable; replay sorts/processes by ts).
let base = Utc::now();
let mut tick = 0i64;
let mut at = || {
tick += 1;
base + chrono::Duration::microseconds(tick)
};
let idea_id = IdeaId::seq(store.funnel.next_idea.max(1));
store.record(Event::IdeaSubmitted {
id: idea_id.clone(),
source: DEMO_SOURCE.to_string(),
text: format!(
"funnel demo: {plans} plan(s) × {nodes_per_plan} nodes × {layers} layers (seed {seed})"
),
refs: vec![],
item_kind: ItemKind::Idea,
ts: at(),
})?;
let specs = build_dep_graph_spec(plans, nodes_per_plan, layers, seed);
let mut plan_ids = Vec::with_capacity(plans);
for (pi, plan_specs) in specs.iter().enumerate() {
let plan_id = PlanId::seq(store.funnel.next_plan.max(1));
store.record(Event::PlanCreated {
id: plan_id.clone(),
idea_id: idea_id.clone(),
summary: format!("demo plan {} (fake dep graph)", pi + 1),
planner: DEMO_SOURCE.to_string(),
ts: at(),
})?;
store.record(Event::PlanStatusChanged {
plan_id: plan_id.clone(),
status: PlanStatus::Active,
why: None,
ts: at(),
})?;
// Nodes first (so edges can reference them), then edges.
for spec in plan_specs {
store.record(Event::NodeAdded {
plan_id: plan_id.clone(),
node_id: spec.node_id.clone(),
kind: spec.kind.clone(),
params: serde_json::Map::new(),
targets: spec.targets.clone(),
prompt_excerpt: None,
ts: at(),
})?;
}
for spec in plan_specs {
for dep in &spec.deps {
store.record(Event::EdgeAdded {
plan_id: plan_id.clone(),
from_node: dep.clone(),
to_node: spec.node_id.clone(),
ts: at(),
})?;
}
}
plan_ids.push(plan_id);
}
store.funnel.promote_ready();
Ok(Some((idea_id, plan_ids)))
}
/// Test-friendly thin wrapper: inject a `size`-plan demo (one plan per "size"
/// is overkill; instead inject a single plan whose node count scales with
/// `size`) into an already-open store. Used by `tests/funnel_demo.rs`.
///
/// Layout chosen so the golden set is easy to reason about:
/// * 1 plan,
/// * `size * 3` nodes,
/// * 3 layers,
/// * fixed seed `0xF1` (so the expected DAG is stable across runs).
pub fn inject_demo(store: &mut Store, size: usize) -> Result<Option<(IdeaId, Vec<PlanId>)>> {
inject_dep_graph(store, 1, size.max(1) * 3, 3, 0xF1)
}
// ─── on-disk fake-repo materialisation ──────────────────────────────────────
/// Write one fake repo dir: `Cargo.toml` (path-deps on lower-indexed siblings),
/// `src/lib.rs`, and a real git repo (`git init` + add + commit). Idempotent:
/// existing files are overwritten, an existing `.git` is left as-is, and the
/// commit step is best-effort (a clean tree re-commit is a harmless no-op).
fn write_fake_repo(dir: &Path, size: usize, i: usize) -> Result<()> {
let src = dir.join("src");
std::fs::create_dir_all(&src).with_context(|| format!("create {}", src.display()))?;
// Path-deps on every LOWER-indexed sibling → a layered (acyclic) dep graph
// that mirrors the funnel DAG. Sibling location depends on the layout.
let name = crate_name(i);
let mut deps = String::new();
for j in 1..i {
let sibling = crate_name(j);
// For size==1 there are no siblings (i==1 only); for size>1 siblings are
// adjacent dirs under the same container.
let rel = format!("../{sibling}");
deps.push_str(&format!("{sibling} = {{ path = \"{rel}\" }}\n"));
}
let cargo = format!(
"[package]\n\
name = \"{name}\"\n\
version = \"0.1.0\"\n\
edition = \"2021\"\n\
publish = false\n\
\n\
# Generated by `nornir funnel demo` (size {size}). Safe to delete via\n\
# `nornir funnel nuke --yes`.\n\
[dependencies]\n\
{deps}"
);
std::fs::write(dir.join("Cargo.toml"), cargo)
.with_context(|| format!("write Cargo.toml in {}", dir.display()))?;
let lib = format!(
"//! Fake demo crate `{name}` — generated by `nornir funnel demo`.\n\
//! Part of a layered path-dep graph mirroring the funnel demo DAG.\n\
\n\
/// Returns this crate's 1-based index in the demo dep graph.\n\
pub fn index() -> usize {{\n {i}\n}}\n"
);
std::fs::write(src.join("lib.rs"), lib)
.with_context(|| format!("write src/lib.rs in {}", dir.display()))?;
git_init_commit(dir)?;
Ok(())
}
/// `git init` + snapshot-commit, idempotent and best-effort. Pure-gix via the
/// `nornir-git` leaf (no `git` subprocess); the commit uses the leaf's fixed
/// deterministic fixture identity so it succeeds in CI/sandbox where `user.name`
/// may be unset. Failures (e.g. nothing to commit) are downgraded to warnings so
/// the demo still produces files.
fn git_init_commit(dir: &Path) -> Result<()> {
if !dir.join(".git").exists() {
let _ = crate::gitio::init(dir);
}
// `commit_all` snapshots the whole working tree on top of HEAD (add-all +
// commit); a clean-tree re-run is a harmless extra commit we swallow.
let _ = crate::gitio::commit_all(dir, "funnel demo snapshot");
Ok(())
}
/// Materialise `--size N` fake repos under the git root AND inject the matching
/// funnel DAG into the warehouse at `funnel_root`. Writes the manifest. Both
/// halves are idempotent.
///
/// Returns the [`DemoManifest`] describing what was created on disk.
pub fn run_demo(funnel_root: &Path, size: usize) -> Result<DemoManifest> {
let size = size.max(1);
let git_root = resolve_git_root(funnel_root);
std::fs::create_dir_all(&git_root)
.with_context(|| format!("create git root {}", git_root.display()))?;
// --- disk half: fake repos + manifest ---
let mut repos = Vec::with_capacity(size);
for i in 1..=size {
let dir = repo_dir(&git_root, size, i);
write_fake_repo(&dir, size, i)?;
repos.push(dir);
}
let container = if size == 1 {
Some(container_dir(&git_root, size))
} else {
Some(container_dir(&git_root, size))
};
let manifest = DemoManifest {
kind: "nornir-funnel-demo".to_string(),
created_at: Utc::now(),
size,
repos,
container,
};
manifest.save(&git_root)?;
// --- warehouse half: the mirroring funnel DAG ---
// One plan whose nodes are the fake repos, edges = path-deps. Layers derived
// from index so node i depends only on lower-indexed nodes (acyclic).
let mut store = Store::open(funnel_root)
.with_context(|| format!("open funnel store at {}", funnel_root.display()))?;
inject_repo_dag(&mut store, size)?;
Ok(manifest)
}
/// Inject a funnel DAG that mirrors the on-disk repo dep graph: one node per
/// fake repo `i`, one edge `j -> i` for each path-dep (`j < i`). Idempotent on
/// the `funnel:demo` source tag.
fn inject_repo_dag(store: &mut Store, size: usize) -> Result<Option<(IdeaId, PlanId)>> {
if store.funnel.ideas.values().any(|i| i.source == DEMO_SOURCE) {
return Ok(None);
}
let base = Utc::now();
let mut tick = 0i64;
let mut at = || {
tick += 1;
base + chrono::Duration::microseconds(tick)
};
let idea_id = IdeaId::seq(store.funnel.next_idea.max(1));
store.record(Event::IdeaSubmitted {
id: idea_id.clone(),
source: DEMO_SOURCE.to_string(),
text: format!("funnel demo: {size} fake repo(s) with a layered path-dep graph"),
refs: vec![],
item_kind: ItemKind::Idea,
ts: at(),
})?;
let plan_id = PlanId::seq(store.funnel.next_plan.max(1));
store.record(Event::PlanCreated {
id: plan_id.clone(),
idea_id: idea_id.clone(),
summary: format!("demo: {size} fake repos (path-dep DAG)"),
planner: DEMO_SOURCE.to_string(),
ts: at(),
})?;
store.record(Event::PlanStatusChanged {
plan_id: plan_id.clone(),
status: PlanStatus::Active,
why: None,
ts: at(),
})?;
// node[i] for fake repo i (1-based). Record nodes one at a time, capturing
// the monotonic seq id each `NodeAdded` consumed (ids must stay stable so
// the edges below reference the right nodes).
let mut ids = Vec::with_capacity(size);
for i in 1..=size {
let nid = NodeId::seq(store.funnel.next_node.max(1));
store.record(Event::NodeAdded {
plan_id: plan_id.clone(),
node_id: nid.clone(),
kind: "code:write".to_string(),
params: serde_json::Map::new(),
targets: vec![format!("{}/src/lib.rs", crate_name(i))],
prompt_excerpt: None,
ts: at(),
})?;
ids.push(nid);
}
// edges: repo i path-deps on every j<i → edge node[j] -> node[i].
for i in 1..=size {
for j in 1..i {
store.record(Event::EdgeAdded {
plan_id: plan_id.clone(),
from_node: ids[j - 1].clone(),
to_node: ids[i - 1].clone(),
ts: at(),
})?;
}
}
store.funnel.promote_ready();
Ok(Some((idea_id, plan_id)))
}
// ─── nuke (disk-only) ───────────────────────────────────────────────────────
/// What a nuke *would* do — the manifest's repos + container — without deleting.
/// Returns `None` when no manifest is present (nothing to nuke).
pub fn nuke_plan(funnel_root: &Path) -> Option<DemoManifest> {
let git_root = resolve_git_root(funnel_root);
DemoManifest::load(&git_root)
}
/// Delete ONLY the paths the manifest recorded (the fake repos + their
/// container + the manifest file). **Disk-only**: the warehouse funnel data is
/// never touched. Returns the list of removed dirs.
///
/// Safe by construction: it deletes exactly the manifest's `repos`/`container`,
/// not a glob — a manually-created dir that happens to match the naming scheme
/// is left alone unless the manifest lists it.
pub fn nuke_disk(funnel_root: &Path) -> Result<Vec<PathBuf>> {
let git_root = resolve_git_root(funnel_root);
let Some(manifest) = DemoManifest::load(&git_root) else {
return Ok(Vec::new());
};
let mut removed = Vec::new();
for repo in &manifest.repos {
if repo.exists() {
std::fs::remove_dir_all(repo)
.with_context(|| format!("remove fake repo {}", repo.display()))?;
removed.push(repo.clone());
}
}
// Remove the container dir if it is now empty (only our repos lived there).
if let Some(container) = &manifest.container {
if container.exists() {
let empty = std::fs::read_dir(container)
.map(|mut d| d.next().is_none())
.unwrap_or(false);
if empty {
let _ = std::fs::remove_dir(container);
}
}
}
// Finally drop the manifest itself.
let _ = std::fs::remove_file(DemoManifest::path(&git_root));
Ok(removed)
}
#[cfg(test)]
mod tests {
use super::*;
/// The dep-graph spec is acyclic: every edge goes lower-layer → higher-layer.
#[test]
fn spec_edges_are_layer_monotone() {
let specs = build_dep_graph_spec(2, 12, 4, 0xABCD);
for plan in &specs {
let layer_of: std::collections::HashMap<&NodeId, usize> =
plan.iter().map(|s| (&s.node_id, s.layer)).collect();
for s in plan {
for dep in &s.deps {
let dl = layer_of[dep];
assert!(
dl < s.layer,
"edge {dep} -> {} not layer-monotone",
s.node_id
);
}
}
}
}
/// Same seed → identical spec (determinism), different seed → may differ.
#[test]
fn spec_is_deterministic() {
let a = build_dep_graph_spec(1, 9, 3, 7);
let b = build_dep_graph_spec(1, 9, 3, 7);
assert_eq!(a.len(), b.len());
for (pa, pb) in a.iter().zip(&b) {
assert_eq!(pa.len(), pb.len());
for (na, nb) in pa.iter().zip(pb) {
assert_eq!(na.node_id, nb.node_id);
assert_eq!(na.deps, nb.deps);
assert_eq!(na.layer, nb.layer);
}
}
}
/// At least one root (layer 0, no deps) always exists, so `promote_ready`
/// has somewhere to start.
#[test]
fn spec_has_a_root() {
let specs = build_dep_graph_spec(1, 20, 5, 1);
assert!(specs[0].iter().any(|s| s.layer == 0 && s.deps.is_empty()));
}
}