2-compile-clean-shard-b — floor consumes the shards (TOP CI priority: dsl_compile_clean_gate is ~47min of the ~50min floor, width-immune). Shard A (#6118) landed partition+algebra+exemplar; B enrolls shard RUNNABLES in ci_floor_plan batches. Acceptance (8): no single serial compile-clean dominator;

dsl/extdeps/shell/shell.dag

@@ -34,7 +34,6 @@ service shell.Env {
       0 => Unit
       1 => Unit "Variable not set"
     }
-
   }
 }
 

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<Nano>) -> 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
     }
   }
 }

dsl/std/realization.dag

@@ -11,6 +11,8 @@ import std.realization_schedule {
   Runnable,
   RunnableSingleClaim,
   RunnableDiscoveryBatch,
+  RunnableCompileCleanShardBatch,
+  compile_clean_shard_batch_label,
   ScheduleWitnessEntry,
   Schedule,
   CostAccount,

dsl/std/realization_schedule.dag

@@ -155,13 +155,15 @@ 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()
   }
 }
 
 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<String>
+      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<List<Runnable>>
@@ -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<String> {
+  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<Runnable>) -> List<String> {
+  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<Runnable>) -> Int {
+  length(xs: schedule_batch_compile_clean_shard_entry_paths(batch: batch))
+}
+
 fn schedule_batch_contains_label(batch: List<Runnable>, 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<Runnable>, target: String) -> Bool
 type ScheduleLensViolation
   = EmptySchedule
   | CompileGateNotFirst { expected: String }
+  | CompileGateMonolithPresent { expected: String }
+  | CompileCleanShardsInsufficient { minimum: Int }
   | CorpusBeforeCompile
   | SingleBatchOnly
 
@@ -228,14 +261,24 @@ 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 =>
       LensVerdictDiagnostic { reason: "schedule_lens_single_batch_only", at: at }
   }
 }
 
-fn schedule_lens_verdict_for_ci_floor<S>(plan: RealizationPlan<S>, compile_gate_fn: String) -> LensVerdict {
+fn schedule_lens_verdict_for_ci_floor<S>(plan: RealizationPlan<S>, 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<S>(plan: RealizationPlan<S>, 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<S>(plan: RealizationPlan<S>, 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
     }
   }

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
+    }
+}

dsl/test/claim/realization_schedule_witness_test.dag

@@ -51,7 +51,8 @@ fn perturbed_schedule_wrong_order(correct: List<List<Runnable>>) -> List<List<Ru
 fn witness_schedule_lens_green_on_synthetic_plan() -> 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

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 }

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()
+}