Step 4A: a host-bound session-demand instrument, with its discriminating pair taken from srv1 and srv2

.github/workflows/fleet-converge.yml

@@ -8,9 +8,9 @@ on:
         options: [srv1, srv2, srv3, srv4]
         type: choice
       mode:
-        description: plan shows membership hunks; apply executes a prior plan artifact; spark_linger enables linger on target spark via FleetSsh typed argv from executor; spark_observe is a read-only observation of both Sparks
+        description: plan shows membership hunks; apply executes a prior plan artifact; spark_linger enables linger on target spark via FleetSsh typed argv from executor; spark_observe is a read-only observation of both Sparks; session_demand_observe is a read-only observation of the dispatched host's own session cgroup population and memory counters
         required: true
-        options: [plan, apply, spark_linger, spark_observe, spark_reboot, spark_grants, spark_durability]
+        options: [plan, apply, spark_linger, spark_observe, spark_reboot, spark_grants, spark_durability, session_demand_observe]
         type: choice
       target:
         description: Spark target host for spark_linger mode (srv5 or srv6)
@@ -266,6 +266,14 @@ jobs:
           cat "$ROOT/target/spark-serving-observe-ci-receipt.txt"
         if: github.event.inputs.mode == 'spark_observe'
         timeout-minutes: 60
+      - name: "Session demand observation (read-only: managed slice population, per-session memory.current, parent slice counters)"
+        id: session_demand_observe
+        run: |
+          ROOT=$(git rev-parse --show-toplevel 2>/dev/null || pwd)
+          "$ROOT/target/release/gunbc" run --source-root "$ROOT/dag" --source-root "$ROOT/src/v2" --entry dag/gunbc/session_demand_observe.dag --function session_demand_observe_ci_wet
+          cat "$ROOT/target/session-demand-observe-receipt.txt"
+        if: github.event.inputs.mode == 'session_demand_observe'
+        timeout-minutes: 60
       - name: "Spark serving reboot (durability probe: the receipt is the converge that follows)"
         id: spark_reboot
         run: |

dag/extdeps/linux/cgroup_v2.dag

@@ -0,0 +1,95 @@
+module extdeps.linux.cgroup_v2
+
+import std.types { Bool, NonEmptyStr, String }
+import extdeps.external_authority { ExternalAuthority }
+import extdeps.uri { Https, Uri }
+
+data extdeps_external_authority_anchor: ExternalAuthority = ExternalAuthority {
+  uri: Uri {
+    scheme: Https
+    locator: "docs.kernel.org/admin-guide/cgroup-v2.html"
+  }
+}
+
+// THE UNIFIED HIERARCHY IS MOUNTED AT ONE PLACE, AND IT IS THE KERNEL'S PLACE, NOT OURS. Every
+// consumer that spells this path itself is a second authority for it (DESIGN 3), which is how a
+// mount point ends up differing by one character between the reader and the thing it reads.
+data cgroup_v2_mount_point: NonEmptyStr = "/sys/fs/cgroup"
+
+// A CHILD CGROUP IS A DIRECTORY; AN INTERFACE FILE IS NOT. The kernel represents the hierarchy in
+// the filesystem, so enumerating children is exactly enumerating SUBDIRECTORIES -- and a plain
+// listing of the directory would return the interface files (memory.current, cgroup.procs, ...)
+// intermixed with the children, which is why this constrains to -type d rather than filtering
+// names afterwards. Filtering by name would encode a guess about which prefixes are interface
+// files; -type d encodes the kernel's actual representation.
+//
+// -mindepth 1 excludes the directory itself; -maxdepth 1 keeps it to immediate children, because a
+// session's own descendants are that session's internal business and counting them would inflate
+// the population. -printf %f yields the bare child name rather than a full path.
+//
+// PathIsDirectory: A CGROUP THAT DOES NOT EXIST IS NOT A CGROUP THAT COULD NOT BE READ. Both make `find` and
+// `cat` exit nonzero, so exit status alone cannot separate "this host has no such slice" from
+// "the slice is there and the read failed" -- and those have opposite meanings for a capacity
+// question: the first says the host hosts nothing of this kind, the second says we are blind.
+// `test -d` is the discriminator the kernel's filesystem representation already affords, and it
+// is total: 0 for a directory, 1 for anything else, with no output to parse.
+service linux.CgroupV2 {
+  operation ListChildCgroups {
+    input { cgroup_path: NonEmptyStr }
+    output {
+      value: String from "stdout"
+      success: Bool from "exit_success"
+    }
+    transport shell {
+      argv: ["find", "{cgroup_path}", "-mindepth", "1", "-maxdepth", "1", "-type", "d", "-printf", "%f\n"]
+    }
+    exit {
+      0 => Unit
+      nonzero => String "cgroup child enumeration failed"
+    }
+  }
+
+  operation ReadInterfaceFile {
+    input { file_path: NonEmptyStr }
+    output {
+      value: String from "stdout"
+      success: Bool from "exit_success"
+    }
+    transport shell {
+      argv: ["cat", "{file_path}"]
+    }
+    exit {
+      0 => Unit
+      nonzero => String "cgroup interface file read failed"
+    }
+  }
+  operation PathIsDirectory {
+    input { path: NonEmptyStr }
+    output {
+      value: String from "stdout"
+      success: Bool from "exit_success"
+    }
+    transport shell {
+      argv: ["test", "-d", "{path}"]
+    }
+    exit {
+      0 => Unit
+      nonzero => String "path is not a directory"
+    }
+  }
+}
+
+// THE PATH IS COMPOSED IN .dag AND INTERPOLATED AS ONE VALUE, NOT ASSEMBLED INSIDE THE ARGV.
+// An argv of ["cat", "{cgroup_path}/{interface_file}"] does not concatenate two interpolations
+// around a literal slash -- the interpolation grammar reads across it and the `/` becomes a
+// DIVISION operator, which fails at evaluation with "unsupported operator Div on String".
+// Measured on srv1-07: ListChildCgroups, whose argv interpolates a single value, executed fine;
+// this operation was the only one carrying two interpolations separated by an operator character.
+// One input, one interpolation, and the composition stays where the type system can see it.
+fn cgroup_v2_child_path(parent: NonEmptyStr, child: NonEmptyStr) -> NonEmptyStr {
+  concat(parent as String, "/", child as String) as NonEmptyStr
+}
+
+fn cgroup_v2_slice_path(slice: NonEmptyStr) -> NonEmptyStr {
+  cgroup_v2_child_path(parent: cgroup_v2_mount_point, child: slice)
+}

