Budget-tree consumer-0 + Track A: instantiate the real srv1 fleet BudgetNode tree from measured numbers + DERIVE the .dag-floor spawn-width from it (construction, not a Bool check)

dsl/gunbc/ci_budget_tree.dag

@@ -0,0 +1,128 @@
+module gunbc.ci_budget_tree
+
+import std.measure { ByteSize, byte_size, measure_scale_fraction_floor, Memory, One }
+import std.types { Int, List, String }
+import std.realization_width { byte_size_saturating_sub }
+import extdeps.accounting.budget { Appropriation, LineItem }
+import product.budget_tree {
+  BudgetNode, BudgetClaim, BudgetPriority, Guaranteed,
+  BudgetChild, BudgetChildFound, BudgetChildMissing,
+  budget_child_named
+}
+import product.compute_fabric { compute_host_ram_bytes_total }
+import gunbc.operator_fleet { srv1_host }
+import gunbc.ci_floor_measurement {
+  HostFixedOverhead, HostFixedOverheadMeasured, HostFixedOverheadUnmeasured,
+  gunbc_ci_host_fixed_overhead
+}
+import gunbc.ci_runner_placement { gunbc_ci_runners_per_host }
+
+
+fn srv1_node_name() -> String {
+  "srv1"
+}
+
+fn srv1_run_pool_name() -> String {
+  "ci_run_pool"
+}
+
+fn srv1_ci_run_name() -> String {
+  "ci_run"
+}
+
+fn srv1_appropriation(b: ByteSize) -> Appropriation<Memory, One> {
+  Appropriation { ceiling: b }
+}
+
+fn srv1_host_ram() -> ByteSize {
+  compute_host_ram_bytes_total(host: srv1_host)
+}
+
+fn srv1_fixed_overhead_bytes() -> ByteSize {
+  match gunbc_ci_host_fixed_overhead {
+    HostFixedOverheadMeasured { bytes: b } => b
+    HostFixedOverheadUnmeasured => srv1_host_ram()
+  }
+}
+
+fn srv1_run_pool_budget() -> ByteSize {
+  byte_size_saturating_sub(total: srv1_host_ram(), reserved: srv1_fixed_overhead_bytes())
+}
+
+fn srv1_runs_per_host() -> Int {
+  gunbc_ci_runners_per_host()
+}
+
+fn per_run_budget_split(pool: ByteSize, r: Int) -> ByteSize {
+  if r <= 0 {
+    measure_scale_fraction_floor(m: pool, num: 0, den: 1)
+  } else {
+    measure_scale_fraction_floor(m: pool, num: 1, den: r)
+  }
+}
+
+fn srv1_per_run_budget() -> ByteSize {
+  per_run_budget_split(pool: srv1_run_pool_budget(), r: srv1_runs_per_host())
+}
+
+fn srv1_overhead_claim() -> BudgetClaim {
+  BudgetClaim {
+    item: LineItem { purpose: "host_fixed_overhead", amount: srv1_fixed_overhead_bytes() },
+    priority: Guaranteed
+  }
+}
+
+fn srv1_ci_run_node_value() -> BudgetNode {
+  BudgetNode {
+    name: srv1_ci_run_name(),
+    appropriation: srv1_appropriation(b: srv1_per_run_budget()),
+    claims: [],
+    children: []
+  }
+}
+
+fn replicate_budget_node(n: Int, node: BudgetNode) -> List<BudgetNode> {
+  if n <= 0 { [] } else { concat([node], replicate_budget_node(n: n - 1, node: node)) }
+}
+
+fn srv1_run_children() -> List<BudgetNode> {
+  replicate_budget_node(n: srv1_runs_per_host(), node: srv1_ci_run_node_value())
+}
+
+fn srv1_run_pool_node_value() -> BudgetNode {
+  BudgetNode {
+    name: srv1_run_pool_name(),
+    appropriation: srv1_appropriation(b: srv1_run_pool_budget()),
+    claims: [],
+    children: srv1_run_children()
+  }
+}
+
+fn build_srv1_budget_tree() -> BudgetNode {
+  BudgetNode {
+    name: srv1_node_name(),
+    appropriation: srv1_appropriation(b: srv1_host_ram()),
+    claims: [srv1_overhead_claim()],
+    children: [srv1_run_pool_node_value()]
+  }
+}
+
+data srv1_budget_tree: BudgetNode = build_srv1_budget_tree()
+
+fn srv1_run_pool_node() -> BudgetChild {
+  budget_child_named(node: srv1_budget_tree, name: srv1_run_pool_name())
+}
+
+fn srv1_ci_run_node() -> BudgetChild {
+  match srv1_run_pool_node() {
+    BudgetChildMissing => BudgetChildMissing
+    BudgetChildFound { node: pool } => budget_child_named(node: pool, name: srv1_ci_run_name())
+  }
+}
+
+fn srv1_floor_memory_budget() -> ByteSize {
+  match srv1_ci_run_node() {
+    BudgetChildMissing => byte_size(0)
+    BudgetChildFound { node: run } => run.appropriation.ceiling
+  }
+}

