needs_replay: make simulated cases real aggregates, report buckets by state

A simulation that lives in JSON never meets the engine. needs_replay
takes a simulation's cases and runs them through the code the site
runs: answers::store_answer for the submission, aggregates::transition
for each desk move, against a real JetStream with its event log and
its refusal of undeclared moves. Buckets are then read back by
replaying each record's events and reported per bucket against
`done_when` and `success`. `--report-only` takes the same scorecard
from any JetStream, a live site's included.

First run, 30 cases from a generated contractor site on a throwaway
in-memory NATS: 30 aggregates, 56 transitions accepted, none refused.
The same routes against the earlier machine that skipped a stage: 8
refused, 8 of 12 enquiries left in `open`.

The local-checkout loaders move from the lint binary to `portal::local`
so both tools share them.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01L4jrCgLiKKHAEFuZUJjckH

Cargo.toml

@@ -111,6 +111,11 @@ name = "question_lint"
 path = "src/bin/question_lint.rs"
 required-features = ["ssr"]
 
+[[bin]]
+name = "needs_replay"
+path = "src/bin/needs_replay.rs"
+required-features = ["ssr"]
+
 [profile.wasm-release]
 inherits = "release"
 opt-level = 'z'

src/bin/needs_replay.rs

@@ -0,0 +1,107 @@
+//! Makes a simulation's cases real: every case becomes an aggregate in
+//! its bucket, created and moved by the same code the running site
+//! uses - `answers::store_answer` for the submission,
+//! `aggregates::transition` for each desk move, against a real
+//! JetStream with its event log and its refusal of undeclared moves.
+//! Then the buckets are read back by replaying each record's events,
+//! and reported per need against `done_when` and `success`.
+//!
+//! The report half runs on its own (`--report-only`) against any
+//! JetStream, which is how the same scorecard is taken from a live
+//! site's buckets.
+//!
+//! Point it at a throwaway NATS, never a site's own: it writes.
+#![cfg(feature = "ssr")]
+
+use std::collections::{BTreeMap, HashMap};
+
+use futures::StreamExt;
+use portal::events::store::{ensure_stream, load_events};
+use portal::local::{load_aggregates_from_dir, load_from_dir, load_needs_from_dir};
+use portal::{aggregates, answers, content, needs};
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let argv: Vec<String> = std::env::args().skip(1).collect();
+    let value = |flag: &str| argv.iter().position(|a| a == flag).and_then(|i| argv.get(i + 1)).cloned();
+    let usage = "usage: needs_replay --path <questions dir> --nats <url> (--cases <cases.jsonl> | --report-only)";
+    let dir = value("--path").ok_or_else(|| anyhow::anyhow!(usage))?;
+    let nats_url = value("--nats").ok_or_else(|| anyhow::anyhow!(usage))?;
+    let cases_path = value("--cases");
+    if cases_path.is_none() && !argv.iter().any(|a| a == "--report-only") {
+        anyhow::bail!(usage);
+    }
+
+    let questions = load_from_dir(&dir)?;
+    let schemas = content::with_builtin_aggregates(load_aggregates_from_dir(&dir)?);
+    let file = needs::parse(&load_needs_from_dir(&dir)?.ok_or_else(|| anyhow::anyhow!("the repo has no needs.yaml"))?)?;
+    let js = async_nats::jetstream::new(async_nats::connect(&nats_url).await?);
+    // The same call the site makes at boot.
+    ensure_stream(&js).await?;
+
+    if let Some(path) = cases_path {
+        let raw = std::fs::read_to_string(&path).map_err(|e| anyhow::anyhow!("reading {path}: {e}"))?;
+        let (mut created, mut moved, mut refused) = (0, 0, 0);
+        for (n, line) in raw.lines().filter(|l| !l.trim().is_empty()).enumerate() {
+            let case: serde_json::Value = serde_json::from_str(line)?;
+            let Some(need) = file.needs.iter().find(|need| case["need"] == need.id.as_str()) else { continue };
+            let Some(schema) = schemas.get(&need.lands_in) else { continue };
+            // Only a case that got as far as submitting leaves a record.
+            let states: Vec<&str> = case["states"].as_array().into_iter().flatten().filter_map(|s| s.as_str()).collect();
+            if states.is_empty() {
+                continue;
+            }
+            let step = needs::path_for(&questions, need).and_then(|p| p.last().cloned());
+            let (page, alternative) = step.map(|s| (s.page, s.alternative)).unwrap_or_default();
+            let id = format!("sim-{}-{}-{n}", case["persona"].as_str().unwrap_or("x"), case["run"]);
+            let now = chrono::Utc::now().timestamp_millis();
+            answers::store_answer(&js, &schemas, &need.lands_in, id.clone(), &page, &alternative, &case["record"], now).await?;
+            created += 1;
+            for target in &states[1..] {
+                match aggregates::transition(&js, schema, &id, target, serde_json::json!({"by": "simulation"}), now).await {
+                    Ok(_) => moved += 1,
+                    Err(e) => {
+                        refused += 1;
+                        eprintln!("REFUSED by the engine: {id} -> {target}: {e:?}");
+                        break;
+                    }
+                }
+            }
+        }
+        println!("{created} aggregate(s) created, {moved} transition(s) accepted, {refused} refused by the engine\n");
+    }
+
+    // Read every bucket back the way the engine knows it: by replaying
+    // each record's own event log.
+    let mut by_bucket: HashMap<String, BTreeMap<String, usize>> = HashMap::new();
+    for bucket in file.needs.iter().map(|n| n.lands_in.clone()).collect::<std::collections::BTreeSet<_>>() {
+        let (Some(schema), Ok(store)) = (schemas.get(&bucket), js.get_key_value(&bucket).await) else { continue };
+        let mut keys = store.keys().await?;
+        while let Some(Ok(id)) = keys.next().await {
+            let events = load_events(&js, &bucket, &id).await?;
+            if let Some(record) = aggregates::replay(schema, &id, &events) {
+                *by_bucket.entry(bucket.clone()).or_default().entry(record.state).or_default() += 1;
+            }
+        }
+    }
+    println!("{:<18} {:>8} {:>9} {:>9}   {}", "bucket", "records", "finished", "success", "by state");
+    let mut buckets: Vec<_> = by_bucket.iter().collect();
+    buckets.sort();
+    for (bucket, states) in buckets {
+        let needs_here: Vec<&needs::Need> = file.needs.iter().filter(|n| &n.lands_in == bucket).collect();
+        let is = |pick: fn(&needs::Need) -> &Vec<String>, state: &String| needs_here.iter().any(|n| pick(n).contains(state));
+        let total: usize = states.values().sum();
+        let finished: usize = states.iter().filter(|(s, _)| is(|n| &n.done_when, s)).map(|(_, c)| c).sum();
+        let success: usize = states.iter().filter(|(s, _)| is(|n| &n.success, s)).map(|(_, c)| c).sum();
+        let pct = |n: usize| if total == 0 { "-".to_string() } else { format!("{:.0}%", n as f64 * 100.0 / total as f64) };
+        println!(
+            "{:<18} {:>8} {:>9} {:>9}   {}",
+            bucket,
+            total,
+            pct(finished),
+            pct(success),
+            states.iter().map(|(s, c)| format!("{s} {c}")).collect::<Vec<_>>().join(", ")
+        );
+    }
+    Ok(())
+}