dag/extdeps/linux/cgroup_v2_memory.dag

@@ -0,0 +1,119 @@
+module extdeps.linux.cgroup_v2_memory
+
+import std.algebra { trim }
+import std.checked_arithmetic { nat_magnitude }
+import extdeps.external_authority { ExternalAuthority }
+import extdeps.uri { Https, Uri }
+import std.measure { ByteSize, byte_size }
+import std.types { Int, NonEmptyStr, String }
+import v2.std.optional { Present, Absent }
+
+data extdeps_external_authority_anchor: ExternalAuthority = ExternalAuthority {
+  uri: Uri {
+    scheme: Https
+    locator: "docs.kernel.org/admin-guide/cgroup-v2.html#memory-interface-files"
+  }
+}
+
+// THE THREE KINDS ARE THE KERNEL'S, NOT OURS. cgroup-v2 documents each memory interface file with a
+// distinct temporal meaning, and a carrier storing every one as a bare ByteSize would erase the
+// distinction that decides whether a number may be compared, summed, or differenced at all:
+//
+//   Instantaneous       -- true at the moment of the read and at no other moment.
+//   LifetimeEpochScoped -- an extremum accumulated since the cgroup was created or the value reset.
+//                          It is NOT "the maximum during my observation window" unless the epoch is
+//                          carried beside it, which is why the epoch is a field rather than prose.
+//   CumulativeCounter   -- monotone since cgroup creation. A single end snapshot is not an interval
+//                          measurement; only a start/end pair yields a delta.
+//
+// Grounding the kinds here rather than in the consumer is DESIGN 3: the upstream spec already draws
+// this line, so re-coining it downstream would nickname a fact the kernel owns.
+type CgroupMemoryMetricKind
+  = Instantaneous
+  | LifetimeEpochScoped
+  | CumulativeCounter
+
+// The interface files this authority models, by their REAL names. Adding a file means adding a
+// variant, so the kind judgment stays exhaustive rather than defaulting.
+type CgroupMemoryInterfaceFile
+  = MemoryCurrent
+  | MemoryPeak
+  | MemoryMax
+  | MemoryHigh
+  | MemorySwapCurrent
+  | MemoryEvents
+
+fn cgroup_memory_interface_file_name(f: CgroupMemoryInterfaceFile) -> NonEmptyStr {
+  match f {
+    MemoryCurrent => "memory.current"
+    MemoryPeak => "memory.peak"
+    MemoryMax => "memory.max"
+    MemoryHigh => "memory.high"
+    MemorySwapCurrent => "memory.swap.current"
+    MemoryEvents => "memory.events"
+  }
+}
+
+fn cgroup_memory_metric_kind(f: CgroupMemoryInterfaceFile) -> CgroupMemoryMetricKind {
+  match f {
+    MemoryCurrent => Instantaneous
+    MemoryPeak => LifetimeEpochScoped
+    MemoryMax => Instantaneous
+    MemoryHigh => Instantaneous
+    MemorySwapCurrent => Instantaneous
+    MemoryEvents => CumulativeCounter
+  }
+}
+
+// A NEGATIVE BODY IS MALFORMED, NOT A MAGNITUDE. parse_int admits negatives, and every file modeled
+// here is documented as a non-negative byte count, so a leading minus is a malformed read rather
+// than a value to take the absolute value of. nat_magnitude is applied only AFTER the sign is
+// excluded, where it is the identity -- using it as the crossing on an unguarded Int would silently
+// turn "-1" into 1.
+fn parse_cgroup_nonnegative_bytes(text: String) -> ByteSize? {
+  match parse_int(s: trim(text)) {
+    Present { value: n } =>
+      if n < 0 {
+        none
+      } else {
+        Present { value: byte_size(count: nat_magnitude(a: n)) }
+      }
+    Absent => none
+  }
+}
+
+// THE `max` SENTINEL IS A THIRD STATE, NOT A LARGE NUMBER. memory.max and memory.high hold either a
+// byte count or the literal string "max", which the kernel documents as "no limit". Rendering that
+// as a huge ByteSize would fabricate a limit the host does not have, and rendering it as absent
+// would conflate it with a read that failed -- so it is its own arm, and an unparseable body is a
+// fourth.
+type CgroupMemoryLimitValue
+  = CgroupMemoryLimited { bytes: ByteSize }
+  | CgroupMemoryUnlimited
+  | CgroupMemoryLimitUnparseable { body: String }
+
+fn parse_cgroup_memory_limit(body: String) -> CgroupMemoryLimitValue {
+  if trim(body) == "max" {
+    CgroupMemoryUnlimited
+  } else {
+    match parse_cgroup_nonnegative_bytes(text: body) {
+      Present { value: b } => CgroupMemoryLimited { bytes: b }
+      Absent => CgroupMemoryLimitUnparseable { body: body }
+    }
+  }
+}
+
+// A COUNT FILE HAS NO `max` ARM. memory.current, memory.peak and memory.swap.current are documented
+// as byte counts with no sentinel, so admitting one here would widen the parse past what the
+// authority says the file can hold. Two arms, not three -- and a body of "max" in a count file is
+// therefore Unparseable, which is the correct reading of a kernel that should never have written it.
+type CgroupMemoryCountValue
+  = CgroupMemoryCount { bytes: ByteSize }
+  | CgroupMemoryCountUnparseable { body: String }
+
+fn parse_cgroup_memory_count(body: String) -> CgroupMemoryCountValue {
+  match parse_cgroup_nonnegative_bytes(text: body) {
+    Present { value: b } => CgroupMemoryCount { bytes: b }
+    Absent => CgroupMemoryCountUnparseable { body: body }
+  }
+}

