Host admission reads occupancy: a held Spark host running its pair worker refuses GPU/memory effects

dag/extdeps/nvidia/system_management_interface.dag

@@ -0,0 +1,139 @@
+module extdeps.nvidia.system_management_interface
+
+import std.types { NonEmptyStr, Bool, Int, List, Unit }
+import std.string_type { String }
+import std.nat { Nat }
+import std.measure { Mebibyte, mebibyte }
+import std.algebra { trim }
+import v2.std.algebra { skip }
+import std.checked_arithmetic { checked_int_magnitude, CheckedNatReady, CheckedNatOverflow }
+import extdeps.external_authority { ExternalAuthority }
+import extdeps.uri { Uri, Https }
+
+data extdeps_external_authority_anchor: ExternalAuthority = ExternalAuthority {
+  uri: Uri {
+    scheme: Https
+    locator: "docs.nvidia.com/deploy/nvidia-smi/index.html"
+  }
+}
+
+// ── WHICH PROCESSES HOLD A GPU CONTEXT, AS nvidia-smi REPORTS THEM ─────────────────────────────
+//
+// `--query-compute-apps` lists one row per process holding a compute context on any device, and
+// `--format=csv,noheader,nounits` makes each row `pid, process_name, used_memory` with used_memory in
+// MiB. No compute process is EMPTY OUTPUT with exit 0 -- not an error and not a header. A device that
+// cannot attribute memory to a process prints a bracketed marker (`[N/A]`, `[Not Supported]`) in that
+// column instead of a number; extdeps.nvidia.management_library records that NVML refuses the
+// device-wide memory query on the unified-memory GB10, so the per-process column is modelled as able
+// to refuse the same way. A refusal is its own arm and carries no number: it is not zero, and a
+// consumer must not read the presence of a process as weightless because its figure was withheld.
+service nvidia_smi.Smi {
+  operation QueryComputeApps {
+    input {}
+    output {
+      value: String from "stdout"
+      success: Bool from "exit_success"
+    }
+    readonly
+    transport shell { argv: ["nvidia-smi", "--query-compute-apps=pid,process_name,used_memory", "--format=csv,noheader,nounits"] }
+    exit {
+      0 => Unit
+      nonzero => String "nvidia-smi --query-compute-apps failed"
+    }
+    mock_response {
+      0 => { value: "", success: true } "hermetic nvidia_smi.Smi.QueryComputeApps: no process holds a compute context"
+    }
+  }
+}
+
+type ComputeAppUsedMemory
+  = UsedMemoryReported { used: Mebibyte }
+  | UsedMemoryWithheld { marker: NonEmptyStr }
+
+type NvidiaComputeApp {
+  pid: Nat
+  process_name: String
+  used: ComputeAppUsedMemory
+}
+
+// A ROW THAT DOES NOT HAVE THE SHAPE IS NOT SKIPPED. Skipping it would read an unparsed GPU user as
+// absent, which is the one direction an occupancy reader must never err in.
+type NvidiaComputeAppsRead
+  = ComputeAppsParsed { apps: List<NvidiaComputeApp> }
+  | ComputeAppsUnparsed { line: String }
+
+// A NEGATIVE OR NON-NUMERIC FIELD IS NOT A ROW. The crossing from Int is the checked one: a runtime Int
+// cast `as Nat` typechecks and then fails at execution.
+fn parse_nat(raw: String) -> Nat? {
+  match parse_int(s: trim(s: raw)) {
+    Absent => none
+    Present { value: n } =>
+      if n < 0 { none } else {
+        match checked_int_magnitude(a: n) {
+          CheckedNatReady { value: m } => Present { value: m }
+          CheckedNatOverflow { cause: _ } => none
+        }
+      }
+  }
+}
+
+fn compute_app_used_memory(raw: String) -> ComputeAppUsedMemory? {
+  let t = trim(s: raw)
+  if starts_with(s: t, prefix: "[") && length(t) > 1 && substring(s: t, start: length(t) - 1, end: length(t)) == "]" {
+    Present { value: UsedMemoryWithheld { marker: t as NonEmptyStr } }
+  } else {
+    match parse_nat(raw: t) {
+      Absent => none
+      Present { value: n } => Present { value: UsedMemoryReported { used: mebibyte(count: n) } }
+    }
+  }
+}
+
+fn compute_app_row(line: String) -> NvidiaComputeApp? {
+  let cells = split(s: line, delimiter: ",")
+  if count(cells) != 3 { none } else {
+    match first(cells) {
+      Absent => none
+      Present { value: raw_pid } =>
+        match parse_nat(raw: raw_pid) {
+          Absent => none
+          Present { value: pid } =>
+            {
+              match first(cells |> skip(n: 1)) {
+                Absent => none
+                Present { value: name } =>
+                  match first(cells |> skip(n: 2)) {
+                    Absent => none
+                    Present { value: raw_used } =>
+                      match compute_app_used_memory(raw: raw_used) {
+                        Absent => none
+                        Present { value: used } => Present { value: NvidiaComputeApp { pid: pid, process_name: trim(s: name), used: used } }
+                      }
+                  }
+              }
+            }
+        }
+    }
+  }
+}
+
+// ONE PASS PER LINE AND ONE FLATTEN, not an accumulator re-copied per row (DESIGN section 6).
+fn parse_compute_apps(text: String) -> NvidiaComputeAppsRead {
+  let lines = flat_map(split(s: text, delimiter: "\n"), l => if trim(s: l) == "" { [] as List<String> } else { [l] })
+  let rows = map(lines, l =>
+    match compute_app_row(line: l) {
+      Absent => ComputeAppsUnparsed { line: l }
+      Present { value: a } => ComputeAppsParsed { apps: [a] }
+    })
+  let unparsed = flat_map(rows, r => match r {
+    ComputeAppsUnparsed { line: l } => [l]
+    ComputeAppsParsed { apps: _ } => [] as List<String>
+  })
+  match first(unparsed) {
+    Present { value: bad } => ComputeAppsUnparsed { line: bad }
+    Absent => ComputeAppsParsed { apps: flat_map(rows, r => match r {
+      ComputeAppsParsed { apps: xs } => xs
+      ComputeAppsUnparsed { line: _ } => [] as List<NvidiaComputeApp>
+    }) }
+  }
+}