src/bin/question_lint.rs

@@ -9,6 +9,7 @@
 //! static binary," not "build a Rust workspace."
 #![cfg(feature = "ssr")]
 
+use portal::local::{load_aggregates_from_dir, load_from_dir, load_needs_from_dir, load_site_from_dir};
 use portal::{aggregates, content, needs};
 
 #[tokio::main]
@@ -108,97 +109,6 @@ async fn main() -> anyhow::Result<()> {
     }
 }
 
-/// The offline counterpart to `content::load_site_from_gitea` -
-/// `site.yaml` lives at the repo root like `aggregates.yaml`, and is
-/// just as optional locally: missing means default branding.
-fn load_site_from_dir(dir: &str) -> anyhow::Result<content::SiteConfig> {
-    let site_path = std::path::Path::new(dir)
-        .parent()
-        .unwrap_or_else(|| std::path::Path::new("."))
-        .join("site.yaml");
-    if !site_path.exists() {
-        return Ok(content::SiteConfig::default());
-    }
-    let raw = std::fs::read_to_string(&site_path)
-        .map_err(|e| anyhow::anyhow!("reading {}: {e}", site_path.display()))?;
-    serde_yaml::from_str(&raw).map_err(|e| anyhow::anyhow!("parsing {}: {e}", site_path.display()))
-}
-
-/// The offline counterpart to `content::load_aggregates_from_gitea` -
-/// `aggregates.yaml` lives at the repo root, one level up from the
-/// pages directory `--path` names, so `dir`'s parent is where it's
-/// looked for. Missing entirely is not an error here (unlike
-/// `--repo` mode, where `aggregates.yaml` is required) - a local
-/// checkout being linted may not have one, and every declared
-/// transition still gets checked against whatever *is* found; an
-/// empty map just means nothing is checked.
-fn load_aggregates_from_dir(
-    dir: &str,
-) -> anyhow::Result<std::collections::HashMap<String, aggregates::AggregateSchema>> {
-    let aggregates_path = std::path::Path::new(dir)
-        .parent()
-        .unwrap_or_else(|| std::path::Path::new("."))
-        .join("aggregates.yaml");
-    if !aggregates_path.exists() {
-        return Ok(std::collections::HashMap::new());
-    }
-    let raw = std::fs::read_to_string(&aggregates_path)
-        .map_err(|e| anyhow::anyhow!("reading {}: {e}", aggregates_path.display()))?;
-    aggregates::parse_aggregates_yaml(&raw)
-}
-
-/// The offline counterpart to `content::load_questions_from_gitea` -
-/// the same recursive tree walk and `content::build_questions`
-/// pipeline (derived ids, relative refs, sections, followup
-/// inference), just reading a local checkout instead of Gitea's API,
-/// for linting a branch that hasn't been pushed yet.
-fn load_from_dir(
-    dir: &str,
-) -> anyhow::Result<std::collections::HashMap<String, content::Question>> {
-    let base = std::path::Path::new(dir);
-    let mut files = Vec::new();
-    collect_yaml(base, base, &mut files)?;
-    content::build_questions(&files)
-}
-
-fn collect_yaml(
-    base: &std::path::Path,
-    dir: &std::path::Path,
-    out: &mut Vec<(String, String)>,
-) -> anyhow::Result<()> {
-    for entry in std::fs::read_dir(dir)? {
-        let path = entry?.path();
-        if path.is_dir() {
-            collect_yaml(base, &path, out)?;
-        } else if path.extension().and_then(|e| e.to_str()) == Some("yaml") {
-            let rel = path
-                .strip_prefix(base)
-                .expect("walked paths sit under their base")
-                .to_string_lossy()
-                .replace('\\', "/");
-            let raw = std::fs::read_to_string(&path)
-                .map_err(|e| anyhow::anyhow!("reading {}: {e}", path.display()))?;
-            out.push((rel, raw));
-        }
-    }
-    Ok(())
-}
-
-/// The offline counterpart to `content::load_needs_from_gitea` -
-/// `needs.yaml` sits at the repo root beside `aggregates.yaml`.
-fn load_needs_from_dir(dir: &str) -> anyhow::Result<Option<String>> {
-    let needs_path = std::path::Path::new(dir)
-        .parent()
-        .unwrap_or_else(|| std::path::Path::new("."))
-        .join("needs.yaml");
-    if !needs_path.exists() {
-        return Ok(None);
-    }
-    std::fs::read_to_string(&needs_path)
-        .map(Some)
-        .map_err(|e| anyhow::anyhow!("reading {}: {e}", needs_path.display()))
-}
-
 /// The business-needs pass (see `portal::needs`): structure always,
 /// then the wording tasks written out and/or an engine's answers
 /// scored, when asked. A repo without `needs.yaml` skips all of it.