dsl/product/budget_tree.dag

@@ -48,6 +48,19 @@ fn node_conserves(node: BudgetNode) -> Bool {
   node_conserves_local(node: node) && all(node.children, c => node_conserves(node: c))
 }
 
+type BudgetChild =
+    BudgetChildFound { node: BudgetNode }
+  | BudgetChildMissing
+
+fn budget_child_named(node: BudgetNode, name: String) -> BudgetChild {
+  fold(node.children, init: BudgetChildMissing, f: fn(acc, c) {
+    match acc {
+      BudgetChildFound { node: found } => BudgetChildFound { node: found }
+      BudgetChildMissing => if c.name == name { BudgetChildFound { node: c } } else { BudgetChildMissing }
+    }
+  })
+}
+
 fn admit_claims(node: BudgetNode) -> AdmissionState<Memory, One> {
   admit_all(appropriation: node.appropriation, candidates: claim_items(claims: node.claims))
 }

dsl/test/claim/ci_budget_tree_witness_test.dag

@@ -0,0 +1,58 @@
+module test.claim.ci_budget_tree_witness
+
+import std.logic { Bool }
+import std.measure { byte_size, byte_size_count }
+import product.budget_tree { node_conserves, budget_child_named, BudgetChildFound, BudgetChildMissing }
+import gunbc.ci_budget_tree {
+  srv1_budget_tree,
+  srv1_host_ram,
+  srv1_run_pool_budget,
+  srv1_per_run_budget,
+  srv1_floor_memory_budget,
+  srv1_ci_run_node,
+  per_run_budget_split,
+}
+
+fn witness_srv1_tree_conserves() -> Bool {
+  node_conserves(node: srv1_budget_tree)
+}
+
+fn witness_floor_budget_is_l2_allocation() -> Bool {
+  byte_size_count(b: srv1_floor_memory_budget()) == byte_size_count(b: srv1_per_run_budget())
+}
+
+fn witness_divide_once_reduces_below_host() -> Bool {
+  let per_run = byte_size_count(b: srv1_floor_memory_budget())
+  per_run > 0
+    && per_run < byte_size_count(b: srv1_run_pool_budget())
+    && byte_size_count(b: srv1_run_pool_budget()) < byte_size_count(b: srv1_host_ram())
+}
+
+fn witness_l2_node_present() -> Bool {
+  match srv1_ci_run_node() {
+    BudgetChildFound { node: _ } => true
+    BudgetChildMissing => false
+  }
+}
+
+fn witness_unknown_child_is_absent() -> Bool {
+  match budget_child_named(node: srv1_budget_tree, name: "__no_such_child__") {
+    BudgetChildFound { node: _ } => false
+    BudgetChildMissing => true
+  }
+}
+
+fn witness_degenerate_r_fails_closed() -> Bool {
+  let zero = per_run_budget_split(pool: byte_size(1000), r: 0)
+  let split = per_run_budget_split(pool: byte_size(1000), r: 10)
+  return byte_size_count(b: zero) == 0 && byte_size_count(b: split) == 100
+}
+
+test fn ci_budget_tree_holds() -> Bool {
+  witness_srv1_tree_conserves()
+    && witness_floor_budget_is_l2_allocation()
+    && witness_divide_once_reduces_below_host()
+    && witness_l2_node_present()
+    && witness_unknown_child_is_absent()
+    && witness_degenerate_r_fails_closed()
+}

src/v2/test/claim/ci_floor_plan_witness_test.dag

