diff --git a/README.md b/README.md index 0d2356ff40..027a6b2a7d 100644 --- a/README.md +++ b/README.md @@ -114,7 +114,7 @@ publishing — all backed by YAML provable contracts that fail CI on drift. | Metric | Count | Source of truth | |-------:|------:|---| | Workspace crates | **81** workspace crates | `cargo metadata --no-deps` (NOT `ls crates/` — 4 are `exclude`d, 1 has no Cargo.toml) | -| Provable contracts | **1983** provable contracts | `contracts/census.json` `.n_files` — the set `pv lint` walks (`pv census`, ONT-001 ONT-1; regenerated by `make contracts`, written by `make readme-sync`, guarded by `scripts/check_readme_claims.sh`) | +| Provable contracts | **1984** provable contracts | `contracts/census.json` `.n_files` — the set `pv lint` walks (`pv census`, ONT-001 ONT-1; regenerated by `make contracts`, written by `make readme-sync`, guarded by `scripts/check_readme_claims.sh`) | | CLI commands | **113** CLI commands | `contracts/apr-cli-commands-v1.yaml` §`commands` (parse the list; `apr --help` also lists `help` itself, and `grep -c '^ - name:'` gives 119 — other same-indent `name:` keys exist in the file) | | Book CLI chapters | **114** chapters | `ls book/src/cli/*.md` | | Book lib chapters | **73** chapters | `ls book/src/lib/*.md` (parity with `pub mod`) | @@ -403,7 +403,7 @@ falsification_tests: prediction: apr validate bad-model.apr exits non-zero ``` -The tree carries 1983 contracts across inference, training, quantization, attention, FFN, +The tree carries 1984 contracts across inference, training, quantization, attention, FFN, tokenization, model formats, CLI safety — and this README itself. ## Migration from old crates diff --git a/ci/explicit-test-commands.d/763-aprender-simulate-pr-flow-contract.cmd b/ci/explicit-test-commands.d/763-aprender-simulate-pr-flow-contract.cmd new file mode 100644 index 0000000000..836bbe9141 --- /dev/null +++ b/ci/explicit-test-commands.d/763-aprender-simulate-pr-flow-contract.cmd @@ -0,0 +1 @@ +cargo test -p aprender-simulate --test pr_flow_contract diff --git a/contracts/apr-cli-commands-v1.yaml b/contracts/apr-cli-commands-v1.yaml index a79469f533..44fafd6f0c 100644 --- a/contracts/apr-cli-commands-v1.yaml +++ b/contracts/apr-cli-commands-v1.yaml @@ -901,7 +901,7 @@ commands: side_effects: [filesystem] # `version` comes from #2527's simular clap conversion, which the batch # takes over #2493's. FALSIFY-CLI-006 caught the omission. - subcommands: [emc-check, emc-validate, list-emc, render, run, validate, verify, version] + subcommands: [course-flow, emc-check, emc-validate, list-emc, pr-flow, render, repo-flow, run, validate, verify, version] - name: cgp category: tools diff --git a/contracts/bin-aprender-simulate--aprender-simulate-v1.yaml b/contracts/bin-aprender-simulate--aprender-simulate-v1.yaml index 7a81ab271c..eda157ec9c 100644 --- a/contracts/bin-aprender-simulate--aprender-simulate-v1.yaml +++ b/contracts/bin-aprender-simulate--aprender-simulate-v1.yaml @@ -69,14 +69,17 @@ shapes: # G0.1 — a violation since batch/ont-10 2026-09-27 (build.rs embeds APR_GIT_SHA); was a warning while RED - {path: "bin:versionString", minCount: 1, maxCount: 1, pattern: "[0-9a-f]{9}"} - path: bin:command - minCount: 9 - maxCount: 9 + minCount: 12 + maxCount: 12 in: + - "course-flow" - "emc-check" - "emc-validate" - "help" - "list-emc" + - "pr-flow" - "render" + - "repo-flow" - "run" - "validate" - "verify" diff --git a/contracts/contracts.nt b/contracts/contracts.nt index c34b6bfbee..c3ed1fb995 100644 --- a/contracts/contracts.nt +++ b/contracts/contracts.nt @@ -9709,6 +9709,13 @@ "corpus-assembly"^^ . "PROPOSED"^^ . "1.0.0"^^ . + . + . + . + "contracts/course-flow-batch-wip-v1.yaml"^^ . + "course-flow-batch-wip-v1"^^ . + "course-flow-batch-wip"^^ . + "1.0.0"^^ . . . . diff --git a/contracts/course-flow-batch-wip-v1.yaml b/contracts/course-flow-batch-wip-v1.yaml new file mode 100644 index 0000000000..3e369efed0 --- /dev/null +++ b/contracts/course-flow-batch-wip-v1.yaml @@ -0,0 +1,47 @@ +metadata: + version: "1.0.0" + created: "2026-10-18" + author: PAIML Engineering + registry: true + references: + - "Little, J.D.C. (1961), A Proof for the Queuing Formula L = lambda W." + - "flow model spec, internal (eq 1)" + description: "course_flow counts the daily human batch gate inside WIP and reports its mean wait" + +name: course-flow-batch-wip +version: "1.0.0" +scope: "crates/aprender-simulate/src/scenarios/course_flow.rs" + +description: | + A daily human batch gate holds finished items until its next release. With + batch_in_wip set, an item holds its WIP slot while it waits there, so Little's + law L = lambda W closes on a window that includes the gate. The mean time an item + spends at the gate is reported as mean_batch_wait_h. + +equations: + little_with_gate: + description: "|lambda W - L| / L <= 0.005 with WIP including the batch wait" + formula: "|lambda * W - L| / L <= 0.005" + + gate_wait_bounded: + description: "mean batch wait lies inside one gate interval" + formula: "0 < mean_batch_wait_h < 24" + +falsification_tests: + - id: FALSIFY-001 + description: "Little's law holds within 0.5% with the batch wait inside WIP" + test: "cargo test -p aprender-simulate --lib littles_law_holds_with_batch_wait_inside_wip" + prediction: "the test passes" + if_fails: "the batch gate is not counted inside WIP as the contract states" + + - id: FALSIFY-002 + description: "mean_batch_wait_h is reported and lies inside one day" + test: "cargo test -p aprender-simulate --lib mean_batch_wait_is_reported_and_inside_a_day" + prediction: "the test passes" + if_fails: "the batch gate is not counted inside WIP as the contract states" + + - id: FALSIFY-003 + description: "counting the batch inside WIP changes the result (not a no-op)" + test: "cargo test -p aprender-simulate --lib batch_in_wip_is_not_a_no_op" + prediction: "the test passes" + if_fails: "the batch gate is not counted inside WIP as the contract states" diff --git a/contracts/shapes.ttl b/contracts/shapes.ttl index 3d3f6cdd2a..a718856fa0 100644 --- a/contracts/shapes.ttl +++ b/contracts/shapes.ttl @@ -908,9 +908,9 @@ ] ; sh:property [ sh:path ; - sh:minCount 9 ; - sh:maxCount 9 ; - sh:in ( "emc-check" "emc-validate" "help" "list-emc" "render" "run" "validate" "verify" "version" ) ; + sh:minCount 12 ; + sh:maxCount 12 ; + sh:in ( "course-flow" "emc-check" "emc-validate" "help" "list-emc" "pr-flow" "render" "repo-flow" "run" "validate" "verify" "version" ) ; ] ; sh:property [ sh:path ; diff --git a/contracts/witness/811c78184eb2609244c59534f015f0be2f90923120a3027c176ad56723d0d850.json b/contracts/witness/811c78184eb2609244c59534f015f0be2f90923120a3027c176ad56723d0d850.json index 184d38c1eb..6bb276cb32 100644 --- a/contracts/witness/811c78184eb2609244c59534f015f0be2f90923120a3027c176ad56723d0d850.json +++ b/contracts/witness/811c78184eb2609244c59534f015f0be2f90923120a3027c176ad56723d0d850.json @@ -1,7 +1,7 @@ { - "census_id_set_sha256": "021e98642f5a6c17b5389d388a0d4b18df2707f513fda99ae35b661dca0b03c3", + "census_id_set_sha256": "7597b4ad189f534dd58cda55c9b1fe29e6339ddb77fd5f40969b4a028eed8f88", "relations_sha256": "811c78184eb2609244c59534f015f0be2f90923120a3027c176ad56723d0d850", - "reasoner_git_sha": "a56c47b937da11a6c94132b89c28d9420a84f372", + "reasoner_git_sha": "461bf9ef1921a8fbea4bc338f21b566ec342a95e", "checkable_n": 87, "result": { "kind": "model", diff --git a/crates/aprender-simulate/src/cli/args.rs b/crates/aprender-simulate/src/cli/args.rs index a7b0da689e..014e33985a 100644 --- a/crates/aprender-simulate/src/cli/args.rs +++ b/crates/aprender-simulate/src/cli/args.rs @@ -62,6 +62,7 @@ impl Cli { /// Available CLI commands. #[derive(Debug, Clone, PartialEq, Subcommand)] +#[allow(clippy::large_enum_variant)] // PrFlow carries the measured-input overrides; parsed once at startup pub enum Commands { /// Run an experiment Run { @@ -95,6 +96,99 @@ pub enum Commands { #[arg(long, default_value_t = 42)] seed: u64, }, + /// Run the aprender PR-flow scenario (FLOW-005) for a named design + PrFlow { + /// Design: D, A, G or E. + #[arg(long, default_value = "D", value_parser = ["D", "A", "G", "E"])] + design: String, + /// Print the cycle-record "sim" block as one JSON line. + #[arg(long)] + json: bool, + /// Override the PR-check runner pool (measured input). + #[arg(long)] + r_pr: Option, + /// Override the merge-queue runner pool (measured input). + #[arg(long)] + r_mq: Option, + /// Override the WIP cap (measured input). + #[arg(long)] + cap: Option, + /// Override concurrent merge-queue groups (measured input). + #[arg(long)] + builds: Option, + /// Override the merge-group flake rate (measured input). + #[arg(long)] + f: Option, + /// Override mean concurrent authors (may be fractional). + #[arg(long)] + n: Option, + /// Measured PR-check runner-minutes per run. + #[arg(long)] + pr_min: Option, + /// Measured merge-queue runner-minutes per run. + #[arg(long)] + mq_min: Option, + /// Measured share of PR runs that go red. + #[arg(long)] + p_red: Option, + /// Measured queue wait per PR run, minutes. + #[arg(long)] + pr_setup: Option, + /// Review wait per PR after green checks, minutes (default 0). + #[arg(long)] + t_rev: Option, + /// Run the review alongside the checks instead of after them. + #[arg(long)] + rev_concurrent: bool, + /// Parallel reviewers (0 = unlimited). + #[arg(long)] + rev_lanes: Option, + /// Merged PRs in the last 24 h, from git (for drift_pct_24h). + #[arg(long)] + observed_24h: Option, + /// Comma-separated run ids the measured inputs came from. + #[arg(long)] + src_runs: Option, + /// Arbiter params file for one repo (ARB-SOT-001 §3); prints {actual, one_change, optimum}. + #[arg(long)] + params: Option, + /// Seeds 1..=N for `--params`. + #[arg(long, default_value_t = 5)] + seeds: u64, + /// Simulated hours per seed for `--params`. + #[arg(long, default_value_t = 400.0)] + hours: f64, + /// Warm-up hours discarded for `--params`. + #[arg(long, default_value_t = 20.0)] + warmup: f64, + /// Commit the sim block is stamped with. + #[arg(long, default_value = "unknown")] + sha: String, + }, + /// Run a FLOW-006 repo case (conflict step, optional merge queue); --json prints the sim block + RepoFlow { + /// Case name, for example rmedia-design or infra-rule. + #[arg(long)] + case: String, + /// JSON file of measured inputs from the ledger; overrides the case. + #[arg(long)] + inputs: Option, + /// Print the sim block as JSON. + #[arg(long)] + json: bool, + }, + /// Run a FLOW-006 course-factory case; --json prints the sim block + CourseFlow { + /// Case: cal-today, design, nine-builders, coursera-24h or render-300. + #[arg(long)] + case: String, + /// JSON file of measured inputs from the ledger; overrides the case. + #[arg(long)] + inputs: Option, + /// Print the sim block as JSON. + #[arg(long)] + json: bool, + }, /// Validate experiment YAML against EDD v2 schema Validate { /// Path to the experiment YAML file. diff --git a/crates/aprender-simulate/src/cli/commands.rs b/crates/aprender-simulate/src/cli/commands.rs index c8a006784c..e36932a4da 100644 --- a/crates/aprender-simulate/src/cli/commands.rs +++ b/crates/aprender-simulate/src/cli/commands.rs @@ -40,6 +40,56 @@ pub fn run_cli(cli: Cli) -> ExitCode { duration, seed, } => render_svg(&domain, &format, &output, fps, duration, seed), + Commands::PrFlow { + design, + json, + r_pr, + r_mq, + cap, + builds, + f, + n, + pr_min, + mq_min, + p_red, + pr_setup, + t_rev, + rev_concurrent, + rev_lanes, + observed_24h: obs24, + src_runs: runs, + params, + seeds, + hours, + warmup, + sha, + } => match params { + Some(path) => pr_flow_params(&path, seeds, hours, warmup), + None => pr_flow( + &design, + json, + &FlowOverrides { + r_pr, + r_mq, + cap, + builds, + f, + n, + pr_min, + mq_min, + p_red, + pr_setup, + t_rev, + rev_concurrent, + rev_lanes, + obs24, + runs, + }, + &sha, + ), + }, + Commands::RepoFlow { case, inputs, json } => repo_flow(&case, inputs.as_deref(), json), + Commands::CourseFlow { case, inputs, json } => course_flow(&case, inputs.as_deref(), json), Commands::Validate { experiment_path } => validate_experiment(&experiment_path), Commands::Verify { experiment_path, @@ -564,3 +614,255 @@ fn render_bouncing_balls( } Ok(()) } + +/// Measured inputs that replace a design's defaults; each is labelled `measured` in the sim block. +#[derive(Default, Clone)] +struct FlowOverrides { + r_pr: Option, + r_mq: Option, + cap: Option, + builds: Option, + f: Option, + n: Option, + pr_min: Option, + mq_min: Option, + p_red: Option, + pr_setup: Option, + t_rev: Option, + rev_concurrent: bool, + rev_lanes: Option, + obs24: Option, + runs: Option, +} + +impl FlowOverrides { + fn apply(&self, c: &mut crate::scenarios::PrFlowConfig) { + c.r_pr = self.r_pr.unwrap_or(c.r_pr); + c.r_mq = self.r_mq.unwrap_or(c.r_mq); + c.cap = self.cap.unwrap_or(c.cap); + c.builds = self.builds.unwrap_or(c.builds); + c.f = self.f.unwrap_or(c.f); + c.n = self.n.unwrap_or(c.n); + c.pr_runner_min = self.pr_min.unwrap_or(c.pr_runner_min); + c.mq_runner_min = self.mq_min.unwrap_or(c.mq_runner_min); + c.p_red = self.p_red.unwrap_or(c.p_red); + c.pr_setup_min = self.pr_setup.unwrap_or(c.pr_setup_min); + c.t_rev_min = self.t_rev.unwrap_or(c.t_rev_min); + if self.rev_concurrent { + c.rev_mode = crate::scenarios::ReviewMode::Concurrent; + } + c.rev_lanes = self.rev_lanes.unwrap_or(c.rev_lanes); + } + + fn cli_flags(&self) -> String { + let parts = [ + self.r_pr.map(|v| format!(" --r-pr {v}")), + self.r_mq.map(|v| format!(" --r-mq {v}")), + self.cap.map(|v| format!(" --cap {v}")), + self.builds.map(|v| format!(" --builds {v}")), + self.f.map(|v| format!(" --f {v}")), + self.n.map(|v| format!(" --n {v}")), + self.pr_min.map(|v| format!(" --pr-min {v}")), + self.mq_min.map(|v| format!(" --mq-min {v}")), + self.p_red.map(|v| format!(" --p-red {v}")), + self.pr_setup.map(|v| format!(" --pr-setup {v}")), + self.t_rev.map(|v| format!(" --t-rev {v}")), + self.rev_concurrent.then(|| " --rev-concurrent".to_string()), + self.rev_lanes.map(|v| format!(" --rev-lanes {v}")), + self.obs24.map(|v| format!(" --observed-24h {v}")), + self.runs.as_ref().map(|v| format!(" --src-runs {v}")), + ]; + parts.into_iter().flatten().collect() + } +} + +/// Print the reference run (5 seeds x 400 h, 20 h warm-up) of a named PR-flow design. +fn pr_flow(design: &str, json: bool, ov: &FlowOverrides, sha: &str) -> ExitCode { + let Some(mut cfg) = crate::scenarios::preset(design) else { + eprintln!("unknown design: {design}"); + return ExitCode::FAILURE; + }; + ov.apply(&mut cfg); + if json { + println!("{}", pr_flow_sim_block(design, &cfg, ov, sha)); + return ExitCode::SUCCESS; + } + let r = crate::scenarios::run_reference(&cfg); + println!( + "design {design}: merged/h {:.2} W mean {:.1} min W p90 {:.1} min L {:.2}", + r.merged_per_h, r.w_mean_min, r.w_p90_min, r.l + ); + println!( + "PRs/pass {:.2} PR util {:.0}% MQ util {:.0}% runs/PR {:.2}", + r.prs_per_pass, + r.pr_util * 100.0, + r.mq_util * 100.0, + r.runs_per_pr + ); + println!( + "stages: run wall {:.1} open->green {:.1} green->armed {:.1} armed->merged {:.1} min", + r.run_wall_min, r.checks_min, r.arm_min, r.mq_min + ); + ExitCode::SUCCESS +} + +/// The cycle-record "sim" block: seeds 1..=5, 400 h, 20 h warm-up, one JSON line. +/// `optimum` compares the design's own reference run with this run; `drift_pct_24h` is null (not computed here). +fn pr_flow_sim_block( + design: &str, + cfg: &crate::scenarios::PrFlowConfig, + ov: &FlowOverrides, + sha: &str, +) -> String { + let runs: Vec<_> = (1..=5) + .map(|s| crate::scenarios::run_pr_flow(cfg, s, 400.0, 20.0)) + .collect(); + let n = runs.len() as f64; + let mean = |f: fn(&crate::scenarios::PrFlowResult) -> f64| runs.iter().map(f).sum::() / n; + let m = mean(|r| r.merged_per_h); + let var = runs + .iter() + .map(|r| (r.merged_per_h - m).powi(2)) + .sum::() + / (n - 1.0); + let ci95 = 1.96 * (var / n).sqrt(); + let avg = crate::scenarios::PrFlowResult { + w_mean_min: mean(|r| r.w_mean_min), + runs_per_pr: mean(|r| r.runs_per_pr), + prs_per_pass: mean(|r| r.prs_per_pass), + run_wall_min: mean(|r| r.run_wall_min), + pass_wait_min: mean(|r| r.pass_wait_min), + pass_wall_min: mean(|r| r.pass_wall_min), + ..Default::default() + }; + let binding = crate::scenarios::binding_station(cfg, &avg); + let law = crate::scenarios::eq7_law(cfg, &avg); + let w_sim = mean(|r| r.w_mean_min); + let base = crate::scenarios::preset(design) + .map_or(0.0, |c| crate::scenarios::run_reference(&c).merged_per_h); + let contract = crate::scenarios::reference_contract_ok(); + let src = |m: bool| if m { "measured" } else { "design_default" }; + serde_json::json!({ + "aprender_sha": sha, + "scenario": "pr-flow", + "cli": format!("aprender-simulate pr-flow --design {design} --json{} --sha {sha}", ov.cli_flags()), + "seeds": [1, 2, 3, 4, 5], + "hours": 400, + "inputs": { + "design": design, "cap": cfg.cap, "n": cfg.n, "t_author_min": cfg.t_author_min, + "r_pr": cfg.r_pr, "r_mq": cfg.r_mq, "builds": cfg.builds, "p_red": cfg.p_red, + "pr_runner_min": cfg.pr_runner_min, "mq_runner_min": cfg.mq_runner_min, + "pr_setup_min": cfg.pr_setup_min, "t_rev_min": cfg.t_rev_min, "f": cfg.f + }, + "inputs_source": { + "t_author_min": "design_default", "n": src(ov.n.is_some()), "cap": src(ov.cap.is_some()), + "r_pr": src(ov.r_pr.is_some()), "r_mq": src(ov.r_mq.is_some()), "p_red": src(ov.p_red.is_some()), + "pr_runner_min": src(ov.pr_min.is_some()), "mq_runner_min": src(ov.mq_min.is_some()), + "pr_setup_min": src(ov.pr_setup.is_some()), "f": src(ov.f.is_some()), "builds": src(ov.builds.is_some()) + }, + "predicted": { + "merged_h": m, "ci95_h": ci95, + "W_min_mean": w_sim, "L": mean(|r| r.l), "rho_a": mean(|r| r.rho_a), "rho_a_assumed": 0.95 + }, + "law": { + "eq": 7, "W_law_min": law.w_law_min(), "gap_pct": (w_sim - law.w_law_min()) / law.w_law_min() * 100.0, + "r": law.r, "w_min": law.w_min, "c_wall_min": law.c_wall_min, "t_fix_min": law.t_fix_min, + "t_arm_min": law.t_arm_min, "p_dirty": law.p_dirty, "t_reb_min": law.t_reb_min, + "wait_mq_min": law.wait_mq_min, "t_pass_min": law.t_pass_min, "w_pass_min": law.w_pass_min, "f": law.f, + "checks_min": law.checks_min(), "rebase_min": law.rebase_min(), "W_mq_min": law.w_mq_min() + }, + "binding_station": binding, + "optimum": { + "design": design, "design_merged_h": base, "this_run_merged_h": m, + "gap_h": base - m, "gap_pct": if base > 0.0 { (base - m) / base * 100.0 } else { 0.0 } + }, + "sim3_contract": if contract { "pass" } else { "fail" }, + "observed_24h": ov.obs24, "source_runs": ov.runs, + "drift_pct_24h": ov.obs24.map(|o| (m - o / 24.0) / (o / 24.0) * 100.0) + }) + .to_string() +} + +/// `pr-flow --params `: one repo's arbiter params in, `{actual, one_change, optimum}` out (one JSON line). +fn pr_flow_params(path: &Path, seeds: u64, hours: f64, warmup: f64) -> ExitCode { + let text = match std::fs::read_to_string(path) { + Ok(t) => t, + Err(e) => { + eprintln!("params: cannot read {}: {e}", path.display()); + return ExitCode::FAILURE; + } + }; + match crate::scenarios::parse_params(&text) { + Ok(p) => { + println!("{}", crate::scenarios::evaluate(&p, seeds, hours, warmup)); + ExitCode::SUCCESS + } + Err(e) => { + eprintln!("{e}"); + ExitCode::FAILURE + } + } +} + +/// Print the reference run of a repo case, optionally with the ledger's inputs, as text or JSON. +fn repo_flow(case: &str, inputs: Option<&Path>, json: bool) -> ExitCode { + let Some(mut cfg) = crate::scenarios::repo_preset(case) else { + eprintln!("unknown case: {case}"); + return ExitCode::FAILURE; + }; + if let Some(p) = inputs { + let applied = std::fs::read_to_string(p) + .map_err(|e| e.to_string()) + .and_then(|s| serde_json::from_str(&s).map_err(|e| e.to_string())) + .and_then(|v| crate::scenarios::apply_inputs(&mut cfg, &v)); + if let Err(e) = applied { + eprintln!("inputs: {e}"); + return ExitCode::FAILURE; + } + } + let r = crate::scenarios::run_repo_reference(&cfg); + if json { + println!("{}", crate::scenarios::sim_block(case, &cfg, &r)); + } else { + println!( + "case {case}: merged/h {:.3} W mean {:.1} min L {:.2} pool {:.0}%", + r.merged_per_h, + r.w_mean_min, + r.l, + r.pr_util * 100.0 + ); + } + ExitCode::SUCCESS +} + +/// Print the reference run of a course-factory case, optionally with the ledger's inputs. +fn course_flow(case: &str, inputs: Option<&Path>, json: bool) -> ExitCode { + let Some(mut cfg) = crate::scenarios::course_preset(case) else { + eprintln!("unknown case: {case}"); + return ExitCode::FAILURE; + }; + if let Some(p) = inputs { + let applied = std::fs::read_to_string(p) + .map_err(|e| e.to_string()) + .and_then(|s| serde_json::from_str(&s).map_err(|e| e.to_string())) + .and_then(|v| crate::scenarios::apply_course_inputs(&mut cfg, &v)); + if let Err(e) = applied { + eprintln!("inputs: {e}"); + return ExitCode::FAILURE; + } + } + let r = crate::scenarios::run_course_reference(&cfg); + if json { + println!("{}", crate::scenarios::course_sim_block(case, &cfg, &r)); + } else { + let (st, u) = r.busiest(); + println!( + "case {case}: courses/day {:.2} shorts/day {:.2} lead {:.1} d busiest {st} {:.0}%", + r.courses_per_day, + r.shorts_per_day, + r.lead_days, + u * 100.0 + ); + } + ExitCode::SUCCESS +} diff --git a/crates/aprender-simulate/src/cli/tests.rs b/crates/aprender-simulate/src/cli/tests.rs index 2305aa9e59..14fa46ef6b 100644 --- a/crates/aprender-simulate/src/cli/tests.rs +++ b/crates/aprender-simulate/src/cli/tests.rs @@ -421,6 +421,16 @@ fn test_every_subcommand_is_reachable() { vec!["aprender-simulate", "emc-validate", "e.emc.yaml"], |c| matches!(c, Commands::EmcValidate { .. }), ), + (vec!["aprender-simulate", "pr-flow"], |c| { + matches!(c, Commands::PrFlow { .. }) + }), + (vec!["aprender-simulate", "repo-flow", "--case", "x"], |c| { + matches!(c, Commands::RepoFlow { .. }) + }), + ( + vec!["aprender-simulate", "course-flow", "--case", "x"], + |c| matches!(c, Commands::CourseFlow { .. }), + ), (vec!["aprender-simulate", "list-emc"], |c| { matches!(c, Commands::ListEmc) }), @@ -435,7 +445,7 @@ fn test_every_subcommand_is_reachable() { // Non-vacuity: the table must cover every variant of `Commands`. assert_eq!( cases.len(), - 9, + 12, "add the new subcommand to this table when Commands grows" ); assert_eq!( diff --git a/crates/aprender-simulate/src/scenarios/arbiter.rs b/crates/aprender-simulate/src/scenarios/arbiter.rs new file mode 100644 index 0000000000..50e0428242 --- /dev/null +++ b/crates/aprender-simulate/src/scenarios/arbiter.rs @@ -0,0 +1,306 @@ +//! SIM-2 (FLOW-005, ARB-SOT-001 §3): the arbiter's `pr-flow --params` surface. +//! +//! Arbiter writes one params file per repo; this module reads it through ONE adapter +//! ([`parse_params`]), runs the PR-flow model for the repo as measured (`actual`), for the +//! design (`optimum`), and for each design change alone (`one_change`), and names the binding +//! station as the single change that moves merged/h the most. +//! +//! The params shape is provisional until infra pins `arbiter-sim-params-v1`: only +//! [`parse_params`] knows it, so the pin edits one function. + +use super::pr_flow::{preset, run_pr_flow, PrFlowConfig, PrFlowResult}; +use serde_json::{json, Value}; + +/// Params fields ARB-SOT-001 §3 item 1 lists, in the order it lists them. +const FIELDS: [&str; 14] = [ + "n", + "t_author", + "cap", + "r_pr", + "pr_runner_min", + "mq_runner_min", + "p_red", + "t_fix", + "t_arm", + "builds", + "batch", + "f", + "p_dirty", + "t_rebase", +]; + +/// Fields the model does not read yet (SIM-6 adds them); reported, never silently dropped. +const UNMODELLED: [&str; 2] = ["p_dirty", "t_rebase"]; + +/// A repo's parsed params. +#[derive(Debug, Clone, PartialEq)] +pub struct Params { + pub repo: String, + pub design: String, + pub cfg: PrFlowConfig, + /// Fields carrying a measured value, out of [`FIELDS`], in percent. + pub measured_pct: f64, + /// Names of the fields that fell back to the design value. + pub design_default: Vec, +} + +/// Read a params file: `{"repo", "design", "params": {field: number | {"value", "source", "window"}}}`. +/// A missing field, or one whose value is not a number, is `design_default`. +/// +/// # Errors +/// A message naming the problem: not JSON, no `repo`, unknown `design`, a negative or non-finite value. +pub fn parse_params(text: &str) -> Result { + let v: Value = serde_json::from_str(text).map_err(|e| format!("params: not JSON: {e}"))?; + let repo = v["repo"] + .as_str() + .ok_or("params: missing `repo`")? + .to_owned(); + let design = v["design"].as_str().unwrap_or("D").to_owned(); + let mut cfg = preset(&design).ok_or_else(|| format!("params: unknown design `{design}`"))?; + let mut measured = 0usize; + let mut design_default = Vec::new(); + for name in FIELDS { + let raw = &v["params"][name]; + let val = raw.get("value").unwrap_or(raw).as_f64(); + match val { + Some(x) if x.is_finite() && x >= 0.0 => { + measured += 1; + set_field(&mut cfg, name, x); + } + Some(x) => { + return Err(format!( + "params: `{name}` = {x} is not a non-negative number" + )) + } + None => design_default.push(name.to_owned()), + } + } + #[allow(clippy::cast_precision_loss)] + let measured_pct = measured as f64 / FIELDS.len() as f64 * 100.0; + Ok(Params { + repo, + design, + cfg, + measured_pct, + design_default, + }) +} + +#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)] +fn set_field(c: &mut PrFlowConfig, name: &str, x: f64) { + match name { + "n" => c.n = x, + "t_author" => c.t_author_min = x, + "cap" => c.cap = x as usize, + "r_pr" => c.r_pr = x as usize, + "pr_runner_min" => c.pr_runner_min = x, + "mq_runner_min" => c.mq_runner_min = x, + "p_red" => c.p_red = x, + "t_fix" => c.t_fix_min = x, + "t_arm" => c.t_arm_min = x, + "builds" => c.builds = x as usize, + "batch" => c.mq_batch = x as usize, + "f" => c.f = x, + _ => {} // p_dirty, t_rebase: see UNMODELLED + } +} + +/// Mean over seeds `1..=seeds`. +fn run_mean(c: &PrFlowConfig, seeds: u64, hours: f64, warmup_h: f64) -> (PrFlowResult, f64) { + let rs: Vec = (1..=seeds) + .map(|s| run_pr_flow(c, s, hours, warmup_h)) + .collect(); + #[allow(clippy::cast_precision_loss)] + let n = rs.len() as f64; + let mean = |f: fn(&PrFlowResult) -> f64| rs.iter().map(f).sum::() / n; + let m = mean(|r| r.merged_per_h); + let var = if rs.len() > 1 { + rs.iter().map(|r| (r.merged_per_h - m).powi(2)).sum::() / (n - 1.0) + } else { + 0.0 + }; + let r = PrFlowResult { + merged_per_h: m, + w_mean_min: mean(|r| r.w_mean_min), + w_p90_min: mean(|r| r.w_p90_min), + l: mean(|r| r.l), + prs_per_pass: mean(|r| r.prs_per_pass), + pr_util: mean(|r| r.pr_util), + mq_util: mean(|r| r.mq_util), + runs_per_pr: mean(|r| r.runs_per_pr), + rho_a: mean(|r| r.rho_a), + run_wall_min: mean(|r| r.run_wall_min), + checks_min: mean(|r| r.checks_min), + arm_min: mean(|r| r.arm_min), + mq_min: mean(|r| r.mq_min), + pass_wait_min: mean(|r| r.pass_wait_min), + pass_wall_min: mean(|r| r.pass_wall_min), + }; + (r, 1.96 * (var / n).sqrt()) +} + +/// The design changes, each as "move this repo's supply fields to the design's". +fn changes(actual: &PrFlowConfig, design: &PrFlowConfig) -> Vec<(&'static str, PrFlowConfig)> { + type Apply = fn(&mut PrFlowConfig, &PrFlowConfig); + let table: [(&'static str, Apply); 8] = [ + ("cap", |a, d| a.cap = d.cap), + ("r_pr", |a, d| a.r_pr = d.r_pr), + ("r_mq", |a, d| a.r_mq = d.r_mq), + ("builds", |a, d| a.builds = d.builds), + ("runner_min", |a, d| { + a.pr_runner_min = d.pr_runner_min; + a.mq_runner_min = d.mq_runner_min; + }), + ("p_red", |a, d| a.p_red = d.p_red), + ("flake", |a, d| a.f = d.f), + ("setup", |a, d| { + a.pr_setup_min = d.pr_setup_min; + a.mq_setup_min = d.mq_setup_min; + }), + ]; + table + .into_iter() + .filter_map(|(name, apply)| { + let mut c = actual.clone(); + apply(&mut c, design); + (&c != actual).then_some((name, c)) + }) + .collect() +} + +/// Run the model for one repo and build the §3 item 5 row (`actual`, `optimum`, `one_change`). +#[must_use] +pub fn evaluate(p: &Params, seeds: u64, hours: f64, warmup_h: f64) -> Value { + let design = preset(&p.design).unwrap_or_else(|| p.cfg.clone()); + let (actual, ci95) = run_mean(&p.cfg, seeds, hours, warmup_h); + let moves = changes(&p.cfg, &design); + let one_change: Vec = moves + .iter() + .map(|(name, c)| json!({"change": name, "merged_h": run_mean(c, seeds, hours, warmup_h).0.merged_per_h})) + .collect(); + let mut opt_cfg = p.cfg.clone(); + for (_, c) in &moves { + opt_cfg = merge_supply(&opt_cfg, c, &p.cfg); + } + let (opt, _) = run_mean(&opt_cfg, seeds, hours, warmup_h); + let binding = one_change + .iter() + .max_by(|a, b| { + a["merged_h"] + .as_f64() + .partial_cmp(&b["merged_h"].as_f64()) + .unwrap_or(std::cmp::Ordering::Equal) + }) + .map_or(Value::Null, |c| c["change"].clone()); + json!({ + "repo": p.repo, + "params_measured_pct": p.measured_pct, + "design_default": p.design_default, + "unmodelled": UNMODELLED, + "actual": { + "merged_h": actual.merged_per_h, "ci95_h": ci95, + "W_min_mean": actual.w_mean_min, "W_min_p90": actual.w_p90_min, "L": actual.l, + "pr_pool_util_pct": actual.pr_util * 100.0, "mq_util_pct": actual.mq_util * 100.0, + "runs_per_pr": actual.runs_per_pr, + "rho_a": actual.rho_a + }, + "binding_station": binding, + "optimum": { + "scenario": p.design, "merged_h": opt.merged_per_h, + "changes": moves.iter().map(|(n, _)| *n).collect::>() + }, + "one_change": one_change + }) +} + +/// `base` with every field that `changed` moved off `orig` taken from `changed`. +fn merge_supply(base: &PrFlowConfig, changed: &PrFlowConfig, orig: &PrFlowConfig) -> PrFlowConfig { + let mut c = base.clone(); + macro_rules! take { ($($f:ident),*) => { $( if changed.$f != orig.$f { c.$f = changed.$f; } )* } } + take!( + cap, + r_pr, + r_mq, + builds, + pr_runner_min, + mq_runner_min, + p_red, + f, + pr_setup_min, + mq_setup_min + ); + c +} + +#[cfg(test)] +mod tests { + use super::*; + + const D_ALL: &str = r#"{"repo":"aprender","design":"D","params":{"n":11.1,"cap":10,"r_pr":28, + "builds":1,"f":0.12,"p_red":{"value":0.62,"source":"jobs.tsv","window":"7d"}}}"#; + + #[test] + fn missing_fields_are_design_defaults_and_counted() { + let p = parse_params(D_ALL).expect("parse"); + assert_eq!(p.cfg.cap, 10); + assert!((p.cfg.p_red - 0.62).abs() < 1e-12); + assert_eq!(p.cfg.pr_runner_min, preset("D").expect("D").pr_runner_min); + assert!(p.design_default.contains(&"t_arm".to_owned())); + assert!((p.measured_pct - 6.0 / 14.0 * 100.0).abs() < 1e-9); + } + + #[test] + fn bad_params_are_refused_by_name() { + assert!(parse_params("nope").is_err()); + assert!(parse_params(r#"{"design":"D"}"#) + .unwrap_err() + .contains("repo")); + assert!(parse_params(r#"{"repo":"x","design":"Z"}"#) + .unwrap_err() + .contains("design")); + assert!(parse_params(r#"{"repo":"x","params":{"cap":-1}}"#) + .unwrap_err() + .contains("cap")); + } + + #[test] + fn a_design_point_repo_has_no_changes_and_equals_the_reference() { + let p = parse_params(r#"{"repo":"x","design":"E"}"#).expect("parse"); + let v = evaluate(&p, 5, 400.0, 20.0); + assert!(v["one_change"].as_array().expect("arr").is_empty()); + assert!(v["binding_station"].is_null()); + let m = v["actual"]["merged_h"].as_f64().expect("num"); + assert!((m - 21.37).abs() <= 21.37 * 0.05, "{m}"); + assert_eq!(v["params_measured_pct"], 0.0); + } + + #[test] + fn the_binding_change_is_the_one_that_moves_merged_h_most() { + let p = parse_params(D_ALL).expect("parse"); + let v = evaluate(&p, 3, 200.0, 10.0); + let rows = v["one_change"].as_array().expect("arr"); + let best = rows + .iter() + .max_by(|a, b| { + a["merged_h"] + .as_f64() + .partial_cmp(&b["merged_h"].as_f64()) + .expect("cmp") + }) + .expect("row"); + assert_eq!(v["binding_station"], best["change"]); + assert!( + v["optimum"]["merged_h"].as_f64().expect("n") + > v["actual"]["merged_h"].as_f64().expect("n") + ); + } + + #[test] + fn same_params_same_bytes() { + let p = parse_params(D_ALL).expect("parse"); + assert_eq!( + evaluate(&p, 2, 100.0, 5.0).to_string(), + evaluate(&p, 2, 100.0, 5.0).to_string() + ); + } +} diff --git a/crates/aprender-simulate/src/scenarios/course_flow.rs b/crates/aprender-simulate/src/scenarios/course_flow.rs new file mode 100644 index 0000000000..3c5dbe3d41 --- /dev/null +++ b/crates/aprender-simulate/src/scenarios/course_flow.rs @@ -0,0 +1,632 @@ +//! Scenario "course-flow": the FLOW-006 course factory and Shorts line (SIM-7, #5511). +//! +//! Stations are multi-server queues with a factory WIP limit over stations 1-6. edX and Shorts +//! launch with no human step; Coursera courses wait for the daily 07:00 UTC submission batch. +//! Shorts take priority over courses on the shared stations (render GPU, Chrome lane). Service +//! times are Erlang-4 (gamma, coefficient of variation 0.5). Time is minutes from 00:00 UTC. + +use std::collections::VecDeque; + +use crate::engine::rng::SimRng; +use crate::engine::scheduler::EventScheduler; +use crate::engine::state::SimEvent; +use crate::engine::SimTime; + +const DAY: f64 = 1440.0; +const OUTLINE: usize = 0; +const BUILD: usize = 1; +const DESIGN: usize = 2; +const RENDER: usize = 3; +const VERIFY: usize = 4; +const CHROME: usize = 5; +const NAMES: [&str; 6] = ["outline", "build", "design", "render", "verify", "chrome"]; + +/// Inputs of the factory. Times are minutes per course; `*_lanes`/`builders`/`gpus` are servers. +#[derive(Debug, Clone, PartialEq)] +pub struct CourseFlowConfig { + pub builders: usize, + pub build_min: f64, + pub design_lanes: usize, + pub design_min: f64, + pub gpus: usize, + pub render_min: f64, + pub chrome_lanes: usize, + pub stage_min: f64, + pub coursera_share: f64, + pub wip: usize, + pub starts_per_day: f64, + pub shorts_per_day: f64, + pub outline_min: f64, + pub verify_min: f64, + /// Count a finished course as WIP until the daily batch releases it (flow model B6). + pub batch_in_wip: bool, +} + +/// What one seeded run measured after the warm-up. +#[derive(Debug, Clone, Default, PartialEq)] +pub struct CourseFlowResult { + pub courses_per_day: f64, + pub shorts_per_day: f64, + pub lead_days: f64, + pub l: f64, + pub util: [f64; 6], + /// Mean hours a built course waits for the daily batch (courses that take the batch path). + pub mean_batch_wait_h: f64, +} + +impl CourseFlowResult { + /// The busiest station and its utilisation. + #[must_use] + pub fn busiest(&self) -> (&'static str, f64) { + let i = (0..6).fold(0, |b, i| if self.util[i] > self.util[b] { i } else { b }); + (NAMES[i], self.util[i]) + } +} + +#[derive(Clone, Copy)] +struct Job { + short: bool, + id: usize, + step: u8, +} + +enum Ev { + Start, + ShortTick, + Done(usize, Job), + Delay(Job), + Batch, +} + +fn pack(e: &Ev) -> SimEvent { + let (name, st, j) = match e { + Ev::Start => ("start", 0, None), + Ev::ShortTick => ("short_tick", 0, None), + Ev::Done(s, j) => ("done", *s, Some(*j)), + Ev::Delay(j) => ("delay", 0, Some(*j)), + Ev::Batch => ("batch", 0, None), + }; + let jv = j.map_or(0u64, |j| { + ((j.id as u64) << 16) | (u64::from(j.step) << 8) | u64::from(j.short) + }); + SimEvent::Custom { + name: name.into(), + data: (jv | ((st as u64) << 56)).to_le_bytes().to_vec(), + } +} + +fn unpack(e: &SimEvent) -> Option { + let SimEvent::Custom { name, data } = e else { + return None; + }; + let v = u64::from_le_bytes(data.as_slice().try_into().ok()?); + let job = Job { + short: v & 0xff == 1, + id: usize::try_from((v >> 16) & 0xffff_ffff).ok()?, + step: ((v >> 8) & 0xff) as u8, + }; + Some(match name.as_str() { + "start" => Ev::Start, + "short_tick" => Ev::ShortTick, + "done" => Ev::Done((v >> 56) as usize, job), + "delay" => Ev::Delay(job), + "batch" => Ev::Batch, + _ => return None, + }) +} + +struct Station { + free: usize, + size: usize, + hi: VecDeque<(Job, f64)>, + lo: VecDeque<(Job, f64)>, + busy_min: f64, +} + +struct Sim { + c: CourseFlowConfig, + rng: SimRng, + sched: EventScheduler, + now: f64, + warm: f64, + st: Vec, + wip: usize, + pending: usize, + started_at: Vec, + coursera: Vec, + batch_wait: Vec<(usize, f64)>, + bw_sum: f64, + bw_n: u64, + sojourn: Vec, + shorts_done: u64, + in_system: usize, + l_area: f64, + last_t: f64, +} + +impl Sim { + fn at(&mut self, dt: f64, e: &Ev) { + self.sched + .schedule(SimTime::from_secs((self.now + dt) * 60.0), pack(e)); + } + + fn erlang4(&mut self, mean: f64) -> f64 { + let p: f64 = (0..4).map(|_| 1.0 - self.rng.gen_f64()).product(); + -mean / 4.0 * p.ln() + } + + fn enqueue(&mut self, s: usize, job: Job, mean: f64) { + let d = self.erlang4(mean); + if job.short { + self.st[s].hi.push_back((job, d)); + } else { + self.st[s].lo.push_back((job, d)); + } + self.dispatch(s); + } + + fn dispatch(&mut self, s: usize) { + while self.st[s].free > 0 { + let Some((job, d)) = self.st[s] + .hi + .pop_front() + .or_else(|| self.st[s].lo.pop_front()) + else { + break; + }; + self.st[s].free -= 1; + if self.now >= self.warm { + self.st[s].busy_min += d; + } + self.at(d, &Ev::Done(s, job)); + } + } + + fn start_course(&mut self) { + let id = self.started_at.len(); + self.started_at.push(self.now); + let cs = self.rng.gen_f64() < self.c.coursera_share; + self.coursera.push(cs); + self.wip += 1; + self.in_system += 1; + let m = self.c.outline_min; + self.enqueue( + OUTLINE, + Job { + short: false, + id, + step: 0, + }, + m, + ); + } + + fn launch(&mut self, id: usize) { + self.in_system -= 1; + if self.now >= self.warm { + self.sojourn.push(self.now - self.started_at[id]); + } + } + + fn leave_wip(&mut self) { + self.wip -= 1; + if self.pending > 0 { + self.pending -= 1; + self.start_course(); + } + } + + fn course_next(&mut self, j: Job) { + let c = self.c.clone(); + let nj = |step| Job { step, ..j }; + match j.step { + 0 => self.enqueue(BUILD, nj(1), c.build_min), + 1 => self.enqueue(DESIGN, nj(2), c.design_min), + 2 => self.enqueue(RENDER, nj(3), c.render_min), + 3 => self.enqueue(VERIFY, nj(4), c.verify_min), + 4 if self.coursera[j.id] => self.enqueue(CHROME, nj(5), c.stage_min), + 4 => self.at(15.0, &Ev::Delay(nj(6))), + 5 => { + if !self.c.batch_in_wip { + self.leave_wip(); + } + self.batch_wait.push((j.id, self.now)); + } + _ => { + self.leave_wip(); + self.launch(j.id); + } + } + } + + /// The 07:00 batch: release every waiting course, and when the wait counts as WIP, free its slot now. + fn release_batch(&mut self) { + for (id, since) in std::mem::take(&mut self.batch_wait) { + if self.now >= self.warm { + self.bw_sum += self.now - since; + self.bw_n += 1; + } + if self.c.batch_in_wip { + self.leave_wip(); + } + self.launch(id); + } + } + + fn short_next(&mut self, j: Job) { + let nj = |step| Job { step, ..j }; + match j.step { + 0 => self.enqueue(RENDER, nj(1), 5.0), + 1 => self.at(20.0, &Ev::Delay(nj(2))), + 2 => self.enqueue(CHROME, nj(3), 5.0), + _ => { + if self.now >= self.warm { + self.shorts_done += 1; + } + } + } + } + + fn step(&mut self, e: &Ev) { + match *e { + Ev::Start => { + if self.wip < self.c.wip { + self.start_course(); + } else { + self.pending += 1; + } + self.at(DAY / self.c.starts_per_day, &Ev::Start); + } + Ev::ShortTick => { + let id = self.started_at.len() + self.shorts_done as usize; + self.at( + 20.0, + &Ev::Delay(Job { + short: true, + id, + step: 0, + }), + ); + self.at(DAY / self.c.shorts_per_day, &Ev::ShortTick); + } + Ev::Done(s, j) => { + self.st[s].free += 1; + self.dispatch(s); + if j.short { + self.short_next(j); + } else { + self.course_next(j); + } + } + Ev::Delay(j) if j.short => self.short_next(j), + Ev::Delay(j) => self.course_next(j), + Ev::Batch => { + self.release_batch(); + self.at(DAY, &Ev::Batch); + } + } + } +} + +/// Run one seed for `days`, discarding the first `warmup_d`. +#[must_use] +pub fn run_course_flow( + c: &CourseFlowConfig, + seed: u64, + days: f64, + warmup_d: f64, +) -> CourseFlowResult { + let size = [1, c.builders, c.design_lanes, c.gpus, 4, c.chrome_lanes]; + let mut s = Sim { + c: c.clone(), + rng: SimRng::new(seed), + sched: EventScheduler::new(), + now: 0.0, + warm: warmup_d * DAY, + st: size + .iter() + .map(|&n| Station { + free: n, + size: n, + hi: VecDeque::new(), + lo: VecDeque::new(), + busy_min: 0.0, + }) + .collect(), + wip: 0, + pending: 0, + started_at: Vec::new(), + coursera: Vec::new(), + batch_wait: Vec::new(), + bw_sum: 0.0, + bw_n: 0, + sojourn: Vec::new(), + shorts_done: 0, + in_system: 0, + l_area: 0.0, + last_t: 0.0, + }; + s.at(0.0, &Ev::Start); + s.at(0.0, &Ev::ShortTick); + s.at(7.0 * 60.0, &Ev::Batch); + let end = days * DAY; + while let Some(ev) = s.sched.next() { + let t = ev.time.as_secs_f64() / 60.0; + if t > end { + break; + } + let from = s.last_t.max(s.warm); + if t > from { + s.l_area += s.in_system as f64 * (t - from); + } + s.last_t = t; + s.now = t; + if let Some(e) = unpack(&ev.event) { + s.step(&e); + } + } + let span = end - s.warm; + let n = s.sojourn.len() as f64; + let mut util = [0.0; 6]; + for (u, st) in util.iter_mut().zip(&s.st) { + *u = st.busy_min / (span * st.size as f64); + } + CourseFlowResult { + courses_per_day: n / span * DAY, + shorts_per_day: s.shorts_done as f64 / span * DAY, + lead_days: s.sojourn.iter().sum::() / n.max(1.0) / DAY, + l: s.l_area / span, + util, + mean_batch_wait_h: s.bw_sum / (s.bw_n.max(1) as f64) / 60.0, + } +} + +/// Named FLOW-006 factory cases. +#[must_use] +pub fn course_preset(name: &str) -> Option { + let mut c = CourseFlowConfig { + builders: 11, + build_min: 2520.0, + design_lanes: 2, + design_min: 180.0, + gpus: 1, + render_min: 100.0, + chrome_lanes: 2, + stage_min: 360.0, + coursera_share: 0.4, + wip: 24, + starts_per_day: 5.0, + shorts_per_day: 5.0, + outline_min: 90.0, + verify_min: 15.0, + batch_in_wip: false, + }; + match name { + "cal-today" => { + c.builders = 1; + c.design_lanes = 1; + c.chrome_lanes = 1; + } + "design" => {} + "nine-builders" => c.builders = 9, + "coursera-24h" => c.stage_min = 1440.0, + "render-300" => c.render_min = 300.0, + _ => return None, + } + Some(c) +} + +/// Names of the FLOW-006 factory cases. +#[must_use] +pub fn course_cases() -> [&'static str; 5] { + [ + "cal-today", + "design", + "nine-builders", + "coursera-24h", + "render-300", + ] +} + +/// The reference run: seeds 1..=5, 300 days, 20 days warm-up, averaged. +#[must_use] +pub fn run_course_reference(c: &CourseFlowConfig) -> CourseFlowResult { + let mut a = CourseFlowResult::default(); + for seed in 1..=5 { + let r = run_course_flow(c, seed, 300.0, 20.0); + a.courses_per_day += r.courses_per_day / 5.0; + a.shorts_per_day += r.shorts_per_day / 5.0; + a.lead_days += r.lead_days / 5.0; + a.l += r.l / 5.0; + a.mean_batch_wait_h += r.mean_batch_wait_h / 5.0; + for i in 0..6 { + a.util[i] += r.util[i] / 5.0; + } + } + a +} + +/// Override the case with the ledger's measured inputs (a JSON object); unknown keys are refused. +/// +/// # Errors +/// Returns the offending key or type when the object holds something the model does not read. +pub fn apply_course_inputs(c: &mut CourseFlowConfig, v: &serde_json::Value) -> Result<(), String> { + let obj = v.as_object().ok_or("inputs must be a JSON object")?; + for (k, x) in obj { + let f = x.as_f64().ok_or_else(|| format!("{k}: not a number"))?; + match k.as_str() { + "builders" => c.builders = f as usize, + "build_min" => c.build_min = f, + "design_lanes" => c.design_lanes = f as usize, + "design_min" => c.design_min = f, + "gpus" => c.gpus = f as usize, + "render_min" => c.render_min = f, + "chrome_lanes" => c.chrome_lanes = f as usize, + "stage_min" => c.stage_min = f, + "coursera_share" => c.coursera_share = f, + "wip" => c.wip = f as usize, + "starts_per_day" => c.starts_per_day = f, + "shorts_per_day" => c.shorts_per_day = f, + "batch_in_wip" => c.batch_in_wip = f != 0.0, + _ => return Err(format!("unknown input: {k}")), + } + } + Ok(()) +} + +/// The `sim` block fields this scenario owns, in the shape the cops publish. +#[must_use] +pub fn course_sim_block( + case: &str, + c: &CourseFlowConfig, + r: &CourseFlowResult, +) -> serde_json::Value { + let (station, util) = r.busiest(); + serde_json::json!({ + "scenario": format!("course-flow:{case}"), + "seeds": 5, "hours": 300.0 * 24.0, "warmup_h": 20.0 * 24.0, + "inputs": { + "builders": c.builders, "build_min": c.build_min, "design_lanes": c.design_lanes, + "design_min": c.design_min, "gpus": c.gpus, "render_min": c.render_min, + "chrome_lanes": c.chrome_lanes, "stage_min": c.stage_min, + "coursera_share": c.coursera_share, "wip": c.wip, + "starts_per_day": c.starts_per_day, "shorts_per_day": c.shorts_per_day, "batch_in_wip": c.batch_in_wip, + }, + "predicted": { + "courses_day": r.courses_per_day, "shorts_day": r.shorts_per_day, + "lead_days": r.lead_days, "L": r.l, "mean_batch_wait_h": r.mean_batch_wait_h, "busiest_util_pct": util * 100.0, + }, + "binding_station": station, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// FLOW-006 reference bands: (case, courses/day, tolerance). + const WANT: [(&str, f64, f64); 5] = [ + ("cal-today", 0.54, 0.10), + ("design", 4.92, 0.05), + ("nine-builders", 4.90, 0.05), + ("coursera-24h", 4.84, 0.05), + ("render-300", 4.70, 0.05), + ]; + + fn got(name: &str) -> CourseFlowResult { + run_course_reference(&course_preset(name).expect("case")) + } + + #[test] + fn courses_per_day_match_the_reference_bands() { + for (name, want, tol) in WANT { + let r = got(name); + assert!( + (r.courses_per_day - want).abs() <= want * tol, + "{name}: {:.3} vs {want}", + r.courses_per_day + ); + } + } + + #[test] + fn shorts_hold_five_a_day_in_every_case() { + for (name, _, _) in WANT { + let r = got(name); + assert!( + (r.shorts_per_day - 5.0).abs() <= 0.1, + "{name}: {:.2}", + r.shorts_per_day + ); + } + } + + #[test] + fn the_binding_station_is_the_one_the_case_squeezes() { + assert_eq!(got("nine-builders").busiest().0, "build"); + assert!(got("nine-builders").util[BUILD] > 0.95); + assert_eq!(got("render-300").busiest().0, "render"); + assert!(got("render-300").util[RENDER] > 0.95); + assert_eq!(got("coursera-24h").busiest().0, "chrome"); + } + + #[test] + fn littles_law_holds_within_half_a_percent() { + for (name, _, _) in WANT.iter().filter(|w| w.0 != "cal-today") { + let r = got(name); + let lam_w = r.courses_per_day * r.lead_days; + assert!( + (lam_w - r.l).abs() <= r.l * 0.005, + "{name}: {lam_w:.3} vs {:.3}", + r.l + ); + } + } + + #[test] + fn same_seed_same_output() { + let c = course_preset("design").expect("case"); + assert_eq!( + run_course_flow(&c, 4, 40.0, 5.0), + run_course_flow(&c, 4, 40.0, 5.0) + ); + } + + #[test] + fn a_reverted_design_shows_red() { + let mut c = course_preset("design").expect("case"); + c.builders = 1; + assert!((run_course_reference(&c).courses_per_day - 4.92).abs() > 4.92 * 0.05); + } + + fn with_batch_in_wip(name: &str) -> CourseFlowConfig { + let mut c = course_preset(name).expect("case"); + c.batch_in_wip = true; + c + } + + /// B6: the daily batch wait is measured and reported, not hidden. + #[test] + fn mean_batch_wait_is_reported_and_inside_a_day() { + let r = got("design"); + assert!( + r.mean_batch_wait_h > 0.5 && r.mean_batch_wait_h < 24.0, + "mean batch wait {:.2} h", + r.mean_batch_wait_h + ); + } + + /// B6: with the batch wait counted inside WIP, Little's law holds within 0.5%, + /// and the WIP cap now binds on it, so it can only cost throughput. + #[test] + fn littles_law_holds_with_batch_wait_inside_wip() { + for name in ["design", "nine-builders", "coursera-24h"] { + let outside = got(name); + let r = run_course_reference(&with_batch_in_wip(name)); + let lam_w = r.courses_per_day * r.lead_days; + assert!( + (lam_w - r.l).abs() <= r.l * 0.005, + "{name}: {lam_w:.3} vs {:.3}", + r.l + ); + assert!(r.l <= 24.0 + 1e-9, "{name}: L {:.2} over the cap", r.l); + assert!( + r.courses_per_day <= outside.courses_per_day + 0.02, + "{name}" + ); + } + } + + /// B6 falsifier: counting the batch wait inside WIP must change the answer, or the flag is dead. + #[test] + fn batch_in_wip_is_not_a_no_op() { + let a = got("coursera-24h"); + let b = run_course_reference(&with_batch_in_wip("coursera-24h")); + assert!((a.courses_per_day - b.courses_per_day).abs() > 1e-6 || (a.l - b.l).abs() > 1e-6); + } + + #[test] + fn unknown_case_is_none() { + assert!(course_preset("nope").is_none()); + } +} diff --git a/crates/aprender-simulate/src/scenarios/mod.rs b/crates/aprender-simulate/src/scenarios/mod.rs index e4558d3bb9..767789fb30 100644 --- a/crates/aprender-simulate/src/scenarios/mod.rs +++ b/crates/aprender-simulate/src/scenarios/mod.rs @@ -8,18 +8,32 @@ //! - Portfolio risk (Monte Carlo `VaR`) //! - Epidemic models (SIR/SEIR) +pub mod arbiter; pub mod bouncing_balls; pub mod climate; +pub mod course_flow; pub mod epidemic; pub mod pendulum; pub mod portfolio; +pub mod pr_flow; +pub mod repo_flow; pub mod rocket; pub mod satellite; +pub use arbiter::{evaluate, parse_params, Params}; pub use bouncing_balls::{BouncingBallsConfig, BouncingBallsState}; pub use climate::{ClimateConfig, ClimateScenario}; +pub use course_flow::{ + apply_course_inputs, course_cases, course_preset, course_sim_block, run_course_flow, + run_course_reference, CourseFlowConfig, CourseFlowResult, +}; pub use epidemic::{SEIRConfig, SEIRScenario, SIRConfig, SIRScenario}; pub use pendulum::{PendulumConfig, PendulumScenario}; pub use portfolio::{PortfolioConfig, PortfolioScenario, VaRResult}; +pub use pr_flow::{ + binding_station, bounds, eq7_law, preset, reference_contract_ok, run_pr_flow, run_reference, + Eq7Law, PrFlowConfig, PrFlowResult, ReviewMode, +}; +pub use repo_flow::{apply_inputs, repo_cases, repo_preset, run_repo_reference, sim_block}; pub use rocket::{RocketConfig, RocketScenario, StageSeparation}; pub use satellite::{OrbitalElements, SatelliteScenario}; diff --git a/crates/aprender-simulate/src/scenarios/pr_flow.rs b/crates/aprender-simulate/src/scenarios/pr_flow.rs new file mode 100644 index 0000000000..5aa549e1ca --- /dev/null +++ b/crates/aprender-simulate/src/scenarios/pr_flow.rs @@ -0,0 +1,958 @@ +//! Scenario "pr-flow": the aprender PR line as a discrete-event model (FLOW-005 SIM-2, #5511). +//! +//! Six components, in minutes, seeded: authors on a pull system, the WIP cap, PR checks on a +//! runner pool (red with `p_red`, then fix and rerun), the arm delay, the merge queue (batch, +//! minimum wait, build limit, runner pool, flake sends the whole group back) and merge. +//! Events ride `engine::scheduler`, time is `SimTime`, randomness is `SimRng`. + +use std::collections::VecDeque; + +use crate::engine::rng::SimRng; +use crate::engine::scheduler::EventScheduler; +use crate::engine::state::SimEvent; +use crate::engine::SimTime; + +/// Where the review sits relative to the PR checks. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ReviewMode { + Serial, + Concurrent, +} + +/// Inputs of the line. Times are minutes; `*_runner_min` is the runner-minutes of one run. +#[derive(Debug, Clone, PartialEq)] +pub struct PrFlowConfig { + pub cap: usize, + /// Mean concurrent authors; the fractional part is one author who thinks 1/frac times slower. + pub n: f64, + pub t_author_min: f64, + pub r_pr: usize, + pub pr_runner_min: f64, + pub pr_jobs: usize, + pub pr_setup_min: f64, + pub p_red: f64, + pub t_fix_min: f64, + pub t_arm_min: f64, + /// Review wait per PR after green checks, serial with arming; 0 = no review station. + pub t_rev_min: f64, + /// Review runs alongside the checks (`Concurrent`) or after them (`Serial`). + pub rev_mode: ReviewMode, + /// Parallel reviewers; 0 = unlimited (no queue at the review station). + pub rev_lanes: usize, + /// Per-tier review minutes, handed out to PRs round-robin; empty = `t_rev_min` for all. + pub rev_tiers: Vec, + pub r_mq: usize, + pub mq_runner_min: f64, + pub mq_jobs: usize, + pub mq_setup_min: f64, + pub builds: usize, + pub mq_batch: usize, + pub mq_wait_min: f64, + pub f: f64, + /// Chance an armed PR is found conflicting (main moved); 0 keeps the FLOW-005 designs. + pub p_dirty: f64, + /// Minutes until a conflicting PR is rebased and rerun (FLOW-006 keep-fresh rule). + pub t_rebase_min: f64, +} + +impl PrFlowConfig { + /// Mean think time of author `a`: the last, fractional author is slowed by its fraction. + fn think_min(&self, a: usize) -> f64 { + let frac = self.n - self.n.floor(); + if frac > 0.0 && a == self.n.floor() as usize { + self.t_author_min / frac + } else { + self.t_author_min + } + } +} + +/// What one seeded run measured after the warm-up. +#[derive(Debug, Clone, Default, PartialEq)] +pub struct PrFlowResult { + pub merged_per_h: f64, + pub w_mean_min: f64, + pub w_p90_min: f64, + pub l: f64, + pub prs_per_pass: f64, + pub pr_util: f64, + pub mq_util: f64, + pub runs_per_pr: f64, + /// Author busy fraction: share of author time not spent blocked on the WIP cap. + pub rho_a: f64, + /// Mean wall of one PR-check run, start to verdict (runner wait + jobs + setup). + pub run_wall_min: f64, + /// Mean open to green checks, merged PRs only (eq 7 terms r(w+c_wall) + (r-1)t_fix). + pub checks_min: f64, + /// Mean green to armed (eq 7 term t_arm, plus any serial review). + pub arm_min: f64, + /// Mean armed to merged (eq 7 term W_mq). + pub mq_min: f64, + /// Mean armed to the start of the first merge pass that holds the PR (the wait in W_mq). + pub pass_wait_min: f64, + /// Mean wall of one merge pass, group formed to verdict (runner wait + jobs + setup). + pub pass_wall_min: f64, +} + +#[derive(Clone, Copy)] +enum Ev { + Author(usize), + PrJob(usize), + PrRun(usize), + Fixed(usize), + Armed(usize), + Reviewed(usize), + Rebased(usize), + MqJob(usize), + MqPass(usize), + MqTimer, +} + +impl Ev { + fn pack(self) -> SimEvent { + let (name, id) = match self { + Self::Author(i) => ("author", i), + Self::PrJob(i) => ("pr_job", i), + Self::PrRun(i) => ("pr_run", i), + Self::Fixed(i) => ("fixed", i), + Self::Armed(i) => ("armed", i), + Self::Reviewed(i) => ("reviewed", i), + Self::Rebased(i) => ("rebased", i), + Self::MqJob(i) => ("mq_job", i), + Self::MqPass(i) => ("mq_pass", i), + Self::MqTimer => ("mq_timer", 0), + }; + SimEvent::Custom { + name: name.into(), + data: (id as u64).to_le_bytes().to_vec(), + } + } + + fn unpack(e: &SimEvent) -> Option { + let SimEvent::Custom { name, data } = e else { + return None; + }; + let id = usize::try_from(u64::from_le_bytes(data.as_slice().try_into().ok()?)).ok()?; + Some(match name.as_str() { + "author" => Self::Author(id), + "pr_job" => Self::PrJob(id), + "pr_run" => Self::PrRun(id), + "fixed" => Self::Fixed(id), + "armed" => Self::Armed(id), + "reviewed" => Self::Reviewed(id), + "rebased" => Self::Rebased(id), + "mq_job" => Self::MqJob(id), + "mq_pass" => Self::MqPass(id), + "mq_timer" => Self::MqTimer, + _ => return None, + }) + } +} + +/// A pool of identical runners with a FIFO of waiting jobs `(owner, minutes)`. +struct Pool { + free: usize, + size: usize, + waiting: VecDeque<(usize, f64)>, + busy_min: f64, +} + +impl Pool { + fn new(size: usize) -> Self { + Self { + free: size, + size, + waiting: VecDeque::new(), + busy_min: 0.0, + } + } +} + +struct Sim { + c: PrFlowConfig, + rng: SimRng, + sched: EventScheduler, + now: f64, + warm: f64, + open: usize, + blocked: VecDeque, + open_at: Vec, + jobs_left: Vec, + pr: Pool, + mq: Pool, + queue: VecDeque, + enq_at: Vec, + groups: Vec>, + group_jobs_left: Vec, + group_at: Vec, + pass_wall: (f64, u64), + active: usize, + timer_at: f64, + sojourn: Vec, + l_area: f64, + rev_free: usize, + rev_wait: VecDeque, + green: Vec, + rev_ok: Vec, + green_at: Vec, + run_at: Vec, + wall: (f64, u64), + stage_sum: [f64; 4], + pass_at: Vec, + blocked_area: f64, + last_t: f64, + runs: u64, + prs_seen: u64, + passes: u64, +} + +impl Sim { + fn at(&mut self, dt: f64, e: Ev) { + self.sched + .schedule(SimTime::from_secs((self.now + dt) * 60.0), e.pack()); + } + + fn exp(&mut self, mean: f64) -> f64 { + -mean * (1.0 - self.rng.gen_f64()).ln() + } + + fn start_run(&mut self, pr: usize) { + self.runs += 1; + self.run_at[pr] = self.now; + self.jobs_left[pr] = self.c.pr_jobs; + let d = self.c.pr_runner_min / self.c.pr_jobs as f64; + for _ in 0..self.c.pr_jobs { + self.pr.waiting.push_back((pr, d)); + } + self.dispatch_pr(); + } + + fn dispatch_pr(&mut self) { + while self.pr.free > 0 { + let Some((pr, d)) = self.pr.waiting.pop_front() else { + break; + }; + self.pr.free -= 1; + if self.now >= self.warm { + self.pr.busy_min += d; + } + self.at(d, Ev::PrJob(pr)); + } + } + + fn open_pr(&mut self, author: usize) { + let pr = self.open_at.len(); + self.open_at.push(self.now); + self.enq_at.push(0.0); + self.jobs_left.push(0); + self.green.push(false); + self.rev_ok.push(false); + self.green_at.push(0.0); + self.run_at.push(0.0); + self.pass_at.push(-1.0); + self.open += 1; + self.prs_seen += 1; + if self.c.rev_mode == ReviewMode::Concurrent { + self.request_review(pr); + } + self.start_run(pr); + let t = self.exp(self.c.think_min(author)); + self.at(t, Ev::Author(author)); + } + + fn form_groups(&mut self) { + while self.active < self.c.builds && !self.queue.is_empty() { + let head = self.enq_at[self.queue[0]]; + let ready = + self.queue.len() >= self.c.mq_batch || self.now >= head + self.c.mq_wait_min; + if !ready { + if self.timer_at < self.now { + self.timer_at = head + self.c.mq_wait_min; + self.at(self.timer_at - self.now, Ev::MqTimer); + } + return; + } + let take = self.c.mq_batch.min(self.queue.len()); + let members: Vec = self.queue.drain(..take).collect(); + for &pr in &members { + if self.pass_at[pr] < self.enq_at[pr] { + self.pass_at[pr] = self.now; + } + } + self.groups.push(members); + let g = self.groups.len() - 1; + self.group_jobs_left.push(self.c.mq_jobs); + self.group_at.push(self.now); + self.active += 1; + let d = self.c.mq_runner_min / self.c.mq_jobs as f64; + for _ in 0..self.c.mq_jobs { + self.mq.waiting.push_back((g, d)); + } + } + self.dispatch_mq(); + } + + fn dispatch_mq(&mut self) { + while self.mq.free > 0 { + let Some((g, d)) = self.mq.waiting.pop_front() else { + break; + }; + self.mq.free -= 1; + if self.now >= self.warm { + self.mq.busy_min += d; + } + self.at(d, Ev::MqJob(g)); + } + } + + fn pass_done(&mut self, g: usize) { + self.active -= 1; + let members = std::mem::take(&mut self.groups[g]); + if self.now >= self.warm { + self.pass_wall.0 += self.now - self.group_at[g]; + self.pass_wall.1 += 1; + } + if self.rng.gen_f64() < self.c.f { + for pr in members.into_iter().rev() { + self.queue.push_front(pr); + } + } else { + if self.now >= self.warm { + self.passes += 1; + } + for pr in members { + self.merge(pr); + } + } + self.form_groups(); + } + + /// Review minutes of `pr`: its round-robin tier, else the flat `t_rev_min`. + fn review_min(&self, pr: usize) -> f64 { + self.c + .rev_tiers + .get(pr % self.c.rev_tiers.len().max(1)) + .copied() + .unwrap_or(self.c.t_rev_min) + } + + /// Ask the review station for `pr`; a zero-minute review finishes inline (no event). + fn request_review(&mut self, pr: usize) { + if self.review_min(pr) <= 0.0 { + self.reviewed(pr); + } else if self.rev_free > 0 { + self.rev_free -= 1; + self.at(self.review_min(pr), Ev::Reviewed(pr)); + } else { + self.rev_wait.push_back(pr); + } + } + + /// A PR-check run reached its verdict: red goes to fix, green goes on. + fn run_done(&mut self, pr: usize) { + if self.now >= self.warm { + self.wall.0 += self.now - self.run_at[pr]; + self.wall.1 += 1; + } + if self.rng.gen_f64() < self.c.p_red { + self.at(self.c.t_fix_min, Ev::Fixed(pr)); + } else { + self.checks_green(pr); + } + } + + fn checks_green(&mut self, pr: usize) { + self.green[pr] = true; + self.green_at[pr] = self.now; + match self.c.rev_mode { + ReviewMode::Serial => self.request_review(pr), + ReviewMode::Concurrent if self.rev_ok[pr] => self.at(self.c.t_arm_min, Ev::Armed(pr)), + ReviewMode::Concurrent => {} + } + } + + fn reviewed(&mut self, pr: usize) { + self.rev_ok[pr] = true; + if self.review_min(pr) > 0.0 { + self.rev_free = self.rev_free.saturating_add(1); + if let Some(next) = self.rev_wait.pop_front() { + self.rev_free -= 1; + self.at(self.review_min(next), Ev::Reviewed(next)); + } + } + if self.green[pr] { + self.at(self.c.t_arm_min, Ev::Armed(pr)); + } + } + + fn merge(&mut self, pr: usize) { + if self.now >= self.warm { + self.sojourn.push(self.now - self.open_at[pr]); + let (open, green, armed) = (self.open_at[pr], self.green_at[pr], self.enq_at[pr]); + let stages = [ + green - open, + armed - green, + self.now - armed, + self.pass_at[pr] - armed, + ]; + for (s, d) in self.stage_sum.iter_mut().zip(stages) { + *s += d; + } + } + self.open -= 1; + if let Some(a) = self.blocked.pop_front() { + self.open_pr(a); + } + } + + fn step(&mut self, e: Ev) { + match e { + Ev::Author(a) if self.open < self.c.cap => self.open_pr(a), + Ev::Author(a) => self.blocked.push_back(a), + Ev::PrJob(pr) => { + self.pr.free += 1; + self.jobs_left[pr] -= 1; + if self.jobs_left[pr] == 0 { + self.at(self.c.pr_setup_min, Ev::PrRun(pr)); + } + self.dispatch_pr(); + } + Ev::PrRun(pr) => self.run_done(pr), + Ev::Reviewed(pr) => self.reviewed(pr), + Ev::Fixed(pr) | Ev::Rebased(pr) => self.start_run(pr), + Ev::Armed(pr) if self.c.p_dirty > 0.0 && self.rng.gen_f64() < self.c.p_dirty => { + self.at(self.c.t_rebase_min, Ev::Rebased(pr)); + } + Ev::Armed(pr) if self.c.builds == 0 => self.merge(pr), + Ev::Armed(pr) => { + self.enq_at[pr] = self.now; + self.queue.push_back(pr); + self.form_groups(); + } + Ev::MqJob(g) => { + self.mq.free += 1; + self.group_jobs_left[g] -= 1; + if self.group_jobs_left[g] == 0 { + self.at(self.c.mq_setup_min, Ev::MqPass(g)); + } + self.dispatch_mq(); + } + Ev::MqPass(g) => self.pass_done(g), + Ev::MqTimer => self.form_groups(), + } + } +} + +/// Run one seed for `hours`, discarding the first `warmup_h`. +#[must_use] +pub fn run_pr_flow(c: &PrFlowConfig, seed: u64, hours: f64, warmup_h: f64) -> PrFlowResult { + let mut s = Sim { + c: c.clone(), + rng: SimRng::new(seed), + sched: EventScheduler::new(), + now: 0.0, + warm: warmup_h * 60.0, + open: 0, + blocked: VecDeque::new(), + open_at: Vec::new(), + jobs_left: Vec::new(), + pr: Pool::new(c.r_pr), + mq: Pool::new(c.r_mq), + queue: VecDeque::new(), + enq_at: Vec::new(), + groups: Vec::new(), + group_jobs_left: Vec::new(), + group_at: Vec::new(), + pass_wall: (0.0, 0), + active: 0, + timer_at: -1.0, + sojourn: Vec::new(), + l_area: 0.0, + blocked_area: 0.0, + rev_free: if c.rev_lanes == 0 { + usize::MAX + } else { + c.rev_lanes + }, + rev_wait: VecDeque::new(), + green: Vec::new(), + rev_ok: Vec::new(), + green_at: Vec::new(), + run_at: Vec::new(), + wall: (0.0, 0), + stage_sum: [0.0; 4], + pass_at: Vec::new(), + last_t: 0.0, + runs: 0, + prs_seen: 0, + passes: 0, + }; + s.timer_at = -1.0; + for a in 0..c.n.ceil() as usize { + let t = s.exp(c.think_min(a)); + s.at(t, Ev::Author(a)); + } + let end = hours * 60.0; + while let Some(ev) = s.sched.next() { + let t = ev.time.as_secs_f64() / 60.0; + if t > end { + break; + } + let from = s.last_t.max(s.warm); + if t > from { + s.l_area += s.open as f64 * (t - from); + s.blocked_area += s.blocked.len() as f64 * (t - from); + } + s.last_t = t; + s.now = t; + if let Some(e) = Ev::unpack(&ev.event) { + s.step(e); + } + } + s.summary(end) +} + +impl Sim { + fn summary(&mut self, end: f64) -> PrFlowResult { + let span = end - self.warm; + let n = self.sojourn.len() as f64; + self.sojourn.sort_by(f64::total_cmp); + let p90 = self + .sojourn + .get(((n * 0.9) as usize).min(self.sojourn.len().saturating_sub(1))); + PrFlowResult { + merged_per_h: n / span * 60.0, + w_mean_min: self.sojourn.iter().sum::() / n.max(1.0), + w_p90_min: p90.copied().unwrap_or(0.0), + l: self.l_area / span, + prs_per_pass: n / (self.passes as f64).max(1.0), + pr_util: self.pr.busy_min / (span * self.pr.size as f64), + mq_util: self.mq.busy_min / (span * self.mq.size as f64), + runs_per_pr: self.runs as f64 / (self.prs_seen as f64).max(1.0), + rho_a: 1.0 - self.blocked_area / (span * self.c.n.ceil().max(1.0)), + run_wall_min: self.wall.0 / (self.wall.1 as f64).max(1.0), + checks_min: self.stage_sum[0] / n.max(1.0), + arm_min: self.stage_sum[1] / n.max(1.0), + mq_min: self.stage_sum[2] / n.max(1.0), + pass_wait_min: self.stage_sum[3] / n.max(1.0), + pass_wall_min: self.pass_wall.0 / (self.pass_wall.1 as f64).max(1.0), + } + } +} + +/// Named FLOW-005 designs. `rm` is runner-minutes per run; the fast tier (24) has a shorter +/// wall. Setup minutes are the overheads the spec leaves open, calibrated to the reference. +#[must_use] +pub fn preset(name: &str) -> Option { + let (cap, r_pr, r_mq, builds, rm, p_red, fast) = match name { + "D" => (30, 38, 4, 1, 72.0, 0.2, false), + "A" => (10, 40, 4, 5, 89.2, 0.2, false), + "G" => (10, 12, 8, 2, 24.0, 0.1, true), + "E" => (30, 12, 4, 1, 24.0, 0.1, true), + _ => return None, + }; + let setup = if fast { 9.0 } else { 7.7 }; + Some(PrFlowConfig { + cap, + n: 11.0, + t_author_min: 30.0, + r_pr, + pr_runner_min: rm, + pr_jobs: 4, + pr_setup_min: if fast { 15.0 } else { setup }, + p_red, + t_fix_min: 10.0, + t_arm_min: 2.0, + t_rev_min: 0.0, + rev_mode: ReviewMode::Serial, + rev_lanes: 0, + rev_tiers: Vec::new(), + r_mq, + mq_runner_min: rm, + mq_jobs: 4, + mq_setup_min: setup, + builds, + mq_batch: 10, + mq_wait_min: 5.0, + f: 0.02, + p_dirty: 0.0, + t_rebase_min: 0.0, + }) +} + +/// Throughput bounds (merged/h) of the four stations; the binding station is the smallest. +/// +/// `cap` is cap/W (eq 2), `supply` is N authors at one PR per think time, `pr_checks` and +/// `merge_queue` are pool capacity over the runner-minutes one merged PR consumes (eq 3, 5, 6). +#[must_use] +pub fn bounds(c: &PrFlowConfig, r: &PrFlowResult) -> [(&'static str, f64); 4] { + let w_h = (r.w_mean_min / 60.0).max(f64::EPSILON); + let pr_per_merge = c.pr_runner_min * r.runs_per_pr.max(1.0); + let mq_per_merge = c.mq_runner_min / r.prs_per_pass.max(1.0); + [ + ("cap", c.cap as f64 / w_h), + ("supply", c.n * 60.0 / c.t_author_min), + ("pr_checks", c.r_pr as f64 * 60.0 / pr_per_merge), + ("merge_queue", c.r_mq as f64 * 60.0 / mq_per_merge), + ] +} + +/// The binding station: argmin of [`bounds`]. No utilisation thresholds. +#[must_use] +pub fn binding_station(c: &PrFlowConfig, r: &PrFlowResult) -> &'static str { + bounds(c, r) + .into_iter() + .min_by(|a, b| a.1.total_cmp(&b.1)) + .map_or("none", |b| b.0) +} + +/// Flow model eq 7, time open, term by term. +/// +/// `W = r(w + c_wall) + (r-1)t_fix + t_arm + p_dirty/(1-p_dirty)(t_reb + r(w + c_wall) + (r-1)t_fix + t_arm) + W_mq` +/// with `W_mq = wait_mq + (w_pass + t_pass)/(1-f)`. This line has no rebase, so `p_dirty = 0`. +#[derive(Debug, Clone, Copy, PartialEq)] +pub struct Eq7Law { + /// Runs per PR, `1/(1-p_red)` (eq 4). + pub r: f64, + /// Runner wait per run: the sim's mean run wall less `c_wall`. + pub w_min: f64, + /// Wall of one run with free runners: job minutes plus setup. + pub c_wall_min: f64, + pub t_fix_min: f64, + /// Green to armed, plus the review minutes when the review is serial. + pub t_arm_min: f64, + pub p_dirty: f64, + pub t_reb_min: f64, + /// Armed to the start of the first merge pass that holds the PR. + pub wait_mq_min: f64, + /// Wall of one merge pass with free runners: job minutes plus setup. + pub t_pass_min: f64, + /// Runner wait inside one merge pass: the sim's mean pass wall less `t_pass`. + pub w_pass_min: f64, + pub f: f64, +} + +impl Eq7Law { + /// Open to green: `r(w + c_wall) + (r-1)t_fix`. + #[must_use] + pub fn checks_min(&self) -> f64 { + self.r * (self.w_min + self.c_wall_min) + (self.r - 1.0) * self.t_fix_min + } + + /// Rebase loop: `p_dirty/(1-p_dirty)(t_reb + checks + t_arm)`. + #[must_use] + pub fn rebase_min(&self) -> f64 { + self.p_dirty / (1.0 - self.p_dirty) * (self.t_reb_min + self.checks_min() + self.t_arm_min) + } + + /// `W_mq = wait_mq + (w_pass + t_pass)/(1-f)`: a flake sends the whole group round again. + #[must_use] + pub fn w_mq_min(&self) -> f64 { + self.wait_mq_min + (self.w_pass_min + self.t_pass_min) / (1.0 - self.f) + } + + /// Eq 7, mean time open. + #[must_use] + pub fn w_law_min(&self) -> f64 { + self.checks_min() + self.t_arm_min + self.rebase_min() + self.w_mq_min() + } +} + +/// Eq 7 evaluated on `c`, with the queue waits (`w`, `wait_mq`, `w_pass`) read from the run `r`. +/// +/// `wait_mq` is the wait for a merge pass to start, not the batch timer: with one build at a +/// time the slot is busy, so an armed PR waits out the pass in flight and any overflow. +#[must_use] +pub fn eq7_law(c: &PrFlowConfig, r: &PrFlowResult) -> Eq7Law { + let c_wall_min = c.pr_runner_min / c.pr_jobs.max(1) as f64 + c.pr_setup_min; + let t_pass_min = c.mq_runner_min / c.mq_jobs.max(1) as f64 + c.mq_setup_min; + let review = match c.rev_mode { + ReviewMode::Serial if c.rev_tiers.is_empty() => c.t_rev_min, + ReviewMode::Serial => c.rev_tiers.iter().sum::() / c.rev_tiers.len() as f64, + ReviewMode::Concurrent => 0.0, + }; + Eq7Law { + r: 1.0 / (1.0 - c.p_red), + w_min: (r.run_wall_min - c_wall_min).max(0.0), + c_wall_min, + t_fix_min: c.t_fix_min, + t_arm_min: c.t_arm_min + review, + p_dirty: 0.0, + t_reb_min: 0.0, + wait_mq_min: r.pass_wait_min, + t_pass_min, + w_pass_min: (r.pass_wall_min - t_pass_min).max(0.0), + f: c.f, + } +} + +/// The reference run: seeds 1..=5, 400 h, 20 h warm-up, averaged. +#[must_use] +pub fn run_reference(c: &PrFlowConfig) -> PrFlowResult { + let mut acc = PrFlowResult::default(); + for seed in 1..=5 { + let r = run_pr_flow(c, seed, 400.0, 20.0); + acc.merged_per_h += r.merged_per_h / 5.0; + acc.w_mean_min += r.w_mean_min / 5.0; + acc.w_p90_min += r.w_p90_min / 5.0; + acc.l += r.l / 5.0; + acc.prs_per_pass += r.prs_per_pass / 5.0; + acc.pr_util += r.pr_util / 5.0; + acc.mq_util += r.mq_util / 5.0; + acc.runs_per_pr += r.runs_per_pr / 5.0; + acc.rho_a += r.rho_a / 5.0; + acc.run_wall_min += r.run_wall_min / 5.0; + acc.checks_min += r.checks_min / 5.0; + acc.arm_min += r.arm_min / 5.0; + acc.mq_min += r.mq_min / 5.0; + acc.pass_wait_min += r.pass_wait_min / 5.0; + acc.pass_wall_min += r.pass_wall_min / 5.0; + } + acc +} + +/// SIM-3 contract: every design's merged/h is within 5% of its reference (D 20.08, A 4.96, G 12.55, E 21.37). +#[must_use] +pub fn reference_contract_ok() -> bool { + [("D", 20.08), ("A", 4.96), ("G", 12.55), ("E", 21.37)] + .iter() + .all(|(d, want)| { + preset(d).is_some_and(|c| (run_reference(&c).merged_per_h - want).abs() <= want * 0.05) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + const WANT: [(&str, f64); 4] = [("D", 20.08), ("A", 4.96), ("G", 12.55), ("E", 21.37)]; + + fn within(got: f64, want: f64, tol: f64) -> bool { + (got - want).abs() <= want * tol + } + + fn roomy() -> PrFlowConfig { + let mut c = preset("D").expect("preset"); + c.cap = 1000; + c.n = 1000.0; + c.r_pr = 10_000; + c.r_mq = 10_000; + c.builds = 100; + c + } + + #[test] + fn six_minutes_of_review_adds_about_two_to_l_at_20_per_hour() { + let mut c = roomy(); + c.t_author_min = 600.0; + c.n = 200.0; + let base = run_reference(&c); + assert!( + within(base.merged_per_h, 20.0, 0.05), + "lambda {}", + base.merged_per_h + ); + c.t_rev_min = 6.0; + let r = run_reference(&c); + let dl = r.l - base.l; + assert!(within(dl, 2.0, 0.1), "dL {dl}"); + } + + #[test] + fn author_busy_fraction_is_one_without_a_cap_and_falls_under_one_with_it() { + let mut c = roomy(); + c.n = 10.0; + let free = run_reference(&c); + assert!(free.rho_a > 0.999, "free {}", free.rho_a); + let mut c = roomy(); + c.n = 60.0; + c.cap = 2; + let capped = run_reference(&c); + assert!(capped.rho_a < 0.5, "capped {}", capped.rho_a); + } + + fn open_line() -> PrFlowConfig { + let mut c = roomy(); + c.t_author_min = 600.0; + c.n = 200.0; + c.p_red = 0.0; + c + } + + #[test] + fn concurrent_review_inside_the_check_wall_removes_the_serial_wait() { + let mut c = open_line(); + let base = run_reference(&c).w_mean_min; + c.t_rev_min = 6.0; + let serial = run_reference(&c).w_mean_min; + c.rev_mode = ReviewMode::Concurrent; + let conc = run_reference(&c).w_mean_min; + assert!( + within(serial - base, 6.0, 0.05), + "serial adds {}", + serial - base + ); + assert!( + within(serial - conc, 6.0, 0.05), + "concurrent saves {}", + serial - conc + ); + } + + #[test] + fn one_lane_queues_reviews_and_tiers_set_each_prs_time() { + let mut c = open_line(); + c.t_rev_min = 6.0; + let free = run_reference(&c).w_mean_min; + c.rev_lanes = 1; + let one = run_reference(&c).w_mean_min; + assert!(one > free + 5.0, "one lane {one} vs unlimited {free}"); + let mut t = open_line(); + t.rev_tiers = vec![2.0, 10.0]; + let tiered = run_reference(&t).w_mean_min - run_reference(&open_line()).w_mean_min; + assert!(within(tiered, 6.0, 0.1), "mean of tiers adds {tiered}"); + } + + #[test] + fn binding_is_cap_when_l_hits_cap() { + let mut c = roomy(); + c.n = 60.0; + c.cap = 2; + let r = run_reference(&c); + assert!(r.l > 0.7 * c.cap as f64, "L {}", r.l); + assert_eq!(binding_station(&c, &r), "cap"); + } + + #[test] + fn binding_is_supply_when_authors_are_starved() { + let mut c = roomy(); + c.n = 1.0; + let r = run_reference(&c); + assert_eq!(binding_station(&c, &r), "supply"); + } + + #[test] + fn binding_is_a_pool_when_the_pool_is_small() { + let mut c = roomy(); + c.n = 60.0; + c.r_pr = 4; + let r = run_reference(&c); + assert_eq!(binding_station(&c, &r), "pr_checks"); + let mut c = roomy(); + c.n = 60.0; + c.r_mq = 1; + c.builds = 1; + let r = run_reference(&c); + assert_eq!(binding_station(&c, &r), "merge_queue"); + } + + #[test] + fn fractional_n_sits_between_its_integer_neighbours() { + let at = |n: f64| { + let mut c = preset("D").expect("preset"); + c.n = n; + c.cap = 100; + run_reference(&c).merged_per_h + }; + let (lo, mid, hi) = (at(11.0), at(11.5), at(12.0)); + assert!(lo < mid && mid < hi, "{lo} {mid} {hi}"); + } + + #[test] + fn merges_per_hour_match_the_reference_within_5_percent() { + for (name, want) in WANT { + let r = run_reference(&preset(name).expect("preset")); + assert!( + within(r.merged_per_h, want, 0.05), + "{name}: {:.2} vs {want}", + r.merged_per_h + ); + } + } + + #[test] + fn littles_law_holds_within_half_a_percent() { + for (name, _) in WANT { + let r = run_reference(&preset(name).expect("preset")); + let lam_w = r.merged_per_h * r.w_mean_min / 60.0; + assert!( + within(lam_w, r.l, 0.005), + "{name}: lambda*W {lam_w:.3} vs L {:.3}", + r.l + ); + } + } + + #[test] + fn eq7_time_open_matches_the_sim_within_2_percent_on_design_d() { + let c = preset("D").expect("preset"); + let r = run_reference(&c); + let law = eq7_law(&c, &r); + assert!( + within(r.w_mean_min, law.w_law_min(), 0.02), + "sim W {:.1} vs law W {:.1} ({law:?})", + r.w_mean_min, + law.w_law_min() + ); + } + + #[test] + fn eq7_terms_match_their_sim_stages() { + let c = preset("D").expect("preset"); + let r = run_reference(&c); + let law = eq7_law(&c, &r); + assert!(within(r.checks_min, law.checks_min(), 0.02), "checks"); + assert!(within(r.arm_min, law.t_arm_min, 0.02), "arm"); + assert!(within(r.mq_min, law.w_mq_min(), 0.02), "W_mq"); + } + + #[test] + fn the_batch_timer_is_not_the_merge_queue_wait_with_one_build() { + // One build at a time keeps the build slot busy: an armed PR waits for the pass in + // flight (and any overflow past the batch size), not for the 5-minute batch timer. + let c = preset("D").expect("preset"); + let r = run_reference(&c); + let timer = Eq7Law { + wait_mq_min: c.mq_wait_min, + ..eq7_law(&c, &r) + }; + assert!( + !within(r.w_mean_min, timer.w_law_min(), 0.02), + "timer-wait law {:.1} matched sim {:.1}", + timer.w_law_min(), + r.w_mean_min + ); + assert!( + r.pass_wait_min > 2.0 * c.mq_wait_min, + "wait {}", + r.pass_wait_min + ); + } + + #[test] + fn same_seed_same_output() { + let c = preset("D").expect("preset"); + assert_eq!( + run_pr_flow(&c, 7, 100.0, 5.0), + run_pr_flow(&c, 7, 100.0, 5.0) + ); + } + + #[test] + fn a_reverted_design_shows_red() { + // D with a single merge-queue runner is not D: the 5% band must reject it. + let mut c = preset("D").expect("preset"); + c.r_mq = 1; + let r = run_reference(&c); + assert!( + !within(r.merged_per_h, 20.08, 0.05), + "mutant passed at {:.2}", + r.merged_per_h + ); + } + + #[test] + fn unknown_preset_is_none() { + assert!(preset("Z").is_none()); + } +} diff --git a/crates/aprender-simulate/src/scenarios/repo_flow.rs b/crates/aprender-simulate/src/scenarios/repo_flow.rs new file mode 100644 index 0000000000..ebc1d6c7af --- /dev/null +++ b/crates/aprender-simulate/src/scenarios/repo_flow.rs @@ -0,0 +1,253 @@ +//! Scenario "repo-flow": the FLOW-006 repo template (SIM-6, #5511). +//! +//! The PR-flow line with the +//! merge queue optional (`builds == 0` merges on arm) and the conflict step: an armed PR is found +//! dirty with `p_dirty` and waits `t_rebase_min` for a rebase and a rerun. One parameter file per +//! repo; inputs come from the KPI report, never typed. + +use super::pr_flow::{run_pr_flow, PrFlowConfig, PrFlowResult, ReviewMode}; + +/// A named repo case: `(name, supply N, t_author, cap, R_pr, runner-min, jobs, setup, p_red, p_dirty, t_rebase)`. +type Case = ( + &'static str, + usize, + f64, + usize, + usize, + f64, + usize, + f64, + f64, + f64, + f64, +); + +const CASES: [Case; 7] = [ + ( + "rmedia-design", + 9, + 25.0, + 10, + 9, + 17.1, + 3, + 3.0, + 0.1, + 0.05, + 60.0, + ), + ( + "infra-today", + 7, + 540.0, + 10, + 6, + 40.0, + 1, + 5.0, + 0.2, + 0.3, + 10000.0, + ), + ("infra-rule", 7, 540.0, 10, 6, 40.0, 1, 5.0, 0.2, 0.3, 60.0), + ( + "infra-design", + 16, + 45.0, + 15, + 6, + 10.0, + 1, + 10.0, + 0.1, + 0.05, + 60.0, + ), + ("infra-33", 16, 45.0, 15, 6, 33.0, 1, 10.0, 0.1, 0.05, 60.0), + ("small-10", 16, 45.0, 10, 12, 20.0, 1, 3.0, 0.1, 0.05, 60.0), + ("small-15", 16, 45.0, 15, 12, 20.0, 1, 3.0, 0.1, 0.05, 60.0), +]; + +/// Names of the FLOW-006 repo cases. +#[must_use] +pub fn repo_cases() -> Vec<&'static str> { + CASES.iter().map(|c| c.0).collect() +} + +/// The configuration of a named repo case; no merge queue (`builds = 0`). +#[must_use] +pub fn repo_preset(name: &str) -> Option { + let c = CASES.iter().find(|c| c.0 == name)?; + Some(PrFlowConfig { + cap: c.3, + n: c.1 as f64, + t_author_min: c.2, + r_pr: c.4, + pr_runner_min: c.5, + pr_jobs: c.6, + pr_setup_min: c.7, + p_red: c.8, + t_fix_min: 10.0, + t_arm_min: 2.0, + r_mq: 1, + mq_runner_min: 0.0, + mq_jobs: 1, + mq_setup_min: 0.0, + builds: 0, + mq_batch: 1, + mq_wait_min: 0.0, + f: 0.0, + p_dirty: c.9, + t_rebase_min: c.10, + t_rev_min: 0.0, + rev_mode: ReviewMode::Serial, + rev_lanes: 0, + rev_tiers: Vec::new(), + }) +} + +/// The reference run of a repo case: seeds 1..=5, 400 h, 20 h warm-up, averaged. +#[must_use] +pub fn run_repo_reference(c: &PrFlowConfig) -> PrFlowResult { + let mut acc = PrFlowResult::default(); + for seed in 1..=5 { + let r = run_pr_flow(c, seed, 400.0, 20.0); + acc.merged_per_h += r.merged_per_h / 5.0; + acc.w_mean_min += r.w_mean_min / 5.0; + acc.w_p90_min += r.w_p90_min / 5.0; + acc.l += r.l / 5.0; + acc.pr_util += r.pr_util / 5.0; + acc.runs_per_pr += r.runs_per_pr / 5.0; + } + acc +} + +/// Override the case with the ledger's measured inputs (a JSON object); unknown keys are refused. +/// +/// # Errors +/// Returns the offending key or type when the object holds something the model does not read. +pub fn apply_inputs(c: &mut PrFlowConfig, v: &serde_json::Value) -> Result<(), String> { + let obj = v.as_object().ok_or("inputs must be a JSON object")?; + for (k, x) in obj { + let f = x.as_f64().ok_or_else(|| format!("{k}: not a number"))?; + match k.as_str() { + "cap" => c.cap = f as usize, + "N" => c.n = f, + "t_author_min" => c.t_author_min = f, + "R_pr" => c.r_pr = f as usize, + "pr_runner_min" => c.pr_runner_min = f, + "p_red" => c.p_red = f, + "t_fix_min" => c.t_fix_min = f, + "t_arm_min" => c.t_arm_min = f, + "p_dirty" => c.p_dirty = f, + "t_rebase_min" => c.t_rebase_min = f, + _ => return Err(format!("unknown input: {k}")), + } + } + Ok(()) +} + +/// The `sim` block fields this scenario owns: scenario, seeds, hours, inputs, predicted, binding. +#[must_use] +pub fn sim_block(case: &str, c: &PrFlowConfig, r: &PrFlowResult) -> serde_json::Value { + let binding = if r.pr_util > 0.85 { + "ci_pool" + } else if r.l >= c.cap as f64 * 0.95 { + "cap" + } else { + "supply" + }; + serde_json::json!({ + "scenario": format!("repo-flow:{case}"), + "seeds": 5, "hours": 400, "warmup_h": 20, + "inputs": { + "cap": c.cap, "N": c.n, "t_author_min": c.t_author_min, "R_pr": c.r_pr, + "pr_runner_min": c.pr_runner_min, "p_red": c.p_red, "t_fix_min": c.t_fix_min, + "t_arm_min": c.t_arm_min, "p_dirty": c.p_dirty, "t_rebase_min": c.t_rebase_min, + }, + "predicted": { + "merged_h": r.merged_per_h, "W_min_mean": r.w_mean_min, "W_min_p90": r.w_p90_min, + "L": r.l, "pr_pool_util_pct": r.pr_util * 100.0, "runs_per_pr": r.runs_per_pr, + }, + "binding_station": binding, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// FLOW-006 reference bands: (case, merged/h, tolerance). + const WANT: [(&str, f64, f64); 7] = [ + ("rmedia-design", 21.13, 0.05), + ("infra-today", 0.142, 0.10), + ("infra-rule", 0.78, 0.05), + ("infra-design", 21.22, 0.05), + ("infra-33", 9.71, 0.05), + ("small-10", 17.48, 0.05), + ("small-15", 21.38, 0.05), + ]; + + fn got(name: &str) -> PrFlowResult { + run_repo_reference(&repo_preset(name).expect("case")) + } + + #[test] + fn merges_per_hour_match_the_reference_bands() { + for (name, want, tol) in WANT { + let r = got(name); + assert!( + (r.merged_per_h - want).abs() <= want * tol, + "{name}: {:.3} vs {want}", + r.merged_per_h + ); + } + } + + #[test] + fn littles_law_holds_within_half_a_percent() { + // infra-today strands PRs past the horizon: they sit in L but never in W, so Little is + // not defined for that unsteady case. + for (name, _, _) in WANT.iter().filter(|w| w.0 != "infra-today") { + let r = got(name); + let lam_w = r.merged_per_h * r.w_mean_min / 60.0; + assert!( + (lam_w - r.l).abs() <= r.l * 0.005, + "{name}: {lam_w:.3} vs {:.3}", + r.l + ); + } + } + + #[test] + fn same_seed_same_output() { + let c = repo_preset("infra-rule").expect("case"); + assert_eq!( + run_pr_flow(&c, 3, 100.0, 5.0), + run_pr_flow(&c, 3, 100.0, 5.0) + ); + } + + #[test] + fn keep_fresh_rule_alone_lifts_infra_fivefold() { + let (a, b) = (got("infra-today"), got("infra-rule")); + assert!(b.merged_per_h > 4.0 * a.merged_per_h); + } + + #[test] + fn a_reverted_rule_shows_red() { + let mut c = repo_preset("infra-rule").expect("case"); + c.t_rebase_min = 10000.0; + let r = run_repo_reference(&c); + assert!( + (r.merged_per_h - 0.78).abs() > 0.78 * 0.05, + "mutant passed at {:.3}", + r.merged_per_h + ); + } + + #[test] + fn unknown_case_is_none() { + assert!(repo_preset("nope").is_none()); + } +} diff --git a/crates/aprender-simulate/tests/cli_surface_gate.rs b/crates/aprender-simulate/tests/cli_surface_gate.rs index 0210f97578..6a8109dce8 100644 --- a/crates/aprender-simulate/tests/cli_surface_gate.rs +++ b/crates/aprender-simulate/tests/cli_surface_gate.rs @@ -18,6 +18,9 @@ const UNKNOWN_SUB: &[&str] = &["no-such-subcommand"]; const SUBCOMMANDS: &[&str] = &[ "run", "render", + "pr-flow", + "repo-flow", + "course-flow", "validate", "verify", "emc-check", diff --git a/crates/aprender-simulate/tests/pr_flow_contract.rs b/crates/aprender-simulate/tests/pr_flow_contract.rs new file mode 100644 index 0000000000..f53f6adb0f --- /dev/null +++ b/crates/aprender-simulate/tests/pr_flow_contract.rs @@ -0,0 +1,128 @@ +//! SIM-3 (FLOW-005, #5511): the `pr-flow` scenario is a contract, run through the shipped binary. +//! +//! Each design must land inside its reference band, Little's law must close, and the same +//! seeds must give the same bytes. A scenario that returned a constant would fail the band for +//! three of the four designs; a scenario that read a clock or an unseeded RNG would fail the +//! byte comparison. + +use std::process::Command; + +const BIN: &str = env!("CARGO_BIN_EXE_aprender-simulate"); + +/// Design, reference merged/h (the +-5% band is the contract). +const WANT: [(&str, f64); 4] = [("D", 20.08), ("A", 4.96), ("G", 12.55), ("E", 21.37)]; + +fn block(design: &str) -> String { + let out = Command::new(BIN) + .args(["pr-flow", "--design", design, "--json", "--sha", "test"]) + .output() + .expect("spawn the binary under test"); + assert!( + out.status.success(), + "pr-flow {design} exited {:?}", + out.status + ); + String::from_utf8(out.stdout).expect("utf-8 json") +} + +fn num(json: &serde_json::Value, path: &[&str]) -> f64 { + path.iter() + .fold(json, |v, k| &v[*k]) + .as_f64() + .unwrap_or(f64::NAN) +} + +#[test] +fn every_design_is_inside_its_reference_band_and_littles_law_closes() { + for (design, want) in WANT { + let v: serde_json::Value = serde_json::from_str(&block(design)).expect("json"); + let got = num(&v, &["predicted", "merged_h"]); + assert!( + (got - want).abs() <= want * 0.05, + "{design}: {got:.2} vs {want}" + ); + let (w_min, l) = ( + num(&v, &["predicted", "W_min_mean"]), + num(&v, &["predicted", "L"]), + ); + let lam_w = got * w_min / 60.0; + assert!( + (lam_w - l).abs() <= l * 0.005, + "{design}: lambda*W {lam_w:.3} vs L {l:.3}" + ); + assert_eq!(v["sim3_contract"], "pass", "{design}"); + } +} + +#[test] +fn same_seeds_give_the_same_bytes() { + for (design, _) in WANT { + assert_eq!(block(design), block(design), "{design}"); + } +} + +#[test] +fn params_file_gives_actual_one_change_and_optimum_with_the_same_bytes() { + let dir = std::path::PathBuf::from(env!("CARGO_TARGET_TMPDIR")); + let file = dir.join("pr-flow-params.json"); + std::fs::write( + &file, + r#"{"repo":"x","design":"D","params":{"cap":10,"r_pr":28,"p_red":0.62}}"#, + ) + .expect("write params"); + let run = |args: &[&str]| { + let out = Command::new(BIN) + .args(["pr-flow", "--params"]) + .arg(&file) + .args(args) + .output() + .expect("spawn"); + ( + out.status.success(), + String::from_utf8(out.stdout).expect("utf-8"), + ) + }; + let flags = ["--seeds", "2", "--hours", "100", "--warmup", "5", "--json"]; + let (ok, a) = run(&flags); + assert!(ok); + assert_eq!(a, run(&flags).1, "same params must give the same bytes"); + let v: serde_json::Value = serde_json::from_str(&a).expect("json"); + for k in ["actual", "optimum", "one_change", "binding_station"] { + assert!(!v[k].is_null() || k == "binding_station", "missing {k}"); + } + assert!(num(&v, &["optimum", "merged_h"]) > num(&v, &["actual", "merged_h"])); + std::fs::write(&file, "not json").expect("write bad"); + let (ok, _) = run(&flags); + assert!(!ok, "a malformed params file must fail, not print a number"); +} + +#[test] +fn json_block_reports_the_author_busy_fraction_beside_the_assumption() { + let out = std::process::Command::new(env!("CARGO_BIN_EXE_aprender-simulate")) + .args(["pr-flow", "--design", "D", "--json"]) + .output() + .expect("run"); + let v: serde_json::Value = serde_json::from_slice(&out.stdout).expect("json"); + let rho = v["predicted"]["rho_a"].as_f64().expect("rho_a present"); + assert!((0.0..=1.0).contains(&rho), "rho_a {rho}"); + assert!((v["predicted"]["rho_a_assumed"].as_f64().expect("assumed") - 0.95).abs() < 1e-9); +} + +#[test] +fn json_law_block_carries_eq7_within_2_percent_of_the_sim_on_design_d() { + let v: serde_json::Value = serde_json::from_str(&block("D")).expect("json"); + let (sim, law) = ( + num(&v, &["predicted", "W_min_mean"]), + num(&v, &["law", "W_law_min"]), + ); + assert!( + (sim - law).abs() <= law * 0.02, + "sim W {sim:.1} vs law W {law:.1}" + ); + let terms = num(&v, &["law", "checks_min"]) + + num(&v, &["law", "t_arm_min"]) + + num(&v, &["law", "rebase_min"]) + + num(&v, &["law", "W_mq_min"]); + assert!((terms - law).abs() < 1e-9, "terms {terms} sum to {law}"); + assert_eq!(num(&v, &["law", "p_dirty"]), 0.0); +} diff --git a/docs/audits/surface_audit.csv b/docs/audits/surface_audit.csv index 10f485d4f2..122777439b 100644 --- a/docs/audits/surface_audit.csv +++ b/docs/audits/surface_audit.csv @@ -416,15 +416,15 @@ apr,apr serve run --no-metrics,4,x86_64-linux;nvidia-cuda,llama.cpp,no,10,apr-co apr,apr shard,6,x86_64-linux,llama.cpp,yes,1,apr-lint-diag,crates/apr-cli/src/extended_commands.rs:1239,high apr,apr shared-cache-lint,1,x86_64-linux,NONE,yes,1,apr-lint-diag,crates/apr-cli/src/extended_commands.rs:1217,high apr,apr showcase,6,x86_64-linux,NONE,yes,10,apr-core-commands,crates/apr-cli/src/tool_commands.rs:5,high -apr,apr sim emc-check,6,x86_64-linux,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:105,low -apr,apr sim emc-validate,6,x86_64-linux,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:110,low -apr,apr sim help,6,x86_64-linux,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:117,low -apr,apr sim list-emc,6,x86_64-linux,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:115,low -apr,apr sim render,6,x86_64-linux,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:71,low -apr,apr sim run,6,x86_64-linux,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:60,low -apr,apr sim validate,6,x86_64-linux,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:92,low -apr,apr sim verify,6,x86_64-linux,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:97,low -apr,apr sim version,6,x86_64-linux,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:119,low +apr,apr sim emc-check,6,x86_64-linux,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:206,low +apr,apr sim emc-validate,6,x86_64-linux,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:211,low +apr,apr sim help,6,x86_64-linux,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:218,low +apr,apr sim list-emc,6,x86_64-linux,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:216,low +apr,apr sim render,6,x86_64-linux,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:79,low +apr,apr sim run,6,x86_64-linux,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:68,low +apr,apr sim validate,6,x86_64-linux,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:193,low +apr,apr sim verify,6,x86_64-linux,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:198,low +apr,apr sim version,6,x86_64-linux,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:220,low apr,apr stamp,2,x86_64-linux,NONE,yes,10,apr-core-commands,crates/apr-cli/src/commands_enum.rs:638,high apr,apr tensors,4,x86_64-linux;nvidia-cuda,safetensors,yes,10,apr-core-commands,crates/apr-cli/src/commands_enum.rs:362,high apr,apr test llm bench,6,x86_64-linux,vllm,no,1,apr-lint-diag,crates/apr-cli/src/extended_commands.rs:1745 (dispatch crates/apr-cli/src/commands/test_llm.rs:482),medium @@ -832,15 +832,15 @@ aprender-score,aprender-score [PATH] (TUI-crate quality scoring),6,UNKNOWN,NONE, aprender-simulate,unserved: GET / (visualization UI; library feature web; no aprender-simulate command starts the server),6,UNKNOWN,NONE,no,0,entrypoint-and-mcp,crates/aprender-simulate/src/visualization/web.rs:88,medium aprender-simulate,unserved: GET /health (visualization; library feature web; no aprender-simulate command starts the server),6,UNKNOWN,NONE,no,0,entrypoint-and-mcp,crates/aprender-simulate/src/visualization/web.rs:89,medium aprender-simulate,unserved: GET /ws (visualization WebSocket; library feature web; no aprender-simulate command starts the server),6,UNKNOWN,NONE,no,0,entrypoint-and-mcp,crates/aprender-simulate/src/visualization/web.rs:82,medium -aprender-simulate,aprender-simulate emc-check,6,UNKNOWN,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:105,high -aprender-simulate,aprender-simulate emc-validate,6,UNKNOWN,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:110,high -aprender-simulate,aprender-simulate help,6,UNKNOWN,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:117,high -aprender-simulate,aprender-simulate list-emc,6,UNKNOWN,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:115,high -aprender-simulate,aprender-simulate render,6,UNKNOWN,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:71,high -aprender-simulate,aprender-simulate run,6,UNKNOWN,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:60,high -aprender-simulate,aprender-simulate validate,6,UNKNOWN,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:92,high -aprender-simulate,aprender-simulate verify,6,UNKNOWN,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:97,high -aprender-simulate,aprender-simulate version,6,UNKNOWN,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:119,high +aprender-simulate,aprender-simulate emc-check,6,UNKNOWN,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:206,high +aprender-simulate,aprender-simulate emc-validate,6,UNKNOWN,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:211,high +aprender-simulate,aprender-simulate help,6,UNKNOWN,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:218,high +aprender-simulate,aprender-simulate list-emc,6,UNKNOWN,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:216,high +aprender-simulate,aprender-simulate render,6,UNKNOWN,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:79,high +aprender-simulate,aprender-simulate run,6,UNKNOWN,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:68,high +aprender-simulate,aprender-simulate validate,6,UNKNOWN,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:193,high +aprender-simulate,aprender-simulate verify,6,UNKNOWN,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:198,high +aprender-simulate,aprender-simulate version,6,UNKNOWN,NONE,no,11,simulation,crates/aprender-simulate/src/cli/args.rs:220,high aprender-rag,aprender-rag demo,6,UNKNOWN,huggingface-transformers,no,8,rag-eval,crates/aprender-rag-cli/src/lib.rs:107,high aprender-rag,aprender-rag eval compare,6,UNKNOWN,huggingface-transformers,no,8,rag-eval,crates/aprender-rag-cli/src/lib.rs:455,high aprender-rag,aprender-rag eval gate,6,UNKNOWN,huggingface-transformers,no,8,rag-eval,crates/aprender-rag-cli/src/lib.rs:466,high