src/lib.rs

@@ -16,6 +16,9 @@ pub mod announce;
 #[cfg(feature = "ssr")]
 pub mod needs;
 
+#[cfg(feature = "ssr")]
+pub mod local;
+
 #[cfg(feature = "ssr")]
 pub mod server;
 

src/local.rs

@@ -0,0 +1,100 @@
+//! Loading a content repo from a local checkout - the offline
+//! counterparts to the Gitea loaders in `content`, for tools that lint
+//! or replay a branch that has not been pushed (question_lint,
+//! needs_replay). `dir` is always the pages directory (`questions/`);
+//! `site.yaml`, `aggregates.yaml` and `needs.yaml` sit one level up.
+#![cfg(feature = "ssr")]
+
+use crate::{aggregates, content};
+
+/// The offline counterpart to `content::load_site_from_gitea` -
+/// `site.yaml` lives at the repo root like `aggregates.yaml`, and is
+/// just as optional locally: missing means default branding.
+pub fn load_site_from_dir(dir: &str) -> anyhow::Result<content::SiteConfig> {
+    let site_path = std::path::Path::new(dir)
+        .parent()
+        .unwrap_or_else(|| std::path::Path::new("."))
+        .join("site.yaml");
+    if !site_path.exists() {
+        return Ok(content::SiteConfig::default());
+    }
+    let raw = std::fs::read_to_string(&site_path)
+        .map_err(|e| anyhow::anyhow!("reading {}: {e}", site_path.display()))?;
+    serde_yaml::from_str(&raw).map_err(|e| anyhow::anyhow!("parsing {}: {e}", site_path.display()))
+}
+
+/// The offline counterpart to `content::load_aggregates_from_gitea` -
+/// `aggregates.yaml` lives at the repo root, one level up from the
+/// pages directory `--path` names, so `dir`'s parent is where it's
+/// looked for. Missing entirely is not an error here (unlike
+/// `--repo` mode, where `aggregates.yaml` is required) - a local
+/// checkout being linted may not have one, and every declared
+/// transition still gets checked against whatever *is* found; an
+/// empty map just means nothing is checked.
+pub fn load_aggregates_from_dir(
+    dir: &str,
+) -> anyhow::Result<std::collections::HashMap<String, aggregates::AggregateSchema>> {
+    let aggregates_path = std::path::Path::new(dir)
+        .parent()
+        .unwrap_or_else(|| std::path::Path::new("."))
+        .join("aggregates.yaml");
+    if !aggregates_path.exists() {
+        return Ok(std::collections::HashMap::new());
+    }
+    let raw = std::fs::read_to_string(&aggregates_path)
+        .map_err(|e| anyhow::anyhow!("reading {}: {e}", aggregates_path.display()))?;
+    aggregates::parse_aggregates_yaml(&raw)
+}
+
+/// The offline counterpart to `content::load_questions_from_gitea` -
+/// the same recursive tree walk and `content::build_questions`
+/// pipeline (derived ids, relative refs, sections, followup
+/// inference), just reading a local checkout instead of Gitea's API,
+/// for linting a branch that hasn't been pushed yet.
+pub fn load_from_dir(
+    dir: &str,
+) -> anyhow::Result<std::collections::HashMap<String, content::Question>> {
+    let base = std::path::Path::new(dir);
+    let mut files = Vec::new();
+    collect_yaml(base, base, &mut files)?;
+    content::build_questions(&files)
+}
+
+fn collect_yaml(
+    base: &std::path::Path,
+    dir: &std::path::Path,
+    out: &mut Vec<(String, String)>,
+) -> anyhow::Result<()> {
+    for entry in std::fs::read_dir(dir)? {
+        let path = entry?.path();
+        if path.is_dir() {
+            collect_yaml(base, &path, out)?;
+        } else if path.extension().and_then(|e| e.to_str()) == Some("yaml") {
+            let rel = path
+                .strip_prefix(base)
+                .expect("walked paths sit under their base")
+                .to_string_lossy()
+                .replace('\\', "/");
+            let raw = std::fs::read_to_string(&path)
+                .map_err(|e| anyhow::anyhow!("reading {}: {e}", path.display()))?;
+            out.push((rel, raw));
+        }
+    }
+    Ok(())
+}
+
+/// The offline counterpart to `content::load_needs_from_gitea` -
+/// `needs.yaml` sits at the repo root beside `aggregates.yaml`.
+pub fn load_needs_from_dir(dir: &str) -> anyhow::Result<Option<String>> {
+    let needs_path = std::path::Path::new(dir)
+        .parent()
+        .unwrap_or_else(|| std::path::Path::new("."))
+        .join("needs.yaml");
+    if !needs_path.exists() {
+        return Ok(None);
+    }
+    std::fs::read_to_string(&needs_path)
+        .map(Some)
+        .map_err(|e| anyhow::anyhow!("reading {}: {e}", needs_path.display()))
+}
+