One binary: the lint and the trainer are commands of portal

portal serve runs a site; portal lint is what iris was; portal plan,
build, report and the rest are what portal-trainer was. One source at
one version, so the three never disagree about what content means.
Both histories are merged in (subtree). The release attaches the bare
binary for a content repo's CI. The trainer's plans, intakes, engines
and reference prompt come along; its runs stay in the old repo.

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

src/lint/check.rs

@@ -0,0 +1,248 @@
+//! Headless schema-check binary - loads content and `aggregates.yaml`
+//! (from a Gitea repo URL or a local directory) and validates them
+//! exactly the way `content::watch_for_reload`/`main.rs`'s boot path
+//! do, with no NATS, OIDC, web server, or JetStream connection
+//! involved. Built once by portal's own deploy workflow and run
+//! directly by `questions`' own CI (same bare-metal runner/host,
+//! published to a stable path - no artifact download needed), rather
+//! than compiled there - keeps that repo's CI coupling to "run a
+//! static binary," not "build a Rust workspace."
+
+use crate::lint::local::{load_aggregates_from_dir, load_from_dir, load_needs_from_dir, load_site_from_dir};
+use crate::lint::needs;
+use crate::{aggregates, content};
+
+pub async fn run(argv: Vec<String>) -> anyhow::Result<()> {
+    let mut args = argv.into_iter();
+    let mut repo: Option<String> = None;
+    let mut branch = "main".to_string();
+    let mut subdir = "questions".to_string();
+    let mut path: Option<String> = None;
+    let mut tasks_out: Option<String> = None;
+    let mut answers_in: Option<String> = None;
+    let mut min_accuracy: f64 = 0.0;
+    let mut skim = false;
+    let mut sim_out: Option<String> = None;
+    let mut show_access = false;
+
+    while let Some(arg) = args.next() {
+        match arg.as_str() {
+            "--repo" => repo = args.next(),
+            "--branch" => branch = args.next().unwrap_or(branch),
+            "--subdir" => subdir = args.next().unwrap_or(subdir),
+            "--path" => path = args.next(),
+            "--needs-tasks" => tasks_out = args.next(),
+            "--skim" => skim = true,
+            "--needs-sim" => sim_out = args.next(),
+            "--access" => show_access = true,
+            "--needs-score" => answers_in = args.next(),
+            "--min-accuracy" => {
+                min_accuracy = args.next().and_then(|v| v.parse().ok()).unwrap_or(min_accuracy)
+            }
+            other => {
+                eprintln!("unknown argument: {other}");
+                std::process::exit(2);
+            }
+        }
+    }
+
+    let (questions, aggregates_map, site, needs_raw) = if let Some(dir) = path {
+        (
+            load_from_dir(&dir)?,
+            load_aggregates_from_dir(&dir)?,
+            load_site_from_dir(&dir)?,
+            load_needs_from_dir(&dir)?,
+        )
+    } else if let Some(repo_url) = repo {
+        // The first storyline of each, as the checks read it; the
+        // loaders have held every list of texts to the site's arity.
+        let questions = content::load_questions_from_gitea(&repo_url, &branch, &subdir).await?.at(0).clone();
+        let aggregates_map = content::load_aggregates_from_gitea(&repo_url, &branch).await?.at(0).clone();
+        // load_site_from_gitea already validates; a missing file is
+        // the default config, same as at portal boot.
+        let site = content::load_site_from_gitea(&repo_url, &branch).await?.at(0).clone();
+        // Optional: a fetch failure means the repo declares no needs.
+        let needs_raw = crate::http::get_text(&format!("{}/needs.yaml?ref={branch}", content::gitea_raw_base(&repo_url)?), crate::http::GITEA_CAP).await.ok();
+        (questions, aggregates_map, site, needs_raw)
+    } else {
+        eprintln!(
+            "usage: iris check (--repo <gitea-url> [--branch main] [--subdir questions] | --path <local-dir>) [--needs-tasks <out.jsonl> [--skim]] [--needs-sim <model.json>] [--access] [--needs-score <answers.jsonl> [--min-accuracy 0.8]]"
+        );
+        std::process::exit(2);
+    };
+
+    if let Err(e) = site.validate() {
+        eprintln!("FAIL: {e}");
+        std::process::exit(1);
+    }
+    let aggregates_map = content::with_builtin_aggregates(aggregates_map);
+    match content::validate_questions(&questions, &aggregates_map) {
+        Ok(()) => {
+            // The runtime's own event desk: a repo announcing pages
+            // (Question.event) should read portal_events somewhere, or
+            // "post what happened" tasks pile up unseen. Warn, don't
+            // fail - the runtime creates the records either way.
+            if questions.values().any(|q| q.event.is_some())
+                && !content::bucket_is_read(&questions, content::EVENTS_BUCKET)
+            {
+                eprintln!(
+                    "WARN: pages carry `event:` but no kv resource reads bucket {:?} - add a desk so summaries get posted",
+                    content::EVENTS_BUCKET
+                );
+            }
+            println!(
+                "OK: {} question(s), {} aggregate(s) valid",
+                questions.len(),
+                aggregates_map.len()
+            );
+            // The access policy the site will compile from this content.
+            // It must type-check against portal's schema: a rule the
+            // schema cannot express would grant nothing at runtime.
+            let policy = crate::access::Policy::from_content(&questions, &aggregates_map);
+            if let Err(e) = policy.validate() {
+                eprintln!("FAIL: access policy does not validate: {e}");
+                std::process::exit(1);
+            }
+            println!("OK: access policy, {} rule(s), validates", policy.rules.len());
+            if show_access {
+                println!("\n{:<16} {:<44} {:<40} {}", "ACTION", "RESOURCE", "WHO", "WHEN");
+                let mut rows = policy.rules.clone();
+                rows.sort_by(|a, b| (&a.resource, &a.action, &a.who).cmp(&(&b.resource, &b.action, &b.who)));
+                for r in rows {
+                    println!("{:<16} {:<44} {:<40} {}", r.action, r.resource, r.who, r.when);
+                }
+                println!();
+            }
+            check_needs(
+                needs_raw.as_deref(),
+                &questions,
+                &aggregates_map,
+                tasks_out.as_deref(),
+                skim,
+                sim_out.as_deref(),
+                answers_in.as_deref(),
+                min_accuracy,
+            )
+        }
+        Err(e) => {
+            eprintln!("FAIL: {e}");
+            std::process::exit(1);
+        }
+    }
+}
+
+/// The business-needs pass (see `crate::lint::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.
+fn check_needs(
+    raw: Option<&str>,
+    questions: &std::collections::HashMap<String, content::Question>,
+    aggregates_map: &std::collections::HashMap<String, aggregates::AggregateSchema>,
+    tasks_out: Option<&str>,
+    skim: bool,
+    sim_out: Option<&str>,
+    answers_in: Option<&str>,
+    min_accuracy: f64,
+) -> anyhow::Result<()> {
+    let Some(raw) = raw else {
+        if tasks_out.is_some() || answers_in.is_some() {
+            eprintln!("FAIL: --needs-tasks/--needs-score given, but the repo has no needs.yaml");
+            std::process::exit(1);
+        }
+        return Ok(());
+    };
+    let file = match needs::parse(raw) {
+        Ok(file) => file,
+        Err(e) => {
+            eprintln!("FAIL: {e}");
+            std::process::exit(1);
+        }
+    };
+    let findings = needs::check(&file, questions, aggregates_map);
+    let mut failed = 0;
+    for finding in &findings {
+        let level = match finding.level {
+            needs::Level::Fail => {
+                failed += 1;
+                "FAIL"
+            }
+            needs::Level::Warn => "WARN",
+        };
+        eprintln!("{level}: need {:?}: {}", finding.need, finding.message);
+    }
+    for need in &file.needs {
+        if let Some(path) = needs::path_for(questions, need) {
+            let steps: Vec<String> = path
+                .iter()
+                .map(|s| format!("{} [{}]", s.page, s.alternative))
+                .collect();
+            println!("need {:?}: {} -> {:?}", need.id, steps.join(" -> "), need.lands_in);
+        }
+    }
+    if failed == 0 {
+        let planned = file.needs.iter().filter(|n| n.planned).count();
+        println!(
+            "OK: {} need(s) structurally met, {} planned, {} persona(s)",
+            file.needs.len() - planned,
+            planned,
+            file.personas.len()
+        );
+    }
+    // The wording pass still runs for the needs that do have a path -
+    // a closed flow elsewhere is no reason to stop grading the rest.
+
+    if let Some(out) = sim_out {
+        let model = needs::sim_model(&file, questions, aggregates_map);
+        std::fs::write(out, serde_json::to_string_pretty(&model)?)
+            .map_err(|e| anyhow::anyhow!("writing {out}: {e}"))?;
+        println!("wrote the simulation model to {out}");
+    }
+    let tasks = needs::tasks(&file, questions, skim);
+    if let Some(out) = tasks_out {
+        let mut lines = String::new();
+        for task in &tasks {
+            lines.push_str(&serde_json::to_string(task)?);
+            lines.push('\n');
+        }
+        std::fs::write(out, lines).map_err(|e| anyhow::anyhow!("writing {out}: {e}"))?;
+        println!("wrote {} wording task(s) to {out}", tasks.len());
+    }
+    if let Some(path) = answers_in {
+        let raw = std::fs::read_to_string(path).map_err(|e| anyhow::anyhow!("reading {path}: {e}"))?;
+        let mut answers = Vec::new();
+        for (n, line) in raw.lines().enumerate().filter(|(_, l)| !l.trim().is_empty()) {
+            answers.push(
+                serde_json::from_str::<needs::Answer>(line)
+                    .map_err(|e| anyhow::anyhow!("{path}:{}: {e}", n + 1))?,
+            );
+        }
+        let score = needs::score(&tasks, &answers, 0.6);
+        for (id, picked, expected, confidence) in &score.misroutes {
+            eprintln!(
+                "MISROUTE: {id}: picked {picked:?}{}, the need is met by {expected:?}",
+                confidence.map(|c| format!(" ({c:.2})")).unwrap_or_default()
+            );
+        }
+        for (id, confidence) in &score.hesitant {
+            eprintln!("HESITANT: {id}: right, at {confidence:.2}");
+        }
+        for id in &score.unanswered {
+            eprintln!("UNANSWERED: {id}");
+        }
+        println!(
+            "wording: {}/{} routed right ({:.0}%)",
+            score.correct,
+            score.total,
+            score.accuracy() * 100.0
+        );
+        if score.accuracy() < min_accuracy {
+            eprintln!("FAIL: below --min-accuracy {min_accuracy}");
+            std::process::exit(1);
+        }
+    }
+    if failed > 0 {
+        eprintln!("FAIL: {failed} unmet business need finding(s)");
+        std::process::exit(1);
+    }
+    Ok(())
+}