dag/gunbc/ci_spec.dag

@@ -551,6 +551,11 @@ data gunbc_ci_spark_serving_observe_target: GunbcRunStepTarget = GunbcRunStepTar
   function: "spark_serving_observe_ci_wet",
 }
 
+data gunbc_ci_session_demand_observe_target: GunbcRunStepTarget = GunbcRunStepTarget {
+  entry: "dag/gunbc/session_demand_observe.dag",
+  function: "session_demand_observe_ci_wet",
+}
+
 data gunbc_ci_spark_serving_converge_slice_target: GunbcRunStepTarget = GunbcRunStepTarget {
   entry: "dag/gunbc/spark/serving_converge_slice_wet.dag",
   function: "spark_serving_converge_slice_ci_wet",
@@ -644,6 +649,7 @@ fn gunbc_run_step_targets() -> List<GunbcRunStepTarget> {
     gunbc_ci_spark_serving_durability_target,
     gunbc_ci_spark_grant_install_target,
     gunbc_ci_spark_serving_observe_target,
+    gunbc_ci_session_demand_observe_target,
     gunbc_ci_spark_serving_converge_slice_target,
     gunbc_ci_collect_fleet_convergence_target,
     gunbc_ci_fleet_host_key_scan_target,
@@ -1061,6 +1067,16 @@ fn gunbc_ci_spark_grant_install_invoke() -> String {
 }
 
 
+fn gunbc_ci_session_demand_observe_invoke() -> String {
+  gunbc_run_step_script(
+    source_roots: witness_layer_roots,
+    entry: gunbc_ci_session_demand_observe_target.entry,
+    function: gunbc_ci_session_demand_observe_target.function,
+    claim_run: false,
+    receipt_rel: "target/session-demand-observe-receipt.txt"
+  )
+}
+
 fn gunbc_ci_spark_serving_observe_invoke() -> String {
   gunbc_run_step_script(
     source_roots: witness_layer_roots,

dag/gunbc/fleet_converge_workflow.dag

@@ -45,6 +45,7 @@ import gunbc.ci_spec {
   gunbc_ci_fleet_converge_apply_invoke,
   gunbc_ci_spark_serving_converge_slice_invoke,
   gunbc_ci_spark_serving_observe_invoke,
+  gunbc_ci_session_demand_observe_invoke,
   gunbc_ci_spark_serving_reboot_invoke,
   gunbc_ci_spark_serving_durability_invoke,
   gunbc_ci_spark_grant_install_invoke,
@@ -62,7 +63,7 @@ data fleet_converge_enrolled_host_options: List<String> = ["srv1", "srv2", "srv3
 
 data fleet_converge_spark_target_options: List<String> = ["srv5", "srv6"]
 
-data fleet_converge_mode_options: List<String> = ["plan", "apply", "spark_linger", "spark_observe", "spark_reboot", "spark_grants", "spark_durability"]
+data fleet_converge_mode_options: List<String> = ["plan", "apply", "spark_linger", "spark_observe", "spark_reboot", "spark_grants", "spark_durability", "session_demand_observe"]
 
 data fleet_converge_host_runner_expression: String = "${{ fromJSON(format('[\"self-hosted\",\"linux\",\"arm64\",\"{0}\"]', github.event.inputs.host)) }}"
 
@@ -123,6 +124,8 @@ data fleet_converge_spark_linger_step_if: String = "github.event.inputs.mode ==
 
 data fleet_converge_spark_observe_step_if: String = "github.event.inputs.mode == 'spark_observe'"
 
+data fleet_converge_session_demand_observe_step_if: String = "github.event.inputs.mode == 'session_demand_observe'"
+
 data fleet_converge_spark_reboot_step_if: String = "github.event.inputs.mode == 'spark_reboot'"
 
 data fleet_converge_spark_durability_step_if: String = "github.event.inputs.mode == 'spark_durability'"
@@ -287,6 +290,26 @@ fn fleet_converge_spark_grants_step() -> Step {
   }
 }
 
+// READ-ONLY, AND THE RUNNER PINNING IS WHAT MAKES IT HOST-BOUND. This job already runs on a runner
+// pinned to the dispatched host, so the producer reads THAT host's own cgroups over LocalShell --
+// no ssh, no fleet key, and no new credential surface. Every read is a find or cat under
+// /sys/fs/cgroup; the step mutates nothing on the host and writes only its own receipt in the
+// workspace. Every actuating step in this job is separately mode-gated, so dispatching this mode
+// executes none of plan, apply, or any spark transition.
+fn fleet_converge_session_demand_observe_step() -> Step {
+  RunStep {
+    name: Present { value: "Session demand observation (read-only: managed slice population, per-session memory.current, parent slice counters)" },
+    id: Present { value: "session_demand_observe" },
+    run: gunbc_ci_session_demand_observe_invoke(),
+    shell: none,
+    env: none,
+    working_directory: none,
+    if_condition: Present { value: fleet_converge_session_demand_observe_step_if },
+    continue_on_error: none,
+    timeout_minutes: Present { value: gunbc_ci_spark_serving_step_timeout_minutes }
+  }
+}
+
 fn fleet_converge_spark_observe_step() -> Step {
   RunStep {
     name: Present { value: "Spark serving observation (read-only: unit state, runtime version, model tags, driver)" },
@@ -330,6 +353,7 @@ fn fleet_converge_job() -> Job {
         fleet_converge_spark_linger_step(),
         fleet_converge_spark_grants_step(),
         fleet_converge_spark_observe_step(),
+        fleet_converge_session_demand_observe_step(),
         fleet_converge_spark_reboot_step(),
         fleet_converge_spark_durability_step(),
         ci_fleet_key_agent_cleanup_step(),
@@ -368,7 +392,7 @@ data fleet_converge_workflow: Workflow = Workflow {
         },
         DispatchInput {
           name: "mode",
-          description: Present { value: "plan shows membership hunks; apply executes a prior plan artifact; spark_linger enables linger on target spark via FleetSsh typed argv from executor; spark_observe is a read-only observation of both Sparks" },
+          description: Present { value: "plan shows membership hunks; apply executes a prior plan artifact; spark_linger enables linger on target spark via FleetSsh typed argv from executor; spark_observe is a read-only observation of both Sparks; session_demand_observe is a read-only observation of the dispatched host's own session cgroup population and memory counters" },
           required: true,
           default: none,
           type: InputChoice { options: fleet_converge_mode_options },