@@ -10,6 +10,8 @@ import v2.workflow.ci_floor_plan {
   witness_floor_tight_budget_backs_off_envelope,
   witness_floor_fail_open_would_exceed_budget,
   witness_floor_unreadable_budget_uses_conservative_envelope,
+  witness_floor_width_governed_by_budget_tree,
+  witness_floor_width_derived_from_tree_is_positive,
 }
 import gunbc.ci_layer_roots { witness_discovery_scan_dirs, witness_layer_roots }
 import v2.std.algebra { length, list_snoc_item }
@@ -118,3 +120,8 @@ test fn ci_floor_run_envelope_witnesses() -> Bool {
     witness_floor_fail_open_would_exceed_budget() &&
     witness_floor_unreadable_budget_uses_conservative_envelope()
 }
+
+test fn ci_floor_spawn_width_from_budget_tree_witnesses() -> Bool {
+  witness_floor_width_governed_by_budget_tree() &&
+    witness_floor_width_derived_from_tree_is_positive()
+}

src/v2/workflow/ci_floor_plan.dag

@@ -13,7 +13,7 @@ import v2.std.dependency { DataDependsOn, ResourceDependsOn, DependencyView }
 import v2.std.node { Atom, Node, TypeNode }
 import v2.std.text { String }
 import std.types { ContentHash, Int }
-import std.measure { HardwareThreadCount, hardware_thread_count, byte_size }
+import std.measure { HardwareThreadCount, hardware_thread_count, hardware_thread_count_value, byte_size, ByteSize, byte_size_count }
 import std.realization_width { width_fold_objective_goals, memory_aware_spawn_width, bounded_host_spawn_width }
 import product.compute_fabric {
   ResourceEnvelope,
@@ -26,6 +26,7 @@ import gunbc.ci_floor_measurement {
   gunbc_ci_floor_per_shard_peak_rss_bytes,
   gunbc_ci_floor_conservative_fallback_width
 }
+import gunbc.ci_budget_tree { srv1_floor_memory_budget }
 import extdeps.cpu.ampere { altra_max_m12830_catalog }
 import std.realization_schedule {
   RealizationPlan,
@@ -242,16 +243,20 @@ fn gunbc_ci_floor_hardware_threads() -> HardwareThreadCount {
   altra_max_m12830_catalog.threads
 }
 
-fn gunbc_ci_floor_spawn_width(memory_budget_bytes: Int) -> HardwareThreadCount {
+fn gunbc_ci_floor_spawn_width_for_budget(memory_budget: ByteSize) -> HardwareThreadCount {
   memory_aware_spawn_width(
     shard_count: gunbc_ci_floor_breadth(),
     hardware_threads: gunbc_ci_floor_hardware_threads(),
-    memory_budget: byte_size(memory_budget_bytes),
+    memory_budget: memory_budget,
     per_shard_peak: gunbc_ci_floor_per_shard_peak_rss_bytes(),
     conservative_fallback_width: gunbc_ci_floor_conservative_fallback_width
   )
 }
 
+fn gunbc_ci_floor_spawn_width() -> HardwareThreadCount {
+  gunbc_ci_floor_spawn_width_for_budget(memory_budget: srv1_floor_memory_budget())
+}
+
 fn gunbc_ci_floor_run_envelope(memory_budget_bytes: Int) -> ResourceEnvelope {
   parallel_run_demand_envelope_from_budget(
     shard_count: gunbc_ci_floor_breadth(),
@@ -287,6 +292,17 @@ fn witness_floor_unreadable_budget_uses_conservative_envelope() -> Bool {
   demand_envelope_thread_count(envelope: envelope) == gunbc_ci_floor_conservative_fallback_width
 }
 
+fn witness_floor_width_governed_by_budget_tree() -> Bool {
+  let tree_width = hardware_thread_count_value(t: gunbc_ci_floor_spawn_width())
+  let cpu_only_width = hardware_thread_count_value(t: bounded_host_spawn_width(shard_count: gunbc_ci_floor_breadth(), hardware_threads: gunbc_ci_floor_hardware_threads()))
+  return tree_width < cpu_only_width
+}
+
+fn witness_floor_width_derived_from_tree_is_positive() -> Bool {
+  byte_size_count(b: srv1_floor_memory_budget()) > 0
+    && hardware_thread_count_value(t: gunbc_ci_floor_spawn_width()) > 0
+}
+
 fn floor_layer_heavy_resolve_count(batch: ExecutorBatch, spec: CiSpec) -> Int {
   fold(batch.jobs, init: 0, f: fn(acc, job) {
     if floor_node_is_heavy_resolve(node: job.node, spec: spec) { acc + 1 } else { acc }