From f36cd2963f673782ee42e4fd2d4f667f6eb2ab8c Mon Sep 17 00:00:00 2001
From: Nicks Hermes <nicks-hermes@invalid.example>
Date: Sat, 15 Aug 2026 20:28:16 -0500
Subject: [PATCH 1/8] feat: record bounded local capacity telemetry

---
 docs/CAPACITY-TELEMETRY.md           |  23 +++++
 docs/README.md                       |   2 +-
 scripts/desired_state.py             |   1 +
 scripts/health.py                    | 130 +++++++++++++++++++++++++++
 scripts/healthcheck.sh               |   2 +-
 scripts/install-worker-controller.sh |   2 +-
 scripts/test_health.py               |  44 +++++++++
 7 files changed, 201 insertions(+), 3 deletions(-)
 create mode 100644 docs/CAPACITY-TELEMETRY.md

diff --git a/docs/CAPACITY-TELEMETRY.md b/docs/CAPACITY-TELEMETRY.md
new file mode 100644
index 0000000..874cc37
--- /dev/null
+++ b/docs/CAPACITY-TELEMETRY.md
@@ -0,0 +1,23 @@
+# Local capacity telemetry
+
+The existing five-minute health timer records a capacity sample only while at least one managed runner is active. Samples remain on the controller host at `/var/lib/ci-fleet/capacity/samples.jsonl`; they are never added to status-reporting or heartbeat payloads.
+
+Each sample contains only:
+
+- timestamp and logical controller/pool ID;
+- host CPU, memory, swap, disk-byte, and inode counters;
+- anonymous per-runner CPU percentage and memory use/limit.
+
+Container IDs, names, repositories, jobs, logs, environment variables, source, network counters, and credentials are not stored. The history directory is mode `0700`, the file is mode `0600`, malformed records are ignored, records older than eight days are discarded, and at most 2,500 samples are retained. Uninstall removes this fleet-owned local history.
+
+## Weekly report
+
+Run on the controller host; reads local state only and changes nothing:
+
+```bash
+sudo /opt/ci-fleet/manager/current/scripts/health.py capacity-report
+```
+
+The JSON report groups the preceding seven days by logical pool and gives nearest-rank p50/p95 values for every observed host and runner metric, plus sample and anonymous runner-observation counts. An empty report means no managed runner was observed during the period.
+
+Review the report before changing runner count, per-runner resources, or pool budget. A capacity change still requires its own reviewed private-configuration change; this public repository does not create or modify that external change.
diff --git a/docs/README.md b/docs/README.md
index 90b0499..8107a48 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -22,7 +22,7 @@ New operator? Use [Operator workflows](OPERATOR-WORKFLOWS.md) to select a suppor
 | Make a project compliant | [Project CI standard](PROJECT-STANDARD.md) and [compliance checklist](COMPLIANCE-CHECKLIST.md) |
 | Split tests across parallel workers | [Project CI standard](PROJECT-STANDARD.md) and the [parallel workflow example](../examples/workflows/parallel-ci.yml.example) |
 | Configure automatic updates and cleanup | [Host maintenance](HOST-MAINTENANCE.md) |
-| Monitor hosts and detect missed reports | [Fleet health monitoring](HEALTH-MONITORING.md) and [authenticated status reporting](STATUS-REPORTING.md) |
+| Monitor hosts and detect missed reports | [Fleet health monitoring](HEALTH-MONITORING.md), [local capacity telemetry](CAPACITY-TELEMETRY.md), and [authenticated status reporting](STATUS-REPORTING.md) |
 | Handle GitHub App, workflow, or deployment secrets | [Secrets model](SECRETS.md) and [security policy](../SECURITY.md) |
 | Review accepted implementation scope | [Design decisions](DESIGN-DECISIONS.md) |
 | Run private CI or deployment for a public project | [Public projects, private delivery, and private configuration](PUBLIC-PRIVATE-CONFIGURATION.md) |