dag/gunbc/compute/host_occupancy.dag

@@ -0,0 +1,190 @@
+module gunbc.compute.host_occupancy
+
+import std.types { String, Bool, Int, NonEmptyStr, List }
+import std.nat { Nat }
+import std.algebra { trim }
+import std.measure { Kibibyte, kibibyte_count, Mebibyte, mebibyte, mebibyte_count }
+import extdeps.linux.proc_meminfo { MeminfoParsed, MeminfoFieldAbsent, parse_proc_meminfo }
+import extdeps.nvidia.system_management_interface {
+  NvidiaComputeApp, ComputeAppUsedMemory, UsedMemoryReported, UsedMemoryWithheld,
+  NvidiaComputeAppsRead, ComputeAppsParsed, ComputeAppsUnparsed, parse_compute_apps,
+}
+import v2.std.operation_argv { ArgvMaterialized, ArgvMaterializationRefused }
+import gunbc.host_operation_exec {
+  HostOperation, SystemctlIsActive, ProcfsReadMeminfo, NvidiaSmiQueryComputeApps,
+  host_operation_materialize_argv, host_operation_label,
+}
+import extdeps.systemd {
+  SystemdUnitActiveState, Active, Inactive, Activating, Deactivating, Reloading, Failed,
+  parse_systemd_unit_active_state, systemd_unit_active_state_wire_label,
+}
+import gunbc.spark.host_effect_quiescence { ArgvRun, ArgvRan, ArgvLegDidNotRun }
+
+// ── WHO IS ALREADY USING THIS HOST: A READING, TAKEN ON THE HOST, BEFORE AN EFFECT IS ADMITTED ─
+//
+// OWNERSHIP IS NOT OCCUPANCY. gunbc.spark.host_commitment `admit_host_held_by_subject` answers
+// whether a subject HOLDS a host; it cannot answer whether that subject's own serving is running on it
+// and holding its memory, because holding is a fact in the event log and occupancy is a fact on the
+// machine. The receipt (2026-09-26, srv8): the Group A pair subject held srv8, so a probe was admitted,
+// while that subject's own DeepSeek V4 pair worker held ~99.7 GB of the unified pool; the host was left
+// with 15 GiB, thrashed, and needed a power cycle. So an effect that declares a memory or GPU need is
+// admitted by TWO independent relations -- held by the subject, and unoccupied by this reading -- and
+// neither stands in for the other.
+//
+// THE READING IS THREE MODELED READS, all through gunbc.host_operation_exec so each argv is derived from
+// its extdeps declaration: `systemctl is-active` per serving unit the caller names, nvidia-smi's compute
+// processes, and /proc/meminfo. THE LEG IS SUPPLIED, as gunbc.spark.host_effect_quiescence supplies it:
+// production hands the leg that reaches the host, a claim hands a recorded one.
+//
+// A GPU PROCESS IS AN OCCUPANT WHETHER OR NOT A KNOWN UNIT OWNS IT (proud-deer-538 ruling 2026-09-26).
+// This reader does not attribute a pid to a unit, and does not need to: a foreign or leftover GPU user
+// -- the stray probe container of the same morning -- occupies the host exactly as the pair worker does,
+// so every compute process is listed as its own occupant beside whichever named units are active. An
+// attribution would only improve the refusal's wording; it can never turn an occupant into a vacancy.
+//
+// UNREAD REFUSES. A leg that did not run, a non-zero read, a state word this reader does not know, an
+// nvidia-smi row it cannot parse: each is HostOccupancyUnread and admission refuses on it. Reading any
+// of them as vacant is the absorbing fallback DESIGN section 5 forbids, and here its cost is a host.
+type HostOccupant
+  = ServingUnitActive { unit: NonEmptyStr, state: SystemdUnitActiveState }
+  | GpuComputeProcess { pid: Nat, process_name: String, used: ComputeAppUsedMemory }
+
+type HostOccupancyReading
+  = HostOccupancyRead { occupants: List<HostOccupant>, available: Kibibyte }
+  | HostOccupancyUnread { cause: String }
+
+// THE ADMISSION. `need` is the memory the effect declares it will use; it is compared against
+// MemAvailable even when nothing is running, because a host can be short with no named occupant.
+type HostOccupancyAdmission
+  = HostUnoccupied { available: Kibibyte }
+  | HostOccupiedBy { occupant: HostOccupant, others: List<HostOccupant>, gpu_used: Mebibyte }
+  | HostMemoryShort { available: Kibibyte, need: Kibibyte }
+  | HostOccupancyNotRead { cause: String }
+
+// WHICH STATES HOLD THE HOST, over the unit-state model extdeps.systemd owns. `inactive` and `failed`
+// hold nothing; a unit that is running, starting, stopping or reloading occupies it. A state word that
+// model does not parse is refused as unread by the reader, never guessed.
+fn unit_state_occupies(state: SystemdUnitActiveState) -> Bool {
+  match state {
+    Inactive => false
+    Failed => false
+    Active => true
+    Activating => true
+    Deactivating => true
+    Reloading => true
+  }
+}
+
+type HostOperationLeg
+  = HostOperationLegRead { stdout: String }
+  | HostOperationLegUnread { cause: String }
+
+// is-active exits 3 for a unit that is not active, and that is an answer, not a failed read -- which
+// is why the exit codes a read accepts are the caller's to state.
+fn host_operation_leg(run: fn(List<String>) -> ArgvRun, operation: HostOperation, accepted_exits: List<Int>) -> HostOperationLeg {
+  match host_operation_materialize_argv(operation: operation) {
+    ArgvMaterializationRefused { at: _, cause: _ } => HostOperationLegUnread { cause: join([host_operation_label(operation: operation), " could not be materialized as an argv"], "") }
+    ArgvMaterialized { argv: argv } =>
+      match run(argv) {
+        ArgvLegDidNotRun { cause: c } => HostOperationLegUnread { cause: join([host_operation_label(operation: operation), " did not run: ", c], "") }
+        ArgvRan { exit_code: code, stdout: out, stderr: err } =>
+          if any(accepted_exits, e => e == code) { HostOperationLegRead { stdout: out } }
+          else { HostOperationLegUnread { cause: join([host_operation_label(operation: operation), " exited ", to_string(code), ": ", trim(s: err)], "") } }
+      }
+  }
+}
+
+type UnitActivity
+  = UnitOccupies { occupant: HostOccupant }
+  | UnitVacant
+  | UnitUnread { cause: String }
+
+fn read_unit_activity(run: fn(List<String>) -> ArgvRun, unit: NonEmptyStr) -> UnitActivity {
+  match host_operation_leg(run: run, operation: SystemctlIsActive { unit: unit }, accepted_exits: [0, 3]) {
+    HostOperationLegUnread { cause: c } => UnitUnread { cause: c }
+    HostOperationLegRead { stdout: out } => {
+      let word = trim(s: out)
+      match parse_systemd_unit_active_state(raw: word) {
+        Absent => UnitUnread { cause: join(["systemctl is-active ", unit as String, " answered a state extdeps.systemd does not model: `", word, "`"], "") }
+        Present { value: state } =>
+          if unit_state_occupies(state: state) { UnitOccupies { occupant: ServingUnitActive { unit: unit, state: state } } } else { UnitVacant }
+      }
+    }
+  }
+}
+
+fn read_host_occupancy(run: fn(List<String>) -> ArgvRun, units: List<NonEmptyStr>) -> HostOccupancyReading {
+  let unit_reads = map(units, u => read_unit_activity(run: run, unit: u))
+  let unit_unread = flat_map(unit_reads, r => match r { UnitUnread { cause: c } => [c]  _ => [] as List<String> })
+  if length(unit_unread) != 0 {
+    HostOccupancyUnread { cause: join(unit_unread, "; ") }
+  } else {
+    let unit_occupants = flat_map(unit_reads, r => match r { UnitOccupies { occupant: o } => [o]  _ => [] as List<HostOccupant> })
+    match host_operation_leg(run: run, operation: NvidiaSmiQueryComputeApps, accepted_exits: [0]) {
+      HostOperationLegUnread { cause: c } => HostOccupancyUnread { cause: c }
+      HostOperationLegRead { stdout: apps_text } =>
+        match parse_compute_apps(text: apps_text) {
+          ComputeAppsUnparsed { line: l } => HostOccupancyUnread { cause: join(["nvidia-smi listed a compute process this reader cannot parse: `", l, "`"], "") }
+          ComputeAppsParsed { apps: apps } =>
+            match host_operation_leg(run: run, operation: ProcfsReadMeminfo, accepted_exits: [0]) {
+              HostOperationLegUnread { cause: c } => HostOccupancyUnread { cause: c }
+              HostOperationLegRead { stdout: meminfo } =>
+                match parse_proc_meminfo(text: meminfo) {
+                  MeminfoFieldAbsent { field: f } => HostOccupancyUnread { cause: join(["/proc/meminfo carries no readable ", f as String, " line"], "") }
+                  MeminfoParsed { meminfo: m } =>
+                    HostOccupancyRead {
+                      occupants: concat(unit_occupants, map(apps, a => GpuComputeProcess { pid: a.pid, process_name: a.process_name, used: a.used })),
+                      available: m.mem_available,
+                    }
+                }
+            }
+        }
+    }
+  }
+}
+
+fn occupant_reported_mebibytes(o: HostOccupant) -> Nat {
+  match o {
+    GpuComputeProcess { pid: _, process_name: _, used: UsedMemoryReported { used: m } } => mebibyte_count(m: m)
+    _ => 0
+  }
+}
+
+fn admit_host_occupancy(reading: HostOccupancyReading, need: Kibibyte) -> HostOccupancyAdmission {
+  match reading {
+    HostOccupancyUnread { cause: c } => HostOccupancyNotRead { cause: c }
+    HostOccupancyRead { occupants: occ, available: avail } =>
+      match first(occ) {
+        Present { value: o } =>
+          HostOccupiedBy {
+            occupant: o,
+            others: occ |> skip(n: 1),
+            gpu_used: mebibyte(count: fold(occ, init: 0, f: (acc, x) => acc + occupant_reported_mebibytes(o: x))),
+          }
+        Absent =>
+          if kibibyte_count(k: avail) < kibibyte_count(k: need) { HostMemoryShort { available: avail, need: need } }
+          else { HostUnoccupied { available: avail } }
+      }
+  }
+}
+
+fn host_occupant_wire(o: HostOccupant) -> String {
+  match o {
+    ServingUnitActive { unit: u, state: s } => join([u as String, " is ", systemd_unit_active_state_wire_label(s: s)], "")
+    GpuComputeProcess { pid: p, process_name: n, used: UsedMemoryReported { used: m } } => join(["pid ", to_string(p), " (", n, ") holds ", to_string(mebibyte_count(m: m)), " MiB of GPU memory"], "")
+    GpuComputeProcess { pid: p, process_name: n, used: UsedMemoryWithheld { marker: k } } => join(["pid ", to_string(p), " (", n, ") holds a GPU context whose memory nvidia-smi withheld as ", k as String], "")
+  }
+}
+
+// None when admitted; otherwise the refusal, prefixed with the host and the effect it refuses.
+fn host_occupancy_refusal(host: String, purpose: String, a: HostOccupancyAdmission) -> String? {
+  match a {
+    HostUnoccupied { available: _ } => none
+    HostOccupiedBy { occupant: o, others: rest, gpu_used: g } =>
+      Present { value: join([host, " is occupied, so ", purpose, " is not admitted: ", join(map(concat([o], rest), x => host_occupant_wire(o: x)), "; "), " (", to_string(mebibyte_count(m: g)), " MiB of GPU memory reported in use); vacate it through the modeled pair vacate before admitting a GPU or memory effect"], "") }
+    HostMemoryShort { available: av, need: nd } =>
+      Present { value: join([host, " has ", to_string(kibibyte_count(k: av)), " KiB available and ", purpose, " declares ", to_string(kibibyte_count(k: nd)), " KiB, so it is not admitted"], "") }
+    HostOccupancyNotRead { cause: c } =>
+      Present { value: join([host, " could not be read for occupancy, so ", purpose, " is not admitted: ", c], "") }
+  }
+}