CONVERGE-0: a merged main revision becomes observed fleet state (incl. offline catch-up)

dag/extdeps/git/git.dag

@@ -907,6 +907,28 @@ service git.Core {
     }
   }
 
+  operation LsRemoteRefInRepo {
+    input { repo: FilePath, remote: String, ref_name: String }
+    output {
+      advertised: String from "stdout"
+      exit_code: Int from "exit_code"
+      stderr: String from "stderr"
+    }
+    readonly
+    transport shell { argv: ["git", "-C", "{repo}", "ls-remote", "{remote}", "{ref_name}"] }
+  }
+
+  operation RevParseInRepo {
+    input { repo: FilePath, target: GitRef }
+    output {
+      revision: String from "stdout"
+      exit_code: Int from "exit_code"
+      stderr: String from "stderr"
+    }
+    readonly
+    transport shell { argv: ["git", "-C", "{repo}", "rev-parse", "{target}"] }
+  }
+
   operation CatFileBlobInRepo {
     input { repo: FilePath, oid: String }
     output {

dag/extdeps/linux/procfs.dag

@@ -23,13 +23,15 @@ type ProcfsPath
   | ProcSelfStatus
   | ProcSelfCgroup
   | ProcSwaps
+  | ProcSysKernelHostname
 
 fn procfs_path_literal(p: ProcfsPath) -> NonEmptyStr {
   match p {
     ProcMeminfo => "/proc/meminfo"
     ProcSelfStatus => "/proc/self/status"
     ProcSelfCgroup => "/proc/self/cgroup"
     ProcSwaps => "/proc/swaps"
+    ProcSysKernelHostname => "/proc/sys/kernel/hostname"
   }
 }
 
@@ -38,6 +40,7 @@ data procfs_paths_read_by_this_repository: List<ProcfsPath> = [
   ProcSelfStatus,
   ProcSelfCgroup,
   ProcSwaps,
+  ProcSysKernelHostname,
 ]
 
 // Provision is a total match over std.os.types.KernelFamily rather than a roster of platforms that happen to have it, so adding a kernel family to that closed enum forces an answer here instead of silently defaulting. LinuxGuestOnWindows answers TRUE deliberately and is the arm most likely to be got wrong by pattern-matching on the name: WSL2 runs a real Linux kernel and synthesizes a real procfs, so a consumer that treated it as a Windows platform would refuse observations the host can actually serve. Darwin and WindowsNt answer false — each has its own mechanism for the same facts (Mach task_info, sysctl, the Windows process API), and modeling those is a separate job this module does not attempt.

dag/extdeps/systemd/systemd.dag

@@ -1,6 +1,6 @@
 module extdeps.systemd
 
-import std.types { NonEmptyStr, String }
+import std.types { NonEmptyStr, String, Int }
 import std.types { List }
 import std.measure { ByteSize, byte_size, byte_size_count }
 import extdeps.external_authority { ExternalAuthority }
@@ -32,6 +32,9 @@ type SystemdUnitProperty
   | MemoryHigh
   | CPUWeight
   | MemoryCurrent
+  | TasksMax
+  | TasksCurrent
+  | ControlGroup
 
 fn systemd_unit_property_wire(property: SystemdUnitProperty) -> NonEmptyStr {
   match property {
@@ -40,6 +43,9 @@ fn systemd_unit_property_wire(property: SystemdUnitProperty) -> NonEmptyStr {
     MemoryHigh => "MemoryHigh" as NonEmptyStr
     CPUWeight => "CPUWeight" as NonEmptyStr
     MemoryCurrent => "MemoryCurrent" as NonEmptyStr
+    TasksMax => "TasksMax" as NonEmptyStr
+    TasksCurrent => "TasksCurrent" as NonEmptyStr
+    ControlGroup => "ControlGroup" as NonEmptyStr
   }
 }
 
@@ -55,6 +61,10 @@ data systemd_cpu_weight_property: NonEmptyStr = systemd_unit_property_wire(prope
 
 data systemd_memory_current_property: NonEmptyStr = systemd_unit_property_wire(property: MemoryCurrent)
 
+data systemd_tasks_max_property: NonEmptyStr = systemd_unit_property_wire(property: TasksMax)
+
+data systemd_tasks_current_property: NonEmptyStr = systemd_unit_property_wire(property: TasksCurrent)
+
 type SystemdCgroupMemoryLimit
   = MemoryLimitUnbounded
   | MemoryLimitBytes { bytes: ByteSize }
@@ -77,6 +87,35 @@ fn systemd_memory_limit_wire_label(lim: SystemdCgroupMemoryLimit) -> String {
   }
 }
 
+type SystemdCgroupTaskLimit
+  = TaskLimitUnbounded
+  | TaskLimitCount { count: Int }
+  | TaskLimitAbsent
+
+// parse_systemd_task_limit_show mirrors parse_systemd_memory_limit_show because a systemd
+// cgroup limit shown by `systemctl show` is one wire shape — a decimal count or the literal
+// "infinity" — differing only in the magnitude it grounds into (ByteSize vs a cardinal task
+// count). 🟡 feature:task-count-measure-carrier dissolve-on: a single limit parse
+// parameterized by magnitude (the SystemdCgroupLimit<M> consolidation) with the count arm
+// grounding into a std.measure TaskCount carrier (the HardwareThreadCount shape) instead of
+// bare Int, at which point both hand parses derive from it and the flat scalar retires.
+fn parse_systemd_task_limit_show(raw: String) -> SystemdCgroupTaskLimit {
+  if raw == "" { TaskLimitAbsent }
+  else { if raw == "infinity" { TaskLimitUnbounded }
+  else { match parse_int(s: raw) {
+    Present { value: n } => TaskLimitCount { count: n }
+    Absent => TaskLimitAbsent
+  } } }
+}
+
+fn systemd_task_limit_wire_label(lim: SystemdCgroupTaskLimit) -> String {
+  match lim {
+    TaskLimitUnbounded => "infinity"
+    TaskLimitCount { count: n } => to_string(n)
+    TaskLimitAbsent => ""
+  }
+}
+
 fn parse_systemd_unit_active_state(raw: String) -> SystemdUnitActiveState? {
   if raw == "active" { Present { value: Active } }
   else { if raw == "inactive" { Present { value: Inactive } }

dag/gunbc/ci_runner_placement.dag

@@ -73,13 +73,19 @@ fn workload_class_cpu_weight(wc: WorkloadClass) -> Int {
   }
 }
 
+// 🟡 feature:task-count-measure-carrier dissolve-on: TaskCount measure carrier — per_runner_tasks_max is a pids-controller
+// COUNT riding a bare Int while its memory siblings carry ByteSize; same dissolution as
+// the ConvergeTarget.PerSlotMemoryCap → PerSlotUnitProperty P3 axis rework (a std.measure
+// TaskCount, the HardwareThreadCount shape), which also retires the to_string() re-serialization
+// at the host_converge knob seam.
 type RunnerHostDeployment {
   identity: HostIdentity
   runner_count: Int
   runner_slice_cap: ByteSize
   per_runner_memory_cap: ByteSize
   per_runner_memory_swap_cap: ByteSize
   per_runner_memory_high_cap: ByteSize
+  per_runner_tasks_max: Int
   build_tokens: HardwareThreadCount
   runner: RunnerSpec
   workload_class: WorkloadClass
@@ -189,6 +195,7 @@ fn conservation_host_deployment(offer: ComputeOffer, nproc: HardwareThreadCount)
     per_runner_memory_cap: gunbc_ci_runner_cgroup_memory_cap,
     per_runner_memory_swap_cap: gunbc_runner_slot_desired().memory_swap_max,
     per_runner_memory_high_cap: gunbc_runner_slot_desired().memory_high,
+    per_runner_tasks_max: gunbc_runner_slot_desired().tasks_max,
     build_tokens: nproc,
     runner: runner_spec_from_offer(offer: offer),
     workload_class: BatchLatencyTolerant,

dag/gunbc/ci_spec.dag

@@ -46,6 +46,7 @@ import gunbc.heal_push_plan {
 }
 import gunbc.ci_failure_class { infra_retry_grep_alternation }
 import gunbc.ci_deploy_access { DeployAccess, ci_deploy_srv1_access }
+import gunbc.fleet_main_revision { fleet_desired_ref_name }
 import gunbc.ci_deploy_target_host { gunbc_ci_deploy_srv1_runner_spec }
 import gunbc.merge_admission_produce {
   ci_repo_root_shell,
@@ -503,6 +504,176 @@ fn gunbc_ci_floor_run(spec: CiSpec) -> String {
   gunbc_ci_floor_only_script(spec: spec)
 }
 
+// Read-only srv1 tasks pre-apply observation (operator ruling 2026-08-09 on PR 8045;
+// manager approval msg_03ec1f78): produces the ONE exact pre-apply receipt (effective
+// TasksMax, pids.max, pids.current, ancestor limit, observer slot membership) via the
+// typed read seams only. workflow_dispatch-gated at the job, so it never rides PR or
+// push events; the printed return value IS the typed receipt.
+data gunbc_ci_srv1_tasks_observation_entry: String = "dag/gunbc/srv1_tasks_preapply_observation.dag"
+data gunbc_ci_srv1_tasks_observation_function: String = "observe_srv1_tasks_preapply_wet"
+
+// Main→desired-ref advancement (CONVERGE-0 loop leg, operator directive 2026-08-09):
+// on a floor-green main push (needs: [ci] IS the floor-green admission — this job exists
+// only downstream of the required check succeeding in the same run), advance
+// refs/fleet/desired to the merged revision by CAS: --force-with-lease names the exact
+// observed prior (gunbc.fleet_main_revision admit_fleet_desired_advance's from side), so
+// a race with a concurrent advance refuses rather than overwriting; ref creation is the
+// from=none arm. Hosts follow the ref; no host is contacted here.
+data gunbc_ci_fleet_reach_probe_shell_emit_scaffold: Disposition = Scaffold {
+  dissolves_to: SingleAuthority,
+  bind: DeclarationRef {
+    module_path: "gunbc.ci_spec",
+    decl_name: "gunbc_ci_fleet_reach_probe_invoke",
+    field: WholeDeclaration
+  }
+}
+
+data gunbc_ci_fleet_reach_probe_shell_emit_dissolution_trigger: String = "🟡 dissolve-on: gunbc_ci_fleet_reach_probe_invoke — concat-built foreign-executor (GitHub Actions run:) shell invoking gunbc run on gunbc.host_reach_identity_probe probe_fleet_reach_wet plus a cat of the typed receipt file; DISSOLVES WHEN bash-emit (#5828 / ROADMAP 6-shell-slice0 / shell→intent Phase 2) binds the probe step to orchestration emit over the typed gunbc.WitnessBin.Run route"
+
+fn gunbc_ci_fleet_reach_probe_invoke() -> String {
+  concat(
+    concat("# ", concat(gunbc_ci_fleet_reach_probe_shell_emit_dissolution_trigger, "\n")),
+    concat(
+      concat(
+        concat("ROOT=", ci_repo_root_shell(), "\n"),
+        gunbc_run_shell(
+          source_roots: witness_layer_roots,
+          entry: "dag/gunbc/host_reach_identity_probe.dag",
+          function: "probe_fleet_reach_wet",
+          rooted: true,
+          claim_run: false
+        )
+      ),
+      "\ncat \"$ROOT/target/fleet-reach-probe-receipt.txt\""
+    )
+  )
+}
+
+// Fleet timer self-install bootstrap (operator verdict join 2, parent verification
+// msg_8d84ae4f): the LIVE invocation wire for fleet_timer_self_install_wet — without
+// this job the self-install route had zero callers, a disconnected mechanism. The job
+// runs on a fleet runner and installs the timer on THE MACHINE IT RUNS ON (self-install:
+// no remote host is contacted); dispatch it once per bootstrap host, thereafter the
+// installed timer sustains itself.
+data gunbc_ci_fleet_timer_install_shell_emit_scaffold: Disposition = Scaffold {
+  dissolves_to: SingleAuthority,
+  bind: DeclarationRef {
+    module_path: "gunbc.ci_spec",
+    decl_name: "gunbc_ci_fleet_timer_install_invoke",
+    field: WholeDeclaration
+  }
+}
+
+data gunbc_ci_fleet_timer_install_shell_emit_dissolution_trigger: String = "🟡 dissolve-on: gunbc_ci_fleet_timer_install_invoke — concat-built foreign-executor (GitHub Actions run:) shell invoking gunbc run on gunbc.fleet_converge_timer_install fleet_timer_self_install_wet; DISSOLVES WHEN bash-emit (#5828 / ROADMAP 6-shell-slice0 / shell→intent Phase 2) binds the step to orchestration emit over the typed gunbc.WitnessBin.Run route"
+
+fn gunbc_ci_fleet_timer_install_invoke() -> String {
+  concat(
+    concat("# ", concat(gunbc_ci_fleet_timer_install_shell_emit_dissolution_trigger, "\n")),
+    concat(
+      concat("ROOT=", ci_repo_root_shell(), "\n"),
+      gunbc_run_shell(
+        source_roots: witness_layer_roots,
+        entry: "dag/gunbc/fleet_converge_timer_install.dag",
+        function: "fleet_timer_self_install_wet",
+        rooted: true,
+        claim_run: false
+      )
+    )
+  )
+}
+
+// Fleet convergence collection (operator verdict join 5, parent verification
+// msg_8d84ae4f): the LIVE invocation wire for collect_fleet_convergence_wet — runs on
+// main pushes downstream of the required ci check, from srv1 (inside the fleet, over
+// the typed SSH probe seam), so the collector->fold composition executes on a real
+// cadence and its collect receipt is printed into the job log.
+data gunbc_ci_collect_fleet_convergence_shell_emit_scaffold: Disposition = Scaffold {
+  dissolves_to: SingleAuthority,
+  bind: DeclarationRef {
+    module_path: "gunbc.ci_spec",
+    decl_name: "gunbc_ci_collect_fleet_convergence_invoke",
+    field: WholeDeclaration
+  }
+}
+
+data gunbc_ci_collect_fleet_convergence_shell_emit_dissolution_trigger: String = "🟡 dissolve-on: gunbc_ci_collect_fleet_convergence_invoke — concat-built foreign-executor (GitHub Actions run:) shell invoking gunbc run on gunbc.fleet_receipt_collector collect_fleet_convergence_wet plus a cat of the typed collect receipt; DISSOLVES WHEN bash-emit (#5828 / ROADMAP 6-shell-slice0 / shell→intent Phase 2) binds the step to orchestration emit over the typed gunbc.WitnessBin.Run route"
+
+fn gunbc_ci_collect_fleet_convergence_invoke() -> String {
+  concat(
+    concat("# ", concat(gunbc_ci_collect_fleet_convergence_shell_emit_dissolution_trigger, "\n")),
+    concat(
+      concat(
+        concat("ROOT=", ci_repo_root_shell(), "\n"),
+        gunbc_run_shell(
+          source_roots: witness_layer_roots,
+          entry: "dag/gunbc/fleet_receipt_collector.dag",
+          function: "collect_fleet_convergence_wet",
+          rooted: true,
+          claim_run: false
+        )
+      ),
+      "\ncat \"$ROOT/target/fleet-converge-collect-receipt.txt\""
+    )
+  )
+}
+
+data gunbc_ci_advance_fleet_desired_shell_emit_scaffold: Disposition = Scaffold {
+  dissolves_to: SingleAuthority,
+  bind: DeclarationRef {
+    module_path: "gunbc.ci_spec",
+    decl_name: "gunbc_ci_advance_fleet_desired_script",
+    field: WholeDeclaration
+  }
+}
+
+data gunbc_ci_advance_fleet_desired_shell_emit_dissolution_trigger: String = "🟡 dissolve-on: gunbc_ci_advance_fleet_desired_script — concat-built foreign-executor (GitHub Actions run:) CAS ref advance (git ls-remote + push --force-with-lease onto gunbc.fleet_main_revision fleet_desired_ref_name); DISSOLVES WHEN the SCM ref-update realization lands as the typed git.Core CAS service op (membership-diff spine classify_change SCM-ref arm) invoked via host_effect_apply"
+
+fn gunbc_ci_advance_fleet_desired_script() -> String {
+  join([
+    "# ", gunbc_ci_advance_fleet_desired_shell_emit_dissolution_trigger, "\n",
+    "set -eu\n",
+    "DESIRED_REF=", fleet_desired_ref_name, "\n",
+    "CURRENT=$(git ls-remote origin \"$DESIRED_REF\" | cut -f1)\n",
+    "if [ -z \"$CURRENT\" ]; then\n",
+    "git push origin \"$GITHUB_SHA:$DESIRED_REF\"\n",
+    "echo \"[fleet-desired] created $DESIRED_REF -> $GITHUB_SHA\"\n",
+    "else\n",
+    "git push --force-with-lease=\"$DESIRED_REF:$CURRENT\" origin \"$GITHUB_SHA:$DESIRED_REF\"\n",
+    "echo \"[fleet-desired] advanced $DESIRED_REF: $CURRENT -> $GITHUB_SHA (CAS expected-prior held)\"\n",
+    "fi\n",
+  ], "")
+}
+
+data gunbc_ci_srv1_tasks_observation_shell_emit_scaffold: Disposition = Scaffold {
+  dissolves_to: SingleAuthority,
+  bind: DeclarationRef {
+    module_path: "gunbc.ci_spec",
+    decl_name: "gunbc_ci_srv1_tasks_observation_invoke",
+    field: WholeDeclaration
+  }
+}
+
+data gunbc_ci_srv1_tasks_observation_shell_emit_dissolution_trigger: String = "🟡 dissolve-on: gunbc_ci_srv1_tasks_observation_invoke — feature:task-count-measure-carrier sibling transport; concat-built foreign-executor (GitHub Actions run:) shell invoking gunbc run on gunbc.srv1_tasks_preapply_observation observe_srv1_tasks_preapply_wet plus a cat of the typed receipt file; DISSOLVES WHEN bash-emit (#5828 / ROADMAP 6-shell-slice0 / shell→intent Phase 2) binds the observation step to orchestration emit over the typed gunbc.WitnessBin.Run route without a medium-as-string concat scaffold"
+
+fn gunbc_ci_srv1_tasks_observation_invoke() -> String {
+  concat(
+    concat("# ", concat(gunbc_ci_srv1_tasks_observation_shell_emit_dissolution_trigger, "\n")),
+    concat(
+    concat(
+      concat("ROOT=", ci_repo_root_shell(), "\n"),
+      gunbc_run_shell(
+        source_roots: witness_layer_roots,
+        entry: gunbc_ci_srv1_tasks_observation_entry,
+        function: gunbc_ci_srv1_tasks_observation_function,
+        rooted: true,
+        claim_run: false
+      )
+    ),
+    "\ncat \"$ROOT/target/srv1-tasks-preapply-receipt.txt\""
+  )
+  )
+}
+
 fn gunbc_ci_deploy_invoke(stage: DeployStage) -> String {
   concat(
     concat("ROOT=", ci_repo_root_shell(), "\n"),