diff --git a/scripts/desired_state.py b/scripts/desired_state.py
index f19af22..ea16beb 100755
--- a/scripts/desired_state.py
+++ b/scripts/desired_state.py
@@ -186,6 +186,7 @@ def build_rendered_env(
         "CI_FLEET_LABELS": ",".join(pool["routing_labels"]),
         "CI_FLEET_MAX_RUNNERS": str(effective_max),
         "CI_FLEET_MIN_RUNNERS": str(controller["min_runners"] if state == "active" else 0),
+        "CI_FLEET_POOL": controller["pool"],
         "CI_FLEET_RUNNER_CPUS": str(controller["runner_resources"]["cpu_cores"]),
         "CI_FLEET_RUNNER_GROUP": pool["runner_group"],
         "CI_FLEET_RUNNER_IMAGE": f"ci-fleet-runner:{short_commit}",
diff --git a/scripts/health.py b/scripts/health.py
index f47b9e1..06907f1 100644
--- a/scripts/health.py
+++ b/scripts/health.py
@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import argparse
+import math
 import http.client
 import json
 import os
@@ -525,6 +526,7 @@ def collect_snapshot(values: dict[str, str], *, root: Path = Path("/"), run: Run
     stale["build_cache"] = _count(run, ["docker", "buildx", "du", "--filter", "until=168h", "--format", "json"]) if docker_ok else 0
     return {
         "controller_id": instance,
+        "pool_id": values.get("CI_FLEET_POOL", instance),
         "desired_state": values.get("CI_FLEET_CONTROLLER_STATE", "active"),
         "disks": {"root": _disk(str(root)), "docker": _disk(str(root / docker_root.lstrip("/")))},
         "cpu": _cpu(root),
@@ -584,6 +586,117 @@ def _write_report(path: Path, report: dict[str, Any]) -> None:
     temporary.replace(path)
 
 
+CAPACITY_RETENTION_SECONDS = 8 * 24 * 60 * 60
+CAPACITY_MAX_SAMPLES = 2500
+
+
+def _quantity_bytes(value: str) -> int:
+    match = re.fullmatch(r"([0-9]+(?:\.[0-9]+)?)\s*([KMGT]?i?B)", value.strip(), re.IGNORECASE)
+    if not match:
+        raise ValueError("invalid Docker memory quantity")
+    units = {"b": 1, "kb": 1000, "mb": 1000**2, "gb": 1000**3, "tb": 1000**4,
+             "kib": 1024, "mib": 1024**2, "gib": 1024**3, "tib": 1024**4}
+    return round(float(match.group(1)) * units[match.group(2).lower()])
+
+
+def _runner_capacity(run: Runner, instance: str) -> list[dict[str, int | float]]:
+    result = run(["docker", "ps", "-q", "--filter", "label=io.randomdevelopment.ci-fleet.managed=true",
+                  "--filter", "label=io.randomdevelopment.ci-fleet.kind=runner",
+                  "--filter", f"label=io.randomdevelopment.ci-fleet.instance={instance}"])
+    ids = [value for value in result.stdout.splitlines() if re.fullmatch(r"[0-9a-f]{12,64}", value)] if result.returncode == 0 else []
+    if not ids:
+        return []
+    result = run(["docker", "stats", "--no-stream", "--format", "{{.CPUPerc}}\t{{.MemUsage}}", *ids])
+    metrics = []
+    for line in result.stdout.splitlines() if result.returncode == 0 else []:
+        try:
+            cpu, memory = line.split("\t", 1)
+            used, limit = memory.split("/", 1)
+            metrics.append({"cpu_percent": float(cpu.removesuffix("%")), "memory_used_bytes": _quantity_bytes(used), "memory_limit_bytes": _quantity_bytes(limit)})
+        except ValueError:
+            continue
+    return metrics
+
+
+def record_capacity(path: Path, snapshot: dict[str, Any], *, run: Runner = _run, now: int | None = None) -> bool:
+    if snapshot.get("runners", {}).get("current", 0) < 1:
+        return False
+    timestamp = int(time.time() if now is None else now)
+    memory = snapshot["memory"]
+    sample = {
+        "timestamp": timestamp, "pool": snapshot.get("pool_id", snapshot["controller_id"]),
+        "host": {
+            "cpu_percent": snapshot["cpu"]["used_percent"],
+            "memory_used_bytes": memory["total_bytes"] - memory["available_bytes"],
+            "swap_used_bytes": snapshot["swap"]["used_bytes"],
+            "disk_used_bytes": {name: value["used_bytes"] for name, value in snapshot["disks"].items()},
+            "inode_used": {name: value["inode_used"] for name, value in snapshot["disks"].items()},
+        },
+        "runners": _runner_capacity(run, snapshot["controller_id"]),
+    }
+    expected_owner = os.getuid() if os.environ.get("CI_FLEET_TESTING") == "1" else 0
+    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+    os.chmod(path.parent, 0o700)
+    retained = []
+    if path.exists():
+        info = path.lstat()
+        if not stat.S_ISREG(info.st_mode) or info.st_uid != expected_owner or stat.S_IMODE(info.st_mode) != 0o600:
+            raise ValueError("capacity history must be root-owned mode 0600")
+        for line in path.read_text().splitlines():
+            try:
+                previous = json.loads(line)
+                if timestamp - CAPACITY_RETENTION_SECONDS <= previous["timestamp"] <= timestamp:
+                    retained.append(previous)
+            except (KeyError, TypeError, ValueError, json.JSONDecodeError):
+                continue
+    retained = (retained + [sample])[-CAPACITY_MAX_SAMPLES:]
+    temporary = path.with_suffix(".tmp")
+    temporary.write_text("".join(json.dumps(value, separators=(",", ":"), sort_keys=True) + "\n" for value in retained))
+    os.chmod(temporary, 0o600)
+    temporary.replace(path)
+    return True
+
+
+def _percentiles(values: list[int | float]) -> dict[str, int | float] | None:
+    if not values:
+        return None
+    ordered = sorted(values)
+    return {name: ordered[max(0, math.ceil(len(ordered) * percentile) - 1)] for name, percentile in (("p50", 0.50), ("p95", 0.95))}
+
+
+def capacity_report(path: Path, *, now: int | None = None) -> dict[str, Any]:
+    timestamp = int(time.time() if now is None else now)
+    if path.exists():
+        info = path.lstat()
+        expected_owner = os.getuid() if os.environ.get("CI_FLEET_TESTING") == "1" else 0
+        if not stat.S_ISREG(info.st_mode) or info.st_uid != expected_owner or stat.S_IMODE(info.st_mode) != 0o600:
+            raise ValueError("capacity history must be root-owned mode 0600")
+    pools: dict[str, dict[str, Any]] = {}
+    for line in path.read_text().splitlines() if path.exists() else []:
+        try:
+            sample = json.loads(line)
+            if not timestamp - 7 * 24 * 60 * 60 <= sample["timestamp"] <= timestamp:
+                continue
+            pool = pools.setdefault(sample["pool"], {"samples": 0, "runner_observations": 0, "values": {}})
+            pool["samples"] += 1
+            values = pool["values"]
+            host = sample["host"]
+            for name in ("cpu_percent", "memory_used_bytes", "swap_used_bytes"):
+                values.setdefault(f"host_{name}", []).append(host[name])
+            for kind in ("disk_used_bytes", "inode_used"):
+                for name, value in host[kind].items():
+                    values.setdefault(f"host_{kind}_{name}", []).append(value)
+            for runner in sample["runners"]:
+                pool["runner_observations"] += 1
+                for name, value in runner.items():
+                    values.setdefault(f"runner_{name}", []).append(value)
+        except (KeyError, TypeError, ValueError, json.JSONDecodeError):
+            continue
+    for pool in pools.values():
+        pool["metrics"] = {name: _percentiles(values) for name, values in sorted(pool.pop("values").items())}
+    return {"schema_version": 1, "generated_at": timestamp, "period_seconds": 7 * 24 * 60 * 60, "pools": pools}
+
+
 def _send_status(
     values: dict[str, str],
     report: dict[str, Any],
@@ -688,6 +801,14 @@ def _local(args: argparse.Namespace) -> int:
     report = evaluate(snapshot, thresholds)
     now = int(time.time())
     report["timestamp"] = now
+    try:
+        capacity_history = getattr(args, "capacity_history", None)
+        if capacity_history is not None:
+            record_capacity(capacity_history, snapshot, now=now)
+    except (OSError, UnicodeError, ValueError):
+        report["checks"].append({"id": "capacity_telemetry", "status": "warning"})
+        if report["exit_code"] == 0:
+            report["status"], report["exit_code"] = "warning", 1
     delivery = 0
     if environment.get("CI_FLEET_HEALTH_SUPPRESS_DELIVERY") != "1":
         if config_invalid:
@@ -721,6 +842,11 @@ def _heartbeats(args: argparse.Namespace) -> int:
     return int(report["exit_code"])
 
 
+def _capacity(args: argparse.Namespace) -> int:
+    print(json.dumps(capacity_report(args.history), indent=2, sort_keys=True))
+    return 0
+
+
 def main() -> int:
     parser = argparse.ArgumentParser(description="Redacted ci-fleet host health")
     commands = parser.add_subparsers(dest="command", required=True)
@@ -728,6 +854,7 @@ def main() -> int:
     local.add_argument("--json", action="store_true")
     local.add_argument("--monitoring-config", type=Path, default=Path("/etc/ci-fleet/monitoring.env"))
     local.add_argument("--output", type=Path, default=Path("/var/lib/ci-fleet/health/latest.json"))
+    local.add_argument("--capacity-history", type=Path, default=Path("/var/lib/ci-fleet/capacity/samples.jsonl"))
     local.set_defaults(handler=_local)
     heartbeats = commands.add_parser("heartbeats")
     heartbeats.add_argument("--config", type=Path, required=True)
@@ -735,6 +862,9 @@ def main() -> int:
     heartbeats.add_argument("--grace-seconds", type=int, default=900)
     heartbeats.add_argument("--json", action="store_true")
     heartbeats.set_defaults(handler=_heartbeats)
+    capacity = commands.add_parser("capacity-report")
+    capacity.add_argument("--history", type=Path, default=Path("/var/lib/ci-fleet/capacity/samples.jsonl"))
+    capacity.set_defaults(handler=_capacity)
     args = parser.parse_args()
     try:
         return args.handler(args)
diff --git a/scripts/healthcheck.sh b/scripts/healthcheck.sh
index c06bf63..89cf1a4 100755
--- a/scripts/healthcheck.sh
+++ b/scripts/healthcheck.sh
@@ -5,7 +5,7 @@ environment=/etc/ci-fleet/ci-fleet.env
 args=(local)
 if [[ ${CI_FLEET_TESTING:-0} == 1 && -n ${CI_FLEET_ROOT_PREFIX:-} ]]; then
   environment="$CI_FLEET_ROOT_PREFIX/etc/ci-fleet/ci-fleet.env"
-  args+=(--monitoring-config "$CI_FLEET_ROOT_PREFIX/etc/ci-fleet/monitoring.env" --output "$CI_FLEET_ROOT_PREFIX/var/lib/ci-fleet/health/latest.json")
+  args+=(--monitoring-config "$CI_FLEET_ROOT_PREFIX/etc/ci-fleet/monitoring.env" --output "$CI_FLEET_ROOT_PREFIX/var/lib/ci-fleet/health/latest.json" --capacity-history "$CI_FLEET_ROOT_PREFIX/var/lib/ci-fleet/capacity/samples.jsonl")
 fi
 if [[ -r $environment ]]; then
   set -a
diff --git a/scripts/install-worker-controller.sh b/scripts/install-worker-controller.sh
index ad15656..ee988b2 100755
--- a/scripts/install-worker-controller.sh
+++ b/scripts/install-worker-controller.sh
@@ -1090,7 +1090,7 @@ perform_uninstall() {
   fi
   remove_systemd_units
   rm -f "$current_link" "$rendered_env" "$state_file"
-  rm -rf -- "$state_root/health"
+  rm -rf -- "$state_root/health" "$state_root/capacity"
   rm -f "$manager_current"
   transaction_active=false
   note "UNINSTALL_OK host_config_preserved=$host_config secrets_preserved=$etc_dir/secrets"
diff --git a/scripts/test_health.py b/scripts/test_health.py
index ed4cf56..47af012 100644
--- a/scripts/test_health.py
+++ b/scripts/test_health.py
@@ -127,6 +127,50 @@ def test_malformed_reconciliation_state_is_observable(self) -> None:
             self.assertEqual(report["configuration"], {"desired_commit": "", "applied_commit": ""})
             self.assertEqual(report["error"]["code"], "reconciliation_invalid")
 
+    def test_capacity_history_is_resource_only_bounded_and_aggregated(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            history = Path(directory) / "capacity" / "samples.jsonl"
+            snapshot = healthy_snapshot()
+            snapshot.update({
+                "cpu": {"logical": 8, "used_percent": 10},
+                "memory": {"total_bytes": 1000, "available_bytes": 600},
+                "swap": {"total_bytes": 100, "used_bytes": 5},
+                "runners": {"current": 1, "busy": 1, "maximum": 2},
+                "secret": "must-not-be-recorded",
+            })
+            for value in snapshot["disks"].values():
+                value.update(used_bytes=200, inode_used=20)
+            runner_cpu = [12.5, 90.0]
+
+            def run(args):
+                output = "a" * 12 + "\n" if args[:3] == ["docker", "ps", "-q"] else f"{runner_cpu.pop(0)}%\t256MiB / 2GiB\n"
+                return health.subprocess.CompletedProcess(args, 0, output, "")
+
+            previous_testing = os.environ.get("CI_FLEET_TESTING")
+            os.environ["CI_FLEET_TESTING"] = "1"
+            try:
+                history.parent.mkdir()
+                history.write_text('{"timestamp":999999}\n' * health.CAPACITY_MAX_SAMPLES)
+                history.chmod(0o600)
+                self.assertTrue(health.record_capacity(history, snapshot, run=run, now=1_000_000))
+                snapshot["cpu"]["used_percent"] = 20
+                self.assertTrue(health.record_capacity(history, snapshot, run=run, now=1_000_300))
+                self.assertEqual(len(history.read_text().splitlines()), health.CAPACITY_MAX_SAMPLES)
+                self.assertEqual(history.stat().st_mode & 0o777, 0o600)
+                self.assertNotIn("must-not-be-recorded", history.read_text())
+                report = health.capacity_report(history, now=1_000_300)
+                pool = report["pools"]["example-ci-01"]
+                self.assertEqual((pool["samples"], pool["runner_observations"]), (2, 2))
+                self.assertEqual(pool["metrics"]["host_cpu_percent"], {"p50": 10, "p95": 20})
+                self.assertEqual(pool["metrics"]["runner_cpu_percent"], {"p50": 12.5, "p95": 90.0})
+                snapshot["runners"]["current"] = 0
+                self.assertFalse(health.record_capacity(history, snapshot, run=run, now=1_000_900))
+            finally:
+                if previous_testing is None:
+                    os.environ.pop("CI_FLEET_TESTING", None)
+                else:
+                    os.environ["CI_FLEET_TESTING"] = previous_testing
+
     def test_external_heartbeats_detect_missing_and_stale_active_hosts(self) -> None:
         controllers = {
             "fresh": {"state": "active", "lifecycle": "stable"},

From d3ab11b15da4e9b3f6d0e38ffe41ec83fbb01466 Mon Sep 17 00:00:00 2001
From: Nicks Hermes <nicks-hermes@invalid.example>
Date: Sat, 15 Aug 2026 21:19:20 -0500
Subject: [PATCH 2/8] fix: harden capacity sampling and retention

---
 docs/CAPACITY-TELEMETRY.md             |   8 +-
 host/systemd/ci-fleet-capacity.service |  21 +++
 host/systemd/ci-fleet-capacity.timer   |  11 ++
 scripts/capacity-sample.sh             |  16 ++
 scripts/health.py                      | 215 +++++++++++++++++--------
 scripts/healthcheck.sh                 |   2 +-
 scripts/install-worker-controller.sh   |  30 +++-
 scripts/test_health.py                 |  81 ++++++++--
 8 files changed, 303 insertions(+), 81 deletions(-)
 create mode 100644 host/systemd/ci-fleet-capacity.service
 create mode 100644 host/systemd/ci-fleet-capacity.timer
 create mode 100755 scripts/capacity-sample.sh

diff --git a/docs/CAPACITY-TELEMETRY.md b/docs/CAPACITY-TELEMETRY.md
index 874cc37..ac72738 100644
--- a/docs/CAPACITY-TELEMETRY.md
+++ b/docs/CAPACITY-TELEMETRY.md
@@ -1,21 +1,21 @@
 # Local capacity telemetry
 
-The existing five-minute health timer records a capacity sample only while at least one managed runner is active. Samples remain on the controller host at `/var/lib/ci-fleet/capacity/samples.jsonl`; they are never added to status-reporting or heartbeat payloads.
+The dedicated capacity timer checks every 30 seconds and records a sample only while at least one managed runner is active. This interval captures normal jobs shorter than the five-minute health period without increasing heartbeat or status-report traffic. Samples remain on the controller host at `/var/lib/ci-fleet/capacity/samples.jsonl`; they are never added to status-reporting or heartbeat payloads.
 
 Each sample contains only:
 
 - timestamp and logical controller/pool ID;
-- host CPU, memory, swap, disk-byte, and inode counters;
+- interval host CPU utilization plus memory, swap, disk-byte, and inode counters;
 - anonymous per-runner CPU percentage and memory use/limit.
 
-Container IDs, names, repositories, jobs, logs, environment variables, source, network counters, and credentials are not stored. The history directory is mode `0700`, the file is mode `0600`, malformed records are ignored, records older than eight days are discarded, and at most 2,500 samples are retained. Uninstall removes this fleet-owned local history.
+Container IDs, names, repositories, jobs, logs, environment variables, source, network counters, and credentials are not stored. The history directory must be a root-owned, non-symlink mode-`0700` directory and the file must be mode `0600`. Malformed records are ignored atomically, records older than eight days are discarded even while the pool is idle, and at most 2,500 samples are retained. Uninstall removes this fleet-owned local history. Runner-stat failures fail the capacity service and appear in the normal health report instead of being recorded as zero observations.
 
 ## Weekly report
 
 Run on the controller host; reads local state only and changes nothing:
 
 ```bash
-sudo /opt/ci-fleet/manager/current/scripts/health.py capacity-report
+sudo /opt/ci-fleet/current/scripts/health.py capacity-report
 ```
 
 The JSON report groups the preceding seven days by logical pool and gives nearest-rank p50/p95 values for every observed host and runner metric, plus sample and anonymous runner-observation counts. An empty report means no managed runner was observed during the period.
diff --git a/host/systemd/ci-fleet-capacity.service b/host/systemd/ci-fleet-capacity.service
new file mode 100644
index 0000000..7904231
--- /dev/null
+++ b/host/systemd/ci-fleet-capacity.service
@@ -0,0 +1,21 @@
+[Unit]
+Description=Sample local ci-fleet capacity
+After=docker.service
+Requires=docker.service
+ConditionPathExists=/etc/ci-fleet/ci-fleet.env
+
+[Service]
+Type=oneshot
+ExecStart=/opt/ci-fleet/current/scripts/capacity-sample.sh
+User=root
+Group=root
+NoNewPrivileges=yes
+PrivateTmp=yes
+ProtectHome=yes
+ProtectSystem=strict
+ReadWritePaths=/var/lib/ci-fleet
+LockPersonality=yes
+MemoryDenyWriteExecute=yes
+RestrictAddressFamilies=AF_UNIX
+SystemCallArchitectures=native
+UMask=0077
diff --git a/host/systemd/ci-fleet-capacity.timer b/host/systemd/ci-fleet-capacity.timer
new file mode 100644
index 0000000..4af5d60
--- /dev/null
+++ b/host/systemd/ci-fleet-capacity.timer
@@ -0,0 +1,11 @@
+[Unit]
+Description=Sample active ci-fleet capacity every 30 seconds
+
+[Timer]
+OnActiveSec=15s
+OnUnitActiveSec=30s
+AccuracySec=1s
+Unit=ci-fleet-capacity.service
+
+[Install]
+WantedBy=timers.target
diff --git a/scripts/capacity-sample.sh b/scripts/capacity-sample.sh
new file mode 100755
index 0000000..a8e1461
--- /dev/null
+++ b/scripts/capacity-sample.sh
@@ -0,0 +1,16 @@
+#!/usr/bin/env bash
+set -Eeuo pipefail
+
+script_dir=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+environment=/etc/ci-fleet/ci-fleet.env
+args=(capacity-sample)
+if [[ ${CI_FLEET_TESTING:-0} == 1 && -n ${CI_FLEET_ROOT_PREFIX:-} ]]; then
+  environment="$CI_FLEET_ROOT_PREFIX/etc/ci-fleet/ci-fleet.env"
+  args+=(--root "$CI_FLEET_ROOT_PREFIX" --history "$CI_FLEET_ROOT_PREFIX/var/lib/ci-fleet/capacity/samples.jsonl")
+fi
+[[ -r $environment ]] || { printf 'CRITICAL capacity_configuration_missing\n' >&2; exit 2; }
+set -a
+# shellcheck disable=SC1090
+. "$environment"
+set +a
+exec python3 "$script_dir/health.py" "${args[@]}"
diff --git a/scripts/health.py b/scripts/health.py
index 06907f1..305bbe0 100644
--- a/scripts/health.py
+++ b/scripts/health.py
@@ -384,16 +384,15 @@ def _memory(root: Path) -> tuple[int, int]:
 
 
 def _cpu(root: Path) -> dict[str, float | int]:
-    # ponytail: cumulative boot-average CPU; persist the prior sample if interval utilization becomes necessary.
     try:
         fields = next(line for line in (root / "proc/stat").read_text().splitlines() if line.startswith("cpu ")).split()[1:]
         counters = [int(value) for value in fields]
         total = sum(counters[:8])
-        used = total - counters[3]
-        percent = round(100 * used / max(total, 1), 1)
+        idle = counters[3]
+        percent = round(100 * (total - idle) / max(total, 1), 1)
     except (OSError, ValueError, IndexError, StopIteration):
-        percent = 0.0
-    return {"logical": max(os.cpu_count() or 1, 1), "used_percent": percent}
+        total, idle, percent = 0, 0, 0.0
+    return {"logical": max(os.cpu_count() or 1, 1), "used_percent": percent, "total_ticks": total, "idle_ticks": idle}
 
 
 def _boot_time(root: Path) -> int:
@@ -498,7 +497,7 @@ def collect_snapshot(values: dict[str, str], *, root: Path = Path("/"), run: Run
             managed["unhealthy"] += int("unhealthy" in status)
             managed["restarting"] += int(state == "restarting" or "restarting" in status)
     oom = run(["journalctl", "--dmesg", "--since=-24h", "--grep=Out of memory|Killed process", "--quiet"])
-    timer_ages = {"health": 900, "cleanup": 172800, "drift": 3600}
+    timer_ages = {"health": 900, "capacity": 120, "cleanup": 172800, "drift": 3600}
     remote_config = bool(re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", values.get("CI_FLEET_CONFIG_REPOSITORY", "")))
     reconciliation = _reconcile_state(root / "var/lib/ci-fleet/reconcile/state.json") if remote_config else None
     if reconciliation and values.get("CI_FLEET_HEALTH_BOOTSTRAP") == "1":
@@ -507,6 +506,7 @@ def collect_snapshot(values: dict[str, str], *, root: Path = Path("/"), run: Run
         timer_ages["reconcile"] = 900
     timers = {name: _unit_state(run, f"ci-fleet-{name}.timer", timer=True, max_age_seconds=age) for name, age in timer_ages.items()}
     service_units = {
+        "capacity": "ci-fleet-capacity.service",
         "cleanup": "ci-fleet-cleanup.service",
         "drift": "ci-fleet-drift.service",
     }
@@ -603,40 +603,39 @@ def _runner_capacity(run: Runner, instance: str) -> list[dict[str, int | float]]
     result = run(["docker", "ps", "-q", "--filter", "label=io.randomdevelopment.ci-fleet.managed=true",
                   "--filter", "label=io.randomdevelopment.ci-fleet.kind=runner",
                   "--filter", f"label=io.randomdevelopment.ci-fleet.instance={instance}"])
-    ids = [value for value in result.stdout.splitlines() if re.fullmatch(r"[0-9a-f]{12,64}", value)] if result.returncode == 0 else []
+    if result.returncode != 0:
+        raise ValueError("managed runner discovery failed")
+    ids = [value for value in result.stdout.splitlines() if re.fullmatch(r"[0-9a-f]{12,64}", value)]
     if not ids:
-        return []
+        raise ValueError("active managed runners have no stat targets")
     result = run(["docker", "stats", "--no-stream", "--format", "{{.CPUPerc}}\t{{.MemUsage}}", *ids])
+    if result.returncode != 0:
+        raise ValueError("managed runner stat collection failed")
     metrics = []
-    for line in result.stdout.splitlines() if result.returncode == 0 else []:
+    for line in result.stdout.splitlines():
         try:
             cpu, memory = line.split("\t", 1)
             used, limit = memory.split("/", 1)
-            metrics.append({"cpu_percent": float(cpu.removesuffix("%")), "memory_used_bytes": _quantity_bytes(used), "memory_limit_bytes": _quantity_bytes(limit)})
-        except ValueError:
-            continue
+            cpu_percent = float(cpu.removesuffix("%"))
+            if not math.isfinite(cpu_percent) or cpu_percent < 0:
+                raise ValueError
+            metrics.append({"cpu_percent": cpu_percent, "memory_used_bytes": _quantity_bytes(used), "memory_limit_bytes": _quantity_bytes(limit)})
+        except ValueError as error:
+            raise ValueError("managed runner stat output is invalid") from error
+    if len(metrics) != len(ids):
+        raise ValueError("managed runner stat output is incomplete")
     return metrics
 
 
-def record_capacity(path: Path, snapshot: dict[str, Any], *, run: Runner = _run, now: int | None = None) -> bool:
-    if snapshot.get("runners", {}).get("current", 0) < 1:
-        return False
-    timestamp = int(time.time() if now is None else now)
-    memory = snapshot["memory"]
-    sample = {
-        "timestamp": timestamp, "pool": snapshot.get("pool_id", snapshot["controller_id"]),
-        "host": {
-            "cpu_percent": snapshot["cpu"]["used_percent"],
-            "memory_used_bytes": memory["total_bytes"] - memory["available_bytes"],
-            "swap_used_bytes": snapshot["swap"]["used_bytes"],
-            "disk_used_bytes": {name: value["used_bytes"] for name, value in snapshot["disks"].items()},
-            "inode_used": {name: value["inode_used"] for name, value in snapshot["disks"].items()},
-        },
-        "runners": _runner_capacity(run, snapshot["controller_id"]),
-    }
+def _capacity_state(path: Path, timestamp: int) -> list[dict[str, Any]]:
     expected_owner = os.getuid() if os.environ.get("CI_FLEET_TESTING") == "1" else 0
-    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
-    os.chmod(path.parent, 0o700)
+    try:
+        parent = path.parent.lstat()
+    except FileNotFoundError:
+        path.parent.mkdir(parents=True, mode=0o700)
+        parent = path.parent.lstat()
+    if not stat.S_ISDIR(parent.st_mode) or parent.st_uid != expected_owner or stat.S_IMODE(parent.st_mode) != 0o700:
+        raise ValueError("capacity history directory must be root-owned mode 0700")
     retained = []
     if path.exists():
         info = path.lstat()
@@ -649,11 +648,43 @@ def record_capacity(path: Path, snapshot: dict[str, Any], *, run: Runner = _run,
                     retained.append(previous)
             except (KeyError, TypeError, ValueError, json.JSONDecodeError):
                 continue
-    retained = (retained + [sample])[-CAPACITY_MAX_SAMPLES:]
+    return retained[-CAPACITY_MAX_SAMPLES:]
+
+
+def _write_capacity(path: Path, samples: list[dict[str, Any]]) -> None:
     temporary = path.with_suffix(".tmp")
-    temporary.write_text("".join(json.dumps(value, separators=(",", ":"), sort_keys=True) + "\n" for value in retained))
+    temporary.write_text("".join(json.dumps(value, separators=(",", ":"), sort_keys=True) + "\n" for value in samples))
     os.chmod(temporary, 0o600)
     temporary.replace(path)
+
+
+def record_capacity(path: Path, snapshot: dict[str, Any], *, run: Runner = _run, now: int | None = None) -> bool:
+    timestamp = int(time.time() if now is None else now)
+    retained = _capacity_state(path, timestamp)
+    if snapshot.get("runners", {}).get("current", 0) < 1:
+        _write_capacity(path, retained)
+        return False
+    pool_id = snapshot.get("pool_id", snapshot["controller_id"])
+    cpu = snapshot["cpu"]
+    memory = snapshot["memory"]
+    host = {
+        "memory_used_bytes": memory["total_bytes"] - memory["available_bytes"],
+        "swap_used_bytes": snapshot["swap"]["used_bytes"],
+        "disk_used_bytes": {name: value["used_bytes"] for name, value in snapshot["disks"].items()},
+        "inode_used": {name: value["inode_used"] for name, value in snapshot["disks"].items()},
+        "cpu_total_ticks": cpu["total_ticks"],
+        "cpu_idle_ticks": cpu["idle_ticks"],
+    }
+    previous_host = next((value.get("host") for value in reversed(retained)
+                          if isinstance(value, dict) and value.get("pool") == pool_id and isinstance(value.get("host"), dict)
+                          and type(value["host"].get("cpu_total_ticks")) is int and type(value["host"].get("cpu_idle_ticks")) is int), None)
+    if previous_host:
+        total_delta = cpu["total_ticks"] - previous_host["cpu_total_ticks"]
+        idle_delta = cpu["idle_ticks"] - previous_host["cpu_idle_ticks"]
+        if total_delta > 0 and 0 <= idle_delta <= total_delta:
+            host["cpu_percent"] = round(100 * (total_delta - idle_delta) / total_delta, 1)
+    sample = {"timestamp": timestamp, "pool": pool_id, "host": host, "runners": _runner_capacity(run, snapshot["controller_id"])}
+    _write_capacity(path, (retained + [sample])[-CAPACITY_MAX_SAMPLES:])
     return True
 
 
@@ -664,39 +695,91 @@ def _percentiles(values: list[int | float]) -> dict[str, int | float] | None:
     return {name: ordered[max(0, math.ceil(len(ordered) * percentile) - 1)] for name, percentile in (("p50", 0.50), ("p95", 0.95))}
 
 
+def _sample_values(sample: Any, timestamp: int) -> tuple[str, dict[str, list[int | float]], int] | None:
+    if not isinstance(sample, dict) or set(sample) != {"timestamp", "pool", "host", "runners"}:
+        return None
+    if type(sample["timestamp"]) is not int or not timestamp - 7 * 24 * 60 * 60 <= sample["timestamp"] <= timestamp:
+        return None
+    if not isinstance(sample["pool"], str) or not re.fullmatch(r"[A-Za-z0-9._-]+", sample["pool"]):
+        return None
+    host, runners = sample["host"], sample["runners"]
+    required_host = {"memory_used_bytes", "swap_used_bytes", "disk_used_bytes", "inode_used", "cpu_total_ticks", "cpu_idle_ticks"}
+    if not isinstance(host, dict) or not required_host <= set(host) <= required_host | {"cpu_percent"} or not isinstance(runners, list):
+        return None
+    values: dict[str, list[int | float]] = {}
+    for name in ("memory_used_bytes", "swap_used_bytes"):
+        if type(host[name]) is not int or host[name] < 0:
+            return None
+        values[f"host_{name}"] = [host[name]]
+    if "cpu_percent" in host:
+        if type(host["cpu_percent"]) not in (int, float) or not math.isfinite(host["cpu_percent"]) or not 0 <= host["cpu_percent"] <= 100:
+            return None
+        values["host_cpu_percent"] = [host["cpu_percent"]]
+    for kind in ("disk_used_bytes", "inode_used"):
+        if not isinstance(host[kind], dict):
+            return None
+        for name, value in host[kind].items():
+            if not isinstance(name, str) or not re.fullmatch(r"[A-Za-z0-9._-]+", name) or type(value) is not int or value < 0:
+                return None
+            values[f"host_{kind}_{name}"] = [value]
+    runner_count = 0
+    for runner in runners:
+        if not isinstance(runner, dict) or set(runner) != {"cpu_percent", "memory_used_bytes", "memory_limit_bytes"}:
+            return None
+        if type(runner["cpu_percent"]) not in (int, float) or not math.isfinite(runner["cpu_percent"]) or runner["cpu_percent"] < 0:
+            return None
+        if any(type(runner[name]) is not int or runner[name] < 0 for name in ("memory_used_bytes", "memory_limit_bytes")):
+            return None
+        runner_count += 1
+        for name, value in runner.items():
+            values.setdefault(f"runner_{name}", []).append(value)
+    return sample["pool"], values, runner_count
+
+
 def capacity_report(path: Path, *, now: int | None = None) -> dict[str, Any]:
     timestamp = int(time.time() if now is None else now)
-    if path.exists():
-        info = path.lstat()
-        expected_owner = os.getuid() if os.environ.get("CI_FLEET_TESTING") == "1" else 0
-        if not stat.S_ISREG(info.st_mode) or info.st_uid != expected_owner or stat.S_IMODE(info.st_mode) != 0o600:
-            raise ValueError("capacity history must be root-owned mode 0600")
     pools: dict[str, dict[str, Any]] = {}
-    for line in path.read_text().splitlines() if path.exists() else []:
-        try:
-            sample = json.loads(line)
-            if not timestamp - 7 * 24 * 60 * 60 <= sample["timestamp"] <= timestamp:
-                continue
-            pool = pools.setdefault(sample["pool"], {"samples": 0, "runner_observations": 0, "values": {}})
-            pool["samples"] += 1
-            values = pool["values"]
-            host = sample["host"]
-            for name in ("cpu_percent", "memory_used_bytes", "swap_used_bytes"):
-                values.setdefault(f"host_{name}", []).append(host[name])
-            for kind in ("disk_used_bytes", "inode_used"):
-                for name, value in host[kind].items():
-                    values.setdefault(f"host_{kind}_{name}", []).append(value)
-            for runner in sample["runners"]:
-                pool["runner_observations"] += 1
-                for name, value in runner.items():
-                    values.setdefault(f"runner_{name}", []).append(value)
-        except (KeyError, TypeError, ValueError, json.JSONDecodeError):
+    for sample in _capacity_state(path, timestamp):
+        parsed = _sample_values(sample, timestamp)
+        if parsed is None:
             continue
+        pool_name, sample_values, runner_count = parsed
+        pool = pools.setdefault(pool_name, {"samples": 0, "runner_observations": 0, "values": {}})
+        pool["samples"] += 1
+        pool["runner_observations"] += runner_count
+        for name, observed in sample_values.items():
+            pool["values"].setdefault(name, []).extend(observed)
     for pool in pools.values():
         pool["metrics"] = {name: _percentiles(values) for name, values in sorted(pool.pop("values").items())}
     return {"schema_version": 1, "generated_at": timestamp, "period_seconds": 7 * 24 * 60 * 60, "pools": pools}
 
 
+def collect_capacity_snapshot(values: dict[str, str], *, root: Path = Path("/"), run: Runner = _run) -> dict[str, Any]:
+    instance = values.get("CI_FLEET_INSTANCE", "")
+    if not re.fullmatch(r"[A-Za-z0-9._-]+", instance):
+        raise ValueError("controller identity is invalid")
+    docker_root_result = run(["docker", "info", "--format", "{{.DockerRootDir}}"])
+    docker_root = docker_root_result.stdout.strip()
+    if docker_root_result.returncode != 0 or not docker_root.startswith("/"):
+        raise ValueError("Docker root directory is unavailable")
+    runners = run(["docker", "ps", "-q", "--filter", "label=io.randomdevelopment.ci-fleet.managed=true",
+                   "--filter", "label=io.randomdevelopment.ci-fleet.kind=runner",
+                   "--filter", f"label=io.randomdevelopment.ci-fleet.instance={instance}"])
+    if runners.returncode != 0:
+        raise ValueError("managed runner discovery failed")
+    current = len([value for value in runners.stdout.splitlines() if re.fullmatch(r"[0-9a-f]{12,64}", value)])
+    memory, swap = _memory_details(root)
+    return {
+        "controller_id": instance,
+        "pool_id": values.get("CI_FLEET_POOL", instance),
+        "cpu": _cpu(root),
+        "memory": memory,
+        "swap": swap,
+        "disks": {"root": _disk(str(root)), "docker": _disk(str(root / docker_root.lstrip("/")))},
+        "runners": {"current": current},
+    }
+
+
 def _send_status(
     values: dict[str, str],
     report: dict[str, Any],
@@ -801,14 +884,6 @@ def _local(args: argparse.Namespace) -> int:
     report = evaluate(snapshot, thresholds)
     now = int(time.time())
     report["timestamp"] = now
-    try:
-        capacity_history = getattr(args, "capacity_history", None)
-        if capacity_history is not None:
-            record_capacity(capacity_history, snapshot, now=now)
-    except (OSError, UnicodeError, ValueError):
-        report["checks"].append({"id": "capacity_telemetry", "status": "warning"})
-        if report["exit_code"] == 0:
-            report["status"], report["exit_code"] = "warning", 1
     delivery = 0
     if environment.get("CI_FLEET_HEALTH_SUPPRESS_DELIVERY") != "1":
         if config_invalid:
@@ -847,6 +922,13 @@ def _capacity(args: argparse.Namespace) -> int:
     return 0
 
 
+def _capacity_sample(args: argparse.Namespace) -> int:
+    snapshot = collect_capacity_snapshot(dict(os.environ), root=args.root)
+    recorded = record_capacity(args.history, snapshot)
+    print(f"CAPACITY_SAMPLE recorded={str(recorded).lower()}")
+    return 0
+
+
 def main() -> int:
     parser = argparse.ArgumentParser(description="Redacted ci-fleet host health")
     commands = parser.add_subparsers(dest="command", required=True)
@@ -854,7 +936,6 @@ def main() -> int:
     local.add_argument("--json", action="store_true")
     local.add_argument("--monitoring-config", type=Path, default=Path("/etc/ci-fleet/monitoring.env"))
     local.add_argument("--output", type=Path, default=Path("/var/lib/ci-fleet/health/latest.json"))
-    local.add_argument("--capacity-history", type=Path, default=Path("/var/lib/ci-fleet/capacity/samples.jsonl"))
     local.set_defaults(handler=_local)
     heartbeats = commands.add_parser("heartbeats")
     heartbeats.add_argument("--config", type=Path, required=True)
@@ -865,6 +946,10 @@ def main() -> int:
     capacity = commands.add_parser("capacity-report")
     capacity.add_argument("--history", type=Path, default=Path("/var/lib/ci-fleet/capacity/samples.jsonl"))
     capacity.set_defaults(handler=_capacity)
+    sample = commands.add_parser("capacity-sample")
+    sample.add_argument("--history", type=Path, default=Path("/var/lib/ci-fleet/capacity/samples.jsonl"))
+    sample.add_argument("--root", type=Path, default=Path("/"))
+    sample.set_defaults(handler=_capacity_sample)
     args = parser.parse_args()
     try:
         return args.handler(args)
diff --git a/scripts/healthcheck.sh b/scripts/healthcheck.sh
index 89cf1a4..c06bf63 100755
--- a/scripts/healthcheck.sh
+++ b/scripts/healthcheck.sh
@@ -5,7 +5,7 @@ environment=/etc/ci-fleet/ci-fleet.env
 args=(local)
 if [[ ${CI_FLEET_TESTING:-0} == 1 && -n ${CI_FLEET_ROOT_PREFIX:-} ]]; then
   environment="$CI_FLEET_ROOT_PREFIX/etc/ci-fleet/ci-fleet.env"
-  args+=(--monitoring-config "$CI_FLEET_ROOT_PREFIX/etc/ci-fleet/monitoring.env" --output "$CI_FLEET_ROOT_PREFIX/var/lib/ci-fleet/health/latest.json" --capacity-history "$CI_FLEET_ROOT_PREFIX/var/lib/ci-fleet/capacity/samples.jsonl")
+  args+=(--monitoring-config "$CI_FLEET_ROOT_PREFIX/etc/ci-fleet/monitoring.env" --output "$CI_FLEET_ROOT_PREFIX/var/lib/ci-fleet/health/latest.json")
 fi
 if [[ -r $environment ]]; then
   set -a
diff --git a/scripts/install-worker-controller.sh b/scripts/install-worker-controller.sh
index ee988b2..94a61b9 100755
--- a/scripts/install-worker-controller.sh
+++ b/scripts/install-worker-controller.sh
@@ -120,6 +120,7 @@ unit_names=(
   ci-fleet-drift.service ci-fleet-drift.timer
 )
 timer_names=(ci-fleet-health.timer ci-fleet-cleanup.timer ci-fleet-drift.timer)
+capacity_unit_names=(ci-fleet-capacity.service ci-fleet-capacity.timer)
 optional_unit_names=(
   ci-fleet-reconcile.service ci-fleet-reconcile.timer
 )
@@ -516,6 +517,9 @@ manager_release_complete() {
   done
   [[ -x "$path/templates/config-repository/scripts/validate.sh" ]] || return 1
   for unit in "${unit_names[@]}"; do [[ -f "$path/host/systemd/$unit" ]] || return 1; done
+  if [[ -e "$path/scripts/capacity-sample.sh" || -e "$path/host/systemd/ci-fleet-capacity.service" || -e "$path/host/systemd/ci-fleet-capacity.timer" ]]; then
+    [[ -x "$path/scripts/capacity-sample.sh" && -f "$path/host/systemd/ci-fleet-capacity.service" && -f "$path/host/systemd/ci-fleet-capacity.timer" ]] || return 1
+  fi
   marker=$(<"$path/.ci-fleet-engine-ref")
   [[ "$marker" == "$expected" ]]
 }
@@ -551,6 +555,14 @@ systemd_matches() {
     systemctl is-enabled --quiet "$unit" || return 1
     systemctl is-active --quiet "$unit" || return 1
   done
+  if [[ -x "$expected_manager/scripts/capacity-sample.sh" ]]; then
+    for unit in "${capacity_unit_names[@]}"; do
+      [[ -f "$systemd_dir/$unit" ]] && cmp -s "$expected_manager/host/systemd/$unit" "$systemd_dir/$unit" || return 1
+    done
+    systemctl is-enabled --quiet ci-fleet-capacity.timer && systemctl is-active --quiet ci-fleet-capacity.timer || return 1
+  else
+    for unit in "${capacity_unit_names[@]}"; do [[ ! -e "$systemd_dir/$unit" ]] || return 1; done
+  fi
 }
 
 drift_count() {
@@ -695,7 +707,7 @@ make_checkpoint() {
     printf '%s\n' "$target" >"$checkpoint_dir/manager-target"
     chmod 0600 "$checkpoint_dir/manager-target"
   fi
-  for unit in "${unit_names[@]}" "${optional_unit_names[@]}"; do
+  for unit in "${unit_names[@]}" "${capacity_unit_names[@]}" "${optional_unit_names[@]}"; do
     [[ ! -f "$systemd_dir/$unit" ]] || install -m 0644 "$systemd_dir/$unit" "$checkpoint_dir/systemd/$unit"
   done
   : >"$checkpoint_dir/enabled-timers"
@@ -704,6 +716,8 @@ make_checkpoint() {
     if systemctl is-enabled --quiet "$timer" 2>/dev/null; then printf '%s\n' "$timer" >>"$checkpoint_dir/enabled-timers"; fi
     if systemctl is-active --quiet "$timer" 2>/dev/null; then printf '%s\n' "$timer" >>"$checkpoint_dir/active-timers"; fi
   done
+  if systemctl is-enabled --quiet ci-fleet-capacity.timer 2>/dev/null; then printf '%s\n' ci-fleet-capacity.timer >>"$checkpoint_dir/enabled-timers"; fi
+  if systemctl is-active --quiet ci-fleet-capacity.timer 2>/dev/null; then printf '%s\n' ci-fleet-capacity.timer >>"$checkpoint_dir/active-timers"; fi
   local opt_name
   for opt_name in "${optional_unit_names[@]}"; do
     case "$opt_name" in *.timer)
@@ -790,6 +804,13 @@ install_systemd_units() {
   install -d -m 0755 "$systemd_dir"
   install -m 0644 "$source/host/systemd/ci-fleet-health.service" "$systemd_dir/"
   install -m 0644 "$source/host/systemd/ci-fleet-health.timer" "$systemd_dir/"
+  if [[ -x "$source/scripts/capacity-sample.sh" ]]; then
+    install -m 0644 "$source/host/systemd/ci-fleet-capacity.service" "$systemd_dir/"
+    install -m 0644 "$source/host/systemd/ci-fleet-capacity.timer" "$systemd_dir/"
+  else
+    systemctl disable --now ci-fleet-capacity.timer >/dev/null 2>&1 || true
+    rm -f "$systemd_dir/ci-fleet-capacity.service" "$systemd_dir/ci-fleet-capacity.timer"
+  fi
   install -m 0644 "$source/host/systemd/ci-fleet-cleanup.service" "$systemd_dir/"
   install -m 0644 "$source/host/systemd/ci-fleet-cleanup.timer" "$systemd_dir/"
   install -m 0644 "$source/host/systemd/ci-fleet-drift.service" "$systemd_dir/"
@@ -803,11 +824,12 @@ install_systemd_units() {
 
 remove_systemd_units() {
   systemctl disable --now "${timer_names[@]}" >/dev/null 2>&1 || true
+  systemctl disable --now ci-fleet-capacity.timer >/dev/null 2>&1 || true
   local unit
   for unit in "${optional_unit_names[@]}"; do
     case "$unit" in *.timer) systemctl disable --now "$unit" >/dev/null 2>&1 || true ;; esac
   done
-  for unit in "${unit_names[@]}" "${optional_unit_names[@]}"; do rm -f "$systemd_dir/$unit"; done
+  for unit in "${unit_names[@]}" "${capacity_unit_names[@]}" "${optional_unit_names[@]}"; do rm -f "$systemd_dir/$unit"; done
   systemctl daemon-reload
 }
 
@@ -887,6 +909,7 @@ PY
   chmod 0600 "$staged_state"
   mv -f "$staged_state" "$state_file"
   systemctl enable --now "${timer_names[@]}" >/dev/null
+  if [[ -f "$systemd_dir/ci-fleet-capacity.timer" ]]; then systemctl enable --now ci-fleet-capacity.timer >/dev/null; fi
   local opt_timer
   for opt_timer in "${optional_unit_names[@]}"; do
     case "$opt_timer" in *.timer)
@@ -908,6 +931,9 @@ restore_systemd_snapshot() {
   for unit in "${unit_names[@]}"; do
     [[ ! -f "$checkpoint_dir/systemd/$unit" ]] || install -m 0644 "$checkpoint_dir/systemd/$unit" "$systemd_dir/$unit" || failed=1
   done
+  for unit in "${capacity_unit_names[@]}"; do
+    [[ ! -f "$checkpoint_dir/systemd/$unit" ]] || install -m 0644 "$checkpoint_dir/systemd/$unit" "$systemd_dir/$unit" || failed=1
+  done
   for unit in "${optional_unit_names[@]}"; do
     [[ ! -f "$checkpoint_dir/systemd/$unit" ]] || install -m 0644 "$checkpoint_dir/systemd/$unit" "$systemd_dir/$unit" || failed=1
   done
diff --git a/scripts/test_health.py b/scripts/test_health.py
index 47af012..0c7b1ac 100644
--- a/scripts/test_health.py
+++ b/scripts/test_health.py
@@ -132,7 +132,7 @@ def test_capacity_history_is_resource_only_bounded_and_aggregated(self) -> None:
             history = Path(directory) / "capacity" / "samples.jsonl"
             snapshot = healthy_snapshot()
             snapshot.update({
-                "cpu": {"logical": 8, "used_percent": 10},
+                "cpu": {"logical": 8, "used_percent": 10, "total_ticks": 100, "idle_ticks": 40},
                 "memory": {"total_bytes": 1000, "available_bytes": 600},
                 "swap": {"total_bytes": 100, "used_bytes": 5},
                 "runners": {"current": 1, "busy": 1, "maximum": 2},
@@ -149,11 +149,11 @@ def run(args):
             previous_testing = os.environ.get("CI_FLEET_TESTING")
             os.environ["CI_FLEET_TESTING"] = "1"
             try:
-                history.parent.mkdir()
+                history.parent.mkdir(mode=0o700)
                 history.write_text('{"timestamp":999999}\n' * health.CAPACITY_MAX_SAMPLES)
                 history.chmod(0o600)
                 self.assertTrue(health.record_capacity(history, snapshot, run=run, now=1_000_000))
-                snapshot["cpu"]["used_percent"] = 20
+                snapshot["cpu"].update(used_percent=20, total_ticks=200, idle_ticks=50)
                 self.assertTrue(health.record_capacity(history, snapshot, run=run, now=1_000_300))
                 self.assertEqual(len(history.read_text().splitlines()), health.CAPACITY_MAX_SAMPLES)
                 self.assertEqual(history.stat().st_mode & 0o777, 0o600)
@@ -161,7 +161,7 @@ def run(args):
                 report = health.capacity_report(history, now=1_000_300)
                 pool = report["pools"]["example-ci-01"]
                 self.assertEqual((pool["samples"], pool["runner_observations"]), (2, 2))
-                self.assertEqual(pool["metrics"]["host_cpu_percent"], {"p50": 10, "p95": 20})
+                self.assertEqual(pool["metrics"]["host_cpu_percent"], {"p50": 90.0, "p95": 90.0})
                 self.assertEqual(pool["metrics"]["runner_cpu_percent"], {"p50": 12.5, "p95": 90.0})
                 snapshot["runners"]["current"] = 0
                 self.assertFalse(health.record_capacity(history, snapshot, run=run, now=1_000_900))
@@ -171,6 +171,69 @@ def run(args):
                 else:
                     os.environ["CI_FLEET_TESTING"] = previous_testing
 
+    def test_capacity_prunes_inactive_history_and_ignores_incomplete_samples(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            history = Path(directory) / "capacity" / "samples.jsonl"
+            history.parent.mkdir(mode=0o700)
+            valid = {
+                "timestamp": 999_999, "pool": "example-ci-01",
+                "host": {"memory_used_bytes": 1, "swap_used_bytes": 0, "disk_used_bytes": {"root": 1},
+                         "inode_used": {"root": 1}, "cpu_total_ticks": 10, "cpu_idle_ticks": 5, "cpu_percent": 50.0},
+                "runners": [{"cpu_percent": 1.0, "memory_used_bytes": 1, "memory_limit_bytes": 2}],
+            }
+            history.write_text(json.dumps({**valid, "timestamp": 1}) + "\n" + json.dumps(valid) + "\n")
+            history.chmod(0o600)
+            previous_testing = os.environ.get("CI_FLEET_TESTING")
+            os.environ["CI_FLEET_TESTING"] = "1"
+            try:
+                self.assertFalse(health.record_capacity(history, {"runners": {"current": 0}}, now=1_000_000))
+                self.assertEqual(len(history.read_text().splitlines()), 1)
+                incomplete = {**valid, "timestamp": 999_998, "host": dict(valid["host"])}
+                incomplete.pop("runners")
+                history.write_text(history.read_text() + json.dumps(incomplete) + "\n")
+                report = health.capacity_report(history, now=1_000_000)
+                self.assertEqual(report["pools"]["example-ci-01"]["samples"], 1)
+                redirected = Path(directory) / "redirected"
+                redirected.mkdir(mode=0o700)
+                other = Path(directory) / "other"
+                other.mkdir(mode=0o700)
+                redirected.rmdir()
+                redirected.symlink_to(other, target_is_directory=True)
+                with self.assertRaisesRegex(ValueError, "directory must be root-owned"):
+                    health.record_capacity(redirected / "samples.jsonl", {"runners": {"current": 0}}, now=1_000_000)
+            finally:
+                if previous_testing is None:
+                    os.environ.pop("CI_FLEET_TESTING", None)
+                else:
+                    os.environ["CI_FLEET_TESTING"] = previous_testing
+
+    def test_capacity_surfaces_runner_stat_failures(self) -> None:
+        snapshot = healthy_snapshot()
+        snapshot.update({
+            "cpu": {"logical": 8, "used_percent": 10, "total_ticks": 100, "idle_ticks": 40},
+            "memory": {"total_bytes": 1000, "available_bytes": 600},
+            "swap": {"total_bytes": 0, "used_bytes": 0},
+            "runners": {"current": 1},
+        })
+        for value in snapshot["disks"].values():
+            value.update(used_bytes=1, inode_used=1)
+        def run(args):
+            return health.subprocess.CompletedProcess(args, 0 if args[:3] == ["docker", "ps", "-q"] else 1,
+                                                      "a" * 12 + "\n" if args[:3] == ["docker", "ps", "-q"] else "", "failed")
+        with tempfile.TemporaryDirectory() as directory:
+            history = Path(directory) / "capacity" / "samples.jsonl"
+            history.parent.mkdir(mode=0o700)
+            previous_testing = os.environ.get("CI_FLEET_TESTING")
+            os.environ["CI_FLEET_TESTING"] = "1"
+            try:
+                with self.assertRaisesRegex(ValueError, "stat collection failed"):
+                    health.record_capacity(history, snapshot, run=run, now=1_000_000)
+            finally:
+                if previous_testing is None:
+                    os.environ.pop("CI_FLEET_TESTING", None)
+                else:
+                    os.environ["CI_FLEET_TESTING"] = previous_testing
+
     def test_external_heartbeats_detect_missing_and_stale_active_hosts(self) -> None:
         controllers = {
             "fresh": {"state": "active", "lifecycle": "stable"},
@@ -222,8 +285,8 @@ def run(args):
                     run=run,
                 )
                 self.assertEqual((snapshot["load_per_cpu"], snapshot["swap_used_percent"]), (3.0, 50))
-                self.assertEqual(set(snapshot["services"]), {"cleanup", "drift"})
-                self.assertEqual(set(snapshot["timers"]), {"health", "cleanup", "drift"})
+                self.assertEqual(set(snapshot["services"]), {"capacity", "cleanup", "drift"})
+                self.assertEqual(set(snapshot["timers"]), {"health", "capacity", "cleanup", "drift"})
                 (root / "var/lib/ci-fleet/reconcile").mkdir(parents=True)
                 (root / "var/lib/ci-fleet/reconcile/state.json").write_text('{"status":"rolled_back","desired_commit":"","applied_commit":"","health":"healthy"}\n')
                 remote = health.collect_snapshot(
@@ -235,9 +298,9 @@ def run(args):
                     root=root,
                     run=run,
                 )
-                self.assertEqual(set(remote["services"]), {"cleanup", "drift", "reconcile"})
+                self.assertEqual(set(remote["services"]), {"capacity", "cleanup", "drift", "reconcile"})
                 self.assertEqual(set(remote["services"].values()), {"ok"})
-                self.assertEqual(set(remote["timers"]), {"health", "cleanup", "drift", "reconcile"})
+                self.assertEqual(set(remote["timers"]), {"health", "capacity", "cleanup", "drift", "reconcile"})
                 self.assertEqual(remote["reconciliation"]["status"], "bootstrap")
                 (root / "etc").mkdir()
                 (root / "etc/debian_version").write_text("13\n")
@@ -340,7 +403,7 @@ def test_status_report_contract_redaction_and_disabled_ssh(self) -> None:
             "software_version": "1" * 40,
             "boot_time": 900,
             "ssh": "disabled",
-            "cpu": {"logical": 8, "used_percent": 25.0},
+            "cpu": {"logical": 8, "used_percent": 25.0, "total_ticks": 0, "idle_ticks": 0},
             "memory": {"total_bytes": 1024, "available_bytes": 768},
             "swap": {"total_bytes": 512, "used_bytes": 0},
             "load": {"one": 0.1, "five": 0.2, "fifteen": 0.3},

From e76754a3e390da211f78477b0c271d6741e1cf3a Mon Sep 17 00:00:00 2001
From: Nicks Hermes <nicks-hermes@invalid.example>
Date: Sat, 15 Aug 2026 23:31:16 -0500
Subject: [PATCH 3/8] fix: preserve complete capacity observations

---
 docs/CAPACITY-TELEMETRY.md                |  2 +-
 engine-capabilities.json                  |  1 +
 scripts/desired_state.py                  |  4 +-
 scripts/health.py                         | 46 +++++++++++++++++++----
 scripts/install-worker-controller.sh      |  2 +-
 scripts/test-install-worker-controller.sh |  4 ++
 scripts/test_desired_state.py             |  7 +++-
 scripts/test_health.py                    | 16 ++++++--
 8 files changed, 66 insertions(+), 16 deletions(-)

diff --git a/docs/CAPACITY-TELEMETRY.md b/docs/CAPACITY-TELEMETRY.md
index ac72738..c46c1ef 100644
--- a/docs/CAPACITY-TELEMETRY.md
+++ b/docs/CAPACITY-TELEMETRY.md
@@ -8,7 +8,7 @@ Each sample contains only:
 - interval host CPU utilization plus memory, swap, disk-byte, and inode counters;
 - anonymous per-runner CPU percentage and memory use/limit.
 
-Container IDs, names, repositories, jobs, logs, environment variables, source, network counters, and credentials are not stored. The history directory must be a root-owned, non-symlink mode-`0700` directory and the file must be mode `0600`. Malformed records are ignored atomically, records older than eight days are discarded even while the pool is idle, and at most 2,500 samples are retained. Uninstall removes this fleet-owned local history. Runner-stat failures fail the capacity service and appear in the normal health report instead of being recorded as zero observations.
+Container IDs, names, repositories, jobs, logs, environment variables, source, network counters, and credentials are not stored. The history directory must be a root-owned, non-symlink mode-`0700` directory and the file must be mode `0600`. Malformed records are ignored atomically, records older than eight days are discarded even while the pool is idle, and at most 24,000 samples (slightly more than eight days at the scheduled 30-second interval) are retained. Uninstall removes this fleet-owned local history. Runner-stat failures fail the capacity service and appear in the normal health report instead of being recorded as zero observations.
 
 ## Weekly report
 
diff --git a/engine-capabilities.json b/engine-capabilities.json
index 12c4d6d..bf07344 100644
--- a/engine-capabilities.json
+++ b/engine-capabilities.json
@@ -1,6 +1,7 @@
 {
   "schema_version": 1,
   "capabilities": {
+    "capacity_telemetry": true,
     "status_reporting_config": true,
     "required_status_reporting": true
   }
diff --git a/scripts/desired_state.py b/scripts/desired_state.py
index ea16beb..e69896c 100755
--- a/scripts/desired_state.py
+++ b/scripts/desired_state.py
@@ -30,6 +30,7 @@
 HOST_OPTIONAL = {"CI_FLEET_RUNNER_TTL"}
 REQUIRED_STATUS_CAPABILITY = "required_status_reporting"
 STATUS_REPORTING_CONFIG_CAPABILITY = "status_reporting_config"
+CAPACITY_TELEMETRY_CAPABILITY = "capacity_telemetry"
 
 
 class DesiredStateError(ValueError):
@@ -186,7 +187,6 @@ def build_rendered_env(
         "CI_FLEET_LABELS": ",".join(pool["routing_labels"]),
         "CI_FLEET_MAX_RUNNERS": str(effective_max),
         "CI_FLEET_MIN_RUNNERS": str(controller["min_runners"] if state == "active" else 0),
-        "CI_FLEET_POOL": controller["pool"],
         "CI_FLEET_RUNNER_CPUS": str(controller["runner_resources"]["cpu_cores"]),
         "CI_FLEET_RUNNER_GROUP": pool["runner_group"],
         "CI_FLEET_RUNNER_IMAGE": f"ci-fleet-runner:{short_commit}",
@@ -195,6 +195,8 @@ def build_rendered_env(
         "CI_FLEET_VERSION": short_commit,
         **validate_host_values(host_values),
     }
+    if CAPACITY_TELEMETRY_CAPABILITY in (engine_capabilities or set()):
+        rendered["CI_FLEET_POOL"] = controller["pool"]
     reporting_configured = "status_reporting" in controller
     reporting_required = (controller.get("status_reporting") or {}).get("enabled") is True
     if reporting_required and REQUIRED_STATUS_CAPABILITY not in (engine_capabilities or set()):
diff --git a/scripts/health.py b/scripts/health.py
index 305bbe0..b85a7aa 100644
--- a/scripts/health.py
+++ b/scripts/health.py
@@ -243,7 +243,7 @@ def build_status_report(snapshot: dict[str, Any], health_report: dict[str, Any],
         },
         "runners": snapshot.get("runners", {"current": 0, "busy": 0, "maximum": 0}),
         "metrics": {
-            "cpu": snapshot.get("cpu", {"logical": 1, "used_percent": 0}),
+            "cpu": {name: snapshot.get("cpu", {}).get(name, default) for name, default in (("logical", 1), ("used_percent", 0))},
             "memory": snapshot.get("memory", {"total_bytes": 0, "available_bytes": 0}),
             "swap": snapshot.get("swap", {"total_bytes": 0, "used_bytes": 0}),
             "disk": {name: {"total_bytes": value.get("total_bytes", 0), "used_bytes": value.get("used_bytes", 0)} for name, value in disks.items()},
@@ -587,7 +587,7 @@ def _write_report(path: Path, report: dict[str, Any]) -> None:
 
 
 CAPACITY_RETENTION_SECONDS = 8 * 24 * 60 * 60
-CAPACITY_MAX_SAMPLES = 2500
+CAPACITY_MAX_SAMPLES = 24000
 
 
 def _quantity_bytes(value: str) -> int:
@@ -658,14 +658,44 @@ def _write_capacity(path: Path, samples: list[dict[str, Any]]) -> None:
     temporary.replace(path)
 
 
+def _update_cpu_baseline(path: Path, timestamp: int, cpu: dict[str, Any]) -> dict[str, int] | None:
+    baseline = path.with_suffix(path.suffix + ".cpu")
+    expected_owner = os.getuid() if os.environ.get("CI_FLEET_TESTING") == "1" else 0
+    previous = None
+    if baseline.exists():
+        info = baseline.lstat()
+        if not stat.S_ISREG(info.st_mode) or info.st_uid != expected_owner or stat.S_IMODE(info.st_mode) != 0o600:
+            raise ValueError("capacity CPU baseline must be root-owned mode 0600")
+        try:
+            value = json.loads(baseline.read_text())
+            if set(value) != {"timestamp", "total_ticks", "idle_ticks"} or any(type(value[name]) is not int or value[name] < 0 for name in value):
+                raise ValueError
+            if value["idle_ticks"] > value["total_ticks"]:
+                raise ValueError
+            previous = value
+        except (TypeError, ValueError, json.JSONDecodeError) as error:
+            raise ValueError("capacity CPU baseline is invalid") from error
+    current = {"timestamp": timestamp, "total_ticks": cpu.get("total_ticks"), "idle_ticks": cpu.get("idle_ticks")}
+    if any(type(current[name]) is not int or current[name] < 0 for name in current) or current["idle_ticks"] > current["total_ticks"]:
+        raise ValueError("current CPU counters are invalid")
+    temporary = baseline.with_suffix(baseline.suffix + ".tmp")
+    temporary.write_text(json.dumps(current, separators=(",", ":"), sort_keys=True) + "\n")
+    os.chmod(temporary, 0o600)
+    temporary.replace(baseline)
+    if previous and 0 < timestamp - previous["timestamp"] <= 120 and current["total_ticks"] >= previous["total_ticks"] and current["idle_ticks"] >= previous["idle_ticks"]:
+        return previous
+    return None
+
+
 def record_capacity(path: Path, snapshot: dict[str, Any], *, run: Runner = _run, now: int | None = None) -> bool:
     timestamp = int(time.time() if now is None else now)
     retained = _capacity_state(path, timestamp)
+    cpu = snapshot["cpu"]
+    previous_host = _update_cpu_baseline(path, timestamp, cpu)
     if snapshot.get("runners", {}).get("current", 0) < 1:
         _write_capacity(path, retained)
         return False
     pool_id = snapshot.get("pool_id", snapshot["controller_id"])
-    cpu = snapshot["cpu"]
     memory = snapshot["memory"]
     host = {
         "memory_used_bytes": memory["total_bytes"] - memory["available_bytes"],
@@ -675,12 +705,9 @@ def record_capacity(path: Path, snapshot: dict[str, Any], *, run: Runner = _run,
         "cpu_total_ticks": cpu["total_ticks"],
         "cpu_idle_ticks": cpu["idle_ticks"],
     }
-    previous_host = next((value.get("host") for value in reversed(retained)
-                          if isinstance(value, dict) and value.get("pool") == pool_id and isinstance(value.get("host"), dict)
-                          and type(value["host"].get("cpu_total_ticks")) is int and type(value["host"].get("cpu_idle_ticks")) is int), None)
     if previous_host:
-        total_delta = cpu["total_ticks"] - previous_host["cpu_total_ticks"]
-        idle_delta = cpu["idle_ticks"] - previous_host["cpu_idle_ticks"]
+        total_delta = cpu["total_ticks"] - previous_host["total_ticks"]
+        idle_delta = cpu["idle_ticks"] - previous_host["idle_ticks"]
         if total_delta > 0 and 0 <= idle_delta <= total_delta:
             host["cpu_percent"] = round(100 * (total_delta - idle_delta) / total_delta, 1)
     sample = {"timestamp": timestamp, "pool": pool_id, "host": host, "runners": _runner_capacity(run, snapshot["controller_id"])}
@@ -706,6 +733,9 @@ def _sample_values(sample: Any, timestamp: int) -> tuple[str, dict[str, list[int
     required_host = {"memory_used_bytes", "swap_used_bytes", "disk_used_bytes", "inode_used", "cpu_total_ticks", "cpu_idle_ticks"}
     if not isinstance(host, dict) or not required_host <= set(host) <= required_host | {"cpu_percent"} or not isinstance(runners, list):
         return None
+    total_ticks, idle_ticks = host["cpu_total_ticks"], host["cpu_idle_ticks"]
+    if any(type(value) is not int or value < 0 for value in (total_ticks, idle_ticks)) or idle_ticks > total_ticks:
+        return None
     values: dict[str, list[int | float]] = {}
     for name in ("memory_used_bytes", "swap_used_bytes"):
         if type(host[name]) is not int or host[name] < 0:
diff --git a/scripts/install-worker-controller.sh b/scripts/install-worker-controller.sh
index 94a61b9..233bba8 100755
--- a/scripts/install-worker-controller.sh
+++ b/scripts/install-worker-controller.sh
@@ -938,7 +938,7 @@ restore_systemd_snapshot() {
     [[ ! -f "$checkpoint_dir/systemd/$unit" ]] || install -m 0644 "$checkpoint_dir/systemd/$unit" "$systemd_dir/$unit" || failed=1
   done
   systemctl daemon-reload || failed=1
-  for timer in "${timer_names[@]}"; do
+  for timer in "${timer_names[@]}" ci-fleet-capacity.timer; do
     if grep -Fxq "$timer" "$checkpoint_dir/enabled-timers"; then systemctl enable "$timer" >/dev/null || failed=1; else systemctl disable "$timer" >/dev/null 2>&1 || true; fi
     if grep -Fxq "$timer" "$checkpoint_dir/active-timers"; then systemctl start "$timer" || failed=1; else systemctl stop "$timer" >/dev/null 2>&1 || true; fi
   done
diff --git a/scripts/test-install-worker-controller.sh b/scripts/test-install-worker-controller.sh
index 445a77d..92e6fe5 100755
--- a/scripts/test-install-worker-controller.sh
+++ b/scripts/test-install-worker-controller.sh
@@ -140,6 +140,7 @@ EOF
 
 cat >"$fake_bin/systemctl" <<'EOF'
 #!/usr/bin/env bash
+[[ -z "${FAKE_SYSTEMCTL_LOG:-}" ]] || printf '%s\n' "$*" >>"$FAKE_SYSTEMCTL_LOG"
 if [[ "${1:-}" == enable && "${2:-}" == --now && ! -f "${CI_FLEET_ROOT_PREFIX:-}/var/lib/ci-fleet/install-state.json" ]]; then
   exit 98
 fi
@@ -577,7 +578,10 @@ grep -Fq 'CI_FLEET_MAX_RUNNERS=2' "$root/etc/ci-fleet/ci-fleet.env" || fail 'upg
 mkdir -p "$root/var/lib/ci-fleet/checkpoints/99999999-incomplete"
 printf 'restarting\n' >"$FAKE_CONTROLLER_STATUS_FILE"
 rm -f "$root/var/lib/ci-fleet/install-state.json" "$root/etc/ci-fleet/ci-fleet.env"
+export FAKE_SYSTEMCTL_LOG=$tmp/rollback-systemctl.log
 expect_success "$installer" --rollback >/dev/null
+grep -Fxq 'start ci-fleet-capacity.timer' "$FAKE_SYSTEMCTL_LOG" || fail 'rollback did not restore the active capacity timer'
+unset FAKE_SYSTEMCTL_LOG
 [[ ! -f "$FAKE_CONTROLLER_STATUS_FILE" ]] || fail 'explicit rollback did not recover a restarting controller'
 grep -Fq 'CI_FLEET_MAX_RUNNERS=1' "$root/etc/ci-fleet/ci-fleet.env" || fail 'rollback did not restore capacity one'
 
diff --git a/scripts/test_desired_state.py b/scripts/test_desired_state.py
index c37a441..75acb36 100755
--- a/scripts/test_desired_state.py
+++ b/scripts/test_desired_state.py
@@ -45,7 +45,7 @@ def render(self, value: dict | None = None, capabilities: set[str] | None = None
             config_repository="example-org/example-fleet-config",
             config_ref=CONFIG_COMMIT,
             docker_gid=998,
-            engine_capabilities={"status_reporting_config"} if capabilities is None else capabilities,
+            engine_capabilities={"status_reporting_config", "capacity_telemetry"} if capabilities is None else capabilities,
         )
 
     def test_active_controller_renders_configured_capacity(self) -> None:
@@ -53,9 +53,14 @@ def test_active_controller_renders_configured_capacity(self) -> None:
         self.assertEqual(environment["CI_FLEET_MAX_RUNNERS"], "1")
         self.assertEqual(environment["CI_FLEET_CONFIGURED_MAX_RUNNERS"], "1")
         self.assertEqual(environment["CI_FLEET_LABELS"], "docker-ci")
+        self.assertEqual(environment["CI_FLEET_POOL"], "trusted-ci")
         self.assertEqual(environment["CI_FLEET_COMMIT"], environment["CI_FLEET_ENGINE_REF"])
         self.assertEqual(metadata["controller_state"], "active")
 
+    def test_legacy_engine_does_not_receive_capacity_pool_variable(self) -> None:
+        environment, _ = self.render(capabilities={"status_reporting_config"})
+        self.assertNotIn("CI_FLEET_POOL", environment)
+
     def test_status_reporting_requires_fixed_host_local_configuration(self) -> None:
         value = config()
         value["controllers"]["example-ci-01"]["status_reporting"] = {
diff --git a/scripts/test_health.py b/scripts/test_health.py
index 0c7b1ac..420a0e6 100644
--- a/scripts/test_health.py
+++ b/scripts/test_health.py
@@ -123,9 +123,11 @@ def test_malformed_reconciliation_state_is_observable(self) -> None:
             snapshot = healthy_snapshot()
             snapshot["controller_id"] = "example-ci-01"
             snapshot["reconciliation"] = reconciliation
+            snapshot["cpu"] = {"logical": 8, "used_percent": 10, "total_ticks": 123, "idle_ticks": 45}
             report = health.build_status_report(snapshot, health.evaluate(snapshot, health.Thresholds()), generated_at=1_000)
             self.assertEqual(report["configuration"], {"desired_commit": "", "applied_commit": ""})
             self.assertEqual(report["error"]["code"], "reconciliation_invalid")
+            self.assertEqual(set(report["metrics"]["cpu"]), {"logical", "used_percent"})
 
     def test_capacity_history_is_resource_only_bounded_and_aggregated(self) -> None:
         with tempfile.TemporaryDirectory() as directory:
@@ -154,17 +156,22 @@ def run(args):
                 history.chmod(0o600)
                 self.assertTrue(health.record_capacity(history, snapshot, run=run, now=1_000_000))
                 snapshot["cpu"].update(used_percent=20, total_ticks=200, idle_ticks=50)
-                self.assertTrue(health.record_capacity(history, snapshot, run=run, now=1_000_300))
+                self.assertTrue(health.record_capacity(history, snapshot, run=run, now=1_000_030))
                 self.assertEqual(len(history.read_text().splitlines()), health.CAPACITY_MAX_SAMPLES)
                 self.assertEqual(history.stat().st_mode & 0o777, 0o600)
                 self.assertNotIn("must-not-be-recorded", history.read_text())
-                report = health.capacity_report(history, now=1_000_300)
+                report = health.capacity_report(history, now=1_000_030)
                 pool = report["pools"]["example-ci-01"]
                 self.assertEqual((pool["samples"], pool["runner_observations"]), (2, 2))
                 self.assertEqual(pool["metrics"]["host_cpu_percent"], {"p50": 90.0, "p95": 90.0})
                 self.assertEqual(pool["metrics"]["runner_cpu_percent"], {"p50": 12.5, "p95": 90.0})
                 snapshot["runners"]["current"] = 0
                 self.assertFalse(health.record_capacity(history, snapshot, run=run, now=1_000_900))
+                snapshot["runners"]["current"] = 1
+                snapshot["cpu"].update(total_ticks=300, idle_ticks=70)
+                runner_cpu.append(25.0)
+                self.assertTrue(health.record_capacity(history, snapshot, run=run, now=1_000_930))
+                self.assertEqual(json.loads(history.read_text().splitlines()[-1])["host"]["cpu_percent"], 80.0)
             finally:
                 if previous_testing is None:
                     os.environ.pop("CI_FLEET_TESTING", None)
@@ -186,11 +193,12 @@ def test_capacity_prunes_inactive_history_and_ignores_incomplete_samples(self) -
             previous_testing = os.environ.get("CI_FLEET_TESTING")
             os.environ["CI_FLEET_TESTING"] = "1"
             try:
-                self.assertFalse(health.record_capacity(history, {"runners": {"current": 0}}, now=1_000_000))
+                self.assertFalse(health.record_capacity(history, {"runners": {"current": 0}, "cpu": {"total_ticks": 20, "idle_ticks": 10}}, now=1_000_000))
                 self.assertEqual(len(history.read_text().splitlines()), 1)
                 incomplete = {**valid, "timestamp": 999_998, "host": dict(valid["host"])}
                 incomplete.pop("runners")
-                history.write_text(history.read_text() + json.dumps(incomplete) + "\n")
+                malformed_counters = {**valid, "timestamp": 999_997, "host": {**valid["host"], "cpu_total_ticks": "10"}}
+                history.write_text(history.read_text() + json.dumps(incomplete) + "\n" + json.dumps(malformed_counters) + "\n")
                 report = health.capacity_report(history, now=1_000_000)
                 self.assertEqual(report["pools"]["example-ci-01"]["samples"], 1)
                 redirected = Path(directory) / "redirected"

From 4a6876496b933a747563428023fb9759c7befd36 Mon Sep 17 00:00:00 2001
From: Nicks Hermes <nicks-hermes@invalid.example>
Date: Sun, 16 Aug 2026 00:02:21 -0500
Subject: [PATCH 4/8] fix: make capacity lifecycle bounded and optional

---
 docs/CAPACITY-TELEMETRY.md           |  2 +-
 scripts/health.py                    | 46 ++++++++++++++++++++--------
 scripts/install-worker-controller.sh |  1 +
 scripts/test_health.py               | 17 +++++++---
 4 files changed, 49 insertions(+), 17 deletions(-)

diff --git a/docs/CAPACITY-TELEMETRY.md b/docs/CAPACITY-TELEMETRY.md
index c46c1ef..89b274b 100644
--- a/docs/CAPACITY-TELEMETRY.md
+++ b/docs/CAPACITY-TELEMETRY.md
@@ -8,7 +8,7 @@ Each sample contains only:
 - interval host CPU utilization plus memory, swap, disk-byte, and inode counters;
 - anonymous per-runner CPU percentage and memory use/limit.
 
-Container IDs, names, repositories, jobs, logs, environment variables, source, network counters, and credentials are not stored. The history directory must be a root-owned, non-symlink mode-`0700` directory and the file must be mode `0600`. Malformed records are ignored atomically, records older than eight days are discarded even while the pool is idle, and at most 24,000 samples (slightly more than eight days at the scheduled 30-second interval) are retained. Uninstall removes this fleet-owned local history. Runner-stat failures fail the capacity service and appear in the normal health report instead of being recorded as zero observations.
+Container IDs, names, repositories, jobs, logs, environment variables, source, network counters, and credentials are not stored. The history directory must be a root-owned, non-symlink mode-`0700` directory and the file must be mode `0600`. Malformed records are ignored atomically, records older than eight days are discarded even while the pool is idle, and history is compacted from at most 26,000 records back to 24,000 (more than eight days at the scheduled 30-second interval). Normal samples append one record; compaction rewrites only when retention, malformed input, or the rotation threshold requires it. Uninstall removes this fleet-owned local history. Runner-stat failures fail the capacity service and appear in the normal health report instead of being recorded as zero observations.
 
 ## Weekly report
 
diff --git a/scripts/health.py b/scripts/health.py
index b85a7aa..4661e8a 100644
--- a/scripts/health.py
+++ b/scripts/health.py
@@ -497,7 +497,10 @@ def collect_snapshot(values: dict[str, str], *, root: Path = Path("/"), run: Run
             managed["unhealthy"] += int("unhealthy" in status)
             managed["restarting"] += int(state == "restarting" or "restarting" in status)
     oom = run(["journalctl", "--dmesg", "--since=-24h", "--grep=Out of memory|Killed process", "--quiet"])
-    timer_ages = {"health": 900, "capacity": 120, "cleanup": 172800, "drift": 3600}
+    timer_ages = {"health": 900, "cleanup": 172800, "drift": 3600}
+    capacity_installed = (root / "etc/systemd/system/ci-fleet-capacity.timer").is_file()
+    if capacity_installed:
+        timer_ages["capacity"] = 120
     remote_config = bool(re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", values.get("CI_FLEET_CONFIG_REPOSITORY", "")))
     reconciliation = _reconcile_state(root / "var/lib/ci-fleet/reconcile/state.json") if remote_config else None
     if reconciliation and values.get("CI_FLEET_HEALTH_BOOTSTRAP") == "1":
@@ -505,11 +508,9 @@ def collect_snapshot(values: dict[str, str], *, root: Path = Path("/"), run: Run
     if remote_config:
         timer_ages["reconcile"] = 900
     timers = {name: _unit_state(run, f"ci-fleet-{name}.timer", timer=True, max_age_seconds=age) for name, age in timer_ages.items()}
-    service_units = {
-        "capacity": "ci-fleet-capacity.service",
-        "cleanup": "ci-fleet-cleanup.service",
-        "drift": "ci-fleet-drift.service",
-    }
+    service_units = {"cleanup": "ci-fleet-cleanup.service", "drift": "ci-fleet-drift.service"}
+    if capacity_installed:
+        service_units["capacity"] = "ci-fleet-capacity.service"
     if remote_config:
         service_units["reconcile"] = "ci-fleet-reconcile.service"
     services = {name: _unit_state(run, unit) for name, unit in service_units.items()}
@@ -588,6 +589,7 @@ def _write_report(path: Path, report: dict[str, Any]) -> None:
 
 CAPACITY_RETENTION_SECONDS = 8 * 24 * 60 * 60
 CAPACITY_MAX_SAMPLES = 24000
+CAPACITY_COMPACT_SAMPLES = 26000
 
 
 def _quantity_bytes(value: str) -> int:
@@ -627,16 +629,19 @@ def _runner_capacity(run: Runner, instance: str) -> list[dict[str, int | float]]
     return metrics
 
 
-def _capacity_state(path: Path, timestamp: int) -> list[dict[str, Any]]:
+def _capacity_state(path: Path, timestamp: int, *, create: bool = False) -> tuple[list[dict[str, Any]], bool]:
     expected_owner = os.getuid() if os.environ.get("CI_FLEET_TESTING") == "1" else 0
     try:
         parent = path.parent.lstat()
     except FileNotFoundError:
+        if not create:
+            return [], False
         path.parent.mkdir(parents=True, mode=0o700)
         parent = path.parent.lstat()
     if not stat.S_ISDIR(parent.st_mode) or parent.st_uid != expected_owner or stat.S_IMODE(parent.st_mode) != 0o700:
         raise ValueError("capacity history directory must be root-owned mode 0700")
     retained = []
+    dirty = False
     if path.exists():
         info = path.lstat()
         if not stat.S_ISREG(info.st_mode) or info.st_uid != expected_owner or stat.S_IMODE(info.st_mode) != 0o600:
@@ -646,9 +651,15 @@ def _capacity_state(path: Path, timestamp: int) -> list[dict[str, Any]]:
                 previous = json.loads(line)
                 if timestamp - CAPACITY_RETENTION_SECONDS <= previous["timestamp"] <= timestamp:
                     retained.append(previous)
+                else:
+                    dirty = True
             except (KeyError, TypeError, ValueError, json.JSONDecodeError):
+                dirty = True
                 continue
-    return retained[-CAPACITY_MAX_SAMPLES:]
+    if len(retained) >= CAPACITY_COMPACT_SAMPLES:
+        retained = retained[-CAPACITY_COMPACT_SAMPLES:]
+        dirty = True
+    return retained, dirty
 
 
 def _write_capacity(path: Path, samples: list[dict[str, Any]]) -> None:
@@ -658,6 +669,12 @@ def _write_capacity(path: Path, samples: list[dict[str, Any]]) -> None:
     temporary.replace(path)
 
 
+def _append_capacity(path: Path, sample: dict[str, Any]) -> None:
+    with path.open("a", encoding="utf-8") as stream:
+        stream.write(json.dumps(sample, separators=(",", ":"), sort_keys=True) + "\n")
+    os.chmod(path, 0o600)
+
+
 def _update_cpu_baseline(path: Path, timestamp: int, cpu: dict[str, Any]) -> dict[str, int] | None:
     baseline = path.with_suffix(path.suffix + ".cpu")
     expected_owner = os.getuid() if os.environ.get("CI_FLEET_TESTING") == "1" else 0
@@ -689,11 +706,12 @@ def _update_cpu_baseline(path: Path, timestamp: int, cpu: dict[str, Any]) -> dic
 
 def record_capacity(path: Path, snapshot: dict[str, Any], *, run: Runner = _run, now: int | None = None) -> bool:
     timestamp = int(time.time() if now is None else now)
-    retained = _capacity_state(path, timestamp)
+    retained, dirty = _capacity_state(path, timestamp, create=True)
     cpu = snapshot["cpu"]
     previous_host = _update_cpu_baseline(path, timestamp, cpu)
     if snapshot.get("runners", {}).get("current", 0) < 1:
-        _write_capacity(path, retained)
+        if dirty:
+            _write_capacity(path, retained[-CAPACITY_MAX_SAMPLES:])
         return False
     pool_id = snapshot.get("pool_id", snapshot["controller_id"])
     memory = snapshot["memory"]
@@ -711,7 +729,10 @@ def record_capacity(path: Path, snapshot: dict[str, Any], *, run: Runner = _run,
         if total_delta > 0 and 0 <= idle_delta <= total_delta:
             host["cpu_percent"] = round(100 * (total_delta - idle_delta) / total_delta, 1)
     sample = {"timestamp": timestamp, "pool": pool_id, "host": host, "runners": _runner_capacity(run, snapshot["controller_id"])}
-    _write_capacity(path, (retained + [sample])[-CAPACITY_MAX_SAMPLES:])
+    if dirty or len(retained) >= CAPACITY_COMPACT_SAMPLES:
+        _write_capacity(path, (retained + [sample])[-CAPACITY_MAX_SAMPLES:])
+    else:
+        _append_capacity(path, sample)
     return True
 
 
@@ -769,7 +790,8 @@ def _sample_values(sample: Any, timestamp: int) -> tuple[str, dict[str, list[int
 def capacity_report(path: Path, *, now: int | None = None) -> dict[str, Any]:
     timestamp = int(time.time() if now is None else now)
     pools: dict[str, dict[str, Any]] = {}
-    for sample in _capacity_state(path, timestamp):
+    samples, _ = _capacity_state(path, timestamp)
+    for sample in samples:
         parsed = _sample_values(sample, timestamp)
         if parsed is None:
             continue
diff --git a/scripts/install-worker-controller.sh b/scripts/install-worker-controller.sh
index 233bba8..d9a48a7 100755
--- a/scripts/install-worker-controller.sh
+++ b/scripts/install-worker-controller.sh
@@ -825,6 +825,7 @@ install_systemd_units() {
 remove_systemd_units() {
   systemctl disable --now "${timer_names[@]}" >/dev/null 2>&1 || true
   systemctl disable --now ci-fleet-capacity.timer >/dev/null 2>&1 || true
+  systemctl stop ci-fleet-capacity.service >/dev/null 2>&1 || true
   local unit
   for unit in "${optional_unit_names[@]}"; do
     case "$unit" in *.timer) systemctl disable --now "$unit" >/dev/null 2>&1 || true ;; esac
diff --git a/scripts/test_health.py b/scripts/test_health.py
index 420a0e6..960f328 100644
--- a/scripts/test_health.py
+++ b/scripts/test_health.py
@@ -155,9 +155,11 @@ def run(args):
                 history.write_text('{"timestamp":999999}\n' * health.CAPACITY_MAX_SAMPLES)
                 history.chmod(0o600)
                 self.assertTrue(health.record_capacity(history, snapshot, run=run, now=1_000_000))
+                history_inode = history.stat().st_ino
                 snapshot["cpu"].update(used_percent=20, total_ticks=200, idle_ticks=50)
                 self.assertTrue(health.record_capacity(history, snapshot, run=run, now=1_000_030))
-                self.assertEqual(len(history.read_text().splitlines()), health.CAPACITY_MAX_SAMPLES)
+                self.assertLessEqual(len(history.read_text().splitlines()), health.CAPACITY_COMPACT_SAMPLES)
+                self.assertEqual(history.stat().st_ino, history_inode)
                 self.assertEqual(history.stat().st_mode & 0o777, 0o600)
                 self.assertNotIn("must-not-be-recorded", history.read_text())
                 report = health.capacity_report(history, now=1_000_030)
@@ -166,7 +168,9 @@ def run(args):
                 self.assertEqual(pool["metrics"]["host_cpu_percent"], {"p50": 90.0, "p95": 90.0})
                 self.assertEqual(pool["metrics"]["runner_cpu_percent"], {"p50": 12.5, "p95": 90.0})
                 snapshot["runners"]["current"] = 0
+                history_before_idle = history.read_bytes()
                 self.assertFalse(health.record_capacity(history, snapshot, run=run, now=1_000_900))
+                self.assertEqual(history.read_bytes(), history_before_idle)
                 snapshot["runners"]["current"] = 1
                 snapshot["cpu"].update(total_ticks=300, idle_ticks=70)
                 runner_cpu.append(25.0)
@@ -181,6 +185,9 @@ def run(args):
     def test_capacity_prunes_inactive_history_and_ignores_incomplete_samples(self) -> None:
         with tempfile.TemporaryDirectory() as directory:
             history = Path(directory) / "capacity" / "samples.jsonl"
+            missing = Path(directory) / "missing" / "samples.jsonl"
+            self.assertEqual(health.capacity_report(missing, now=1_000_000)["pools"], {})
+            self.assertFalse(missing.parent.exists())
             history.parent.mkdir(mode=0o700)
             valid = {
                 "timestamp": 999_999, "pool": "example-ci-01",
@@ -293,8 +300,10 @@ def run(args):
                     run=run,
                 )
                 self.assertEqual((snapshot["load_per_cpu"], snapshot["swap_used_percent"]), (3.0, 50))
-                self.assertEqual(set(snapshot["services"]), {"capacity", "cleanup", "drift"})
-                self.assertEqual(set(snapshot["timers"]), {"health", "capacity", "cleanup", "drift"})
+                self.assertEqual(set(snapshot["services"]), {"cleanup", "drift"})
+                self.assertEqual(set(snapshot["timers"]), {"health", "cleanup", "drift"})
+                (root / "etc/systemd/system").mkdir(parents=True)
+                (root / "etc/systemd/system/ci-fleet-capacity.timer").write_text("fixture\n")
                 (root / "var/lib/ci-fleet/reconcile").mkdir(parents=True)
                 (root / "var/lib/ci-fleet/reconcile/state.json").write_text('{"status":"rolled_back","desired_commit":"","applied_commit":"","health":"healthy"}\n')
                 remote = health.collect_snapshot(
@@ -310,7 +319,7 @@ def run(args):
                 self.assertEqual(set(remote["services"].values()), {"ok"})
                 self.assertEqual(set(remote["timers"]), {"health", "capacity", "cleanup", "drift", "reconcile"})
                 self.assertEqual(remote["reconciliation"]["status"], "bootstrap")
-                (root / "etc").mkdir()
+                (root / "etc").mkdir(exist_ok=True)
                 (root / "etc/debian_version").write_text("13\n")
                 debian = health.collect_snapshot({"CI_FLEET_CONTROLLER_STATE": "disabled", "CI_FLEET_HEALTH_BOOTSTRAP": "1"}, root=root, run=run)
                 self.assertIn("updates", debian["services"])

From 60b64ee6ccbf15cc9c23c37aea973d0c496b0fd4 Mon Sep 17 00:00:00 2001
From: Nicks Hermes <nicks-hermes@invalid.example>
Date: Sun, 16 Aug 2026 00:57:46 -0500
Subject: [PATCH 5/8] fix: stop capacity service on legacy downgrade

---
 scripts/install-worker-controller.sh      | 1 +
 scripts/test-install-worker-controller.sh | 3 +++
 2 files changed, 4 insertions(+)

diff --git a/scripts/install-worker-controller.sh b/scripts/install-worker-controller.sh
index d9a48a7..7d349aa 100755
--- a/scripts/install-worker-controller.sh
+++ b/scripts/install-worker-controller.sh
@@ -809,6 +809,7 @@ install_systemd_units() {
     install -m 0644 "$source/host/systemd/ci-fleet-capacity.timer" "$systemd_dir/"
   else
     systemctl disable --now ci-fleet-capacity.timer >/dev/null 2>&1 || true
+    systemctl stop ci-fleet-capacity.service >/dev/null 2>&1 || true
     rm -f "$systemd_dir/ci-fleet-capacity.service" "$systemd_dir/ci-fleet-capacity.timer"
   fi
   install -m 0644 "$source/host/systemd/ci-fleet-cleanup.service" "$systemd_dir/"
diff --git a/scripts/test-install-worker-controller.sh b/scripts/test-install-worker-controller.sh
index 92e6fe5..c8cd1ce 100755
--- a/scripts/test-install-worker-controller.sh
+++ b/scripts/test-install-worker-controller.sh
@@ -694,7 +694,10 @@ legacy_ref=$(write_config active 1 1 "$legacy_engine_ref" omit)
 export FAKE_ENGINE_REF=$legacy_engine_ref
 export FAKE_RUNNER_IMAGE=ci-fleet-runner:${legacy_engine_ref:0:12}
 export FAKE_CONTROLLER_IMAGE=ci-fleet-controller:${legacy_engine_ref:0:12}
+export FAKE_SYSTEMCTL_LOG=$tmp/legacy-systemctl.log
 expect_success "$installer" --upgrade "${base_args[@]}" --ref "$legacy_ref" >/dev/null
+grep -Fxq 'stop ci-fleet-capacity.service' "$FAKE_SYSTEMCTL_LOG" || fail 'legacy upgrade did not stop the in-flight capacity service'
+unset FAKE_SYSTEMCTL_LOG
 [[ $(readlink -f "$adopt_root/opt/ci-fleet/current") == "$adopt_root/opt/ci-fleet/releases/$legacy_engine_ref" ]] || fail 'upgrade could not restore a pre-health-contract engine'
 
 grep -Fq 'Issue #7' "$repo_root/docs/DESIGN-DECISIONS.md" || fail 'isolated proof approval is not recorded'

From a9f3e1b7fcaab9152423eb77cebc7ff08ca65a77 Mon Sep 17 00:00:00 2001
From: Nicks Hermes <nicks-hermes@invalid.example>
Date: Sun, 16 Aug 2026 01:42:00 -0500
Subject: [PATCH 6/8] fix: preserve telemetry lifecycle guarantees

---
 docs/DESIRED-STATE.md                     |  2 +-
 scripts/desired_state.py                  |  3 ++
 scripts/health.py                         |  6 +++-
 scripts/install-worker-controller.sh      | 39 +++++++++++++++++------
 scripts/test-install-worker-controller.sh | 15 +++++++++
 scripts/test_health.py                    |  6 ++++
 6 files changed, 59 insertions(+), 12 deletions(-)

diff --git a/docs/DESIRED-STATE.md b/docs/DESIRED-STATE.md
index dca40f5..b702f8a 100644
--- a/docs/DESIRED-STATE.md
+++ b/docs/DESIRED-STATE.md
@@ -88,7 +88,7 @@ The installer:
 10. starts the controller only when its desired state is active and verifies runtime health;
 11. atomically records redacted installation state, then enables the maintenance timers.
 
-A successful second `--install` run reports `NO_CHANGE` and performs no unnecessary replacement. A successful engine upgrade advances both the runtime release and the maintenance installer-manager to the same pinned commit; rollback restores both.
+A successful second `--install` run reports `NO_CHANGE` and performs no unnecessary replacement. Engine upgrades advance the runtime release and maintenance installer-manager together. A deliberate runtime downgrade retains a newer compatible manager so a later reviewed upgrade can still be rendered and activated; maintenance units always come from the selected runtime. Rollback restores both.
 
 ## Adopt an existing controller
 
diff --git a/scripts/desired_state.py b/scripts/desired_state.py
index e69896c..aaaa41f 100755
--- a/scripts/desired_state.py
+++ b/scripts/desired_state.py
@@ -295,6 +295,8 @@ def command_validate_engine_capabilities(args: argparse.Namespace) -> None:
         raise DesiredStateError("selected engine does not support status reporting configuration")
     if args.require_status_reporting and REQUIRED_STATUS_CAPABILITY not in capabilities:
         raise DesiredStateError("selected engine does not advertise required status reporting")
+    if args.require_capacity_telemetry and CAPACITY_TELEMETRY_CAPABILITY not in capabilities:
+        raise DesiredStateError("selected engine does not advertise capacity telemetry")
     print("ENGINE_CAPABILITIES_OK")
 
 
@@ -332,6 +334,7 @@ def parse_args() -> argparse.Namespace:
     capabilities.add_argument("--manifest", type=Path, required=True)
     capabilities.add_argument("--require-status-reporting-config", action="store_true")
     capabilities.add_argument("--require-status-reporting", action="store_true")
+    capabilities.add_argument("--require-capacity-telemetry", action="store_true")
     capabilities.set_defaults(function=command_validate_engine_capabilities)
     return parser.parse_args()
 
diff --git a/scripts/health.py b/scripts/health.py
index 4661e8a..0b0b4a8 100644
--- a/scripts/health.py
+++ b/scripts/health.py
@@ -18,6 +18,7 @@
 from pathlib import Path
 from typing import Any, Callable
 
+sys.dont_write_bytecode = True
 from status_auth import sign_headers
 
 
@@ -642,8 +643,11 @@ def _capacity_state(path: Path, timestamp: int, *, create: bool = False) -> tupl
         raise ValueError("capacity history directory must be root-owned mode 0700")
     retained = []
     dirty = False
-    if path.exists():
+    try:
         info = path.lstat()
+    except FileNotFoundError:
+        info = None
+    if info is not None:
         if not stat.S_ISREG(info.st_mode) or info.st_uid != expected_owner or stat.S_IMODE(info.st_mode) != 0o600:
             raise ValueError("capacity history must be root-owned mode 0600")
         for line in path.read_text().splitlines():
diff --git a/scripts/install-worker-controller.sh b/scripts/install-worker-controller.sh
index 7d349aa..0822318 100755
--- a/scripts/install-worker-controller.sh
+++ b/scripts/install-worker-controller.sh
@@ -508,6 +508,12 @@ runtime_release_complete() {
   [[ "$actual_digest" == "$stored_digest" ]]
 }
 
+engine_has_capacity_telemetry() {
+  local path=$1
+  [[ -f "$path/engine-capabilities.json" ]] && python3 "$repo_root/scripts/desired_state.py" validate-engine-capabilities \
+    --manifest "$path/engine-capabilities.json" --require-capacity-telemetry >/dev/null 2>&1
+}
+
 manager_release_complete() {
   local path=$1 expected=$2 require_status=${3:-0} require_schema=${4:-0} marker required unit
   runtime_release_complete "$path" "$expected" "$require_status" "$require_schema" || return 1
@@ -517,7 +523,7 @@ manager_release_complete() {
   done
   [[ -x "$path/templates/config-repository/scripts/validate.sh" ]] || return 1
   for unit in "${unit_names[@]}"; do [[ -f "$path/host/systemd/$unit" ]] || return 1; done
-  if [[ -e "$path/scripts/capacity-sample.sh" || -e "$path/host/systemd/ci-fleet-capacity.service" || -e "$path/host/systemd/ci-fleet-capacity.timer" ]]; then
+  if engine_has_capacity_telemetry "$path" || [[ -e "$path/scripts/capacity-sample.sh" || -e "$path/host/systemd/ci-fleet-capacity.service" || -e "$path/host/systemd/ci-fleet-capacity.timer" ]]; then
     [[ -x "$path/scripts/capacity-sample.sh" && -f "$path/host/systemd/ci-fleet-capacity.service" && -f "$path/host/systemd/ci-fleet-capacity.timer" ]] || return 1
   fi
   marker=$(<"$path/.ci-fleet-engine-ref")
@@ -542,22 +548,27 @@ managed_images_match() {
 }
 
 systemd_matches() {
-  local expected_manager unit
+  local expected_manager manager_commit unit
   expected_manager=$manager_releases/$engine_ref
-  manager_release_complete "$expected_manager" "$engine_ref" "$status_reporting_required" "$status_reporting_configured" || return 1
   [[ -L "$manager_current" ]] || return 1
+  if ! engine_has_capacity_telemetry "$release_dir" && engine_has_capacity_telemetry "$(readlink -f "$manager_current")"; then
+    expected_manager=$(readlink -f "$manager_current")
+  fi
+  [[ -f "$expected_manager/.ci-fleet-engine-ref" ]] || return 1
+  manager_commit=$(<"$expected_manager/.ci-fleet-engine-ref")
+  manager_release_complete "$expected_manager" "$manager_commit" "$status_reporting_required" "$status_reporting_configured" || return 1
   [[ $(readlink -f "$manager_current") == $(readlink -f "$expected_manager") ]] || return 1
   for unit in "${unit_names[@]}"; do
     [[ -f "$systemd_dir/$unit" ]] || return 1
-    cmp -s "$expected_manager/host/systemd/$unit" "$systemd_dir/$unit" || return 1
+    cmp -s "$release_dir/host/systemd/$unit" "$systemd_dir/$unit" || return 1
   done
   for unit in "${timer_names[@]}"; do
     systemctl is-enabled --quiet "$unit" || return 1
     systemctl is-active --quiet "$unit" || return 1
   done
-  if [[ -x "$expected_manager/scripts/capacity-sample.sh" ]]; then
+  if [[ -x "$release_dir/scripts/capacity-sample.sh" ]]; then
     for unit in "${capacity_unit_names[@]}"; do
-      [[ -f "$systemd_dir/$unit" ]] && cmp -s "$expected_manager/host/systemd/$unit" "$systemd_dir/$unit" || return 1
+      [[ -f "$systemd_dir/$unit" ]] && cmp -s "$release_dir/host/systemd/$unit" "$systemd_dir/$unit" || return 1
     done
     systemctl is-enabled --quiet ci-fleet-capacity.timer && systemctl is-active --quiet ci-fleet-capacity.timer || return 1
   else
@@ -647,8 +658,16 @@ install_release() {
 }
 
 install_manager() {
-  local manager_commit manager_release archive staged_manager
+  local manager_commit manager_release archive staged_manager current_manager current_manager_commit
   manager_commit=$engine_ref
+  current_manager=$(readlink -f "$manager_current" 2>/dev/null || true)
+  current_manager_commit=
+  [[ -z "$current_manager" || ! -f "$current_manager/.ci-fleet-engine-ref" ]] || current_manager_commit=$(<"$current_manager/.ci-fleet-engine-ref")
+  if ! engine_has_capacity_telemetry "$release_dir" && [[ -n "$current_manager_commit" ]] \
+    && engine_has_capacity_telemetry "$current_manager" \
+    && manager_release_complete "$current_manager" "$current_manager_commit" "$status_reporting_required" "$status_reporting_configured"; then
+    return
+  fi
   [[ "$manager_commit" =~ ^[0-9a-f]{40}$ ]] || die 'installer manager commit is invalid'
   runtime_release_complete "$release_dir" "$manager_commit" "$status_reporting_required" "$status_reporting_configured" || die 'desired engine release is unavailable for installer manager activation'
   manager_release=$manager_releases/$manager_commit
@@ -809,7 +828,7 @@ install_systemd_units() {
     install -m 0644 "$source/host/systemd/ci-fleet-capacity.timer" "$systemd_dir/"
   else
     systemctl disable --now ci-fleet-capacity.timer >/dev/null 2>&1 || true
-    systemctl stop ci-fleet-capacity.service >/dev/null 2>&1 || true
+    [[ ! -f "$systemd_dir/ci-fleet-capacity.service" ]] || systemctl stop ci-fleet-capacity.service
     rm -f "$systemd_dir/ci-fleet-capacity.service" "$systemd_dir/ci-fleet-capacity.timer"
   fi
   install -m 0644 "$source/host/systemd/ci-fleet-cleanup.service" "$systemd_dir/"
@@ -826,7 +845,7 @@ install_systemd_units() {
 remove_systemd_units() {
   systemctl disable --now "${timer_names[@]}" >/dev/null 2>&1 || true
   systemctl disable --now ci-fleet-capacity.timer >/dev/null 2>&1 || true
-  systemctl stop ci-fleet-capacity.service >/dev/null 2>&1 || true
+  [[ ! -f "$systemd_dir/ci-fleet-capacity.service" ]] || systemctl stop ci-fleet-capacity.service
   local unit
   for unit in "${optional_unit_names[@]}"; do
     case "$unit" in *.timer) systemctl disable --now "$unit" >/dev/null 2>&1 || true ;; esac
@@ -882,7 +901,7 @@ activate_candidate() {
   ln -sfn "$release_dir" "$temporary/current"
   mv -Tf "$temporary/current" "$current_link"
   install_manager
-  install_systemd_units "$(readlink -f "$manager_current")"
+  install_systemd_units "$release_dir"
   if [[ "$target_state" == active ]]; then
     compose "$release_dir" "$rendered_env" up -d --no-deps controller
     sleep "${CI_FLEET_STARTUP_WAIT_SECONDS:-2}"
diff --git a/scripts/test-install-worker-controller.sh b/scripts/test-install-worker-controller.sh
index c8cd1ce..5ebf4e4 100755
--- a/scripts/test-install-worker-controller.sh
+++ b/scripts/test-install-worker-controller.sh
@@ -141,6 +141,10 @@ EOF
 cat >"$fake_bin/systemctl" <<'EOF'
 #!/usr/bin/env bash
 [[ -z "${FAKE_SYSTEMCTL_LOG:-}" ]] || printf '%s\n' "$*" >>"$FAKE_SYSTEMCTL_LOG"
+if [[ -n "${FAKE_FAIL_CAPACITY_STOP_ONCE:-}" && -f "$FAKE_FAIL_CAPACITY_STOP_ONCE" && "$*" == 'stop ci-fleet-capacity.service' ]]; then
+  rm -f "$FAKE_FAIL_CAPACITY_STOP_ONCE"
+  exit 96
+fi
 if [[ "${1:-}" == enable && "${2:-}" == --now && ! -f "${CI_FLEET_ROOT_PREFIX:-}/var/lib/ci-fleet/install-state.json" ]]; then
   exit 98
 fi
@@ -232,6 +236,7 @@ for dockerfile in "$repo_root/controller/Dockerfile" "$repo_root/runner/Dockerfi
 done
 grep -Fq '    user: "0:0"' "$repo_root/deploy/compose.yaml" || fail 'controller cannot read the required root-owned mode-0600 GitHub App PEM'
 grep -Fq 'export PYTHONDONTWRITEBYTECODE=1' "$repo_root/scripts/install-worker-controller.sh" || fail 'managed validation may write Python bytecode into the immutable manager release'
+grep -Fq 'sys.dont_write_bytecode = True' "$repo_root/scripts/health.py" || fail 'manual capacity reports may write Python bytecode into the immutable runtime release'
 grep -Fq '    trap - ERR' "$repo_root/scripts/install-worker-controller.sh" || fail 'warning health subprocess inherits the transactional rollback trap'
 grep -Fq "CI_FLEET_COMMIT: \${CI_FLEET_COMMIT:-unknown}" "$repo_root/deploy/compose.yaml" || fail 'runner build lacks engine provenance argument'
 config_repo=$tmp/config-repo
@@ -695,10 +700,20 @@ export FAKE_ENGINE_REF=$legacy_engine_ref
 export FAKE_RUNNER_IMAGE=ci-fleet-runner:${legacy_engine_ref:0:12}
 export FAKE_CONTROLLER_IMAGE=ci-fleet-controller:${legacy_engine_ref:0:12}
 export FAKE_SYSTEMCTL_LOG=$tmp/legacy-systemctl.log
+export FAKE_FAIL_CAPACITY_STOP_ONCE=$tmp/fail-capacity-stop-once
+: >"$FAKE_FAIL_CAPACITY_STOP_ONCE"
+expect_failure 'ROLLBACK_RESTORED' "$installer" --upgrade "${base_args[@]}" --ref "$legacy_ref"
+unset FAKE_FAIL_CAPACITY_STOP_ONCE
 expect_success "$installer" --upgrade "${base_args[@]}" --ref "$legacy_ref" >/dev/null
 grep -Fxq 'stop ci-fleet-capacity.service' "$FAKE_SYSTEMCTL_LOG" || fail 'legacy upgrade did not stop the in-flight capacity service'
 unset FAKE_SYSTEMCTL_LOG
 [[ $(readlink -f "$adopt_root/opt/ci-fleet/current") == "$adopt_root/opt/ci-fleet/releases/$legacy_engine_ref" ]] || fail 'upgrade could not restore a pre-health-contract engine'
+[[ $(readlink -f "$adopt_root/opt/ci-fleet/manager/current") == "$adopt_root/opt/ci-fleet/manager/releases/$engine_ref" ]] || fail 'legacy downgrade replaced the forward-compatible manager'
+[[ ! -e "$adopt_root/etc/systemd/system/ci-fleet-capacity.timer" ]] || fail 'legacy runtime retained capacity units'
+retained_installer=$adopt_root/opt/ci-fleet/manager/current/scripts/install-worker-controller.sh
+expect_success "$retained_installer" --upgrade "${base_args[@]}" --ref "$ref_one" >/dev/null
+[[ -f "$adopt_root/etc/systemd/system/ci-fleet-capacity.timer" ]] || fail 'upgrade back from legacy did not restore capacity units'
+grep -Fq 'CI_FLEET_POOL=trusted-ci' "$adopt_root/etc/ci-fleet/ci-fleet.env" || fail 'upgrade back from legacy did not restore the telemetry pool'
 
 grep -Fq 'Issue #7' "$repo_root/docs/DESIGN-DECISIONS.md" || fail 'isolated proof approval is not recorded'
 if grep -Fq '/etc/ci-fleet/ci-fleet.env.before-max2' "$repo_root/docs/CAPACITY-PROMOTION.md"; then fail 'capacity runbook still edits rendered host state'; fi
diff --git a/scripts/test_health.py b/scripts/test_health.py
index 960f328..601056e 100644
--- a/scripts/test_health.py
+++ b/scripts/test_health.py
@@ -216,6 +216,12 @@ def test_capacity_prunes_inactive_history_and_ignores_incomplete_samples(self) -
                 redirected.symlink_to(other, target_is_directory=True)
                 with self.assertRaisesRegex(ValueError, "directory must be root-owned"):
                     health.record_capacity(redirected / "samples.jsonl", {"runners": {"current": 0}}, now=1_000_000)
+                dangling_target = Path(directory) / "outside.jsonl"
+                dangling = history.parent / "dangling.jsonl"
+                dangling.symlink_to(dangling_target)
+                with self.assertRaisesRegex(ValueError, "history must be root-owned"):
+                    health.record_capacity(dangling, {"runners": {"current": 0}}, now=1_000_000)
+                self.assertFalse(dangling_target.exists())
             finally:
                 if previous_testing is None:
                     os.environ.pop("CI_FLEET_TESTING", None)

From 3df6cb4d48f860c18aea66e3048c04ef18918b1c Mon Sep 17 00:00:00 2001
From: Nicks Hermes <nicks-hermes@invalid.example>
Date: Sun, 16 Aug 2026 01:53:49 -0500
Subject: [PATCH 7/8] test: restore current engine fixture after downgrade

---
 scripts/test-install-worker-controller.sh | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/scripts/test-install-worker-controller.sh b/scripts/test-install-worker-controller.sh
index 5ebf4e4..9465fcf 100755
--- a/scripts/test-install-worker-controller.sh
+++ b/scripts/test-install-worker-controller.sh
@@ -711,6 +711,9 @@ unset FAKE_SYSTEMCTL_LOG
 [[ $(readlink -f "$adopt_root/opt/ci-fleet/manager/current") == "$adopt_root/opt/ci-fleet/manager/releases/$engine_ref" ]] || fail 'legacy downgrade replaced the forward-compatible manager'
 [[ ! -e "$adopt_root/etc/systemd/system/ci-fleet-capacity.timer" ]] || fail 'legacy runtime retained capacity units'
 retained_installer=$adopt_root/opt/ci-fleet/manager/current/scripts/install-worker-controller.sh
+export FAKE_ENGINE_REF=$engine_ref
+export FAKE_RUNNER_IMAGE=ci-fleet-runner:${engine_ref:0:12}
+export FAKE_CONTROLLER_IMAGE=ci-fleet-controller:${engine_ref:0:12}
 expect_success "$retained_installer" --upgrade "${base_args[@]}" --ref "$ref_one" >/dev/null
 [[ -f "$adopt_root/etc/systemd/system/ci-fleet-capacity.timer" ]] || fail 'upgrade back from legacy did not restore capacity units'
 grep -Fq 'CI_FLEET_POOL=trusted-ci' "$adopt_root/etc/ci-fleet/ci-fleet.env" || fail 'upgrade back from legacy did not restore the telemetry pool'

From 71f56fb85bd2281ad669edaaf2fda6009cbca5f8 Mon Sep 17 00:00:00 2001
From: Nicks Hermes <nicks-hermes@invalid.example>
Date: Sun, 16 Aug 2026 02:37:58 -0500
Subject: [PATCH 8/8] fix: propagate capacity stop failures during rollback

---
 scripts/install-worker-controller.sh      | 8 ++++++--
 scripts/test-install-worker-controller.sh | 7 +++++++
 2 files changed, 13 insertions(+), 2 deletions(-)

diff --git a/scripts/install-worker-controller.sh b/scripts/install-worker-controller.sh
index 0822318..dda8853 100755
--- a/scripts/install-worker-controller.sh
+++ b/scripts/install-worker-controller.sh
@@ -828,7 +828,9 @@ install_systemd_units() {
     install -m 0644 "$source/host/systemd/ci-fleet-capacity.timer" "$systemd_dir/"
   else
     systemctl disable --now ci-fleet-capacity.timer >/dev/null 2>&1 || true
-    [[ ! -f "$systemd_dir/ci-fleet-capacity.service" ]] || systemctl stop ci-fleet-capacity.service
+    if [[ -f "$systemd_dir/ci-fleet-capacity.service" ]] && ! systemctl stop ci-fleet-capacity.service; then
+      return 1
+    fi
     rm -f "$systemd_dir/ci-fleet-capacity.service" "$systemd_dir/ci-fleet-capacity.timer"
   fi
   install -m 0644 "$source/host/systemd/ci-fleet-cleanup.service" "$systemd_dir/"
@@ -845,7 +847,9 @@ install_systemd_units() {
 remove_systemd_units() {
   systemctl disable --now "${timer_names[@]}" >/dev/null 2>&1 || true
   systemctl disable --now ci-fleet-capacity.timer >/dev/null 2>&1 || true
-  [[ ! -f "$systemd_dir/ci-fleet-capacity.service" ]] || systemctl stop ci-fleet-capacity.service
+  if [[ -f "$systemd_dir/ci-fleet-capacity.service" ]] && ! systemctl stop ci-fleet-capacity.service; then
+    return 1
+  fi
   local unit
   for unit in "${optional_unit_names[@]}"; do
     case "$unit" in *.timer) systemctl disable --now "$unit" >/dev/null 2>&1 || true ;; esac
diff --git a/scripts/test-install-worker-controller.sh b/scripts/test-install-worker-controller.sh
index 9465fcf..d093c12 100755
--- a/scripts/test-install-worker-controller.sh
+++ b/scripts/test-install-worker-controller.sh
@@ -717,6 +717,13 @@ export FAKE_CONTROLLER_IMAGE=ci-fleet-controller:${engine_ref:0:12}
 expect_success "$retained_installer" --upgrade "${base_args[@]}" --ref "$ref_one" >/dev/null
 [[ -f "$adopt_root/etc/systemd/system/ci-fleet-capacity.timer" ]] || fail 'upgrade back from legacy did not restore capacity units'
 grep -Fq 'CI_FLEET_POOL=trusted-ci' "$adopt_root/etc/ci-fleet/ci-fleet.env" || fail 'upgrade back from legacy did not restore the telemetry pool'
+rollback_stop_ref=$(write_config active 2 2)
+export FAKE_FAIL_UP_ONCE=$tmp/rollback-stop-fail-up
+export FAKE_FAIL_CAPACITY_STOP_ONCE=$tmp/rollback-stop-failure
+: >"$FAKE_FAIL_UP_ONCE"
+: >"$FAKE_FAIL_CAPACITY_STOP_ONCE"
+expect_failure 'ROLLBACK_FAILED' "$retained_installer" --upgrade "${base_args[@]}" --ref "$rollback_stop_ref"
+unset FAKE_FAIL_UP_ONCE FAKE_FAIL_CAPACITY_STOP_ONCE
 
 grep -Fq 'Issue #7' "$repo_root/docs/DESIGN-DECISIONS.md" || fail 'isolated proof approval is not recorded'
 if grep -Fq '/etc/ci-fleet/ci-fleet.env.before-max2' "$repo_root/docs/CAPACITY-PROMOTION.md"; then fail 'capacity runbook still edits rendered host state'; fi