diff --git a/dsl/extdeps/shell/shell.dag b/dsl/extdeps/shell/shell.dag index 0209cb20df3..dc91b3e720b 100644 --- a/dsl/extdeps/shell/shell.dag +++ b/dsl/extdeps/shell/shell.dag @@ -34,7 +34,6 @@ service shell.Env { 0 => Unit 1 => Unit "Variable not set" } - } } diff --git a/dsl/gunbc/accelerator_demo_plan.dag b/dsl/gunbc/accelerator_demo_plan.dag index 3ddd60ec16d..2e3c4775b6d 100644 --- a/dsl/gunbc/accelerator_demo_plan.dag +++ b/dsl/gunbc/accelerator_demo_plan.dag @@ -18,6 +18,7 @@ import std.realization_schedule { RealizationObjective, Runnable, RunnableKernelWorkload, + RunnableCompileCleanShardBatch, Schedule, CostAccount, cost_account_predicted_zero, @@ -193,6 +194,7 @@ fn demo_plan_fused_op_count(plan: RealizationPlan) -> Int { RunnableKernelWorkload { fused_op_count: count } => count RunnableSingleClaim { entry: _, function: _, profile: _ } => 0 RunnableDiscoveryBatch { source_roots: _, scan_dirs: _, explicit_entries: _, skip_unaffected_node_frontier: _, exclude_substrings: _, discovery_scope_dirs: _, spawn_width_cap: _, profile: _ } => 0 + RunnableCompileCleanShardBatch { entry_paths: _, spawn_width_cap: _, profile: _ } => 0 } } } diff --git a/dsl/std/realization.dag b/dsl/std/realization.dag index 959fa31eaf5..0661bac9201 100644 --- a/dsl/std/realization.dag +++ b/dsl/std/realization.dag @@ -11,6 +11,8 @@ import std.realization_schedule { Runnable, RunnableSingleClaim, RunnableDiscoveryBatch, + RunnableCompileCleanShardBatch, + compile_clean_shard_batch_label, ScheduleWitnessEntry, Schedule, CostAccount, diff --git a/dsl/std/realization_schedule.dag b/dsl/std/realization_schedule.dag index 66b654592d8..cb8f4e04fc5 100644 --- a/dsl/std/realization_schedule.dag +++ b/dsl/std/realization_schedule.dag @@ -155,6 +155,7 @@ fn runnable_profile(r: Runnable) -> RunnableResourceProfile { match r { RunnableSingleClaim { entry: _, function: _, profile: p } => p RunnableDiscoveryBatch { source_roots: _, scan_dirs: _, explicit_entries: _, skip_unaffected_node_frontier: _, exclude_substrings: _, discovery_scope_dirs: _, spawn_width_cap: _, profile: p } => p + RunnableCompileCleanShardBatch { entry_paths: _, spawn_width_cap: _, profile: p } => p RunnableKernelWorkload { fused_op_count: _ } => runnable_resource_profile_negligible() } } @@ -162,6 +163,7 @@ fn runnable_profile(r: Runnable) -> RunnableResourceProfile { fn runnable_forbids_corpus_co_residence(r: Runnable) -> Bool { match r { RunnableDiscoveryBatch { source_roots: _, scan_dirs: _, explicit_entries: _, skip_unaffected_node_frontier: _, exclude_substrings: _, discovery_scope_dirs: _, spawn_width_cap: _, profile: _ } => false + RunnableCompileCleanShardBatch { entry_paths: _, spawn_width_cap: _, profile: p } => runnable_excludes_corpus_co_residence(profile: p) RunnableSingleClaim { entry: _, function: _, profile: p } => runnable_excludes_corpus_co_residence(profile: p) RunnableKernelWorkload { fused_op_count: _ } => false } @@ -179,8 +181,15 @@ type Runnable spawn_width_cap: Int profile: RunnableResourceProfile } + | RunnableCompileCleanShardBatch { + entry_paths: List + spawn_width_cap: Int + profile: RunnableResourceProfile + } | RunnableKernelWorkload { fused_op_count: Int } +data compile_clean_shard_batch_label: String = "__compile_clean_shard_batch__" + data runnable_kernel_workload_dissolution_trigger: String = "dissolve-on: execution lane attaches extdeps.languages kernel_surface rows to RunnableKernelWorkload when fused_op_count alone is insufficient; op_count is structural on substrate (not a string-label nickname)." type Schedule = List> @@ -197,10 +206,32 @@ fn runnable_step_label(r: Runnable) -> String { match r { RunnableSingleClaim { entry: _, function: f, profile: _ } => f RunnableDiscoveryBatch { source_roots: _, scan_dirs: _, explicit_entries: _, skip_unaffected_node_frontier: _, exclude_substrings: _, discovery_scope_dirs: _, spawn_width_cap: _, profile: _ } => "__discovery_corpus__" + RunnableCompileCleanShardBatch { entry_paths: _, spawn_width_cap: _, profile: _ } => compile_clean_shard_batch_label RunnableKernelWorkload { fused_op_count: _ } => "__kernel_workload__" } } +fn runnable_compile_clean_shard_entry_paths(r: Runnable) -> List { + match r { + RunnableCompileCleanShardBatch { entry_paths: paths, spawn_width_cap: _, profile: _ } => paths + _ => [] + } +} + +fn runnable_compile_clean_shard_count(r: Runnable) -> Int { + length(xs: runnable_compile_clean_shard_entry_paths(r: r)) +} + +fn schedule_batch_compile_clean_shard_entry_paths(batch: List) -> List { + fold(batch, init: [], f: fn(acc, r) { + concat(acc, runnable_compile_clean_shard_entry_paths(r: r)) + }) +} + +fn schedule_batch_compile_clean_shard_count(batch: List) -> Int { + length(xs: schedule_batch_compile_clean_shard_entry_paths(batch: batch)) +} + fn schedule_batch_contains_label(batch: List, target: String) -> Bool { fold(batch, init: false, f: fn(acc, r) { acc || (runnable_step_label(r: r) == target) @@ -210,6 +241,8 @@ fn schedule_batch_contains_label(batch: List, target: String) -> Bool type ScheduleLensViolation = EmptySchedule | CompileGateNotFirst { expected: String } + | CompileGateMonolithPresent { expected: String } + | CompileCleanShardsInsufficient { minimum: Int } | CorpusBeforeCompile | SingleBatchOnly @@ -228,6 +261,16 @@ fn schedule_lens_violation_diagnostic( reason: "schedule_lens_compile_gate_not_first:" + expected, at: at } + CompileGateMonolithPresent { expected: expected } => + LensVerdictDiagnostic { + reason: "schedule_lens_compile_gate_monolith_present:" + expected, + at: at + } + CompileCleanShardsInsufficient { minimum: _ } => + LensVerdictDiagnostic { + reason: "schedule_lens_compile_clean_shards_insufficient", + at: at + } CorpusBeforeCompile => LensVerdictDiagnostic { reason: "schedule_lens_corpus_before_compile", at: at } SingleBatchOnly => @@ -235,7 +278,7 @@ fn schedule_lens_violation_diagnostic( } } -fn schedule_lens_verdict_for_ci_floor(plan: RealizationPlan, compile_gate_fn: String) -> LensVerdict { +fn schedule_lens_verdict_for_ci_floor(plan: RealizationPlan, compile_gate_fn: String, compile_clean_sharded: Bool) -> LensVerdict { if plan.schedule.count() == 0 { Violation { diagnostic: schedule_lens_violation_diagnostic( @@ -250,7 +293,7 @@ fn schedule_lens_verdict_for_ci_floor(plan: RealizationPlan, compile_gate_ compile_gate_fn: compile_gate_fn ) } - } else { + } else if !compile_clean_sharded { let batch0 = plan.schedule.first() if batch0.count() != 1 { Violation { @@ -286,6 +329,43 @@ fn schedule_lens_verdict_for_ci_floor(plan: RealizationPlan, compile_gate_ Holds } } + } else { + let batch0 = plan.schedule.first() + let shard_count = schedule_batch_compile_clean_shard_count(batch: batch0) + if schedule_batch_contains_label(batch: batch0, target: compile_gate_fn) { + Violation { + diagnostic: schedule_lens_violation_diagnostic( + kind: CompileGateMonolithPresent { expected: compile_gate_fn }, + compile_gate_fn: compile_gate_fn + ) + } + } else if shard_count < 2 { + Violation { + diagnostic: schedule_lens_violation_diagnostic( + kind: CompileCleanShardsInsufficient { minimum: 2 }, + compile_gate_fn: compile_gate_fn + ) + } + } else if schedule_batch_contains_label(batch: batch0, target: "__discovery_corpus__") { + Violation { + diagnostic: schedule_lens_violation_diagnostic( + kind: CorpusBeforeCompile, + compile_gate_fn: compile_gate_fn + ) + } + } else { + let batch1 = plan.schedule.skip(n: 1).first() + if batch1.count() < 2 { + Violation { + diagnostic: schedule_lens_violation_diagnostic( + kind: SingleBatchOnly, + compile_gate_fn: compile_gate_fn + ) + } + } else { + Holds + } + } } } @@ -327,6 +407,7 @@ fn runnable_eq(left: Runnable, right: Runnable) -> Bool { RunnableSingleClaim { entry: re, function: rf, profile: rp } => (le == re) && (lf == rf) && runnable_resource_profile_eq(left: lp, right: rp) RunnableDiscoveryBatch { source_roots: _, scan_dirs: _, explicit_entries: _, skip_unaffected_node_frontier: _, exclude_substrings: _, discovery_scope_dirs: _, spawn_width_cap: _, profile: _ } => false + RunnableCompileCleanShardBatch { entry_paths: _, spawn_width_cap: _, profile: _ } => false RunnableKernelWorkload { fused_op_count: _ } => false } RunnableDiscoveryBatch { source_roots: lsr, scan_dirs: lsd, explicit_entries: lex, skip_unaffected_node_frontier: lskip, exclude_substrings: lex2, discovery_scope_dirs: lsc, spawn_width_cap: lwc, profile: lp } => match right { @@ -340,11 +421,22 @@ fn runnable_eq(left: Runnable, right: Runnable) -> Bool { && string_list_eq(left: lsc, right: rsc) && (lwc == rwc) && runnable_resource_profile_eq(left: lp, right: rp) + RunnableCompileCleanShardBatch { entry_paths: _, spawn_width_cap: _, profile: _ } => false + RunnableKernelWorkload { fused_op_count: _ } => false + } + RunnableCompileCleanShardBatch { entry_paths: lpaths, spawn_width_cap: lwc, profile: lp } => match right { + RunnableSingleClaim { entry: _, function: _, profile: _ } => false + RunnableDiscoveryBatch { source_roots: _, scan_dirs: _, explicit_entries: _, skip_unaffected_node_frontier: _, exclude_substrings: _, discovery_scope_dirs: _, spawn_width_cap: _, profile: _ } => false + RunnableCompileCleanShardBatch { entry_paths: rpaths, spawn_width_cap: rwc, profile: rp } => + string_list_eq(left: lpaths, right: rpaths) + && (lwc == rwc) + && runnable_resource_profile_eq(left: lp, right: rp) RunnableKernelWorkload { fused_op_count: _ } => false } RunnableKernelWorkload { fused_op_count: lleft } => match right { RunnableSingleClaim { entry: _, function: _, profile: _ } => false RunnableDiscoveryBatch { source_roots: _, scan_dirs: _, explicit_entries: _, skip_unaffected_node_frontier: _, exclude_substrings: _, discovery_scope_dirs: _, spawn_width_cap: _, profile: _ } => false + RunnableCompileCleanShardBatch { entry_paths: _, spawn_width_cap: _, profile: _ } => false RunnableKernelWorkload { fused_op_count: rcount } => lleft == rcount } } diff --git a/dsl/test/claim/dsl_compile_clean_shard_b_witness_test.dag b/dsl/test/claim/dsl_compile_clean_shard_b_witness_test.dag new file mode 100644 index 00000000000..231fdd23fab --- /dev/null +++ b/dsl/test/claim/dsl_compile_clean_shard_b_witness_test.dag @@ -0,0 +1,102 @@ +module test.claim.dsl_compile_clean_shard_b_witness + +import tools.dsl_compile_clean_partition { + compose_compile_clean_shard_verdicts, + compose_compile_clean_shard_verdicts_as_reduce, + compile_clean_shard_verdict_from_bool, + compile_clean_shard_verdict_to_reduce, + ShardClean, + ShardFailed, + witness_empty_compose_is_unknown_reduce +} +import tools.dsl_compile_clean_shard_transport { + run_shard_compile_verdict, + run_shard_a_exemplar_compile, + witness_broken_staged_shard_compile_red, + shard_a_exemplar_entry_path +} +import v2.workflow.compile_clean_shard_roster { + compile_clean_sharding_enabled, + compile_clean_whole_tree_rollback_env, + compile_clean_whole_tree_rollback_disposition, + compile_clean_shard_roster_has_multiple, + compile_clean_shard_roster_count, + witness_shard_roster_derives_from_witness_layer_roots, + witness_shard_roster_paths_duplicate_free, + witness_shard_roster_coverage, + witness_dropped_shard_breaks_coverage, + compile_clean_shard_entry_paths, + witness_compose_all_clean_matches_reduce, + witness_compose_failure_is_fail_closed +} +import std.disposition { Scaffold } +import v2.workflow.ci_floor_plan { + gunbc_ci_floor_batches, + witness_floor_batch_one_uses_compile_clean_shards, + witness_floor_batch_one_has_no_compile_gate_monolith, + witness_floor_shard_roster_coverage_in_plan, + witness_floor_dropped_shard_red_control, + gunbc_ci_floor_compile_clean_shard_count, + gunbc_ci_floor_schedule_lens_holds +} + +fn witness_transport_verdict_bridge_green() -> Bool { + run_shard_a_exemplar_compile() + && (run_shard_compile_verdict(entry_rel_path: shard_a_exemplar_entry_path) == ShardClean) +} + +fn witness_transport_verdict_bridge_red_on_failure() -> Bool { + let v = compile_clean_shard_verdict_from_bool( + entry_path: shard_a_exemplar_entry_path, + ok: false + ) + v == ShardFailed { entry_path: shard_a_exemplar_entry_path } +} + +fn witness_broken_module_shard_reds_composed_gate() -> Bool { + witness_broken_staged_shard_compile_red() + && (compose_compile_clean_shard_verdicts( + verdicts: [ShardClean, ShardFailed { entry_path: "staged/broken.dag" }] + ) != ShardClean) +} + +fn witness_plan_shard_count_matches_roster() -> Bool { + gunbc_ci_floor_compile_clean_shard_count() == compile_clean_shard_roster_count() +} + +test fn compile_clean_shard_b_floor_plan_enrollment() -> Bool { + compile_clean_sharding_enabled() + && witness_floor_batch_one_uses_compile_clean_shards() + && witness_floor_batch_one_has_no_compile_gate_monolith() + && witness_plan_shard_count_matches_roster() + && witness_floor_shard_roster_coverage_in_plan() + && gunbc_ci_floor_schedule_lens_holds() +} + +test fn compile_clean_shard_b_verdict_algebra_and_transport() -> Bool { + witness_transport_verdict_bridge_green() + && witness_transport_verdict_bridge_red_on_failure() + && witness_broken_module_shard_reds_composed_gate() + && witness_compose_all_clean_matches_reduce() + && witness_compose_failure_is_fail_closed() + && witness_empty_compose_is_unknown_reduce() +} + +test fn compile_clean_shard_b_roster_and_coverage_red() -> Bool { + witness_shard_roster_derives_from_witness_layer_roots() + && witness_shard_roster_paths_duplicate_free() + && compile_clean_shard_roster_has_multiple() + && witness_dropped_shard_breaks_coverage( + full: compile_clean_shard_entry_paths(), + plan_paths: compile_clean_shard_entry_paths() + ) + && witness_floor_dropped_shard_red_control() +} + +test fn compile_clean_whole_tree_rollback_env_is_named() -> Bool { + compile_clean_whole_tree_rollback_env == "GUNBC_CI_COMPILE_CLEAN_WHOLE_TREE" + && match compile_clean_whole_tree_rollback_disposition { + Scaffold { dissolves_to: _, bind: _ } => true + _ => false + } +} diff --git a/dsl/test/claim/realization_schedule_witness_test.dag b/dsl/test/claim/realization_schedule_witness_test.dag index 5d8d28358c0..e98c7c3a86e 100644 --- a/dsl/test/claim/realization_schedule_witness_test.dag +++ b/dsl/test/claim/realization_schedule_witness_test.dag @@ -51,7 +51,8 @@ fn perturbed_schedule_wrong_order(correct: List>) -> List Bool { match schedule_lens_verdict_for_ci_floor( plan: synthetic_correct_plan(), - compile_gate_fn: compile_gate_fn + compile_gate_fn: compile_gate_fn, + compile_clean_sharded: false ) { Holds => true Violation { diagnostic: _ } => false @@ -67,7 +68,7 @@ fn witness_schedule_lens_red_when_corpus_runs_first_on_synthetic() -> Bool { schedule: perturbed_schedule_wrong_order(correct: synthetic_correct_schedule()) total: cost_account_predicted_zero() } - match schedule_lens_verdict_for_ci_floor(plan: perturbed, compile_gate_fn: compile_gate_fn) { + match schedule_lens_verdict_for_ci_floor(plan: perturbed, compile_gate_fn: compile_gate_fn, compile_clean_sharded: false) { Holds => false Violation { diagnostic: _ } => true NotApplicable { diagnostic: _ } => false diff --git a/dsl/tools/dsl_compile_clean_partition.dag b/dsl/tools/dsl_compile_clean_partition.dag index 8f342431b30..45119f20966 100644 --- a/dsl/tools/dsl_compile_clean_partition.dag +++ b/dsl/tools/dsl_compile_clean_partition.dag @@ -48,6 +48,14 @@ fn compose_compile_clean_shard_verdicts( ) } +fn compile_clean_shard_verdict_from_bool(entry_path: String, ok: Bool) -> CompileCleanShardVerdict { + if ok { + ShardClean + } else { + ShardFailed { entry_path: entry_path } + } +} + fn compile_clean_shard_verdict_to_reduce(v: CompileCleanShardVerdict) -> ReduceVerdict { match v { ShardClean => Reduced { last_parts: 0 } diff --git a/dsl/tools/dsl_compile_clean_perturb_witness.dag b/dsl/tools/dsl_compile_clean_perturb_witness.dag new file mode 100644 index 00000000000..a1d33cfd750 --- /dev/null +++ b/dsl/tools/dsl_compile_clean_perturb_witness.dag @@ -0,0 +1,24 @@ +module tools.dsl_compile_clean_perturb_witness + +import tools.dsl_compile_clean_transport { + run_perturb_optional_skew_red_receipt, + run_perturb_unresolved_import_red_receipt, + run_cross_tree_import_green_receipt, + run_dsl_shell_primary_precedence_receipt +} + +func compile_clean_perturb_optional_skew_passes() -> Bool { + run_perturb_optional_skew_red_receipt() +} + +func compile_clean_perturb_unresolved_import_passes() -> Bool { + run_perturb_unresolved_import_red_receipt() +} + +func compile_clean_perturb_cross_tree_import_passes() -> Bool { + run_cross_tree_import_green_receipt() +} + +func compile_clean_perturb_dsl_shell_primary_passes() -> Bool { + run_dsl_shell_primary_precedence_receipt() +} diff --git a/dsl/tools/dsl_compile_clean_shard_transport.dag b/dsl/tools/dsl_compile_clean_shard_transport.dag index 637a541ddd3..66e1ac2fe0a 100644 --- a/dsl/tools/dsl_compile_clean_shard_transport.dag +++ b/dsl/tools/dsl_compile_clean_shard_transport.dag @@ -14,7 +14,12 @@ import tools.host_prelude { ensure_gunbc_built } import tools.dsl_compile_clean_transport { dcc_pool_index_flags, dcc_source_root_flags, - shell_serializes_dcc_common_compile_invocation + shell_serializes_dcc_common_compile_invocation, + perturb_module_source +} +import tools.dsl_compile_clean_partition { + CompileCleanShardVerdict, + compile_clean_shard_verdict_from_bool } data shard_a_exemplar_entry_path: String = "dsl/std/logic.dag" @@ -52,12 +57,66 @@ fn single_module_shard_compile_program(entry_rel_path: String) -> ShellProgram { } fn run_shard_a_exemplar_compile() -> Bool { + run_shard_compile(entry_rel_path: shard_a_exemplar_entry_path) +} + +fn run_shard_compile(entry_rel_path: String) -> Bool { let result = shell.Exec.Run( - script: serialize_bash(p: single_module_shard_compile_program(entry_rel_path: shard_a_exemplar_entry_path)) + script: serialize_bash(p: single_module_shard_compile_program(entry_rel_path: entry_rel_path)) + ) + result.success +} + +fn run_shard_compile_verdict(entry_rel_path: String) -> CompileCleanShardVerdict { + compile_clean_shard_verdict_from_bool( + entry_path: entry_rel_path, + ok: run_shard_compile(entry_rel_path: entry_rel_path) ) +} + +data broken_shard_module_source: String = perturb_module_source + +fn single_broken_module_shard_compile_program() -> ShellProgram { + let root = var_ref(name: "ROOT") + let gunbc = var_ref(name: "GUNBC") + let stage = var_ref(name: "SHARD_STAGE") + let out = var_ref(name: "OUT") + let tmp = var_ref(name: "TMP") + ShellProgram { + set_e: true, + statements: concat(ensure_gunbc_built(), [ + command(words: [lit(text: "cd"), root]), + assign(name: "TMP", value: CmdSubst { body: command(words: [lit(text: "mktemp"), lit(text: "-d")]) }), + assign(name: "SHARD_STAGE", value: CmdSubst { body: command(words: [lit(text: "mktemp"), lit(text: "-d")]) }), + assign(name: "OUT", value: CmdSubst { body: command(words: [lit(text: "mktemp"), lit(text: "-d")]) }), + Heredoc { + path: Concat { parts: [stage, lit(text: "/_broken_shard.dag")] }, + delimiter: "BROKEN_SHARD", + body: broken_shard_module_source + }, + command(words: concat( + [gunbc, lit(text: "compile"), lit(text: "--source-root"), stage], + concat(dcc_source_root_flags(), concat(dcc_pool_index_flags(), [ + lit(text: "--output-dir"), out, + lit(text: "--target"), lit(text: "dag") + ])) + )), + command(words: [lit(text: "rm"), lit(text: "-rf"), out]), + command(words: [lit(text: "rm"), lit(text: "-rf"), stage]), + command(words: [lit(text: "rm"), lit(text: "-rf"), tmp]) + ]) + } +} + +fn run_broken_staged_shard_compile() -> Bool { + let result = shell.Exec.Run(script: serialize_bash(p: single_broken_module_shard_compile_program())) result.success } +fn witness_broken_staged_shard_compile_red() -> Bool { + !run_broken_staged_shard_compile() +} + data dcc_shard_transport_invocation_substring_witness_disposition: Disposition = Scaffold { dissolves_to: RealizationDispatch, bind: DeclarationRef { diff --git a/src/v1/stage0/src/bin/claim_executor.rs b/src/v1/stage0/src/bin/claim_executor.rs index 01fb15e42c8..7f1e99bbab3 100644 --- a/src/v1/stage0/src/bin/claim_executor.rs +++ b/src/v1/stage0/src/bin/claim_executor.rs @@ -16,7 +16,8 @@ use v1_compiler::cli_run::{ DEFAULT_SLOWEST_WITNESS_ATTRIBUTION_N, }; use v1_compiler::v1_interpreter::{ - color_enabled, paint, run_in_context_with_args, sgr, ExecutionMode, InterpContext, Value, + color_enabled, eval_data_item_value, paint, run_in_context_with_args, sgr, ExecutionMode, + InterpContext, Value, }; #[derive(Clone)] @@ -35,6 +36,10 @@ enum Runnable { discovery_scope_dirs: Vec, spawn_width_cap: usize, }, + CompileCleanShardBatch { + entry_paths: Vec, + spawn_width_cap: usize, + }, } fn require_value(args: &[String], idx: usize, flag: &str) -> Result { @@ -223,6 +228,28 @@ fn runnable_from_value(value: &Value, ctx: &InterpContext) -> Result { + let entry_paths = match ctx.field(fields, "entry_paths") { + Some(v) => str_list_from_value(v, ctx)?, + None => { + return Err( + "RunnableCompileCleanShardBatch missing field `entry_paths`".to_string() + ) + } + }; + let spawn_width_cap = match ctx.field(fields, "spawn_width_cap") { + Some(Value::Int(n)) => (*n).max(0) as usize, + Some(_) | None => 0, + }; + Ok(Runnable::CompileCleanShardBatch { + entry_paths, + spawn_width_cap, + }) + } other => Err(format!( "expected a ClaimRef record or Runnable variant, got {}", ctx.format_value(other) @@ -287,6 +314,10 @@ enum BatchUnit { discovery_scope_dirs: Vec, spawn_width_cap: usize, }, + CompileCleanShardBatch { + entry_paths: Vec, + spawn_width_cap: usize, + }, } /// Partition a batch's runnables into resolve-groups, preserving first-appearance order so the @@ -349,6 +380,13 @@ fn group_batch_units(batch: &[Runnable]) -> Vec { discovery_scope_dirs: discovery_scope_dirs.clone(), spawn_width_cap: *spawn_width_cap, }), + Runnable::CompileCleanShardBatch { + entry_paths, + spawn_width_cap, + } => units.push(BatchUnit::CompileCleanShardBatch { + entry_paths: entry_paths.clone(), + spawn_width_cap: *spawn_width_cap, + }), } } units @@ -439,6 +477,15 @@ fn run_batch_unit( spawn_width, spawn_width_cap, )], + BatchUnit::CompileCleanShardBatch { + entry_paths, + spawn_width_cap, + } => vec![run_compile_clean_shard_batch_node( + &source_roots, + entry_paths, + spawn_width, + spawn_width_cap, + )], BatchUnit::SharedClaims { entry, functions, .. } => run_shared_entry_claims(&source_roots, &entry, &functions), @@ -753,6 +800,150 @@ fn emit_slowest_witness_attribution(source_roots: &[String], summary: &Discovery } } +const COMPILE_CLEAN_SHARD_TRANSPORT: &str = "dsl/tools/dsl_compile_clean_shard_transport.dag"; +const PERTURB_SHARD_BROKEN_SOURCE_DATA: &str = "broken_shard_module_source"; + +fn perturb_shard_broken_module_source(source_roots: &[String]) -> Result { + let (graph, source_indices) = resolve_entry_graph(source_roots, COMPILE_CLEAN_SHARD_TRANSPORT) + .map_err(|msg| format!("resolve {COMPILE_CLEAN_SHARD_TRANSPORT}: {msg}"))?; + let ctx = make_eval_context(&graph, source_indices, ExecutionMode::Hermetic); + let value = eval_data_item_value(&ctx, PERTURB_SHARD_BROKEN_SOURCE_DATA).map_err(|e| { + format!("eval {PERTURB_SHARD_BROKEN_SOURCE_DATA} from {COMPILE_CLEAN_SHARD_TRANSPORT}: {e}") + })?; + match value { + Some(Value::Str(s)) => Ok(s), + Some(other) => Err(format!( + "{PERTURB_SHARD_BROKEN_SOURCE_DATA} is {}, not String", + ctx.format_value(&other) + )), + None => Err(format!( + "{PERTURB_SHARD_BROKEN_SOURCE_DATA} not found in {COMPILE_CLEAN_SHARD_TRANSPORT}" + )), + } +} + +fn run_single_shard_compile(ctx: &InterpContext, entry_path: &str) -> Result { + let value = run_in_context_with_args( + ctx, + "run_shard_compile", + &[( + Some("entry_rel_path".to_string()), + Value::Str(entry_path.to_string()), + )], + false, + ) + .map_err(|e| format!("run_shard_compile({entry_path}): {e}"))?; + match value { + Value::Bool(b) => Ok(b), + other => Err(format!( + "run_shard_compile({entry_path}) returned {}, not Bool", + ctx.format_value(&other) + )), + } +} + +fn run_compile_clean_shard_batch_node( + source_roots: &[String], + entry_paths: Vec, + spawn_width: usize, + spawn_width_cap: usize, +) -> ClaimResult { + let width = spawn_width.max(1).min(spawn_width_cap.max(1)); + let shard_count = entry_paths.len(); + let label = format!("compile-clean-shard-batch[{shard_count} modules, width={width}]"); + let batch_start = Instant::now(); + + if entry_paths.is_empty() { + return ClaimResult { + function: label, + ok: false, + detail: "compile-clean-shard:empty entry_paths — fail-closed".to_string(), + wall_nanos: batch_start.elapsed().as_nanos(), + resolve_nanos: 0, + corpus_resolve_nanos: 0, + corpus_eval_nanos: 0, + corpus_witnesses: 0, + }; + } + + let mut shards: Vec> = vec![Vec::new(); width]; + for (i, path) in entry_paths.iter().enumerate() { + shards[i % width].push(path.clone()); + } + + let mut handles = Vec::new(); + for shard in shards { + if shard.is_empty() { + continue; + } + let roots = source_roots.to_vec(); + handles.push(thread::spawn(move || { + let (graph, source_indices) = + match resolve_entry_graph(&roots, COMPILE_CLEAN_SHARD_TRANSPORT) { + Ok(pair) => pair, + Err(msg) => { + return vec![format!( + "resolve failed for {COMPILE_CLEAN_SHARD_TRANSPORT}: {msg}" + )]; + } + }; + let ctx = make_eval_context(&graph, source_indices, ExecutionMode::Wet); + let mut failures = Vec::new(); + for path in shard { + match run_single_shard_compile(&ctx, &path) { + Ok(true) => {} + Ok(false) => failures.push(format!("{path}: compile returned false")), + Err(msg) => failures.push(msg), + } + } + failures + })); + } + + let mut all_failures = Vec::new(); + for handle in handles { + match handle.join() { + Ok(failures) => all_failures.extend(failures), + Err(_) => all_failures.push("compile-clean shard thread panicked".to_string()), + } + } + + let wall_nanos = batch_start.elapsed().as_nanos(); + let wall_ms = wall_nanos / 1_000_000; + eprintln!( + "[measurement] compile-clean shard batch: {shard_count} module shard(s), width={width}, wall {wall_ms}ms" + ); + + if all_failures.is_empty() { + ClaimResult { + function: label, + ok: true, + detail: String::new(), + wall_nanos, + resolve_nanos: 0, + corpus_resolve_nanos: 0, + corpus_eval_nanos: 0, + corpus_witnesses: shard_count, + } + } else { + ClaimResult { + function: label, + ok: false, + detail: format!( + "{} of {} compile-clean shard(s) failed: {}", + all_failures.len(), + shard_count, + all_failures.join("; ") + ), + wall_nanos, + resolve_nanos: 0, + corpus_resolve_nanos: 0, + corpus_eval_nanos: 0, + corpus_witnesses: shard_count, + } + } +} + #[allow(clippy::too_many_arguments)] fn run_discovery_batch_node( source_roots: Vec, @@ -847,7 +1038,7 @@ fn eval_plan( ) -> Result>, String> { let (plan_graph, plan_indices) = resolve_entry_graph(source_roots, plan_entry) .map_err(|msg| format!("resolve failed for plan {}:\n{}", plan_entry, msg))?; - let plan_ctx = make_eval_context(&plan_graph, plan_indices, ExecutionMode::Hermetic); + let plan_ctx = make_eval_context(&plan_graph, plan_indices, ExecutionMode::Wet); let plan_value = run_value(&plan_ctx, plan_function).map_err(|msg| { format!( "plan eval failed ({}::{}): {}", @@ -1516,6 +1707,43 @@ fn perturb_function_to_false(path: &Path, function: &str) -> Result<(), String> fs::write(path, out).map_err(|e| format!("write {}: {e}", path.display())) } +const PERTURB_SHARD_BROKEN_ENTRY: &str = "_perturb_compile_clean_shard_red.dag"; + +#[derive(Clone, Debug, PartialEq, Eq)] +enum PerturbPlant { + SingleClaim { entry: String, function: String }, + CompileCleanShardBrokenEntry { rel_path: String }, +} + +fn perturb_plant_target(batch: &[Runnable]) -> Result { + match batch.first() { + Some(Runnable::CompileCleanShardBatch { .. }) => { + Ok(PerturbPlant::CompileCleanShardBrokenEntry { + rel_path: PERTURB_SHARD_BROKEN_ENTRY.to_string(), + }) + } + Some(Runnable::SingleClaim { + entry, function, .. + }) if !entry.is_empty() => Ok(PerturbPlant::SingleClaim { + entry: entry.clone(), + function: function.clone(), + }), + _ => Err( + "batch 1 has no plantable gating node (SingleClaim or CompileCleanShardBatch)" + .to_string(), + ), + } +} + +fn perturb_plant_label(plant: &PerturbPlant) -> String { + match plant { + PerturbPlant::SingleClaim { function, .. } => function.clone(), + PerturbPlant::CompileCleanShardBrokenEntry { rel_path } => { + format!("compile-clean-shard-batch[{rel_path}]") + } + } +} + fn run_perturb_check( source_roots: &[String], plan_entry: &str, @@ -1536,14 +1764,10 @@ fn run_perturb_check( ); return Err(ExitCode::from(2)); } - let (gating_entry, gating_function) = match batches[0].first() { - Some(Runnable::SingleClaim { - entry, function, .. - }) if !entry.is_empty() => (entry.clone(), function.clone()), - _ => { - eprintln!( - "claim_executor: --perturb-check: batch 1 has no plantable SingleClaim gating node" - ); + let plant = match perturb_plant_target(&batches[0]) { + Ok(p) => p, + Err(msg) => { + eprintln!("claim_executor: --perturb-check: {msg}"); return Err(ExitCode::from(2)); } }; @@ -1557,11 +1781,31 @@ fn run_perturb_check( return Err(ExitCode::from(2)); } - let gating_path = remap_entry_for_temp(primary, &temp_src, &gating_entry); - if let Err(e) = perturb_function_to_false(&gating_path, &gating_function) { - let _ = fs::remove_dir_all(&tmp); - eprintln!("claim_executor: --perturb-check: plant gating->false failed: {e}"); - return Err(ExitCode::from(2)); + match &plant { + PerturbPlant::SingleClaim { entry, function } => { + let gating_path = remap_entry_for_temp(primary, &temp_src, entry); + if let Err(e) = perturb_function_to_false(&gating_path, function) { + let _ = fs::remove_dir_all(&tmp); + eprintln!("claim_executor: --perturb-check: plant gating->false failed: {e}"); + return Err(ExitCode::from(2)); + } + } + PerturbPlant::CompileCleanShardBrokenEntry { rel_path } => { + let broken_path = temp_src.join(rel_path); + let module_source = match perturb_shard_broken_module_source(source_roots) { + Ok(s) => s, + Err(e) => { + let _ = fs::remove_dir_all(&tmp); + eprintln!("claim_executor: --perturb-check: {e}"); + return Err(ExitCode::from(2)); + } + }; + if let Err(e) = fs::write(&broken_path, module_source) { + let _ = fs::remove_dir_all(&tmp); + eprintln!("claim_executor: --perturb-check: plant broken shard entry failed: {e}"); + return Err(ExitCode::from(2)); + } + } } let temp_root = temp_src.to_string_lossy().into_owned(); @@ -1610,6 +1854,21 @@ fn run_perturb_check( discovery_scope_dirs: discovery_scope_dirs.clone(), spawn_width_cap: *spawn_width_cap, }, + Runnable::CompileCleanShardBatch { + entry_paths, + spawn_width_cap, + } => match &plant { + PerturbPlant::CompileCleanShardBrokenEntry { rel_path } => { + Runnable::CompileCleanShardBatch { + entry_paths: vec![rel_path.clone()], + spawn_width_cap: *spawn_width_cap, + } + } + _ => Runnable::CompileCleanShardBatch { + entry_paths: entry_paths.clone(), + spawn_width_cap: *spawn_width_cap, + }, + }, }) .collect() }) @@ -1617,7 +1876,7 @@ fn run_perturb_check( eprintln!( "claim_executor: --perturb-check: planted batch-1 gating witness `{}` -> false; re-walking", - gating_function + perturb_plant_label(&plant) ); let outcome = run_walk(&[temp_root], &remapped, 1); let _ = fs::remove_dir_all(&tmp); @@ -1832,6 +2091,39 @@ mod tests { } } + fn compile_clean_shard_batch() -> Runnable { + Runnable::CompileCleanShardBatch { + entry_paths: vec!["dsl/std/logic.dag".to_string()], + spawn_width_cap: 4, + } + } + + #[test] + fn perturb_plant_target_prefers_shard_batch_when_first() { + let batch = vec![compile_clean_shard_batch(), single("gate.dag", "g1")]; + assert_eq!( + perturb_plant_target(&batch), + Ok(PerturbPlant::CompileCleanShardBrokenEntry { + rel_path: PERTURB_SHARD_BROKEN_ENTRY.to_string(), + }) + ); + } + + #[test] + fn perturb_plant_target_uses_single_claim_for_monolith_batch() { + let batch = vec![single( + "dsl/tools/dsl_compile_clean_gate.dag", + "dsl_compile_clean_gate_passes", + )]; + assert_eq!( + perturb_plant_target(&batch), + Ok(PerturbPlant::SingleClaim { + entry: "dsl/tools/dsl_compile_clean_gate.dag".to_string(), + function: "dsl_compile_clean_gate_passes".to_string(), + }) + ); + } + /// Flatten the grouped units back to the (entry, function) claims they will execute, in the /// order each unit runs them. `Discovery` contributes no SingleClaim pairs; an empty-entry /// sentinel surfaces as `("", function)`. This is the verdict-preservation oracle: grouping is @@ -1851,6 +2143,7 @@ mod tests { out.push((String::new(), function.clone())); } BatchUnit::Discovery { .. } => {} + BatchUnit::CompileCleanShardBatch { .. } => {} } } out @@ -2124,4 +2417,35 @@ mod tests { "second call must cache-hit — resolve_entry_graph must NOT fire again" ); } + + #[test] + fn perturb_shard_broken_module_source_reads_dag_authority() { + let root = workspace_root(); + let source_roots = vec![ + root.join("dsl").to_string_lossy().into_owned(), + root.join("src/v2").to_string_lossy().into_owned(), + ]; + std::env::set_current_dir(&root).expect("chdir to workspace root"); + let source = perturb_shard_broken_module_source(&source_roots) + .expect("read broken_shard_module_source from shard transport"); + assert!( + source.contains("perturb_compile_clean_red"), + "expected authority module body" + ); + assert!( + source.contains("Some { value: s }"), + "expected broken Optional skew from perturb_module_source authority" + ); + } + + #[test] + fn compile_clean_shard_batch_empty_entry_paths_fails_closed() { + let result = run_compile_clean_shard_batch_node(&[], vec![], 1, 4); + assert!(!result.ok, "empty shard roster must fail closed"); + assert!( + result.detail.contains("empty entry_paths"), + "detail must name empty roster: {}", + result.detail + ); + } } diff --git a/src/v1/stage0/src/std_http_path.rs b/src/v1/stage0/src/std_http_path.rs index 4a36b0a2aa3..4e953160365 100644 --- a/src/v1/stage0/src/std_http_path.rs +++ b/src/v1/stage0/src/std_http_path.rs @@ -62,3 +62,41 @@ pub fn last_path_param(template: Rc) -> Option { } } } + +#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)] +pub struct PathParamBinding { + pub name: String, + pub value: String, +} + +pub fn path_param_value(params: Rc>>, name: String) -> String { + params.iter().cloned().fold( + "".to_string(), + |acc: String, binding: Rc| { + if (acc.clone() != "".to_string()) { + acc.clone() + } else { + if (binding.name.clone() == name.clone()) { + binding.value.clone() + } else { + acc.clone() + } + } + }, + ) +} + +pub fn render_path_template( + template: Rc, + params: Rc>>, +) -> String { + template.tokens.clone().iter().cloned().fold( + "".to_string(), + |acc: String, tok: Rc| match (*tok.clone()).clone() { + UrlPathToken::LiteralToken { text: t, .. } => v1_rt::concat(acc.clone(), t.clone()), + UrlPathToken::ParamToken { name: n, .. } => { + v1_rt::concat(acc.clone(), path_param_value(params.clone(), n.clone())) + } + }, + ) +} diff --git a/src/v1/stage0/src/std_measure.rs b/src/v1/stage0/src/std_measure.rs index 86ab6826f9d..e613cd4c07c 100644 --- a/src/v1/stage0/src/std_measure.rs +++ b/src/v1/stage0/src/std_measure.rs @@ -195,8 +195,21 @@ pub type ByteSize = Rc>; pub type Kibibyte = Rc>; +pub type Mebibyte = Rc>; + pub type Gibibyte = Rc>; +pub fn mebibyte(count: Nat) -> Mebibyte { + Rc::new(Measure { + count: count, + _phantom: std::marker::PhantomData, + }) +} + +pub fn mebibyte_count(m: Mebibyte) -> Nat { + measure_count(m) +} + pub fn gibibyte(count: Nat) -> Gibibyte { Rc::new(Measure { count: count, @@ -222,6 +235,8 @@ pub type Hertz = Rc>; pub type HardwareThreadCount = Rc>; +pub type Millicore = Rc>; + pub type Watt = Rc>; pub type Celsius = Rc>; @@ -320,6 +335,17 @@ pub fn hardware_thread_count_value(t: HardwareThreadCount) -> Nat { measure_count(t) } +pub fn millicore(count: Nat) -> Millicore { + Rc::new(Measure { + count: count, + _phantom: std::marker::PhantomData, + }) +} + +pub fn millicore_count(m: Millicore) -> Nat { + measure_count(m) +} + pub type Bandwidth = Rc>; pub fn bandwidth(count: Nat) -> Bandwidth { diff --git a/src/v1/stage0/src/std_realization_schedule.rs b/src/v1/stage0/src/std_realization_schedule.rs index bd2ca24e794..b7c12e9efc9 100644 --- a/src/v1/stage0/src/std_realization_schedule.rs +++ b/src/v1/stage0/src/std_realization_schedule.rs @@ -209,6 +209,7 @@ pub fn runnable_profile(r: Rc) -> Rc { match (*r).clone() { Runnable::RunnableSingleClaim { profile: p, .. } => p.clone(), Runnable::RunnableDiscoveryBatch { profile: p, .. } => p.clone(), + Runnable::RunnableCompileCleanShardBatch { profile: p, .. } => p.clone(), Runnable::RunnableKernelWorkload { fused_op_count: _, .. } => runnable_resource_profile_negligible(), @@ -218,6 +219,9 @@ pub fn runnable_profile(r: Rc) -> Rc { pub fn runnable_forbids_corpus_co_residence(r: Rc) -> bool { match (*r).clone() { Runnable::RunnableDiscoveryBatch { .. } => false, + Runnable::RunnableCompileCleanShardBatch { profile: p, .. } => { + runnable_excludes_corpus_co_residence(p.clone()) + } Runnable::RunnableSingleClaim { profile: p, .. } => { runnable_excludes_corpus_co_residence(p.clone()) } @@ -245,11 +249,25 @@ pub enum Runnable { spawn_width_cap: i64, profile: Rc, }, + RunnableCompileCleanShardBatch { + entry_paths: Rc>, + spawn_width_cap: i64, + profile: Rc, + }, RunnableKernelWorkload { fused_op_count: i64, }, } +pub fn compile_clean_shard_batch_label() -> String { + thread_local! { + static CACHED: String = { + "__compile_clean_shard_batch__".to_string() + }; + } + CACHED.with(|c: &String| c.clone()) +} + pub fn runnable_kernel_workload_dissolution_trigger() -> String { thread_local! { static CACHED: String = { @@ -274,12 +292,41 @@ pub fn runnable_step_label(r: Rc) -> String { match (*r).clone() { Runnable::RunnableSingleClaim { function: f, .. } => f.clone(), Runnable::RunnableDiscoveryBatch { .. } => "__discovery_corpus__".to_string(), + Runnable::RunnableCompileCleanShardBatch { .. } => compile_clean_shard_batch_label(), Runnable::RunnableKernelWorkload { fused_op_count: _, .. } => "__kernel_workload__".to_string(), } } +pub fn runnable_compile_clean_shard_entry_paths(r: Rc) -> Rc> { + match (*r).clone() { + Runnable::RunnableCompileCleanShardBatch { + entry_paths: paths, .. + } => paths.clone(), + _ => Rc::new(vec![]), + } +} + +pub fn runnable_compile_clean_shard_count(r: Rc) -> i64 { + (runnable_compile_clean_shard_entry_paths(r).len() as i64) +} + +pub fn schedule_batch_compile_clean_shard_entry_paths( + batch: Rc>>, +) -> Rc> { + batch + .iter() + .cloned() + .fold(Rc::new(vec![]), |acc: Rc>, r: Rc| { + v1_rt::concat(acc, runnable_compile_clean_shard_entry_paths(r.clone())) + }) +} + +pub fn schedule_batch_compile_clean_shard_count(batch: Rc>>) -> i64 { + (schedule_batch_compile_clean_shard_entry_paths(batch).len() as i64) +} + pub fn schedule_batch_contains_label(batch: Rc>>, target: String) -> bool { batch .iter() @@ -294,21 +341,11 @@ pub fn schedule_batch_contains_label(batch: Rc>>, target: Strin pub enum ScheduleLensViolation { EmptySchedule, CompileGateNotFirst { expected: String }, + CompileGateMonolithPresent { expected: String }, + CompileCleanShardsInsufficient { minimum: i64 }, CorpusBeforeCompile, SingleBatchOnly, } -impl ScheduleLensViolation { - pub fn expected(&self) -> String { - match self { - ScheduleLensViolation::EmptySchedule => panic!("no expected on unit variant"), - ScheduleLensViolation::CompileGateNotFirst { - expected: __val, .. - } => __val.clone(), - ScheduleLensViolation::CorpusBeforeCompile => panic!("no expected on unit variant"), - ScheduleLensViolation::SingleBatchOnly => panic!("no expected on unit variant"), - } - } -} pub fn schedule_lens_module() -> String { thread_local! { @@ -338,6 +375,19 @@ pub fn schedule_lens_violation_diagnostic( reason: ("schedule_lens_compile_gate_not_first:".to_string() + &expected.clone()), at: at, }), + ScheduleLensViolation::CompileGateMonolithPresent { + expected: expected, .. + } => Rc::new(LensVerdictDiagnostic { + reason: ("schedule_lens_compile_gate_monolith_present:".to_string() + + &expected.clone()), + at: at, + }), + ScheduleLensViolation::CompileCleanShardsInsufficient { minimum: _, .. } => { + Rc::new(LensVerdictDiagnostic { + reason: "schedule_lens_compile_clean_shards_insufficient".to_string(), + at: at, + }) + } ScheduleLensViolation::CorpusBeforeCompile => Rc::new(LensVerdictDiagnostic { reason: "schedule_lens_corpus_before_compile".to_string(), at: at, @@ -353,6 +403,7 @@ pub fn schedule_lens_violation_diagnostic( pub fn schedule_lens_verdict_for_ci_floor( plan: Rc>, compile_gate_fn: String, + compile_clean_sharded: bool, ) -> Rc { if (((*plan.schedule).clone().len() as i64) == 0) { Rc::new(LensVerdict::Violation { @@ -370,37 +421,76 @@ pub fn schedule_lens_verdict_for_ci_floor( ), }) } else { - { - let batch0 = (*plan.schedule).clone().first().cloned(); - if ((batch0.clone().expect("fail-closed: Optional receiver for method count (empty Optional at runtime)").len() as i64) != 1) { - Rc::new(LensVerdict::Violation { + if !compile_clean_sharded { + { + let batch0 = (*plan.schedule).clone().first().cloned(); + if ((batch0.clone().expect("fail-closed: Optional receiver for method count (empty Optional at runtime)").len() as i64) != 1) { + Rc::new(LensVerdict::Violation { diagnostic: schedule_lens_violation_diagnostic(Rc::new(ScheduleLensViolation::CompileGateNotFirst { expected: compile_gate_fn.clone(), }), compile_gate_fn.clone()), }) - } else { - if !schedule_batch_contains_label(batch0.clone().expect("fail-closed: an optional value flowed into non-optional parameter 0 of schedule_batch_contains_label (empty Optional at runtime)"), compile_gate_fn.clone()) { - Rc::new(LensVerdict::Violation { + } else { + if !schedule_batch_contains_label(batch0.clone().expect("fail-closed: an optional value flowed into non-optional parameter 0 of schedule_batch_contains_label (empty Optional at runtime)"), compile_gate_fn.clone()) { + Rc::new(LensVerdict::Violation { diagnostic: schedule_lens_violation_diagnostic(Rc::new(ScheduleLensViolation::CompileGateNotFirst { expected: compile_gate_fn.clone(), }), compile_gate_fn.clone()), +}) + } else { + if schedule_batch_contains_label(batch0.clone().expect("fail-closed: an optional value flowed into non-optional parameter 0 of schedule_batch_contains_label (empty Optional at runtime)"), "__discovery_corpus__".to_string()) { + Rc::new(LensVerdict::Violation { + diagnostic: schedule_lens_violation_diagnostic(Rc::new(ScheduleLensViolation::CorpusBeforeCompile), compile_gate_fn.clone()), +}) + } else { + { + let batch1 = (*plan.schedule).clone().get(1 as usize).cloned(); +if ((batch1.expect("fail-closed: Optional receiver for method count (empty Optional at runtime)").len() as i64) < 2) { + Rc::new(LensVerdict::Violation { + diagnostic: schedule_lens_violation_diagnostic(Rc::new(ScheduleLensViolation::SingleBatchOnly), compile_gate_fn.clone()), +}) + } else { + Rc::new(LensVerdict::Holds) + } +} + } + } + } + } + } else { + { + let batch0 = (*plan.schedule).clone().first().cloned(); + let shard_count = schedule_batch_compile_clean_shard_count(batch0.clone().expect("fail-closed: an optional value flowed into non-optional parameter 0 of schedule_batch_compile_clean_shard_count (empty Optional at runtime)")); + if schedule_batch_contains_label(batch0.clone().expect("fail-closed: an optional value flowed into non-optional parameter 0 of schedule_batch_contains_label (empty Optional at runtime)"), compile_gate_fn.clone()) { + Rc::new(LensVerdict::Violation { + diagnostic: schedule_lens_violation_diagnostic(Rc::new(ScheduleLensViolation::CompileGateMonolithPresent { + expected: compile_gate_fn.clone(), +}), compile_gate_fn.clone()), }) } else { - if schedule_batch_contains_label(batch0.clone().expect("fail-closed: an optional value flowed into non-optional parameter 0 of schedule_batch_contains_label (empty Optional at runtime)"), "__discovery_corpus__".to_string()) { + if (shard_count < 2) { Rc::new(LensVerdict::Violation { - diagnostic: schedule_lens_violation_diagnostic(Rc::new(ScheduleLensViolation::CorpusBeforeCompile), compile_gate_fn.clone()), + diagnostic: schedule_lens_violation_diagnostic(Rc::new(ScheduleLensViolation::CompileCleanShardsInsufficient { + minimum: 2, +}), compile_gate_fn.clone()), }) } else { - { - let batch1 = (*plan.schedule).clone().get(1 as usize).cloned(); + if schedule_batch_contains_label(batch0.clone().expect("fail-closed: an optional value flowed into non-optional parameter 0 of schedule_batch_contains_label (empty Optional at runtime)"), "__discovery_corpus__".to_string()) { + Rc::new(LensVerdict::Violation { + diagnostic: schedule_lens_violation_diagnostic(Rc::new(ScheduleLensViolation::CorpusBeforeCompile), compile_gate_fn.clone()), +}) + } else { + { + let batch1 = (*plan.schedule).clone().get(1 as usize).cloned(); if ((batch1.expect("fail-closed: Optional receiver for method count (empty Optional at runtime)").len() as i64) < 2) { - Rc::new(LensVerdict::Violation { + Rc::new(LensVerdict::Violation { diagnostic: schedule_lens_violation_diagnostic(Rc::new(ScheduleLensViolation::SingleBatchOnly), compile_gate_fn.clone()), }) - } else { - Rc::new(LensVerdict::Holds) - } + } else { + Rc::new(LensVerdict::Holds) + } } + } } } } @@ -493,6 +583,7 @@ pub fn runnable_eq(left: Rc, right: Rc) -> bool { && runnable_resource_profile_eq(lp.clone(), rp.clone())) } Runnable::RunnableDiscoveryBatch { .. } => false, + Runnable::RunnableCompileCleanShardBatch { .. } => false, Runnable::RunnableKernelWorkload { fused_op_count: _, .. } => false, @@ -529,6 +620,28 @@ pub fn runnable_eq(left: Rc, right: Rc) -> bool { && (lwc.clone() == rwc.clone())) && runnable_resource_profile_eq(lp.clone(), rp.clone())) } + Runnable::RunnableCompileCleanShardBatch { .. } => false, + Runnable::RunnableKernelWorkload { + fused_op_count: _, .. + } => false, + }, + Runnable::RunnableCompileCleanShardBatch { + entry_paths: lpaths, + spawn_width_cap: lwc, + profile: lp, + .. + } => match (*right).clone() { + Runnable::RunnableSingleClaim { .. } => false, + Runnable::RunnableDiscoveryBatch { .. } => false, + Runnable::RunnableCompileCleanShardBatch { + entry_paths: rpaths, + spawn_width_cap: rwc, + profile: rp, + .. + } => { + ((string_list_eq(lpaths.clone(), rpaths.clone()) && (lwc.clone() == rwc.clone())) + && runnable_resource_profile_eq(lp.clone(), rp.clone())) + } Runnable::RunnableKernelWorkload { fused_op_count: _, .. } => false, @@ -539,6 +652,7 @@ pub fn runnable_eq(left: Rc, right: Rc) -> bool { } => match (*right).clone() { Runnable::RunnableSingleClaim { .. } => false, Runnable::RunnableDiscoveryBatch { .. } => false, + Runnable::RunnableCompileCleanShardBatch { .. } => false, Runnable::RunnableKernelWorkload { fused_op_count: rcount, .. diff --git a/src/v2/test/claim/ci_floor_plan_witness_test.dag b/src/v2/test/claim/ci_floor_plan_witness_test.dag index 16d6e1bb434..a11a776880a 100644 --- a/src/v2/test/claim/ci_floor_plan_witness_test.dag +++ b/src/v2/test/claim/ci_floor_plan_witness_test.dag @@ -2,12 +2,17 @@ module v2.test.claim.ci_floor_plan_witness import std.realization { Runnable, RunnableSingleClaim, RunnableDiscoveryBatch, + RunnableCompileCleanShardBatch, + compile_clean_shard_batch_label, } import v2.workflow.ci_floor_plan { gunbc_ci_floor_batches, corpus_runnable, gunbc_ci_floor_schedule_without_serialization, gunbc_ci_floor_plan_serializes_heavy_resolves, gunbc_ci_floor_plan_co_runs_heavy_without_serialization, + gunbc_ci_floor_compile_clean_shard_count, + witness_floor_batch_one_uses_compile_clean_shards, + witness_floor_batch_one_has_no_compile_gate_monolith, gate_runnable, witness_floor_corpus_exclusion_edges_exist, witness_floor_compile_gate_honest_heavy_profile, @@ -32,6 +37,7 @@ fn runnable_function(r: Runnable) -> String { match r { RunnableSingleClaim { entry: _, function: f, profile: _ } => f RunnableDiscoveryBatch { source_roots: _, scan_dirs: _, explicit_entries: _, skip_unaffected_node_frontier: _, exclude_substrings: _, discovery_scope_dirs: _, spawn_width_cap: _, profile: _ } => corpus_sentinel + RunnableCompileCleanShardBatch { entry_paths: _, spawn_width_cap: _, profile: _ } => compile_clean_shard_batch_label RunnableKernelWorkload { fused_op_count: _ } => "__kernel_workload__" } } @@ -45,9 +51,9 @@ fn list_contains_str(xs: List, target: String) -> Bool { } fn witness_compile_root_runs_first() -> Bool { - let batches = gunbc_ci_floor_batches() - (batches.first().length() == 1) && - list_contains_str(xs: batch_functions(b: batches.first()), target: "dsl_compile_clean_gate_passes") + witness_floor_batch_one_uses_compile_clean_shards() + && witness_floor_batch_one_has_no_compile_gate_monolith() + && (gunbc_ci_floor_compile_clean_shard_count() >= 2) } fn witness_corpus_not_before_compile() -> Bool { diff --git a/src/v2/test/claim/realization_schedule_witness.dag b/src/v2/test/claim/realization_schedule_witness.dag index c65450c10a3..bc9840bb57f 100644 --- a/src/v2/test/claim/realization_schedule_witness.dag +++ b/src/v2/test/claim/realization_schedule_witness.dag @@ -18,6 +18,7 @@ import v2.workflow.ci_floor_plan { ci_floor_target_digest, ci_floor_compile_gate_fn, } +import v2.workflow.compile_clean_shard_roster { compile_clean_sharding_enabled } fn witness_batches_are_plan_schedule() -> Bool { schedule_generates_identical_schedule( @@ -36,7 +37,8 @@ fn witness_batches_identical_to_inner_schedule() -> Bool { fn witness_schedule_lens_holds_on_floor_plan() -> Bool { match schedule_lens_verdict_for_ci_floor( plan: gunbc_ci_floor_realization_plan(), - compile_gate_fn: ci_floor_compile_gate_fn + compile_gate_fn: ci_floor_compile_gate_fn, + compile_clean_sharded: compile_clean_sharding_enabled() ) { Holds => true Violation { diagnostic: _ } => false @@ -60,7 +62,7 @@ fn witness_schedule_lens_red_when_corpus_runs_first_on_live_floor() -> Bool { schedule: perturbed_schedule_wrong_order(inner: inner) total: cost_account_predicted_zero() } - match schedule_lens_verdict_for_ci_floor(plan: perturbed, compile_gate_fn: ci_floor_compile_gate_fn) { + match schedule_lens_verdict_for_ci_floor(plan: perturbed, compile_gate_fn: ci_floor_compile_gate_fn, compile_clean_sharded: compile_clean_sharding_enabled()) { Holds => false Violation { diagnostic: _ } => true NotApplicable { diagnostic: _ } => false diff --git a/src/v2/workflow/ci_floor_plan.dag b/src/v2/workflow/ci_floor_plan.dag index 4c3e9df4261..1ce4229a0f7 100644 --- a/src/v2/workflow/ci_floor_plan.dag +++ b/src/v2/workflow/ci_floor_plan.dag @@ -10,7 +10,7 @@ import gunbc.ci_spec { import v2.std.algebra { length, list_snoc_item, skip, zip_map } import v2.std.collection { List } import v2.std.dependency { DataDependsOn, ResourceDependsOn, DependencyView } -import v2.std.node { Atom, Node, TypeNode } +import v2.std.node { Atom, Node, TypeNode, OccurrenceId, node_occurrence_minted } import v2.std.text { String } import std.types { ContentHash, Int } import std.measure { HardwareThreadCount, hardware_thread_count, hardware_thread_count_value, byte_size, ByteSize, byte_size_count } @@ -39,6 +39,10 @@ import std.realization_schedule { Runnable, RunnableSingleClaim, RunnableDiscoveryBatch, + RunnableCompileCleanShardBatch, + compile_clean_shard_batch_label, + schedule_batch_contains_label, + schedule_batch_compile_clean_shard_entry_paths, RunnableResourceProfile, runnable_resource_profile, runnable_resource_profile_negligible, @@ -51,6 +55,15 @@ import std.realization_schedule { cost_account_predicted_zero, schedule_lens_verdict_for_ci_floor } +import v2.workflow.compile_clean_shard_roster { + compile_clean_shard_batch_runnable, + compile_clean_shard_batch_profile, + compile_clean_shard_entry_paths, + compile_clean_shard_roster_count, + compile_clean_sharding_enabled, + witness_shard_roster_coverage, + witness_dropped_shard_breaks_coverage +} import v2.workflow.executor { ExecutorBatch, ExecutorJob, @@ -99,12 +112,40 @@ fn gate_node(g: Gate) -> Node { } } +data ci_floor_compile_perturb_entry: String = "dsl/tools/dsl_compile_clean_perturb_witness.dag" + +data floor_compile_clean_shard_barrier_node: Node = Node { + kind: TypeNode { connective: Atom { identity: ^gunbc_floor_compile_clean_shard } }, + children: [], + occurrence_id: node_occurrence_minted(id: OccurrenceId { value: 0 }) +} + +fn compile_clean_shard_barrier_node_count() -> Int { + if compile_clean_sharding_enabled() { 1 } else { 0 } +} + fn floor_compile_root_node() -> Node { - gate_node(g: DslCompileCleanGate) + if compile_clean_sharding_enabled() { + floor_compile_clean_shard_barrier_node + } else { + gate_node(g: DslCompileCleanGate) + } +} + +fn floor_is_compile_clean_shard_node(node: Node) -> Bool { + compile_clean_sharding_enabled() + && (node.kind == floor_compile_clean_shard_barrier_node.kind) +} + +fn floor_gate_enrolled_in_plan(g: Gate) -> Bool { + match g { + DslCompileCleanGate => !compile_clean_sharding_enabled() + _ => true + } } fn floor_resource_anchor_node(node: Node) -> Bool { - (node == floor_compile_root_node()) || (node == floor_corpus_node) || (node == execution_corpus_node) + floor_is_compile_clean_shard_node(node: node) || (node == floor_corpus_node) || (node == execution_corpus_node) } fn gate_runnable_profile(g: Gate) -> RunnableResourceProfile { @@ -242,9 +283,22 @@ fn execution_corpus_runnable() -> Runnable { } fn floor_nodes(spec: CiSpec) -> List { - fold(spec.gates, init: [floor_corpus_node, execution_corpus_node, floor_grounding_whole_tree_node], f: fn(acc, g) { - list_snoc_item(xs: acc, item: gate_node(g: g)) - }) + if compile_clean_sharding_enabled() { + let gate_nodes = fold(spec.gates, init: [], f: fn(acc, g) { + match g { + DslCompileCleanGate => acc + _ => list_snoc_item(xs: acc, item: gate_node(g: g)) + } + }) + concat( + [floor_corpus_node, execution_corpus_node, floor_grounding_whole_tree_node], + concat([floor_compile_clean_shard_barrier_node], gate_nodes) + ) + } else { + fold(spec.gates, init: [floor_corpus_node, execution_corpus_node, floor_grounding_whole_tree_node], f: fn(acc, g) { + list_snoc_item(xs: acc, item: gate_node(g: g)) + }) + } } fn floor_depends_on_compile(dependent: Node) -> DependencyView { @@ -272,32 +326,40 @@ fn floor_corpus_exclusion_resource_edges(spec: CiSpec) -> List { floor_resource_depends_on(source: execution_corpus_node, dependent: floor_grounding_whole_tree_node) ] fold(spec.gates, init: after_floor_corpus, f: fn(acc, g) { - let node = gate_node(g: g) - if floor_resource_anchor_node(node: node) { + if !floor_gate_enrolled_in_plan(g: g) { acc - } else if runnable_excludes_corpus_co_residence(profile: gate_runnable_profile(g: g)) { - list_snoc_item( - xs: list_snoc_item( - xs: acc, - item: floor_resource_depends_on(source: floor_corpus_node, dependent: node) - ), - item: floor_resource_depends_on(source: execution_corpus_node, dependent: node) - ) } else { - acc + let node = gate_node(g: g) + if floor_resource_anchor_node(node: node) { + acc + } else if runnable_excludes_corpus_co_residence(profile: gate_runnable_profile(g: g)) { + list_snoc_item( + xs: list_snoc_item( + xs: acc, + item: floor_resource_depends_on(source: floor_corpus_node, dependent: node) + ), + item: floor_resource_depends_on(source: execution_corpus_node, dependent: node) + ) + } else { + acc + } } }) } fn floor_heavy_resolve_chain_resource_edges(spec: CiSpec) -> List { let heavy_nodes = fold(spec.gates, init: [], f: fn(acc, g) { - let node = gate_node(g: g) - if floor_resource_anchor_node(node: node) { + if !floor_gate_enrolled_in_plan(g: g) { acc - } else if runnable_heavy_whole_tree_resolve(profile: gate_runnable_profile(g: g)) { - list_snoc_item(xs: acc, item: node) } else { - acc + let node = gate_node(g: g) + if floor_resource_anchor_node(node: node) { + acc + } else if runnable_heavy_whole_tree_resolve(profile: gate_runnable_profile(g: g)) { + list_snoc_item(xs: acc, item: node) + } else { + acc + } } }) let heavy_nodes = list_snoc_item(xs: heavy_nodes, item: floor_grounding_whole_tree_node) @@ -372,6 +434,35 @@ fn floor_dependencies(spec: CiSpec) -> List { floor_dependencies_with(spec: spec, serialize_heavy: true) } +fn compile_clean_perturb_runnables() -> List { + let entry = ci_floor_compile_perturb_entry + let profile = runnable_resource_profile_negligible() + [ + RunnableSingleClaim { entry: entry, function: "compile_clean_perturb_optional_skew_passes", profile: profile }, + RunnableSingleClaim { entry: entry, function: "compile_clean_perturb_unresolved_import_passes", profile: profile }, + RunnableSingleClaim { entry: entry, function: "compile_clean_perturb_cross_tree_import_passes", profile: profile }, + RunnableSingleClaim { entry: entry, function: "compile_clean_perturb_dsl_shell_primary_passes", profile: profile } + ] +} + +fn compile_clean_batch_runnables() -> List { + concat( + [ + compile_clean_shard_batch_runnable( + spawn_width_cap: hardware_thread_count_value(t: gunbc_ci_floor_spawn_width_from_tree()) + ) + ], + compile_clean_perturb_runnables() + ) +} + +fn batch_jobs_are_all_compile_shards(jobs: List) -> Bool { + (length(xs: jobs) == compile_clean_shard_barrier_node_count()) + && fold(jobs, init: true, f: fn(ok, job) { + ok && floor_is_compile_clean_shard_node(node: job.node) + }) +} + fn floor_unrunnable_sentinel() -> Runnable { RunnableSingleClaim { entry: "", @@ -387,6 +478,10 @@ fn runnable_for_node(node: Node, spec: CiSpec) -> Runnable { execution_corpus_runnable() } else if node == floor_grounding_whole_tree_node { grounding_whole_tree_runnable() + } else if floor_is_compile_clean_shard_node(node: node) { + compile_clean_shard_batch_runnable( + spawn_width_cap: hardware_thread_count_value(t: gunbc_ci_floor_spawn_width_from_tree()) + ) } else { fold(spec.gates, init: floor_unrunnable_sentinel(), f: fn(acc, g) { if node == gate_node(g: g) { gate_runnable(g: g) } else { acc } @@ -395,9 +490,14 @@ fn runnable_for_node(node: Node, spec: CiSpec) -> Runnable { } fn runnables_for_batch(batch: ExecutorBatch, spec: CiSpec) -> List { - fold(batch.jobs, init: [], f: fn(acc, job) { - list_snoc_item(xs: acc, item: runnable_for_node(node: job.node, spec: spec)) - }) + if compile_clean_sharding_enabled() + && batch_jobs_are_all_compile_shards(jobs: batch.jobs) { + compile_clean_batch_runnables() + } else { + fold(batch.jobs, init: [], f: fn(acc, job) { + list_snoc_item(xs: acc, item: runnable_for_node(node: job.node, spec: spec)) + }) + } } fn floor_schedule_for_dependencies(dependencies: List) -> List> { @@ -441,7 +541,8 @@ fn gunbc_ci_floor_batches() -> List> { fn gunbc_ci_floor_schedule_lens_holds() -> Bool { match schedule_lens_verdict_for_ci_floor( plan: gunbc_ci_floor_realization_plan(), - compile_gate_fn: ci_floor_compile_gate_fn + compile_gate_fn: ci_floor_compile_gate_fn, + compile_clean_sharded: compile_clean_sharding_enabled() ) { Holds => true Violation { diagnostic: _ } => false @@ -535,6 +636,7 @@ fn floor_batch_heavy_gate_resolve_count(batch: List) -> Int { fold(batch, init: 0, f: fn(acc, r) { match r { RunnableDiscoveryBatch { source_roots: _, scan_dirs: _, explicit_entries: _, skip_unaffected_node_frontier: _, exclude_substrings: _, discovery_scope_dirs: _, spawn_width_cap: _, profile: _ } => acc + RunnableCompileCleanShardBatch { entry_paths: _, spawn_width_cap: _, profile: _ } => acc RunnableKernelWorkload { fused_op_count: _ } => acc RunnableSingleClaim { entry: _, function: f, profile: p } => if f == ci_floor_compile_gate_fn { @@ -548,6 +650,38 @@ fn floor_batch_heavy_gate_resolve_count(batch: List) -> Int { }) } +fn floor_batch_zero_compile_clean_shard_entry_paths() -> List { + schedule_batch_compile_clean_shard_entry_paths(batch: gunbc_ci_floor_batches().first()) +} + +fn gunbc_ci_floor_compile_clean_shard_count() -> Int { + length(xs: floor_batch_zero_compile_clean_shard_entry_paths()) +} + +fn witness_floor_batch_one_uses_compile_clean_shards() -> Bool { + let batch0 = gunbc_ci_floor_batches().first() + schedule_batch_contains_label(batch: batch0, target: compile_clean_shard_batch_label) +} + +fn witness_floor_batch_one_has_no_compile_gate_monolith() -> Bool { + let batch0 = gunbc_ci_floor_batches().first() + !schedule_batch_contains_label(batch: batch0, target: ci_floor_compile_gate_fn) +} + +fn witness_floor_shard_roster_coverage_in_plan() -> Bool { + witness_shard_roster_coverage( + full: compile_clean_shard_entry_paths(), + plan_paths: floor_batch_zero_compile_clean_shard_entry_paths() + ) +} + +fn witness_floor_dropped_shard_red_control() -> Bool { + witness_dropped_shard_breaks_coverage( + full: compile_clean_shard_entry_paths(), + plan_paths: floor_batch_zero_compile_clean_shard_entry_paths() + ) +} + fn floor_plan_at_most_one_heavy_gate_resolve_per_batch(batches: List>) -> Bool { fold(batches, init: true, f: fn(ok, batch) { ok && (floor_batch_heavy_gate_resolve_count(batch: batch) <= 1) @@ -569,13 +703,22 @@ fn witness_floor_corpus_exclusion_edges_exist() -> Bool { } fn witness_floor_compile_gate_honest_heavy_profile() -> Bool { - runnable_heavy_whole_tree_resolve(profile: gate_runnable_profile(g: DslCompileCleanGate)) - && runnable_excludes_corpus_co_residence(profile: gate_runnable_profile(g: DslCompileCleanGate)) + if compile_clean_sharding_enabled() { + !runnable_heavy_whole_tree_resolve(profile: compile_clean_shard_batch_profile()) + && runnable_excludes_corpus_co_residence(profile: compile_clean_shard_batch_profile()) + } else { + runnable_heavy_whole_tree_resolve(profile: gate_runnable_profile(g: DslCompileCleanGate)) + && runnable_excludes_corpus_co_residence(profile: gate_runnable_profile(g: DslCompileCleanGate)) + } } fn witness_floor_no_corpus_resource_edge_to_compile_anchor() -> Bool { fold(floor_corpus_exclusion_resource_edges(spec: gunbc_ci_spec), init: true, f: fn(ok, edge) { - ok && (edge.dependent != floor_compile_root_node()) + if compile_clean_sharding_enabled() { + ok && !floor_is_compile_clean_shard_node(node: edge.dependent) + } else { + ok && (edge.dependent != floor_compile_root_node()) + } }) } diff --git a/src/v2/workflow/compile_clean_shard_roster.dag b/src/v2/workflow/compile_clean_shard_roster.dag new file mode 100644 index 00000000000..c535453ffdd --- /dev/null +++ b/src/v2/workflow/compile_clean_shard_roster.dag @@ -0,0 +1,156 @@ +module v2.workflow.compile_clean_shard_roster + +import extdeps.shell +import std.disposition { Disposition, Scaffold, SingleAuthority } +import std.decl_ref { DeclarationRef, WholeDeclaration } +import gunbc.ci_layer_roots { witness_layer_roots } +import tools.dsl_compile_clean_partition { + CompileCleanModuleShard, + compile_clean_partition_boundary, + compose_compile_clean_shard_verdicts, + compose_compile_clean_shard_verdicts_as_reduce, + compile_clean_shard_verdict_to_reduce, + ShardClean, + ShardFailed +} +import v2.lens.module_graph { module_declaration_facts_live, ModuleDeclarationFact } +import std.measure { HardwareThreadCount, hardware_thread_count_value } +import std.realization_schedule { + Runnable, + RunnableCompileCleanShardBatch, + RunnableResourceProfile, + runnable_resource_profile, + runnable_memory_substantial +} +import gunbc.ci_floor_measurement { gunbc_ci_floor_per_shard_peak_rss_bytes } +import std.reducible { verdict_eq } +import std.types { String, Bool, Int, List, list_length, NonEmptyStr } + +data compile_clean_whole_tree_rollback_env: NonEmptyStr = "GUNBC_CI_COMPILE_CLEAN_WHOLE_TREE" + +data compile_clean_whole_tree_rollback_dissolution_trigger: String = "dissolve-on: GUNBC_CI_COMPILE_CLEAN_WHOLE_TREE rollback removed after sharded compile-clean floor passes clean on main for N consecutive runs without rollback; delete monolith branch from floor_nodes, runnables_for_batch, and schedule_lens." + +data compile_clean_whole_tree_rollback_disposition: Disposition = Scaffold { + dissolves_to: SingleAuthority, + bind: DeclarationRef { + module_path: "v2.workflow.compile_clean_shard_roster", + decl_name: "compile_clean_sharding_enabled", + field: WholeDeclaration + } +} + +fn compile_clean_whole_tree_rollback_active() -> Bool { + match shell.Env.Get(name: compile_clean_whole_tree_rollback_env).value { + Present { value: v } => (v == "1") || (v == "true") + Absent => false + } +} + +fn compile_clean_sharding_enabled() -> Bool { + !compile_clean_whole_tree_rollback_active() +} + +data compile_clean_shard_transport_entry: String = "dsl/tools/dsl_compile_clean_shard_transport.dag" + +fn compile_clean_module_shard_from_decl(decl: ModuleDeclarationFact) -> CompileCleanModuleShard { + CompileCleanModuleShard { entry_path: decl.path } +} + +fn compile_clean_module_shard_roster() -> List { + fold( + module_declaration_facts_live(pool_roots: witness_layer_roots), + init: [], + f: fn(acc, decl) { + concat(acc, [compile_clean_module_shard_from_decl(decl: decl)]) + } + ) +} + +fn compile_clean_shard_entry_paths() -> List { + fold(compile_clean_module_shard_roster(), init: [], f: fn(acc, shard) { + concat(acc, [shard.entry_path]) + }) +} + +fn compile_clean_shard_spawn_width_cap(t: HardwareThreadCount) -> Int { + hardware_thread_count_value(t: t) +} + +fn compile_clean_shard_batch_profile() -> RunnableResourceProfile { + runnable_resource_profile( + heavy_whole_tree_resolve: false, + spawns_host_compiler: true, + memory: runnable_memory_substantial(predicted_peak: gunbc_ci_floor_per_shard_peak_rss_bytes()) + ) +} + +fn compile_clean_shard_batch_runnable(spawn_width_cap: Int) -> Runnable { + RunnableCompileCleanShardBatch { + entry_paths: compile_clean_shard_entry_paths(), + spawn_width_cap: spawn_width_cap, + profile: compile_clean_shard_batch_profile() + } +} + +fn compile_clean_shard_roster_count() -> Int { + list_length(items: compile_clean_module_shard_roster()) +} + +fn compile_clean_shard_roster_nonempty() -> Bool { + compile_clean_shard_roster_count() > 0 +} + +fn compile_clean_shard_roster_has_multiple() -> Bool { + compile_clean_shard_roster_count() >= 2 +} + +fn witness_shard_roster_derives_from_witness_layer_roots() -> Bool { + let boundary = compile_clean_partition_boundary() + boundary.pool_roots == witness_layer_roots +} + +fn list_contains_path(xs: List, target: String) -> Bool { + fold(xs, init: false, f: fn(found, p) { found || (p == target) }) +} + +fn path_occurrence_count(xs: List, target: String) -> Int { + fold(xs, init: 0, f: fn(n, p) { if p == target { n + 1 } else { n } }) +} + +fn witness_shard_roster_paths_duplicate_free() -> Bool { + let paths = compile_clean_shard_entry_paths() + fold(paths, init: true, f: fn(ok, path) { + ok && (path_occurrence_count(xs: paths, target: path) == 1) + }) +} + +fn witness_shard_roster_coverage(full: List, plan_paths: List) -> Bool { + fold(full, init: true, f: fn(ok, path) { + ok && list_contains_path(xs: plan_paths, target: path) + }) +} + +fn witness_dropped_shard_breaks_coverage(full: List, plan_paths: List) -> Bool { + if list_length(items: full) < 2 { + false + } else { + let truncated_plan = plan_paths.skip(n: 1) + witness_shard_roster_coverage(full: full, plan_paths: plan_paths) + && !witness_shard_roster_coverage(full: full, plan_paths: truncated_plan) + } +} + +fn witness_compose_all_clean_matches_reduce() -> Bool { + let vs = [ShardClean, ShardClean] + verdict_eq( + a: compose_compile_clean_shard_verdicts_as_reduce(verdicts: vs), + b: compile_clean_shard_verdict_to_reduce( + v: compose_compile_clean_shard_verdicts(verdicts: vs) + ) + ) +} + +fn witness_compose_failure_is_fail_closed() -> Bool { + let failed = ShardFailed { entry_path: "dsl/std/logic.dag" } + compose_compile_clean_shard_verdicts(verdicts: [ShardClean, failed]) != ShardClean +}