src/lint/local.rs

@@ -0,0 +1,105 @@
+//! 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.
+
+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()))?;
+    // The first storyline, held to the arity (content::variants).
+    let files = vec![("site.yaml".to_string(), raw)];
+    content::arity(&files)?;
+    serde_yaml::from_str(&content::storyline(&files, 0)[0].1).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()))?;
+    // The first storyline, held to the arity (content::variants).
+    let files = vec![("aggregates.yaml".to_string(), raw)];
+    content::arity(&files)?;
+    aggregates::parse_aggregates_yaml(&content::storyline(&files, 0)[0].1)
+}
+
+/// 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 `check`'s needs.yaml fetch -
+/// `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()))
+}
+

src/lint/mod.rs

@@ -0,0 +1,53 @@
+//! iris - the shield over a portal site. A content push is an incoming
+//! wormhole; nothing comes through until it checks out.
+//!
+//! `iris check` loads a content repo (a Gitea URL or a local checkout)
+//! exactly the way the running site does, validates it, and holds it to
+//! the business needs it declares in `needs.yaml` - a path within
+//! budget for every kind of visitor, a desk for every handling group,
+//! every finished state reachable, no record stranding. It also exports
+//! what a trainer needs to grade the wording: personas as typed choice
+//! tasks, and a resolved simulation model.
+//!
+//! `iris replay` runs simulated cases through the site's real aggregate
+//! engine on a JetStream and reports every bucket by state; with
+//! `--report-only` it takes the same scorecard from a live site's
+//! buckets.
+//!
+//! Every type iris reads is portal's own, pinned to the portal release
+//! it matches, so the lint and the site never disagree about what a
+//! page is.
+
+mod check;
+mod local;
+mod needs;
+mod replay;
+
+/// The lint, as `portal lint ...`: `argv` is everything after `portal`.
+pub async fn run(mut argv: Vec<String>) -> anyhow::Result<()> {
+    // `iris --path questions` reads as `iris check --path questions`,
+    // so the content repos' one-line CI keeps working as it was.
+    match argv.first().map(String::as_str) {
+        Some("check") => {
+            argv.remove(0);
+            check::run(argv).await
+        }
+        Some("replay") => {
+            argv.remove(0);
+            replay::run(argv).await
+        }
+        Some("--version") | Some("-V") => {
+            println!("portal lint v{}", env!("CARGO_PKG_VERSION"));
+            Ok(())
+        }
+        Some(flag) if flag.starts_with("--") => check::run(argv).await,
+        _ => {
+            eprintln!(
+                "iris {v} - matches portal v{v}\n\nusage:\n  iris check (--repo <gitea-url> [--branch main] [--subdir questions] | --path <dir>) [--needs-tasks <out.jsonl> [--skim]] [--needs-sim <model.json>] [--access] [--needs-score <answers.jsonl> [--min-accuracy 0.8]]\n  iris replay --path <dir> --nats <url> (--cases <cases.jsonl> | --report-only)",
+                v = env!("CARGO_PKG_VERSION")
+            );
+            std::process::exit(2)
+        }
+    }
+}
+

src/lint/replay.rs

@@ -0,0 +1,105 @@
+//! 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.
+
+use std::collections::{BTreeMap, HashMap};
+
+use futures::StreamExt;
+use crate::events::store::{ensure_stream, load_events};
+use crate::lint::local::{load_aggregates_from_dir, load_from_dir, load_needs_from_dir};
+use crate::lint::needs;
+use crate::{aggregates, answers, content};
+
+pub async fn run(argv: Vec<String>) -> anyhow::Result<()> {
+    let value = |flag: &str| argv.iter().position(|a| a == flag).and_then(|i| argv.get(i + 1)).cloned();
+    let usage = "usage: iris 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, 0).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(())
+}