diff --git a/deploy/status-receiver/ci-fleet-status-receiver.service b/deploy/status-receiver/ci-fleet-status-receiver.service
new file mode 100644
index 00000000..bd49bf2e
--- /dev/null
+++ b/deploy/status-receiver/ci-fleet-status-receiver.service
@@ -0,0 +1,32 @@
+[Unit]
+Description=ci-fleet authenticated status receiver
+After=network.target
+
+[Service]
+Type=simple
+User=ci-fleet-status
+Group=ci-fleet-status
+ExecStart=/usr/bin/python3 /opt/ci-fleet-status/current/status_receiver.py --auth-config /etc/ci-fleet-status/auth.json --database /var/lib/ci-fleet-status/status.db --bind 127.0.0.1 --port 8080
+Restart=on-failure
+RestartSec=5s
+LogRateLimitIntervalSec=30s
+LogRateLimitBurst=20
+UMask=0077
+NoNewPrivileges=yes
+ProtectSystem=strict
+ProtectHome=yes
+PrivateTmp=yes
+PrivateDevices=yes
+ProtectKernelTunables=yes
+ProtectKernelModules=yes
+ProtectControlGroups=yes
+RestrictSUIDSGID=yes
+LockPersonality=yes
+MemoryDenyWriteExecute=yes
+CapabilityBoundingSet=
+RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
+ReadOnlyPaths=/etc/ci-fleet-status
+ReadWritePaths=/var/lib/ci-fleet-status
+
+[Install]
+WantedBy=multi-user.target
diff --git a/deploy/status-receiver/nginx-location.conf.example b/deploy/status-receiver/nginx-location.conf.example
new file mode 100644
index 00000000..5e6a4987
--- /dev/null
+++ b/deploy/status-receiver/nginx-location.conf.example
@@ -0,0 +1,10 @@
+# Terminate TLS in the existing reverse proxy. Keep the application on loopback.
+location / {
+    client_max_body_size 32k;
+    proxy_connect_timeout 5s;
+    proxy_read_timeout 20s;
+    proxy_send_timeout 20s;
+    proxy_pass http://127.0.0.1:8080;
+    proxy_set_header Host $host;
+    proxy_set_header X-Forwarded-Proto https;
+}
diff --git a/docs/README.md b/docs/README.md
index bf5d1e60..72c197cc 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -62,6 +62,7 @@ These pages are normative for compatible projects and hosts:
 - [Host maintenance standard](HOST-MAINTENANCE.md)
 - [Fleet health monitoring](HEALTH-MONITORING.md)
 - [Authenticated controller status reporting](STATUS-REPORTING.md)
+- [Status receiver deployment](STATUS-RECEIVER-DEPLOYMENT.md)
 - [Git-authored controller desired state](DESIRED-STATE.md)
 - [Secrets model](SECRETS.md)
 - [Security policy](../SECURITY.md)
diff --git a/docs/STATUS-RECEIVER-DEPLOYMENT.md b/docs/STATUS-RECEIVER-DEPLOYMENT.md
new file mode 100644
index 00000000..1cdfddca
--- /dev/null
+++ b/docs/STATUS-RECEIVER-DEPLOYMENT.md
@@ -0,0 +1,165 @@
+# Status receiver deployment
+
+This runbook prepares the existing authenticated status receiver for a dedicated
+Linux host. It does not create a host, provision credentials, change a controller,
+or deploy anything by itself.
+
+## Target and boundaries
+
+Use a small always-on VM or LXC separate from every runner controller: 2 vCPU,
+2–4 GiB RAM, and about 20 GiB disk. Confirm the next unused infrastructure ID
+from the live hypervisor inventory immediately before creation; repository state
+is not inventory evidence.
+
+The receiver host has no Docker socket, runner credentials, deployment
+credentials, or inbound connection to a controller. The Python application binds
+only to `127.0.0.1:8080`; an existing reverse proxy terminates HTTPS. Controllers
+submit outbound HTTPS. The read API is authenticated and has no mutation route.
+
+## Install from a reviewed commit
+
+On the prepared receiver host, use the supported `/usr/bin/python3` version 3.9
+or newer, check out the exact reviewed commit, and verify a clean tree. The
+installer creates the unprivileged `ci-fleet-status` account, release and state
+directories, a hardened systemd unit, and an atomic `current` link. It does not
+create credentials.
+
+```bash
+ref=$(git rev-parse HEAD)
+test -z "$(git status --porcelain)"
+sudo ./scripts/install-status-receiver.sh --install --ref "$ref"
+```
+
+A second identical invocation returns `NO_CHANGE`. Before activation, verify the
+unit definition:
+
+```bash
+sudo systemd-analyze verify \
+  /etc/systemd/system/ci-fleet-status-receiver.service
+```
+
+## One-time secret provisioning boundary
+
+Provision one independent 32–128 byte signing key per controller and one distinct
+32–128 byte visible-ASCII read token. Values never belong in Git, command
+arguments, chat, logs, issues, PRs, fixtures, or artifacts.
+
+An authorized human uses an approved secret manager or controlled provisioning
+workstation to place the same controller key at these host-local paths:
+
+- receiver: `/etc/ci-fleet-status/controller-keys/<controller-id>.key`;
+- controller: `/etc/ci-fleet/secrets/status-reporting.key`.
+
+The read token exists only at `/etc/ci-fleet-status/read-api.token`. On the
+receiver, every key, token, and `auth.json` is owned by `ci-fleet-status` with
+mode `0600`; both containing directories are mode `0700`. On the controller, the
+signing key is root-owned mode `0600`. Verify ownership, type, and mode without
+printing content. Stop and remove only the newly provisioned files if any check
+fails. Delete any provisioning-workstation copy after both destinations are
+verified. Do not enable SSH to provision or verify the controller.
+
+Create receiver-local `auth.json` with an editor that does not log content. It
+contains only controller-to-key-path mappings and the read-token path; use the
+fictional shape in [status reporting](STATUS-REPORTING.md#receiver). Never put a
+secret value in that JSON file.
+
+## HTTPS and activation
+
+Install the location block from
+`deploy/status-receiver/nginx-location.conf.example` in the existing HTTPS
+reverse proxy. Supply the real public certificate and private endpoint only in
+private infrastructure configuration. Do not expose port 8080.
+
+After receiver-local credential metadata and reverse-proxy configuration pass:
+
+```bash
+sudo systemctl daemon-reload
+sudo systemctl enable --now ci-fleet-status-receiver.service
+sudo systemctl is-active --quiet ci-fleet-status-receiver.service
+sudo ./scripts/install-status-receiver.sh --check
+python3 - <<'PY'
+import json
+import urllib.request
+
+response = urllib.request.urlopen("http://127.0.0.1:8080/healthz", timeout=5)
+assert json.load(response) == {"status": "ok"}
+PY
+```
+
+Configure the controller's private monitoring policy with the HTTPS
+`/v1/status` URL and its host-local signing-key path. Preserve its existing
+identity, routing, capacity, resources, scale-to-zero behavior, and disabled SSH.
+A reporting failure must remain warning-only and must not interrupt runner
+management or reconciliation.
+
+## Verification
+
+1. Submit one scheduled report and confirm HTTP 202 without printing its body or
+   authorization headers.
+2. Read `/v1/controllers/<controller-id>` with the read token loaded from its
+   file by the client process, not placed in an argument or environment dump.
+3. Confirm an invalid signature, stale timestamp, replayed nonce, wrong
+   controller identity, and oversized payload are rejected.
+4. Restart the receiver and confirm `/healthz`, authenticated reads, and retained
+   bounded history recover.
+5. Stop the receiver for longer than one reporting interval. Confirm the
+   controller records only a reporting warning and continues reconciliation and
+   runner lifecycle; then restart the receiver and confirm reporting resumes.
+6. Stop the controller or take it offline. Confirm external monitoring of the
+   separate receiver still works and alerts on the latest report age.
+7. Confirm the receiver listens only on loopback and port 8080 is unreachable
+   externally. Confirm SSH remains disabled on the controller.
+
+The receiver suppresses request logs. Keep systemd journal retention bounded by
+the host's reviewed journald policy and monitor service restart count. SQLite
+retention is enforced independently by age and per-controller count. The status
+database is disposable. For database corruption, stop the service, quarantine the
+database and its journal files outside service-writable state, and start with a
+fresh database:
+
+```bash
+sudo systemctl stop ci-fleet-status-receiver.service
+quarantine="/var/lib/ci-fleet-status-installer/quarantine/$(date -u +%Y%m%dT%H%M%SZ)"
+sudo install -d -o root -g root -m 0700 "$quarantine"
+sudo find /var/lib/ci-fleet-status -maxdepth 1 -type f -name 'status.db*' \
+  -exec mv -t "$quarantine" -- {} +
+sudo systemctl start ci-fleet-status-receiver.service
+```
+
+A full host-loss recovery reinstalls the reviewed release and reprovisions
+credentials. Back up the database only if an operator separately decides that
+short status history is durable evidence.
+
+## Upgrade and rollback
+
+From a clean checkout at the newer reviewed commit:
+
+```bash
+ref=$(git rev-parse HEAD)
+sudo ./scripts/install-status-receiver.sh --upgrade --ref "$ref"
+sudo ./scripts/install-status-receiver.sh --check
+```
+
+The upgrade stages an immutable release, records the previous release, switches
+the symlink atomically, and restarts only an already-active service. If health,
+ingestion, read access, or retention verification fails:
+
+```bash
+sudo ./scripts/install-status-receiver.sh --rollback
+sudo ./scripts/install-status-receiver.sh --check
+```
+
+Rollback restores the selected release's application files and systemd unit. It
+preserves `auth.json`, keys, read token, database, reverse-proxy configuration,
+and journal policy. Reverse-proxy or schema changes require their own reviewed
+compatibility and rollback plan.
+
+## Test coverage
+
+`scripts/test-install-status-receiver.sh` exercises clean install, idempotent
+rerun, upgrade, check, and rollback in an isolated root. Receiver tests cover
+restart/key reload, incorrect secret permissions, authentication, controller
+isolation, replay/freshness, request and payload bounds, strict schema/redaction,
+retention, loopback binding, and read-only routes. Health and installer tests
+cover warning-only reporter outages, disabled SSH reporting, and preservation of
+runner lifecycle during delivery failure.
diff --git a/docs/STATUS-REPORTING.md b/docs/STATUS-REPORTING.md
index c37a530c..69b2d8b5 100644
--- a/docs/STATUS-REPORTING.md
+++ b/docs/STATUS-REPORTING.md
@@ -86,6 +86,10 @@ CI_FLEET_HEALTH_STATUS_KEY_FILE=/etc/ci-fleet/secrets/status-reporting.key
 ```
 
 The URL must be HTTPS with the exact `/v1/status` path and no embedded credentials, query, or fragment.
+Schema-v3 desired state may require reporting with only the fixed host-local
+configuration reference `/etc/ci-fleet/monitoring.env`; endpoint and key values
+remain outside Git. A required but missing or unsafe host-local configuration is
+reported as a redacted delivery warning and does not interrupt runner lifecycle.
 
 ## Threat model
 
@@ -121,3 +125,7 @@ Reports never contain:
 - Docker socket access or any host-control capability.
 
 The local full health result remains available for recovery, but only the status schema's allowlisted summary leaves the controller.
+
+For dedicated-host packaging, one-time secret boundaries, activation,
+verification, upgrade, and rollback, see
+[Status receiver deployment](STATUS-RECEIVER-DEPLOYMENT.md).
diff --git a/engine-capabilities.json b/engine-capabilities.json
new file mode 100644
index 00000000..12c4d6de
--- /dev/null
+++ b/engine-capabilities.json
@@ -0,0 +1,7 @@
+{
+  "schema_version": 1,
+  "capabilities": {
+    "status_reporting_config": true,
+    "required_status_reporting": true
+  }
+}
diff --git a/scripts/desired_state.py b/scripts/desired_state.py
index 42c66bbd..f19af22e 100755
--- a/scripts/desired_state.py
+++ b/scripts/desired_state.py
@@ -28,12 +28,40 @@
     "CI_FLEET_GITHUB_APP_PRIVATE_KEY_FILE",
 }
 HOST_OPTIONAL = {"CI_FLEET_RUNNER_TTL"}
+REQUIRED_STATUS_CAPABILITY = "required_status_reporting"
+STATUS_REPORTING_CONFIG_CAPABILITY = "status_reporting_config"
 
 
 class DesiredStateError(ValueError):
     """A safe operator-facing desired-state error."""
 
 
+def load_engine_capabilities(path: Path) -> set[str]:
+    def reject_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
+        value: dict[str, Any] = {}
+        for name, item in pairs:
+            if name in value:
+                raise ValueError("duplicate capability key")
+            value[name] = item
+        return value
+
+    try:
+        metadata = path.lstat()
+        if not stat.S_ISREG(metadata.st_mode):
+            raise DesiredStateError("engine capability declaration must be a regular file")
+        value = json.loads(path.read_text(encoding="utf-8"), object_pairs_hook=reject_duplicates)
+    except FileNotFoundError as exc:
+        raise DesiredStateError("engine capability declaration is missing") from exc
+    except (json.JSONDecodeError, ValueError) as exc:
+        raise DesiredStateError("engine capability declaration is malformed") from exc
+    if not isinstance(value, dict) or set(value) != {"schema_version", "capabilities"} or value.get("schema_version") != 1:
+        raise DesiredStateError("engine capability declaration is malformed")
+    capabilities = value.get("capabilities")
+    if not isinstance(capabilities, dict) or any(type(supported) is not bool for supported in capabilities.values()):
+        raise DesiredStateError("engine capability declaration is malformed")
+    return {name for name, supported in capabilities.items() if supported}
+
+
 def load_template_validator():
     spec = importlib.util.spec_from_file_location("ci_fleet_template_validator", TEMPLATE_VALIDATOR)
     if spec is None or spec.loader is None:
@@ -127,6 +155,7 @@ def build_rendered_env(
     config_repository: str,
     config_ref: str,
     docker_gid: int,
+    engine_capabilities: set[str] | None = None,
 ) -> tuple[dict[str, str], dict[str, Any]]:
     controller, pool = select_controller(config, controller_id)
     engine_commit = controller["engine_ref"]
@@ -165,6 +194,14 @@ def build_rendered_env(
         "CI_FLEET_VERSION": short_commit,
         **validate_host_values(host_values),
     }
+    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()):
+        raise DesiredStateError("selected engine does not advertise required status reporting")
+    if reporting_configured and STATUS_REPORTING_CONFIG_CAPABILITY not in (engine_capabilities or set()):
+        raise DesiredStateError("selected engine does not support status reporting configuration")
+    if reporting_required:
+        rendered["CI_FLEET_STATUS_REPORTING_REQUIRED"] = "1"
     for name, value in rendered.items():
         if not SAFE_ENV_VALUE.fullmatch(value):
             raise DesiredStateError(f"rendered value for {name} contains unsafe characters")
@@ -183,6 +220,8 @@ def build_rendered_env(
         "config_ref": config_ref,
         "engine_ref": engine_commit,
         "engine_repository": config["organization"]["delivery_engine"],
+        "status_reporting_configured": reporting_configured,
+        "status_reporting_required": reporting_required,
     }
     return rendered, metadata
 
@@ -221,6 +260,7 @@ def command_extract_host(args: argparse.Namespace) -> None:
 def command_render(args: argparse.Namespace) -> None:
     config = load_and_validate_config(args.config)
     host_values = parse_env(args.host_config, allow_unknown=False)
+    capabilities = load_engine_capabilities(args.engine_capabilities) if args.engine_capabilities else set()
     values, metadata = build_rendered_env(
         config,
         args.controller,
@@ -228,6 +268,7 @@ def command_render(args: argparse.Namespace) -> None:
         config_repository=args.config_repository,
         config_ref=args.config_ref,
         docker_gid=args.docker_gid,
+        engine_capabilities=capabilities,
     )
     write_private(args.output, render_env(values))
     write_private(args.metadata_output, json.dumps(metadata, indent=2, sort_keys=True) + "\n")
@@ -238,6 +279,22 @@ def command_render(args: argparse.Namespace) -> None:
     )
 
 
+def command_engine(args: argparse.Namespace) -> None:
+    config = load_and_validate_config(args.config)
+    controller, _ = select_controller(config, args.controller)
+    print(controller["engine_ref"])
+    print(config["organization"]["delivery_engine"])
+
+
+def command_validate_engine_capabilities(args: argparse.Namespace) -> None:
+    capabilities = load_engine_capabilities(args.manifest)
+    if args.require_status_reporting_config and STATUS_REPORTING_CONFIG_CAPABILITY not in capabilities:
+        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")
+    print("ENGINE_CAPABILITIES_OK")
+
+
 def parse_args() -> argparse.Namespace:
     parser = argparse.ArgumentParser(description=__doc__)
     subparsers = parser.add_subparsers(dest="command", required=True)
@@ -258,9 +315,21 @@ def parse_args() -> argparse.Namespace:
     render.add_argument("--config-repository", required=True)
     render.add_argument("--config-ref", required=True)
     render.add_argument("--docker-gid", type=int, required=True)
+    render.add_argument("--engine-capabilities", type=Path)
     render.add_argument("--output", type=Path, required=True)
     render.add_argument("--metadata-output", type=Path, required=True)
     render.set_defaults(function=command_render)
+
+    engine = subparsers.add_parser("engine", help="select the immutable engine for one controller")
+    engine.add_argument("--config", type=Path, required=True)
+    engine.add_argument("--controller", required=True)
+    engine.set_defaults(function=command_engine)
+
+    capabilities = subparsers.add_parser("validate-engine-capabilities", help="validate an engine capability declaration")
+    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.set_defaults(function=command_validate_engine_capabilities)
     return parser.parse_args()
 
 
diff --git a/scripts/health.py b/scripts/health.py
index 7f847fbf..f47b9e1d 100644
--- a/scripts/health.py
+++ b/scripts/health.py
@@ -557,11 +557,12 @@ def collect_snapshot(values: dict[str, str], *, root: Path = Path("/"), run: Run
 
 
 def load_monitoring_config(path: Path) -> dict[str, str]:
-    if not path.exists():
+    try:
+        info = path.lstat()
+    except FileNotFoundError:
         return {}
-    info = path.stat()
     expected_owner = os.getuid() if os.environ.get("CI_FLEET_TESTING") == "1" else 0
-    if info.st_uid != expected_owner or stat.S_IMODE(info.st_mode) & 0o077:
+    if not stat.S_ISREG(info.st_mode) or info.st_uid != expected_owner or stat.S_IMODE(info.st_mode) & 0o077:
         raise ValueError(f"monitoring configuration must be root-owned mode 0600: {path}")
     values: dict[str, str] = {}
     for number, raw in enumerate(path.read_text().splitlines(), 1):
@@ -671,18 +672,33 @@ def _send_heartbeat(
 
 
 def _local(args: argparse.Namespace) -> int:
-    values = dict(os.environ)
-    values.update(load_monitoring_config(args.monitoring_config))
+    environment = dict(os.environ)
+    values = dict(environment)
+    config_invalid = False
+    try:
+        values.update(load_monitoring_config(args.monitoring_config))
+        thresholds = thresholds_from(values)
+    except (OSError, UnicodeError, ValueError):
+        if environment.get("CI_FLEET_STATUS_REPORTING_REQUIRED") != "1":
+            raise
+        config_invalid = True
+        values = environment
+        thresholds = thresholds_from(values)
     snapshot = collect_snapshot(values)
-    report = evaluate(snapshot, thresholds_from(values))
+    report = evaluate(snapshot, thresholds)
     now = int(time.time())
     report["timestamp"] = now
     delivery = 0
-    if values.get("CI_FLEET_HEALTH_SUPPRESS_DELIVERY") != "1":
-        delivery = (
-            _send_status(values, build_status_report(snapshot, report, generated_at=now), now=now)
-            if values.get("CI_FLEET_HEALTH_STATUS_URL") else _send_heartbeat(values, report)
-        )
+    if environment.get("CI_FLEET_HEALTH_SUPPRESS_DELIVERY") != "1":
+        if config_invalid:
+            delivery = 1
+        elif values.get("CI_FLEET_HEALTH_STATUS_URL") or values.get("CI_FLEET_STATUS_REPORTING_REQUIRED") == "1":
+            delivery = (
+                _send_status(values, build_status_report(snapshot, report, generated_at=now), now=now)
+                if values.get("CI_FLEET_HEALTH_STATUS_URL") else 1
+            )
+        else:
+            delivery = _send_heartbeat(values, report)
     if delivery:
         severity = "critical" if delivery == 2 else "warning"
         report["checks"].append({"id": "status_delivery", "status": severity})
diff --git a/scripts/install-status-receiver.sh b/scripts/install-status-receiver.sh
new file mode 100755
index 00000000..baccb466
--- /dev/null
+++ b/scripts/install-status-receiver.sh
@@ -0,0 +1,384 @@
+#!/usr/bin/env bash
+set -Eeuo pipefail
+
+usage() {
+  echo "usage: $0 (--install|--upgrade|--rollback|--check) [--ref COMMIT]" >&2
+  exit 2
+}
+
+mode=
+ref=
+while (($#)); do
+  case "$1" in
+    --install|--upgrade|--rollback|--check) [[ -z "$mode" ]] || usage; mode=${1#--}; shift ;;
+    --ref) (($# >= 2)) || usage; ref=$2; shift 2 ;;
+    *) usage ;;
+  esac
+done
+[[ -n "$mode" ]] || usage
+
+root=${CI_FLEET_STATUS_ROOT:-}
+test_mode=${CI_FLEET_STATUS_TEST_MODE:-0}
+if [[ (-n "$root" && "$test_mode" != 1) || (-z "$root" && "$test_mode" != 0) ]]; then
+  echo "alternate root requires CI_FLEET_STATUS_TEST_MODE=1 and a nonempty root" >&2
+  exit 1
+fi
+if [[ -z "$root" && $EUID -ne 0 ]]; then
+  echo "run as root" >&2
+  exit 1
+fi
+
+repo_root=$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)
+install_root="$root/opt/ci-fleet-status"
+state_root="$root/var/lib/ci-fleet-status"
+metadata_root="$root/var/lib/ci-fleet-status-installer"
+config_root="$root/etc/ci-fleet-status"
+unit_path="$root/etc/systemd/system/ci-fleet-status-receiver.service"
+current="$install_root/current"
+unit_target="$current/ci-fleet-status-receiver.service"
+previous="$metadata_root/previous-ref"
+restart_required="$metadata_root/restart-required"
+previous_was_active="$metadata_root/previous-was-active"
+mkdir -p "$root/run/lock"
+lock_directory="$root/run/lock/ci-fleet-status"
+expected_lock_uid=0
+[[ -z "$root" ]] || expected_lock_uid=$EUID
+if ! mkdir -m 0700 "$lock_directory" 2>/dev/null; then
+  [[ -d "$lock_directory" && ! -L "$lock_directory" ]]
+  [[ $(stat -c '%u:%a' "$lock_directory") == "$expected_lock_uid:700" ]]
+fi
+exec 9<"$lock_directory"
+flock 9
+
+current_ref() {
+  local link release releases target
+  [[ -L "$current" ]] || return 1
+  link=$(readlink "$current")
+  [[ "$link" =~ ^releases/([0-9a-f]{40})$ ]] || return 1
+  release=${BASH_REMATCH[1]}
+  releases=$(realpath -e "$install_root/releases") || return 1
+  target=$(realpath -e "$current") || return 1
+  [[ -d "$target" && $(dirname "$target") == "$releases" && "$target" == "$releases/$release" ]] || return 1
+  printf '%s\n' "$release"
+}
+
+write_metadata() {
+  local destination=$1 value=$2 temporary
+  temporary=$(mktemp "$metadata_root/.metadata.XXXXXX")
+  printf '%s\n' "$value" >"$temporary"
+  chmod 0600 "$temporary"
+  mv -Tf "$temporary" "$destination"
+}
+
+link_unit() {
+  local installed
+  installed=$(current_ref) || { echo "receiver is not installed" >&2; exit 1; }
+  validate_release "$install_root/releases/$installed"
+  ensure_systemd_directory
+  ln -sfn "$unit_target" "$unit_path.new"
+  mv -Tf "$unit_path.new" "$unit_path"
+}
+
+activate() {
+  local target=$1 record_previous=${2:-1} old=
+  validate_release "$install_root/releases/$target" || { echo "release not installed safely: $target" >&2; exit 1; }
+  old=$(current_ref || true)
+  if [[ "$record_previous" == 1 && -n "$old" && "$old" != "$target" ]]; then
+    write_metadata "$previous" "$old"
+  fi
+  ln -sfn "releases/$target" "$current.new"
+  mv -Tf "$current.new" "$current"
+}
+
+restart_live_service() {
+  local force=${1:-0}
+  local installed
+  installed=$(current_ref) || { echo "receiver is not installed" >&2; exit 1; }
+  validate_release "$install_root/releases/$installed"
+  ensure_systemd_directory
+  if [[ -n "$root" ]]; then
+    if [[ -n ${CI_FLEET_STATUS_TEST_FAIL_RELOAD_ONCE:-} && -f "$CI_FLEET_STATUS_TEST_FAIL_RELOAD_ONCE" ]]; then
+      rm -f "$CI_FLEET_STATUS_TEST_FAIL_RELOAD_ONCE"
+      return 1
+    fi
+    printf '%s\n' "$force" >"$root/run/ci-fleet-status-last-restart-force"
+    return
+  fi
+  systemctl daemon-reload
+  if [[ "$force" == 1 ]] || systemctl is-active --quiet ci-fleet-status-receiver.service; then
+    systemctl restart ci-fleet-status-receiver.service
+  fi
+}
+
+managed_uid=0
+managed_gid=0
+expected_release_uid=0
+if [[ -n "$root" ]]; then
+  [[ ${CI_FLEET_STATUS_TEST_EXPECTED_OWNER:-} =~ ^[0-9]+$ ]] || {
+    echo "test mode requires CI_FLEET_STATUS_TEST_EXPECTED_OWNER" >&2
+    exit 1
+  }
+  managed_uid=$EUID
+  managed_gid=$(id -g)
+  expected_release_uid=$CI_FLEET_STATUS_TEST_EXPECTED_OWNER
+fi
+managed_directory() {
+  local path=$1 create=${2:-0}
+  if [[ -L "$path" ]]; then
+    echo "unsafe managed release directory: $path" >&2
+    exit 1
+  elif [[ -e "$path" ]]; then
+    [[ -d "$path" && $(stat -c '%u:%a' "$path") == "$managed_uid:755" ]] || {
+      echo "unsafe managed release directory: $path" >&2
+      exit 1
+    }
+  elif [[ "$create" == 1 ]]; then
+    install -d -m 0755 "$path"
+    [[ $(stat -c '%u:%a' "$path") == "$managed_uid:755" ]]
+  else
+    return 1
+  fi
+}
+
+ensure_systemd_directory() {
+  local path mode owner
+  path=$(dirname "$unit_path")
+  if [[ -L "$path" ]]; then
+    echo "unsafe systemd directory: $path" >&2
+    exit 1
+  elif [[ ! -e "$path" ]]; then
+    [[ "$test_mode" == 1 ]] || { echo "systemd directory is missing: $path" >&2; exit 1; }
+    install -d -m 0755 "$path"
+  fi
+  [[ -d "$path" ]] || { echo "unsafe systemd directory: $path" >&2; exit 1; }
+  owner=$(stat -c %u "$path")
+  mode=$(stat -c %a "$path")
+  [[ "$owner" == "$managed_uid" && $((8#$mode & 0300)) == $((8#0300)) && $((8#$mode & 022)) == 0 ]] || {
+    echo "unsafe systemd directory: $path" >&2
+    exit 1
+  }
+}
+
+validate_release() {
+  local release=$1 entry expected name mode stored_digest actual_digest
+  local -a entries=()
+  [[ ! -L "$release" && -d "$release" && $(stat -c '%u:%a' "$release") == "$expected_release_uid:755" ]] || {
+    echo "unsafe receiver release: $release" >&2
+    return 1
+  }
+  mapfile -d '' entries < <(find "$release" -mindepth 1 -maxdepth 1 -print0)
+  ((${#entries[@]} == 4)) || { echo "unexpected receiver release contents: $release" >&2; return 1; }
+  for expected in status_receiver.py:755 status_auth.py:644 ci-fleet-status-receiver.service:644 .ci-fleet-tree-sha256:644; do
+    name=${expected%%:*}
+    mode=${expected##*:}
+    entry=$release/$name
+    [[ ! -L "$entry" && -f "$entry" && $(stat -c '%u:%a' "$entry" 2>/dev/null) == "$expected_release_uid:$mode" ]] || {
+      echo "unsafe receiver artifact: $entry" >&2
+      return 1
+    }
+  done
+  stored_digest=$(<"$release/.ci-fleet-tree-sha256")
+  [[ "$stored_digest" =~ ^[0-9a-f]{64}$ ]] || { echo "invalid receiver release digest: $release" >&2; return 1; }
+  actual_digest=$(cd "$release" && sha256sum status_receiver.py status_auth.py ci-fleet-status-receiver.service | sha256sum | cut -d' ' -f1)
+  [[ "$actual_digest" == "$stored_digest" ]] || { echo "modified receiver release: $release" >&2; return 1; }
+}
+
+validate_service_account() {
+  local passwd_record=$1 group_record=$2 groups=$3 account uid gid home shell group group_gid
+  IFS=: read -r account _ uid gid _ home shell <<<"$passwd_record"
+  IFS=: read -r group _ group_gid _ <<<"$group_record"
+  [[ "$account" == ci-fleet-status && "$group" == ci-fleet-status ]]
+  [[ "$uid" =~ ^[0-9]+$ && "$gid" =~ ^[0-9]+$ && "$uid" != 0 && "$gid" != 0 && "$gid" == "$group_gid" ]]
+  [[ "$home" == /nonexistent && "$shell" == /usr/sbin/nologin ]]
+  [[ "$groups" == "$gid" ]] || { echo "ci-fleet-status has unexpected supplementary groups" >&2; return 1; }
+}
+
+ensure_owned_directory() {
+  local path=$1 uid=$2 gid=$3 expected_mode=$4 create=${5:-0}
+  if [[ -L "$path" ]]; then
+    echo "unsafe receiver directory: $path" >&2
+    return 1
+  elif [[ ! -e "$path" ]]; then
+    [[ "$create" == 1 ]] || { echo "receiver directory is missing: $path" >&2; return 1; }
+    if [[ -z "$root" ]]; then
+      install -d -o "$uid" -g "$gid" -m "$expected_mode" "$path"
+    else
+      install -d -m "$expected_mode" "$path"
+    fi
+  fi
+  [[ -d "$path" && $(stat -c '%u:%g:%a' "$path") == "$uid:$gid:$expected_mode" ]] || {
+    echo "unsafe receiver directory: $path" >&2
+    return 1
+  }
+}
+
+validate_runtime_directories() {
+  local create=${1:-0}
+  ensure_owned_directory "$state_root" "$service_uid" "$service_gid" 700 "$create"
+  ensure_owned_directory "$config_root" "$service_uid" "$service_gid" 700 "$create"
+  ensure_owned_directory "$metadata_root" "$managed_uid" "$managed_gid" 700 "$create"
+}
+
+python=/usr/bin/python3
+if [[ "$test_mode" == 1 && -n ${CI_FLEET_STATUS_TEST_PYTHON:-} ]]; then
+  python=$CI_FLEET_STATUS_TEST_PYTHON
+fi
+"$python" -c 'import sys; raise SystemExit(sys.version_info < (3, 9))' || {
+  echo "Python 3.9 or newer is required" >&2
+  exit 1
+}
+sqlite_version=$("$python" -c 'import sqlite3; print(".".join(map(str, sqlite3.sqlite_version_info)))') || {
+  echo "Python SQLite version detection failed" >&2
+  exit 1
+}
+if [[ "$test_mode" == 1 && -n ${CI_FLEET_STATUS_TEST_SQLITE_VERSION:-} ]]; then
+  sqlite_version=$CI_FLEET_STATUS_TEST_SQLITE_VERSION
+fi
+IFS=. read -r sqlite_major sqlite_minor sqlite_patch sqlite_extra <<<"$sqlite_version"
+if [[ -n ${sqlite_extra:-} || ! ${sqlite_major:-} =~ ^[0-9]+$ || ! ${sqlite_minor:-} =~ ^[0-9]+$ || ! ${sqlite_patch:-} =~ ^[0-9]+$ ]] ||
+   ((sqlite_major < 3 || (sqlite_major == 3 && sqlite_minor < 25))); then
+  echo "SQLite 3.25.0 or newer is required by the Python receiver (found $sqlite_version)" >&2
+  exit 1
+fi
+
+if [[ "$mode" == install || "$mode" == upgrade ]]; then
+  managed_directory "$install_root" 1
+  managed_directory "$install_root/releases" 1
+else
+  managed_directory "$install_root" || { echo "status receiver is not installed" >&2; exit 1; }
+  managed_directory "$install_root/releases" || { echo "status receiver is not installed" >&2; exit 1; }
+fi
+
+service_uid=$managed_uid
+service_gid=$managed_gid
+if [[ -n ${CI_FLEET_STATUS_TEST_ACCOUNT_GROUPS:-} ]]; then
+  validate_service_account 'ci-fleet-status:x:12345:12345::/nonexistent:/usr/sbin/nologin' \
+    'ci-fleet-status:x:12345:' "$CI_FLEET_STATUS_TEST_ACCOUNT_GROUPS"
+elif [[ -z "$root" && ("$mode" == check || "$mode" == rollback) ]]; then
+  passwd_record=$(getent passwd ci-fleet-status) || { echo "ci-fleet-status account is missing" >&2; exit 1; }
+  group_record=$(getent group ci-fleet-status) || { echo "ci-fleet-status group is missing" >&2; exit 1; }
+  validate_service_account "$passwd_record" "$group_record" "$(id -G ci-fleet-status)"
+  service_uid=$(id -u ci-fleet-status)
+  service_gid=$(id -g ci-fleet-status)
+fi
+
+if [[ "$mode" == check || "$mode" == rollback ]]; then
+  validate_runtime_directories
+fi
+
+if [[ "$mode" == check ]]; then
+  installed=$(current_ref) || { echo "receiver is not installed" >&2; exit 1; }
+  validate_release "$install_root/releases/$installed"
+  ensure_systemd_directory
+  [[ -L "$unit_path" && $(readlink "$unit_path") == "$unit_target" ]]
+  [[ ! -f "$restart_required" ]] || { echo "receiver restart is pending" >&2; exit 1; }
+  [[ "$test_mode" == 1 ]] || systemctl is-active --quiet ci-fleet-status-receiver.service || {
+    echo "status receiver service is not active" >&2
+    exit 1
+  }
+  echo "CHECK_OK $installed"
+  exit
+fi
+
+if [[ "$mode" == rollback ]]; then
+  [[ -s "$previous" ]] || { echo "no rollback release recorded" >&2; exit 1; }
+  target=$(<"$previous")
+  [[ "$target" =~ ^[0-9a-f]{40}$ ]] || { echo "invalid rollback release" >&2; exit 1; }
+  validate_release "$install_root/releases/$target"
+  ensure_systemd_directory
+  force=0
+  [[ -f "$restart_required" || -f "$previous_was_active" ]] && force=1
+  activate "$target" 0
+  link_unit
+  restart_live_service "$force"
+  rm -f "$restart_required" "$previous_was_active"
+  echo "ROLLED_BACK $target"
+  exit
+fi
+
+[[ "$ref" =~ ^[0-9a-f]{40}$ ]] || usage
+head=$(git -C "$repo_root" rev-parse HEAD)
+[[ "$head" == "$ref" ]] || { echo "--ref must equal the reviewed checkout HEAD" >&2; exit 1; }
+inputs=(
+  scripts/install-status-receiver.sh
+  scripts/status_auth.py
+  scripts/status_receiver.py
+  deploy/status-receiver/ci-fleet-status-receiver.service
+)
+git -C "$repo_root" diff --quiet HEAD -- "${inputs[@]}" || {
+  echo "reviewed receiver inputs differ from HEAD" >&2
+  exit 1
+}
+if [[ -z "$root" ]]; then
+  getent passwd ci-fleet-status >/dev/null || \
+    useradd --system --user-group --home /nonexistent --shell /usr/sbin/nologin ci-fleet-status
+  validate_service_account "$(getent passwd ci-fleet-status)" "$(getent group ci-fleet-status)" "$(id -G ci-fleet-status)"
+  service_uid=$(id -u ci-fleet-status)
+  service_gid=$(id -g ci-fleet-status)
+fi
+validate_runtime_directories 1
+ensure_systemd_directory
+existing=$(current_ref || true)
+if [[ "$mode" == install && -n "$existing" && "$existing" != "$ref" ]]; then
+  echo "use --upgrade to change an active release" >&2
+  exit 1
+fi
+release="$install_root/releases/$ref"
+if [[ -d "$release" ]]; then
+  validate_release "$release"
+  cmp -s "$repo_root/scripts/status_receiver.py" "$release/status_receiver.py"
+  cmp -s "$repo_root/scripts/status_auth.py" "$release/status_auth.py"
+  cmp -s "$repo_root/deploy/status-receiver/ci-fleet-status-receiver.service" \
+    "$release/ci-fleet-status-receiver.service"
+else
+  staging=$(mktemp -d "$install_root/releases/.staging.XXXXXX")
+  trap 'rm -rf "$staging"' EXIT
+  chmod 0755 "$staging"
+  install -m 0755 "$repo_root/scripts/status_receiver.py" "$staging/status_receiver.py"
+  install -m 0644 "$repo_root/scripts/status_auth.py" "$staging/status_auth.py"
+  install -m 0644 "$repo_root/deploy/status-receiver/ci-fleet-status-receiver.service" \
+    "$staging/ci-fleet-status-receiver.service"
+  (cd "$staging" && sha256sum status_receiver.py status_auth.py ci-fleet-status-receiver.service | sha256sum | cut -d' ' -f1) \
+    >"$staging/.ci-fleet-tree-sha256"
+  chmod 0644 "$staging/.ci-fleet-tree-sha256"
+  mv -T "$staging" "$release"
+  trap - EXIT
+fi
+validate_release "$release"
+
+if [[ "$existing" == "$ref" ]]; then
+  link_unit
+  if [[ -f "$restart_required" ]]; then
+    restart_live_service 1
+    rm -f "$restart_required"
+    echo UPGRADED
+    exit
+  fi
+  restart_live_service
+  echo NO_CHANGE
+  exit
+fi
+
+if [[ "$mode" == upgrade && -z "$root" ]]; then
+  if systemctl is-active --quiet ci-fleet-status-receiver.service; then
+    write_metadata "$restart_required" 1
+    write_metadata "$previous_was_active" 1
+  else
+    rm -f "$restart_required" "$previous_was_active"
+  fi
+elif [[ "$mode" != upgrade ]]; then
+  rm -f "$restart_required" "$previous_was_active"
+fi
+
+activate "$ref"
+link_unit
+force=0
+[[ -f "$restart_required" ]] && force=1
+restart_live_service "$force"
+rm -f "$restart_required"
+if [[ "$mode" == upgrade ]]; then
+  echo UPGRADED
+else
+  rm -f "$restart_required" "$previous_was_active"
+  echo INSTALLED
+fi
diff --git a/scripts/install-worker-controller.sh b/scripts/install-worker-controller.sh
index 6eb9f240..ad156566 100755
--- a/scripts/install-worker-controller.sh
+++ b/scripts/install-worker-controller.sh
@@ -181,10 +181,15 @@ resolve_config() {
 }
 
 validate_candidate_config_commit() {
-  local tree_paths=$temporary/config-tree-paths
+  local tree_paths=$temporary/config-tree-paths evidence=$temporary/engine-rollout-evidence.json
+  local args=(--config "$candidate_config" --strict --tree-paths "$tree_paths")
   git -C "$config_source_checkout" ls-tree -rz --name-only "$config_ref" >"$tree_paths" || die 'cannot inspect the configuration commit tree'
+  if git -C "$config_source_checkout" cat-file -e "$config_ref:engine-rollout-evidence.json" 2>/dev/null; then
+    git -C "$config_source_checkout" show "$config_ref:engine-rollout-evidence.json" >"$evidence" || die 'cannot read engine rollout evidence'
+    args+=(--rollout-evidence "$evidence")
+  fi
   python3 "$repo_root/templates/config-repository/scripts/validate.py" \
-    --config "$candidate_config" --strict --tree-paths "$tree_paths" || die 'configuration commit validation failed'
+    "${args[@]}" || die 'configuration commit validation failed'
   python3 "$repo_root/scripts/scan_committed_secrets.py" \
     --repository "$config_source_checkout" --commit "$config_ref" || die 'configuration commit secret scan failed'
 }
@@ -254,10 +259,50 @@ docker_gid() {
   stat -c '%g' /var/run/docker.sock
 }
 
+select_engine() {
+  local -a selected
+  mapfile -t selected < <(python3 "$repo_root/scripts/desired_state.py" engine \
+    --config "$candidate_config" --controller "$controller_id")
+  [[ ${#selected[@]} == 2 ]] || die 'selected engine metadata is incomplete'
+  engine_ref=${selected[0]}
+  engine_repository=${selected[1]}
+  [[ "$engine_repository" == RandomDevelopment/ci-fleet ]] || die 'delivery engine repository is not the fixed reviewed public engine'
+  release_dir=$releases_dir/$engine_ref
+}
+
+prepare_engine_capabilities() {
+  local checkout resolved manifest_mode
+  engine_capabilities=$temporary/engine-capabilities.json
+  if runtime_release_complete "$release_dir" "$engine_ref"; then
+    if [[ -f "$release_dir/engine-capabilities.json" ]]; then
+      cp "$release_dir/engine-capabilities.json" "$engine_capabilities"
+    else
+      rm -f "$engine_capabilities"
+    fi
+    return
+  fi
+  if is_git_checkout "$repo_root" && git -C "$repo_root" cat-file -e "$engine_ref^{commit}" 2>/dev/null; then
+    manifest_mode=$(git -C "$repo_root" ls-tree "$engine_ref" -- engine-capabilities.json | awk '{print $1}')
+    [[ "$manifest_mode" == 100644 ]] || { rm -f "$engine_capabilities"; return; }
+    git -C "$repo_root" show "$engine_ref:engine-capabilities.json" >"$engine_capabilities" 2>/dev/null || rm -f "$engine_capabilities"
+    return
+  fi
+  checkout=$temporary/engine-capabilities-repository
+  git init -q "$checkout"
+  git -C "$checkout" remote add origin "https://github.com/${engine_repository}.git"
+  GIT_TERMINAL_PROMPT=0 git -C "$checkout" fetch -q --depth=1 origin "$engine_ref" || die 'pinned ci-fleet engine commit could not be fetched for capability validation'
+  resolved=$(git -C "$checkout" rev-parse 'FETCH_HEAD^{commit}')
+  [[ "$resolved" == "$engine_ref" ]] || die 'fetched ci-fleet engine commit does not match desired state'
+  manifest_mode=$(git -C "$checkout" ls-tree FETCH_HEAD -- engine-capabilities.json | awk '{print $1}')
+  [[ "$manifest_mode" == 100644 ]] || { rm -f "$engine_capabilities"; return; }
+  git -C "$checkout" show "FETCH_HEAD:engine-capabilities.json" >"$engine_capabilities" 2>/dev/null || rm -f "$engine_capabilities"
+}
+
 render_candidate() {
-  local -a metadata_values
+  local -a metadata_values capability_args=()
   candidate_env=$temporary/ci-fleet.env
   candidate_metadata=$temporary/metadata.json
+  [[ ! -f "$engine_capabilities" ]] || capability_args=(--engine-capabilities "$engine_capabilities")
   python3 "$repo_root/scripts/desired_state.py" render \
     --config "$candidate_config" \
     --controller "$controller_id" \
@@ -265,6 +310,7 @@ render_candidate() {
     --config-repository "$config_identity" \
     --config-ref "$config_ref" \
     --docker-gid "$(docker_gid)" \
+    "${capability_args[@]}" \
     --output "$candidate_env" \
     --metadata-output "$candidate_metadata"
   mapfile -t metadata_values < <(python3 - "$candidate_metadata" <<'PY'
@@ -273,14 +319,15 @@ import sys
 value = json.load(open(sys.argv[1], encoding="utf-8"))
 for key in ("controller_state", "engine_ref", "engine_repository"):
     print(value[key])
+print(1 if value["status_reporting_required"] else 0)
+print(1 if value["status_reporting_configured"] else 0)
 PY
   )
-  [[ ${#metadata_values[@]} == 3 ]] || die 'rendered controller metadata is incomplete'
+  [[ ${#metadata_values[@]} == 5 ]] || die 'rendered controller metadata is incomplete'
   target_state=${metadata_values[0]}
-  engine_ref=${metadata_values[1]}
-  engine_repository=${metadata_values[2]}
-  [[ "$engine_repository" == RandomDevelopment/ci-fleet ]] || die 'delivery engine repository is not the fixed reviewed public engine'
-  release_dir=$releases_dir/$engine_ref
+  [[ ${metadata_values[1]} == "$engine_ref" && ${metadata_values[2]} == "$engine_repository" ]] || die 'rendered engine metadata changed during validation'
+  status_reporting_required=${metadata_values[3]}
+  status_reporting_configured=${metadata_values[4]}
 }
 
 compose() {
@@ -432,9 +479,17 @@ PY
 }
 
 runtime_release_complete() {
-  local path=$1 expected=$2 marker required stored_digest actual_digest
+  local path=$1 expected=$2 require_status=${3:-0} require_schema=${4:-0} marker required stored_digest actual_digest
+  local -a capability_args=()
   [[ -d "$path" && -f "$path/.ci-fleet-engine-ref" && -f "$path/.ci-fleet-tree-sha256" && -f "$path/deploy/compose.yaml" ]] || return 1
   [[ -x "$path/scripts/preflight.sh" && -x "$path/scripts/healthcheck.sh" && -x "$path/scripts/cleanup.sh" ]] || return 1
+  if [[ -e "$path/engine-capabilities.json" || "$require_status" == 1 || "$require_schema" == 1 ]]; then
+    [[ ! -L "$path/engine-capabilities.json" && -f "$path/engine-capabilities.json" ]] || return 1
+    [[ "$require_schema" != 1 ]] || capability_args+=(--require-status-reporting-config)
+    [[ "$require_status" != 1 ]] || capability_args+=(--require-status-reporting)
+    python3 "$repo_root/scripts/desired_state.py" validate-engine-capabilities \
+      --manifest "$path/engine-capabilities.json" "${capability_args[@]}" >/dev/null || return 1
+  fi
   if grep -Fq 'scripts/health.py' "$path/scripts/healthcheck.sh"; then
     [[ -f "$path/scripts/health.py" ]] || return 1
     if grep -Fq 'build_status_report' "$path/scripts/health.py"; then
@@ -453,8 +508,8 @@ runtime_release_complete() {
 }
 
 manager_release_complete() {
-  local path=$1 expected=$2 marker required unit
-  runtime_release_complete "$path" "$expected" || return 1
+  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
   [[ -x "$path/scripts/install-worker-controller.sh" && -x "$path/scripts/check-installed-state.sh" ]] || return 1
   for required in scripts/desired_state.py scripts/scan_committed_secrets.py templates/config-repository/fleet.schema.json templates/config-repository/scripts/validate.py; do
     [[ -f "$path/$required" ]] || return 1
@@ -466,7 +521,7 @@ manager_release_complete() {
 }
 
 release_matches() {
-  runtime_release_complete "$release_dir" "$engine_ref" || return 1
+  runtime_release_complete "$release_dir" "$engine_ref" "$status_reporting_required" "$status_reporting_configured" || return 1
   [[ -L "$current_link" ]] || return 1
   [[ $(readlink -f "$current_link") == $(readlink -f "$release_dir") ]]
 }
@@ -485,7 +540,7 @@ managed_images_match() {
 systemd_matches() {
   local expected_manager unit
   expected_manager=$manager_releases/$engine_ref
-  manager_release_complete "$expected_manager" "$engine_ref" || return 1
+  manager_release_complete "$expected_manager" "$engine_ref" "$status_reporting_required" "$status_reporting_configured" || return 1
   [[ -L "$manager_current" ]] || return 1
   [[ $(readlink -f "$manager_current") == $(readlink -f "$expected_manager") ]] || return 1
   for unit in "${unit_names[@]}"; do
@@ -550,7 +605,7 @@ PY
 
 install_release() {
   local archive checkout resolved staged_release
-  if runtime_release_complete "$release_dir" "$engine_ref"; then
+  if runtime_release_complete "$release_dir" "$engine_ref" "$status_reporting_required" "$status_reporting_configured"; then
     return
   fi
   install -d -m 0755 "$releases_dir"
@@ -575,7 +630,7 @@ install_release() {
   chmod 0644 "$staged_release/.ci-fleet-engine-ref"
   release_tree_digest "$staged_release" >"$staged_release/.ci-fleet-tree-sha256"
   chmod 0644 "$staged_release/.ci-fleet-tree-sha256"
-  runtime_release_complete "$staged_release" "$engine_ref" || die 'staged engine release is incomplete'
+  runtime_release_complete "$staged_release" "$engine_ref" "$status_reporting_required" "$status_reporting_configured" || die 'staged engine release is incomplete'
   atomic_replace_directory "$staged_release" "$release_dir"
 }
 
@@ -583,9 +638,9 @@ install_manager() {
   local manager_commit manager_release archive staged_manager
   manager_commit=$engine_ref
   [[ "$manager_commit" =~ ^[0-9a-f]{40}$ ]] || die 'installer manager commit is invalid'
-  runtime_release_complete "$release_dir" "$manager_commit" || die 'desired engine release is unavailable for installer manager activation'
+  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
-  if ! manager_release_complete "$manager_release" "$manager_commit"; then
+  if ! manager_release_complete "$manager_release" "$manager_commit" "$status_reporting_required" "$status_reporting_configured"; then
     install -d -m 0755 "$manager_releases"
     archive=$temporary/manager.tar
     tar -cf "$archive" -C "$release_dir" .
@@ -595,7 +650,7 @@ install_manager() {
     tar -xf "$archive" -C "$staged_manager"
     printf '%s\n' "$manager_commit" >"$staged_manager/.ci-fleet-engine-ref"
     chmod 0644 "$staged_manager/.ci-fleet-engine-ref"
-    manager_release_complete "$staged_manager" "$manager_commit" || die 'staged installer manager release is incomplete'
+    manager_release_complete "$staged_manager" "$manager_commit" "$status_reporting_required" "$status_reporting_configured" || die 'staged installer manager release is incomplete'
     atomic_replace_directory "$staged_manager" "$manager_release"
   fi
   install -d -m 0755 "$manager_root"
@@ -1058,6 +1113,8 @@ case "$mode" in
     validate_candidate_config_commit
     prepare_host_config
     verify_host_files
+    select_engine
+    prepare_engine_capabilities
     render_candidate
     if [[ "$mode" == check ]]; then perform_check; else perform_converge; fi
     ;;
diff --git a/scripts/status_receiver.py b/scripts/status_receiver.py
index dd84950f..d0914259 100644
--- a/scripts/status_receiver.py
+++ b/scripts/status_receiver.py
@@ -61,6 +61,9 @@ def __init__(
         self._write_lock = threading.Lock()
         self._last_attempt: dict[str, int] = {}
         self._clock = time.time
+        self._monotonic = time.monotonic
+        self._health_checked_at = float("-inf")
+        self._health_integrity_ok = False
         with closing(self._connect()) as connection, connection:
             connection.executescript("""
                 CREATE TABLE IF NOT EXISTS reports (
@@ -76,6 +79,9 @@ def __init__(
                     authenticated_at INTEGER NOT NULL,
                     PRIMARY KEY (controller, nonce)
                 );
+                CREATE TABLE IF NOT EXISTS health_write_probe (
+                    checked_at INTEGER NOT NULL
+                );
             """)
         os.chmod(self.database, 0o600)
         self.expire()
@@ -113,33 +119,33 @@ def submit(self, body: bytes, headers: Mapping[str, str], *, now: int) -> None:
         if abs(now - generated_at) > self.max_clock_skew_seconds:
             raise StatusError(409, "report_time_stale")
         encoded = json.dumps(report, separators=(",", ":"), sort_keys=True)
-        with self._write_lock, closing(self._connect()) as connection, connection:
-            if connection.execute("SELECT 1 FROM nonces WHERE controller=? AND nonce=?", (controller, nonce)).fetchone():
-                raise StatusError(409, "replayed_report")
-            last_attempt = self._last_attempt.get(controller)
-            if last_attempt is not None and now - last_attempt < 1:
-                raise StatusError(429, "submission_too_frequent")
+        with self._write_lock, closing(self._connect()) as connection:
+            with connection:
+                if connection.execute("SELECT 1 FROM nonces WHERE controller=? AND nonce=?", (controller, nonce)).fetchone():
+                    raise StatusError(409, "replayed_report")
+                last_attempt = self._last_attempt.get(controller)
+                if last_attempt is not None and now - last_attempt < 1:
+                    raise StatusError(429, "submission_too_frequent")
+                try:
+                    connection.execute("INSERT INTO nonces VALUES (?, ?, ?)", (controller, nonce, authenticated_at))
+                except sqlite3.IntegrityError as error:
+                    raise StatusError(409, "replayed_report") from error
+                last = connection.execute(
+                    "SELECT generated_at, received_at FROM reports WHERE controller=? ORDER BY generated_at DESC LIMIT 1",
+                    (controller,),
+                ).fetchone()
+                if last and generated_at <= last[0]:
+                    raise StatusError(409, "stale_report")
+                if last and now - last[1] < self.min_interval_seconds:
+                    raise StatusError(429, "submission_too_frequent")
+                connection.execute("INSERT INTO reports VALUES (?, ?, ?, ?)", (controller, generated_at, now, encoded))
+                connection.execute("DELETE FROM reports WHERE received_at < ?", (now - self.retention_seconds,))
+                connection.execute(
+                    "DELETE FROM reports WHERE controller=? AND rowid NOT IN (SELECT rowid FROM reports WHERE controller=? ORDER BY generated_at DESC LIMIT ?)",
+                    (controller, controller, self.history_limit),
+                )
+                connection.execute("DELETE FROM nonces WHERE authenticated_at < ?", (now - self.max_clock_skew_seconds,))
             self._last_attempt[controller] = now
-            try:
-                connection.execute("INSERT INTO nonces VALUES (?, ?, ?)", (controller, nonce, authenticated_at))
-                connection.commit()
-            except sqlite3.IntegrityError as error:
-                raise StatusError(409, "replayed_report") from error
-            last = connection.execute(
-                "SELECT generated_at, received_at FROM reports WHERE controller=? ORDER BY generated_at DESC LIMIT 1",
-                (controller,),
-            ).fetchone()
-            if last and generated_at <= last[0]:
-                raise StatusError(409, "stale_report")
-            if last and now - last[1] < self.min_interval_seconds:
-                raise StatusError(429, "submission_too_frequent")
-            connection.execute("INSERT INTO reports VALUES (?, ?, ?, ?)", (controller, generated_at, now, encoded))
-            connection.execute("DELETE FROM reports WHERE received_at < ?", (now - self.retention_seconds,))
-            connection.execute(
-                "DELETE FROM reports WHERE controller=? AND rowid NOT IN (SELECT rowid FROM reports WHERE controller=? ORDER BY generated_at DESC LIMIT ?)",
-                (controller, controller, self.history_limit),
-            )
-            connection.execute("DELETE FROM nonces WHERE authenticated_at < ?", (now - self.max_clock_skew_seconds,))
 
     def _expire(self, connection: sqlite3.Connection, now: int) -> None:
         connection.execute("DELETE FROM reports WHERE received_at < ?", (now - self.retention_seconds,))
@@ -285,6 +291,32 @@ def list_latest(self, read_token: str) -> list[dict[str, Any]]:
             """).fetchall()
         return [json.loads(row[0]) for row in rows]
 
+    def health(self) -> None:
+        with self._write_lock, closing(self._connect()) as connection:
+            now = self._monotonic()
+            if now - self._health_checked_at >= 60:
+                try:
+                    if connection.execute("PRAGMA quick_check(1)").fetchone() != ("ok",):
+                        raise sqlite3.DatabaseError("database health check failed")
+                    connection.execute(
+                        "SELECT controller, generated_at, received_at, payload FROM reports LIMIT 0"
+                    )
+                    connection.execute(
+                        "SELECT controller, nonce, authenticated_at FROM nonces LIMIT 0"
+                    )
+                    connection.execute("INSERT INTO health_write_probe VALUES (?)", (int(self._clock()),))
+                    connection.commit()
+                    connection.execute("DELETE FROM health_write_probe")
+                    connection.commit()
+                except sqlite3.Error:
+                    self._health_integrity_ok = False
+                    self._health_checked_at = now
+                    raise sqlite3.DatabaseError("database health check failed") from None
+                self._health_integrity_ok = True
+                self._health_checked_at = now
+            if not self._health_integrity_ok:
+                raise sqlite3.DatabaseError("database health check failed")
+
 
 class _BoundedHTTPServer(http.server.ThreadingHTTPServer):
     daemon_threads = True
@@ -359,12 +391,18 @@ def do_POST(self) -> None:
             except (ValueError, StatusError) as error:
                 failure = error if isinstance(error, StatusError) else StatusError(400, "invalid_request")
                 self.send_json(failure.status, {"error": failure.code})
+            except (OSError, sqlite3.Error):
+                self.send_json(503, {"error": "unavailable"})
 
         def do_GET(self) -> None:
             path = urllib.parse.urlsplit(self.path)
             try:
                 if path.query or path.fragment:
                     raise StatusError(400, "invalid_request")
+                if path.path == "/healthz":
+                    receiver.health()
+                    self.send_json(200, {"status": "ok"})
+                    return
                 if path.path == "/v1/controllers":
                     value = {"schema_version": 1, "controllers": receiver.list_latest(self.bearer())}
                 elif path.path.startswith("/v1/controllers/"):
@@ -382,6 +420,8 @@ def do_GET(self) -> None:
                 self.send_json(200, value)
             except StatusError as error:
                 self.send_json(error.status, {"error": error.code})
+            except (OSError, sqlite3.Error):
+                self.send_json(503, {"error": "unavailable"})
 
         def log_message(self, format: str, *args: Any) -> None:
             pass
diff --git a/scripts/test-install-status-receiver.sh b/scripts/test-install-status-receiver.sh
new file mode 100755
index 00000000..d653369c
--- /dev/null
+++ b/scripts/test-install-status-receiver.sh
@@ -0,0 +1,223 @@
+#!/usr/bin/env bash
+set -Eeuo pipefail
+
+repo_root=$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)
+installer="$repo_root/scripts/install-status-receiver.sh"
+unit="$repo_root/deploy/status-receiver/ci-fleet-status-receiver.service"
+test -x "$installer"
+test -f "$unit"
+
+tmp=$(mktemp -d)
+trap 'rm -rf "$tmp"' EXIT
+source_tree="$tmp/source"
+root="$tmp/root"
+mkdir -p "$source_tree/scripts" "$source_tree/deploy/status-receiver"
+mkdir -p "$root/run/lock"
+chmod 1777 "$root/run/lock"
+cp "$installer" "$source_tree/scripts/"
+cp "$repo_root/scripts/status_receiver.py" "$repo_root/scripts/status_auth.py" "$source_tree/scripts/"
+cp "$unit" "$source_tree/deploy/status-receiver/"
+git -C "$source_tree" init -q
+git -C "$source_tree" config user.name test
+git -C "$source_tree" config user.email test@example.invalid
+git -C "$source_tree" add .
+git -C "$source_tree" commit -qm initial
+first=$(git -C "$source_tree" rev-parse HEAD)
+
+run() {
+  CI_FLEET_STATUS_TEST_MODE=1 CI_FLEET_STATUS_TEST_EXPECTED_OWNER="$(id -u)" CI_FLEET_STATUS_ROOT="$root" \
+    "$source_tree/scripts/install-status-receiver.sh" "$@"
+}
+
+assert_systemd_mode() { test "$(stat -c %a "$root/etc/systemd/system")" = 750; }
+
+if CI_FLEET_STATUS_ROOT="$root" "$source_tree/scripts/install-status-receiver.sh" --check >/dev/null 2>&1; then
+  echo "alternate root accepted without explicit test mode" >&2
+  exit 1
+fi
+mkdir "$tmp/attacker-lock-target"
+chmod 0777 "$tmp/attacker-lock-target"
+ln -s "$tmp/attacker-lock-target" "$root/run/lock/ci-fleet-status"
+if run --install --ref "$first" >/dev/null 2>&1; then
+  echo "symlinked installer lock directory was accepted" >&2
+  exit 1
+fi
+test "$(stat -c %a "$tmp/attacker-lock-target")" = 777
+rm "$root/run/lock/ci-fleet-status"
+mkdir -p "$root/opt/ci-fleet-status" "$tmp/untrusted-releases"
+chmod 0755 "$root/opt/ci-fleet-status" "$tmp/untrusted-releases"
+ln -s "$tmp/untrusted-releases" "$root/opt/ci-fleet-status/releases"
+if run --install --ref "$first" >/dev/null 2>&1; then
+  echo "symlinked release root was accepted" >&2
+  exit 1
+fi
+rm -rf "$root/opt/ci-fleet-status"
+mkdir -p "$root/etc/systemd/system"
+chmod 0750 "$root/etc/systemd/system"
+export CI_FLEET_STATUS_TEST_FAIL_RELOAD_ONCE=$tmp/fail-reload-once
+: >"$CI_FLEET_STATUS_TEST_FAIL_RELOAD_ONCE"
+if run --install --ref "$first" >/dev/null 2>&1; then
+  echo "simulated daemon reload failure was accepted" >&2
+  exit 1
+fi
+test "$(readlink "$root/opt/ci-fleet-status/current")" = "releases/$first"
+test "$(run --install --ref "$first")" = NO_CHANGE
+assert_systemd_mode
+test "$(readlink "$root/opt/ci-fleet-status/current")" = "releases/$first"
+test -f "$root/opt/ci-fleet-status/current/status_receiver.py"
+external="$tmp/external/$first"
+mkdir -p "$external"
+cp "$source_tree/scripts/status_receiver.py" "$source_tree/scripts/status_auth.py" "$external/"
+ln -sfn "$external" "$root/opt/ci-fleet-status/current"
+if run --check >/dev/null 2>&1; then
+  echo "external same-basename release was accepted" >&2
+  exit 1
+fi
+test "$(run --install --ref "$first")" = INSTALLED
+assert_systemd_mode
+test "$(readlink "$root/opt/ci-fleet-status/current")" = "releases/$first"
+test "$(stat -c %a "$root/var/lib/ci-fleet-status")" = 700
+test "$(stat -c %a "$root/run/lock")" = 1777
+test "$(stat -c %a "$root/opt/ci-fleet-status/releases/$first")" = 755
+cp "$source_tree/deploy/status-receiver/ci-fleet-status-receiver.service" "$tmp/first-unit"
+rm "$root/etc/systemd/system/ci-fleet-status-receiver.service"
+printf '%s\n' 'ExecStart=python3 --bind 127.0.0.1' >"$root/etc/systemd/system/ci-fleet-status-receiver.service"
+test "$(run --install --ref "$first")" = NO_CHANGE
+assert_systemd_mode
+test -L "$root/etc/systemd/system/ci-fleet-status-receiver.service"
+cmp "$source_tree/deploy/status-receiver/ci-fleet-status-receiver.service" \
+  "$root/etc/systemd/system/ci-fleet-status-receiver.service"
+printf '%s\n' 1 >"$root/var/lib/ci-fleet-status-installer/restart-required"
+printf '%s\n' 1 >"$root/var/lib/ci-fleet-status-installer/previous-was-active"
+test "$(run --upgrade --ref "$first")" = UPGRADED
+test ! -e "$root/var/lib/ci-fleet-status-installer/restart-required"
+test -e "$root/var/lib/ci-fleet-status-installer/previous-was-active"
+
+printf '\n# dirty\n' >>"$source_tree/scripts/status_auth.py"
+if run --upgrade --ref "$first" >/dev/null 2>&1; then
+  echo "dirty reviewed input was installed" >&2
+  exit 1
+fi
+git -C "$source_tree" checkout -q -- scripts/status_auth.py
+
+for version in 3.24.9 3.25.0 3.99.0; do
+  if output=$(CI_FLEET_STATUS_TEST_SQLITE_VERSION="$version" run --install --ref "$first" 2>&1); then
+    [[ "$version" != 3.24.9 ]] || { echo "old SQLite was accepted" >&2; exit 1; }
+  else
+    [[ "$version" == 3.24.9 && "$output" == *"SQLite 3.25.0 or newer is required"* ]] || {
+      echo "supported SQLite $version was rejected" >&2
+      exit 1
+    }
+  fi
+done
+
+lock="$root/run/lock/ci-fleet-status"
+ready="$tmp/lock-ready"
+(flock 9; : >"$ready"; sleep 1) 9<"$lock" &
+lock_pid=$!
+for _ in {1..20}; do [[ -e "$ready" ]] && break; sleep 0.05; done
+if CI_FLEET_STATUS_TEST_MODE=1 CI_FLEET_STATUS_ROOT="$root" \
+  timeout 0.1 "$source_tree/scripts/install-status-receiver.sh" --check >/dev/null 2>&1; then
+  echo "overlapping installer did not wait for lock" >&2
+  exit 1
+else
+  test "$?" = 124
+fi
+wait "$lock_pid"
+
+printf '\n# upgraded unit\n' >>"$source_tree/deploy/status-receiver/ci-fleet-status-receiver.service"
+git -C "$source_tree" add .
+git -C "$source_tree" commit -qm upgrade
+second=$(git -C "$source_tree" rev-parse HEAD)
+if run --install --ref "$second" >/dev/null 2>&1; then
+  echo "install mode changed an active release" >&2
+  exit 1
+fi
+printf '%s\n' 1 >"$root/var/lib/ci-fleet-status-installer/restart-required"
+test "$(run --upgrade --ref "$second")" = UPGRADED
+test ! -e "$root/var/lib/ci-fleet-status-installer/restart-required"
+test -e "$root/var/lib/ci-fleet-status-installer/previous-was-active"
+test "$(cat "$root/run/ci-fleet-status-last-restart-force")" = 1
+assert_systemd_mode
+test "$(readlink "$root/opt/ci-fleet-status/current")" = "releases/$second"
+test "$(cat "$root/var/lib/ci-fleet-status-installer/previous-ref")" = "$first"
+test "$(stat -c %a "$root/var/lib/ci-fleet-status-installer")" = 700
+cmp "$source_tree/deploy/status-receiver/ci-fleet-status-receiver.service" \
+  "$root/etc/systemd/system/ci-fleet-status-receiver.service"
+printf '%s\n' 1 >"$root/var/lib/ci-fleet-status-installer/restart-required"
+if run --check >/dev/null 2>&1; then
+  echo "pending receiver restart was accepted" >&2
+  exit 1
+fi
+rm "$root/var/lib/ci-fleet-status-installer/restart-required"
+test "$(run --check)" = "CHECK_OK $second"
+assert_systemd_mode
+rm "$root/etc/systemd/system/ci-fleet-status-receiver.service"
+printf '%s\n' drift >"$root/etc/systemd/system/ci-fleet-status-receiver.service"
+test "$(run --rollback)" = "ROLLED_BACK $first"
+test ! -e "$root/var/lib/ci-fleet-status-installer/previous-was-active"
+assert_systemd_mode
+test -L "$root/etc/systemd/system/ci-fleet-status-receiver.service"
+test "$(readlink "$root/opt/ci-fleet-status/current")" = "releases/$first"
+test "$(cat "$root/var/lib/ci-fleet-status-installer/previous-ref")" = "$first"
+cmp "$tmp/first-unit" "$root/etc/systemd/system/ci-fleet-status-receiver.service"
+test "$(run --upgrade --ref "$second")" = UPGRADED
+test "$(run --rollback)" = "ROLLED_BACK $first"
+assert_systemd_mode
+
+active="$root/opt/ci-fleet-status/releases/$first"
+printf '\n# modified\n' >>"$active/status_receiver.py"
+if run --check >/dev/null 2>&1; then echo "modified artifact was accepted" >&2; exit 1; fi
+git -C "$source_tree" show "$first:scripts/status_receiver.py" >"$active/status_receiver.py"
+chmod 0755 "$active/status_receiver.py"
+chmod 0664 "$active/status_auth.py"
+if run --check >/dev/null 2>&1; then echo "writable artifact was accepted" >&2; exit 1; fi
+chmod 0644 "$active/status_auth.py"
+mv "$active/status_auth.py" "$tmp/status_auth.py.real"
+ln -s "$tmp/status_auth.py.real" "$active/status_auth.py"
+if run --rollback >/dev/null 2>&1; then echo "symlinked artifact was accepted" >&2; exit 1; fi
+rm "$active/status_auth.py"
+mv "$tmp/status_auth.py.real" "$active/status_auth.py"
+if CI_FLEET_STATUS_TEST_MODE=1 CI_FLEET_STATUS_TEST_EXPECTED_OWNER=99999 CI_FLEET_STATUS_ROOT="$root" \
+  "$source_tree/scripts/install-status-receiver.sh" --check >/dev/null 2>&1; then
+  echo "service-owned release directory was accepted" >&2
+  exit 1
+fi
+test "$(run --check)" = "CHECK_OK $first"
+test "$(CI_FLEET_STATUS_TEST_MODE=1 CI_FLEET_STATUS_TEST_EXPECTED_OWNER="$(id -u)" \
+  CI_FLEET_STATUS_TEST_ACCOUNT_GROUPS=12345 CI_FLEET_STATUS_ROOT="$root" \
+  "$source_tree/scripts/install-status-receiver.sh" --check)" = "CHECK_OK $first"
+if CI_FLEET_STATUS_TEST_MODE=1 CI_FLEET_STATUS_TEST_EXPECTED_OWNER="$(id -u)" \
+  CI_FLEET_STATUS_TEST_ACCOUNT_GROUPS='12345 99999' CI_FLEET_STATUS_ROOT="$root" \
+  "$source_tree/scripts/install-status-receiver.sh" --check >/dev/null 2>&1; then
+  echo "service account supplementary groups were accepted" >&2
+  exit 1
+fi
+for directory in "$root/etc/ci-fleet-status" "$root/var/lib/ci-fleet-status"; do
+  mv "$directory" "$directory.real"
+  ln -s "${directory##*/}.real" "$directory"
+  if run --check >/dev/null 2>&1; then
+    echo "symlinked receiver directory was accepted: $directory" >&2
+    exit 1
+  fi
+  rm "$directory"
+  mv "$directory.real" "$directory"
+done
+if grep -F '%F' "$source_tree/scripts/install-status-receiver.sh" >/dev/null; then
+  echo "locale-dependent stat file types remain" >&2
+  exit 1
+fi
+
+grep -F -- '--bind 127.0.0.1' "$unit" >/dev/null
+grep -F 'User=ci-fleet-status' "$unit" >/dev/null
+grep -F 'NoNewPrivileges=yes' "$unit" >/dev/null
+grep -F 'ProtectSystem=strict' "$unit" >/dev/null
+grep -F 'ReadWritePaths=/var/lib/ci-fleet-status' "$unit" >/dev/null
+grep -F 'useradd --system --user-group' "$installer" >/dev/null
+grep -F 'restart-required' "$installer" >/dev/null
+grep -F 'getent passwd ci-fleet-status' "$installer" >/dev/null
+grep -F "\$gid\" != 0" "$installer" >/dev/null
+grep -F '/usr/bin/python3' "$installer" >/dev/null
+grep -F 'SQLite 3.25.0 or newer is required' "$installer" >/dev/null
+
+echo STATUS_RECEIVER_INSTALL_TESTS_OK
diff --git a/scripts/test-install-worker-controller.sh b/scripts/test-install-worker-controller.sh
index b75e5799..445a77d0 100755
--- a/scripts/test-install-worker-controller.sh
+++ b/scripts/test-install-worker-controller.sh
@@ -14,6 +14,8 @@ export REAL_STAT
 REAL_STAT=$(command -v stat)
 export REAL_TAR
 REAL_TAR=$(command -v tar)
+export REAL_GIT
+REAL_GIT=$(command -v git)
 
 cat >"$fake_bin/docker" <<'EOF'
 #!/usr/bin/env bash
@@ -174,6 +176,15 @@ exec "$REAL_TAR" "$@"
 EOF
 chmod 700 "$fake_bin/tar"
 
+cat >"$fake_bin/git" <<'EOF'
+#!/usr/bin/env bash
+if [[ -n ${FAKE_FAIL_GIT_FETCH:-} && " $* " == *" fetch "* ]]; then
+  exit 90
+fi
+exec "$REAL_GIT" "$@"
+EOF
+chmod 700 "$fake_bin/git"
+
 export PATH="$fake_bin:$PATH"
 export FAKE_DOCKER_STATE=$tmp/docker-controller-running
 export FAKE_CONTROLLER_STATUS_FILE=$tmp/docker-controller-status
@@ -229,11 +240,11 @@ git -C "$config_repo" config user.email fixture@example.invalid
 
 write_config() {
   local state=$1 maximum=$2 budget=$3
-  local desired_engine=${4:-$engine_ref}
-  python3 - "$repo_root/templates/config-repository/fleet.json" "$config_repo/fleet.json" "$desired_engine" "$state" "$maximum" "$budget" <<'PY'
+  local desired_engine=${4:-$engine_ref} reporting=${5:-false}
+  python3 - "$repo_root/templates/config-repository/fleet.json" "$config_repo/fleet.json" "$config_repo/engine-rollout-evidence.json" "$desired_engine" "$state" "$maximum" "$budget" "$reporting" <<'PY'
 import json
 import sys
-source, target, engine_ref, state, maximum, budget = sys.argv[1:]
+source, target, evidence_target, engine_ref, state, maximum, budget, reporting = sys.argv[1:]
 value = json.load(open(source, encoding="utf-8"))
 value["organization"]["slug"] = "fixture-org"
 value["runner_pools"]["trusted-ci"]["allowed_repositories"] = ["fixture-org/example-app"]
@@ -242,12 +253,31 @@ controller = value["controllers"]["example-ci-01"]
 controller["engine_ref"] = engine_ref
 controller["state"] = state
 controller["max_runners"] = int(maximum)
+if reporting == "omit":
+    controller.pop("status_reporting", None)
+else:
+    controller["status_reporting"] = {
+        "enabled": reporting == "true",
+        "config_file": "/etc/ci-fleet/monitoring.env",
+    }
 value["runner_pools"]["trusted-ci"]["capacity_budget"] = int(budget)
 with open(target, "w", encoding="utf-8") as handle:
     json.dump(value, handle, indent=2)
     handle.write("\n")
+with open(evidence_target, "w", encoding="utf-8") as handle:
+    json.dump({
+        "schema_version": 1,
+        "status_reporting_engine_capabilities": {
+            "example-ci-01": {
+                "engine_ref": engine_ref,
+                "status_reporting_config": True,
+                "required_status_reporting": True,
+            },
+        },
+    }, handle, indent=2)
+    handle.write("\n")
 PY
-  git -C "$config_repo" add fleet.json
+  git -C "$config_repo" add fleet.json engine-rollout-evidence.json
   git -C "$config_repo" commit -q -m "fixture $state $maximum"
   git -C "$config_repo" rev-parse HEAD
 }
@@ -351,6 +381,8 @@ second=$(expect_success "$installer" --install "${base_args[@]}" --ref "$ref_one
 grep -Fq 'NO_CHANGE' <<<"$second" || fail 'idempotent rerun changed the host'
 check=$(expect_success "$installer" --check "${base_args[@]}" --ref "$ref_one")
 grep -Fq 'HEALTH last=' <<<"$check" || fail 'check output omitted the last redacted health result'
+installed_installer=$root/opt/ci-fleet/manager/current/scripts/install-worker-controller.sh
+expect_success env FAKE_FAIL_GIT_FETCH=1 "$installed_installer" --check "${base_args[@]}" --ref "$ref_one" >/dev/null
 [[ ! -d "$root/opt/ci-fleet/manager/releases/$engine_ref/templates/config-repository/scripts/__pycache__" ]] || fail 'manager validation wrote Python bytecode into the immutable release'
 python3 - "$install_state" <<'PY'
 import json
@@ -435,6 +467,10 @@ printf '\n# tampered runtime fixture\n' >>"$active_release/scripts/preflight.sh"
 expect_failure 'DRIFT engine_release' "$installer" --check "${base_args[@]}" --ref "$ref_one"
 expect_success "$installer" --install "${base_args[@]}" --ref "$ref_one" >/dev/null
 if grep -Fq 'tampered runtime fixture' "$active_release/scripts/preflight.sh"; then fail 'modified runtime release was reused'; fi
+printf '{"schema_version":1,"capabilities":null}\n' >"$active_release/engine-capabilities.json"
+expect_failure 'DRIFT engine_release' "$installer" --check "${base_args[@]}" --ref "$ref_one"
+expect_success "$installer" --install "${base_args[@]}" --ref "$ref_one" >/dev/null
+python3 "$repo_root/scripts/desired_state.py" validate-engine-capabilities --manifest "$active_release/engine-capabilities.json" >/dev/null || fail 'engine capability declaration was not repaired'
 rm -f "$active_release/deploy/compose.yaml"
 export FAKE_FAIL_TAR_ONCE=$tmp/fail-tar-once
 : >"$FAKE_FAIL_TAR_ONCE"
@@ -646,7 +682,11 @@ unset FAKE_RUNNER_STATE_ONCE FAKE_COMPOSE_LOG
 
 # Public pre-health engine fixture; do not depend on a local remote-tracking ref.
 legacy_engine_ref=af9c0c13cd12866ce75dd6c43a4cda01915507e1
-legacy_ref=$(write_config active 1 1 "$legacy_engine_ref")
+legacy_disabled_ref=$(write_config active 1 1 "$legacy_engine_ref" false)
+expect_failure 'selected engine does not support status reporting configuration' "$installer" --upgrade "${base_args[@]}" --ref "$legacy_disabled_ref"
+legacy_required_ref=$(write_config active 1 1 "$legacy_engine_ref" true)
+expect_failure 'selected engine does not advertise required status reporting' "$installer" --upgrade "${base_args[@]}" --ref "$legacy_required_ref"
+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}
diff --git a/scripts/test_desired_state.py b/scripts/test_desired_state.py
index 5e7eab51..c37a441f 100755
--- a/scripts/test_desired_state.py
+++ b/scripts/test_desired_state.py
@@ -12,6 +12,7 @@
 from desired_state import (
     DesiredStateError,
     build_rendered_env,
+    load_engine_capabilities,
     load_and_validate_config,
     parse_env,
     validate_host_values,
@@ -36,7 +37,7 @@ def host_values() -> dict[str, str]:
 
 
 class DesiredStateTests(unittest.TestCase):
-    def render(self, value: dict | None = None):
+    def render(self, value: dict | None = None, capabilities: set[str] | None = None):
         return build_rendered_env(
             value or config(),
             "example-ci-01",
@@ -44,6 +45,7 @@ def render(self, value: dict | 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,
         )
 
     def test_active_controller_renders_configured_capacity(self) -> None:
@@ -54,6 +56,68 @@ def test_active_controller_renders_configured_capacity(self) -> None:
         self.assertEqual(environment["CI_FLEET_COMMIT"], environment["CI_FLEET_ENGINE_REF"])
         self.assertEqual(metadata["controller_state"], "active")
 
+    def test_status_reporting_requires_fixed_host_local_configuration(self) -> None:
+        value = config()
+        value["controllers"]["example-ci-01"]["status_reporting"] = {
+            "enabled": True,
+            "config_file": "/etc/ci-fleet/monitoring.env",
+        }
+        environment, _ = self.render(value, {"status_reporting_config", "required_status_reporting"})
+        self.assertEqual(environment["CI_FLEET_STATUS_REPORTING_REQUIRED"], "1")
+        value["controllers"]["example-ci-01"]["status_reporting"]["config_file"] = "https://example.invalid/v1/status"
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "fleet.json"
+            path.write_text(json.dumps(value), encoding="utf-8")
+            with self.assertRaisesRegex(DesiredStateError, "fixed host-local monitoring"):
+                load_and_validate_config(path)
+        value["controllers"]["example-ci-01"]["status_reporting"] = None
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "fleet.json"
+            path.write_text(json.dumps(value), encoding="utf-8")
+            with self.assertRaisesRegex(DesiredStateError, "must be an object"):
+                load_and_validate_config(path)
+
+    def test_status_reporting_requires_engine_capability(self) -> None:
+        value = config()
+        value["controllers"]["example-ci-01"]["status_reporting"] = {
+            "enabled": True,
+            "config_file": "/etc/ci-fleet/monitoring.env",
+        }
+        with self.assertRaisesRegex(DesiredStateError, "does not advertise"):
+            self.render(value, set())
+        with tempfile.TemporaryDirectory() as directory:
+            manifest = Path(directory) / "engine-capabilities.json"
+            manifest.write_text("not json", encoding="utf-8")
+            with self.assertRaisesRegex(DesiredStateError, "malformed"):
+                load_engine_capabilities(manifest)
+            manifest.write_text('{"schema_version":1,"schema_version":1,"capabilities":{}}', encoding="utf-8")
+            with self.assertRaisesRegex(DesiredStateError, "malformed"):
+                load_engine_capabilities(manifest)
+            manifest.unlink()
+            with self.assertRaisesRegex(DesiredStateError, "missing"):
+                load_engine_capabilities(manifest)
+
+    def test_omitted_status_reporting_accepts_older_engine(self) -> None:
+        value = config()
+        value["controllers"]["example-ci-01"].pop("status_reporting", None)
+        environment, metadata = self.render(value, set())
+        self.assertNotIn("CI_FLEET_STATUS_REPORTING_REQUIRED", environment)
+        self.assertFalse(metadata["status_reporting_configured"])
+        self.assertFalse(metadata["status_reporting_required"])
+
+    def test_disabled_status_reporting_requires_schema_capability(self) -> None:
+        value = config()
+        value["controllers"]["example-ci-01"]["status_reporting"] = {
+            "enabled": False,
+            "config_file": "/etc/ci-fleet/monitoring.env",
+        }
+        with self.assertRaisesRegex(DesiredStateError, "does not support status reporting configuration"):
+            self.render(value, set())
+        environment, metadata = self.render(value, {"status_reporting_config"})
+        self.assertNotIn("CI_FLEET_STATUS_REPORTING_REQUIRED", environment)
+        self.assertTrue(metadata["status_reporting_configured"])
+        self.assertFalse(metadata["status_reporting_required"])
+
     def test_drained_controller_renders_zero_effective_capacity(self) -> None:
         value = config()
         value["controllers"]["example-ci-01"]["state"] = "drained"
diff --git a/scripts/test_health.py b/scripts/test_health.py
index 2cf89d3e..ed4cf561 100644
--- a/scripts/test_health.py
+++ b/scripts/test_health.py
@@ -425,6 +425,72 @@ def opener(request, timeout):
                 else:
                     os.environ["CI_FLEET_TESTING"] = old
 
+    def test_required_status_reporting_fails_closed_without_host_local_values(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            output = Path(directory) / "health.json"
+            monitoring = Path(directory) / "monitoring.env"
+            monitoring.write_text("CI_FLEET_HEALTH_SUPPRESS_DELIVERY=1\n")
+            monitoring.chmod(0o600)
+            old_collect = health.collect_snapshot
+            old_required = os.environ.get("CI_FLEET_STATUS_REPORTING_REQUIRED")
+            old_testing = os.environ.get("CI_FLEET_TESTING")
+            os.environ["CI_FLEET_STATUS_REPORTING_REQUIRED"] = "1"
+            os.environ["CI_FLEET_TESTING"] = "1"
+            setattr(health, "collect_snapshot", lambda _values: healthy_snapshot())
+            try:
+                result = health._local(health.argparse.Namespace(
+                    monitoring_config=monitoring, output=output, json=True,
+                ))
+                report = json.loads(output.read_text())
+                self.assertEqual(result, 1)
+                self.assertEqual(report["checks"][-1], {"id": "status_delivery", "status": "warning"})
+            finally:
+                setattr(health, "collect_snapshot", old_collect)
+                if old_required is None:
+                    os.environ.pop("CI_FLEET_STATUS_REPORTING_REQUIRED", None)
+                else:
+                    os.environ["CI_FLEET_STATUS_REPORTING_REQUIRED"] = old_required
+                if old_testing is None:
+                    os.environ.pop("CI_FLEET_TESTING", None)
+                else:
+                    os.environ["CI_FLEET_TESTING"] = old_testing
+
+    def test_required_status_reporting_redacts_invalid_host_local_config(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            output = Path(directory) / "health.json"
+            monitoring = Path(directory) / "monitoring.env"
+            old_collect = health.collect_snapshot
+            old_required = os.environ.get("CI_FLEET_STATUS_REPORTING_REQUIRED")
+            old_testing = os.environ.get("CI_FLEET_TESTING")
+            os.environ["CI_FLEET_STATUS_REPORTING_REQUIRED"] = "1"
+            os.environ["CI_FLEET_TESTING"] = "1"
+            setattr(health, "collect_snapshot", lambda _values: healthy_snapshot())
+            try:
+                for contents, mode in (
+                    ("not-an-env-line\n", 0o600),
+                    ("CI_FLEET_HEALTH_STATUS_URL=https://example.invalid/v1/status\n", 0o644),
+                    ("CI_FLEET_HEALTH_DISK_WARN_PERCENT=abc\n", 0o600),
+                ):
+                    monitoring.write_text(contents)
+                    monitoring.chmod(mode)
+                    result = health._local(health.argparse.Namespace(
+                        monitoring_config=monitoring, output=output, json=True,
+                    ))
+                    report = json.loads(output.read_text())
+                    self.assertEqual(result, 1)
+                    self.assertEqual(report["checks"][-1], {"id": "status_delivery", "status": "warning"})
+                    self.assertNotIn("example.invalid", output.read_text())
+            finally:
+                setattr(health, "collect_snapshot", old_collect)
+                if old_required is None:
+                    os.environ.pop("CI_FLEET_STATUS_REPORTING_REQUIRED", None)
+                else:
+                    os.environ["CI_FLEET_STATUS_REPORTING_REQUIRED"] = old_required
+                if old_testing is None:
+                    os.environ.pop("CI_FLEET_TESTING", None)
+                else:
+                    os.environ["CI_FLEET_TESTING"] = old_testing
+
     def test_expired_active_resources_and_stopped_capacity_are_observable(self) -> None:
         cleanup = "KEEP container runner state=running expired=1 (routine cleanup never removes active containers)\nWOULD_REMOVE volume old expired=1\n"
         run = lambda args: health.subprocess.CompletedProcess(args, 0, cleanup, "")
diff --git a/scripts/test_status_receiver.py b/scripts/test_status_receiver.py
index c756a55d..ff67a417 100644
--- a/scripts/test_status_receiver.py
+++ b/scripts/test_status_receiver.py
@@ -8,6 +8,7 @@
 import threading
 import time
 import unittest
+import urllib.error
 import urllib.request
 from pathlib import Path
 
@@ -116,6 +117,27 @@ def submit() -> None:
         thread.join()
         self.assertEqual(errors, [])
 
+    def test_concurrent_duplicate_reports_allow_one_success(self) -> None:
+        body, headers = self.signed(valid_report())
+        barrier = threading.Barrier(3)
+        outcomes = []
+
+        def submit() -> None:
+            barrier.wait()
+            try:
+                self.receiver.submit(body, headers, now=1_000)
+                outcomes.append("accepted")
+            except status_receiver.StatusError as error:
+                outcomes.append((error.status, error.code))
+
+        threads = [threading.Thread(target=submit) for _ in range(2)]
+        for thread in threads:
+            thread.start()
+        barrier.wait()
+        for thread in threads:
+            thread.join()
+        self.assertCountEqual(outcomes, ["accepted", (409, "replayed_report")])
+
     def test_read_token_cannot_reuse_controller_key(self) -> None:
         with self.assertRaisesRegex(ValueError, "read token"):
             status_receiver.StatusReceiver(
@@ -166,6 +188,54 @@ def test_replay_stale_authentication_and_stale_report_are_rejected(self) -> None
         stale_body, stale_headers = self.signed(valid_report(), timestamp=1_001, nonce="c" * 32)
         self.assert_status_error(409, "stale_report", lambda: self.receiver.submit(stale_body, stale_headers, now=1_001))
 
+    def test_persistence_failure_rolls_back_nonce_and_allows_retry(self) -> None:
+        body, headers = self.signed(valid_report())
+        with sqlite3.connect(self.receiver.database) as connection:
+            connection.execute("""
+                CREATE TRIGGER fail_report BEFORE INSERT ON reports
+                BEGIN SELECT RAISE(ABORT, 'test persistence failure'); END
+            """)
+        with self.assertRaises(sqlite3.Error):
+            self.receiver.submit(body, headers, now=1_000)
+        with sqlite3.connect(self.receiver.database) as connection:
+            self.assertEqual(connection.execute("SELECT COUNT(*) FROM nonces").fetchone()[0], 0)
+            self.assertEqual(connection.execute("SELECT COUNT(*) FROM reports").fetchone()[0], 0)
+            connection.execute("DROP TRIGGER fail_report")
+        self.receiver.submit(body, headers, now=1_000)
+        self.assertEqual(self.receiver.latest("example-ci-01", "reader-token"), valid_report())
+
+    def test_http_503_can_retry_the_same_signed_report(self) -> None:
+        server = status_receiver.create_server("127.0.0.1", 0, self.receiver)
+        thread = threading.Thread(target=server.serve_forever)
+        thread.start()
+        self.addCleanup(thread.join)
+        self.addCleanup(server.server_close)
+        self.addCleanup(server.shutdown)
+        now = int(time.time())
+        report = valid_report(generated_at=now)
+        report["controller"]["boot_time"] = now - 100
+        report["reconciliation"]["last_success_at"] = now - 10
+        body, headers = self.signed(report, timestamp=now)
+        request = urllib.request.Request(
+            f"http://127.0.0.1:{server.server_port}/v1/status",
+            data=body, headers=headers, method="POST",
+        )
+        with sqlite3.connect(self.receiver.database) as connection:
+            connection.execute("""
+                CREATE TRIGGER fail_report BEFORE INSERT ON reports
+                BEGIN SELECT RAISE(ABORT, 'test persistence failure'); END
+            """)
+        with self.assertRaises(urllib.error.HTTPError) as caught:
+            urllib.request.urlopen(request)
+        self.assertEqual(caught.exception.code, 503)
+        with sqlite3.connect(self.receiver.database) as connection:
+            connection.execute("DROP TRIGGER fail_report")
+        with urllib.request.urlopen(request) as response:
+            self.assertEqual(response.status, 202)
+        with self.assertRaises(urllib.error.HTTPError) as caught:
+            urllib.request.urlopen(request)
+        self.assertEqual(caught.exception.code, 409)
+
     def test_rejected_submissions_close_database_connections(self) -> None:
         connections = []
 
@@ -205,9 +275,11 @@ def test_payload_and_submission_frequency_are_bounded(self) -> None:
         self.assert_status_error(429, "submission_too_frequent", lambda: limited.submit(second_body, second_headers, now=1_001))
         third_body, third_headers = self.signed(valid_report(generated_at=1_002), timestamp=1_002, nonce="c" * 32)
         self.assert_status_error(429, "submission_too_frequent", lambda: limited.submit(third_body, third_headers, now=1_001))
+        with sqlite3.connect(limited.database) as connection:
+            self.assertEqual(connection.execute("SELECT COUNT(*) FROM nonces").fetchone()[0], 1)
+        limited.submit(second_body, second_headers, now=1_031)
         with sqlite3.connect(limited.database) as connection:
             self.assertEqual(connection.execute("SELECT COUNT(*) FROM nonces").fetchone()[0], 2)
-        self.assert_status_error(409, "replayed_report", lambda: limited.submit(second_body, second_headers, now=1_031))
 
     def test_schema_compatibility_and_malformed_metrics(self) -> None:
         future = valid_report()
@@ -354,6 +426,27 @@ def test_http_post_and_read_only_api(self) -> None:
         request = urllib.request.Request(base + "/v1/status", data=body, headers=headers, method="POST")
         with urllib.request.urlopen(request) as response:
             self.assertEqual(response.status, 202)
+        submit = self.receiver.submit
+        self.receiver.submit = lambda *_args, **_kwargs: (_ for _ in ()).throw(sqlite3.OperationalError("unavailable"))
+        try:
+            with self.assertRaises(urllib.error.HTTPError) as caught:
+                urllib.request.urlopen(request)
+            self.assertEqual(caught.exception.code, 503)
+            self.assertEqual(json.load(caught.exception), {"error": "unavailable"})
+        finally:
+            self.receiver.submit = submit
+        request = urllib.request.Request(base + "/healthz")
+        with urllib.request.urlopen(request) as response:
+            self.assertEqual(json.load(response), {"status": "ok"})
+        health = self.receiver.health
+        self.receiver.health = lambda: (_ for _ in ()).throw(sqlite3.OperationalError("unavailable"))
+        try:
+            with self.assertRaises(urllib.error.HTTPError) as caught:
+                urllib.request.urlopen(request)
+            self.assertEqual(caught.exception.code, 503)
+            self.assertEqual(json.load(caught.exception), {"error": "unavailable"})
+        finally:
+            self.receiver.health = health
         request = urllib.request.Request(base + "/v1/controllers", headers={"Authorization": "Bearer reader-token"})
         with urllib.request.urlopen(request) as response:
             payload = json.load(response)
@@ -364,6 +457,93 @@ def test_http_post_and_read_only_api(self) -> None:
         self.assertEqual(payload["latest"], report)
         self.assertEqual(payload["history"], [report])
 
+    def test_health_requires_receiver_schema(self) -> None:
+        queries: list[str] = []
+        now = [0.0]
+        connect = self.receiver._connect
+
+        def traced_connect() -> sqlite3.Connection:
+            connection = connect()
+            connection.set_trace_callback(queries.append)
+            return connection
+
+        self.receiver._connect = traced_connect
+        self.receiver._monotonic = lambda: now[0]
+        self.receiver.health()
+        self.receiver.health()
+        self.assertEqual(sum("quick_check" in query.lower() for query in queries), 1)
+        with sqlite3.connect(self.receiver.database) as connection:
+            connection.execute("DROP TABLE nonces")
+        now[0] = 60.0
+        with self.assertRaises(sqlite3.Error):
+            self.receiver.health()
+
+    def test_health_write_probe_leaves_no_application_rows(self) -> None:
+        with sqlite3.connect(self.receiver.database) as connection:
+            before = tuple(connection.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] for table in ("reports", "nonces"))
+        self.receiver.health()
+        with sqlite3.connect(self.receiver.database) as connection:
+            after = tuple(connection.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] for table in ("reports", "nonces"))
+            self.assertEqual(connection.execute("SELECT COUNT(*) FROM health_write_probe").fetchone()[0], 0)
+        self.assertEqual(after, before)
+
+    def test_health_rejects_commit_failure(self) -> None:
+        class FailingCommit(sqlite3.Connection):
+            def commit(self) -> None:
+                raise sqlite3.OperationalError("test commit failure")
+
+        self.receiver._connect = lambda: sqlite3.connect(self.receiver.database, factory=FailingCommit)
+        with self.assertRaisesRegex(sqlite3.DatabaseError, "health check failed"):
+            self.receiver.health()
+
+    def test_health_rejects_read_only_storage_and_caches_failure(self) -> None:
+        queries: list[str] = []
+
+        def read_only_connect() -> sqlite3.Connection:
+            connection = sqlite3.connect(f"file:{self.receiver.database}?mode=ro", uri=True)
+            connection.set_trace_callback(queries.append)
+            return connection
+
+        self.receiver._connect = read_only_connect
+        for _ in range(2):
+            with self.assertRaisesRegex(sqlite3.DatabaseError, "health check failed"):
+                self.receiver.health()
+        self.assertEqual(sum("insert" in query.lower() and "health_write_probe" in query.lower() for query in queries), 1)
+
+    def test_health_caches_failure_then_rechecks_and_recovers(self) -> None:
+        checks = 0
+        results = iter([("corrupt",), ("ok",)])
+
+        class Connection:
+            def execute(self, query: str, parameters=()):
+                nonlocal checks
+                if "quick_check" in query.lower():
+                    checks += 1
+                    result = next(results)
+                    return type("Result", (), {"fetchone": lambda self: result})()
+                return self
+
+            def commit(self) -> None:
+                pass
+
+            def close(self) -> None:
+                pass
+
+        now = 0.0
+        self.receiver._connect = Connection
+        self.receiver._monotonic = lambda: now
+        with self.assertRaises(sqlite3.DatabaseError):
+            self.receiver.health()
+        now = 59.0
+        with self.assertRaises(sqlite3.DatabaseError):
+            self.receiver.health()
+        self.assertEqual(checks, 1)
+        now = 60.0
+        self.receiver.health()
+        now = 119.0
+        self.receiver.health()
+        self.assertEqual(checks, 2)
+
     def test_read_api_authentication_and_controller_listing(self) -> None:
         self.submit(valid_report())
         self.assert_status_error(401, "read_authentication_failed", lambda: self.receiver.latest("example-ci-01", "wrong"))
diff --git a/scripts/validate.sh b/scripts/validate.sh
index 56f2fa8d..9660aaa2 100755
--- a/scripts/validate.sh
+++ b/scripts/validate.sh
@@ -23,10 +23,12 @@ python3 scripts/test_health.py
 python3 scripts/test_status_receiver.py
 python3 scripts/test_quickstart.py
 python3 -m json.tool schemas/status-report-v1.json >/dev/null
+python3 scripts/desired_state.py validate-engine-capabilities --manifest engine-capabilities.json --require-status-reporting-config --require-status-reporting >/dev/null
 python3 .github/actions/plan/plan.py --plan examples/project/scripts/ci/plan.json --group fast >/dev/null
 python3 .github/actions/plan/plan.py --plan examples/project/scripts/ci/plan.json --group full >/dev/null
 scripts/test-capacity-preflight.sh
 scripts/test-install-worker-controller.sh
+scripts/test-install-status-receiver.sh
 
 tmp=$(mktemp)
 trap 'rm -f "$tmp"' EXIT
diff --git a/templates/config-repository/.github/workflows/validate.yml b/templates/config-repository/.github/workflows/validate.yml
index fe492185..8537d641 100644
--- a/templates/config-repository/.github/workflows/validate.yml
+++ b/templates/config-repository/.github/workflows/validate.yml
@@ -56,8 +56,20 @@ jobs:
           done
 
       - name: Validate reference configurations
+        env:
+          BASE_SHA: ${{ github.event.pull_request.base.sha || github.event.before }}
         run: |
-          ./scripts/validate.sh
+          if [[ -n "$BASE_SHA" && ! "$BASE_SHA" =~ ^0+$ ]] && git cat-file -e "$BASE_SHA:fleet.json" 2>/dev/null; then
+            git show "$BASE_SHA:fleet.json" >"$RUNNER_TEMP/previous-fleet.json"
+            args=(--previous-config "$RUNNER_TEMP/previous-fleet.json")
+            if git cat-file -e "$BASE_SHA:engine-rollout-evidence.json" 2>/dev/null; then
+              git show "$BASE_SHA:engine-rollout-evidence.json" >"$RUNNER_TEMP/previous-rollout-evidence.json"
+              args+=(--previous-rollout-evidence "$RUNNER_TEMP/previous-rollout-evidence.json")
+            fi
+            ./scripts/validate.sh "${args[@]}"
+          else
+            ./scripts/validate.sh
+          fi
           ./scripts/validate.sh --config examples/multi-host/fleet.json
 
       - name: Prove initialized configurations pass strict policy
diff --git a/templates/config-repository/README.md b/templates/config-repository/README.md
index 5476d06a..e54b315e 100644
--- a/templates/config-repository/README.md
+++ b/templates/config-repository/README.md
@@ -62,6 +62,20 @@ The initializer refuses to replace a configured file unless `--force` is explici
 - a zero managed minimum and reviewed maximum runner capacity;
 - CPU and memory available to each ephemeral runner.
 
+`status_reporting` is deliberately omitted from initialized and reference
+configurations. For an existing controller, roll out schema support in three
+separately reviewed, integrated changes: first update only `engine_ref` and prove
+routine reconciliation has activated that engine; then record the controller ID,
+proven active ref, and reviewed `status_reporting_config` and
+`required_status_reporting` capability booleans in
+`engine-rollout-evidence.json`; only then add the optional `status_reporting`
+object without changing `engine_ref`. Enabling required delivery also requires the
+prior evidence to record both capabilities as `true`. Transition validation
+rejects introducing or enabling the property without the corresponding evidence.
+This staging prevents an older active manager from rejecting the new property before it
+can upgrade itself. Endpoint and key values remain host-local and never enter
+Git.
+
 The controller ID is how a target host selects its declaration. A location is a non-sensitive logical slug such as `primary-site` or `remote-site`, never an address. Runtime-generated configuration and credentials remain host-local.
 
 ### Pool capacity is infrastructure policy
diff --git a/templates/config-repository/engine-rollout-evidence.json b/templates/config-repository/engine-rollout-evidence.json
new file mode 100644
index 00000000..934e58ab
--- /dev/null
+++ b/templates/config-repository/engine-rollout-evidence.json
@@ -0,0 +1,4 @@
+{
+  "schema_version": 1,
+  "status_reporting_engine_capabilities": {}
+}
diff --git a/templates/config-repository/fleet.schema.json b/templates/config-repository/fleet.schema.json
index 606536ad..c08e7920 100644
--- a/templates/config-repository/fleet.schema.json
+++ b/templates/config-repository/fleet.schema.json
@@ -80,7 +80,16 @@
         "engine_ref": {"type": "string", "pattern": "^(?!0{40}$)[0-9a-f]{40}$"},
         "min_runners": {"const": 0},
         "max_runners": {"type": "integer", "minimum": 1},
-        "runner_resources": {"$ref": "#/$defs/runner_resources"}
+        "runner_resources": {"$ref": "#/$defs/runner_resources"},
+        "status_reporting": {
+          "type": "object",
+          "additionalProperties": false,
+          "required": ["enabled", "config_file"],
+          "properties": {
+            "enabled": {"type": "boolean"},
+            "config_file": {"const": "/etc/ci-fleet/monitoring.env"}
+          }
+        }
       }
     },
     "runner_resources": {
diff --git a/templates/config-repository/scripts/test_policy.py b/templates/config-repository/scripts/test_policy.py
index fc6d0380..77b0ad75 100755
--- a/templates/config-repository/scripts/test_policy.py
+++ b/templates/config-repository/scripts/test_policy.py
@@ -6,13 +6,14 @@
 import copy
 import json
 import re
+import shutil
 import subprocess
 import sys
 import tempfile
 import unittest
 from pathlib import Path
 
-from validate import Validation, load_json, scan_secret_material, scan_tree_path_list, validate_config
+from validate import Validation, load_json, scan_secret_material, scan_tree_path_list, validate_config, validate_rollout_evidence, validate_transition
 
 
 ROOT = Path(__file__).resolve().parents[1]
@@ -71,6 +72,344 @@ def assert_delivery_engine_contract(self, value: str, accepted: bool) -> None:
     def test_reference_configuration_is_valid(self) -> None:
         self.assertEqual(errors_for(reference_config()), [])
 
+    def test_status_reporting_null_is_rejected(self) -> None:
+        config = copy.deepcopy(reference_config())
+        first_controller(config)["status_reporting"] = None
+        self.assert_rejected(config, "must be an object")
+
+    def test_initializer_omits_status_reporting_for_staged_adoption(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            output = Path(directory) / "fleet.json"
+            subprocess.run([
+                sys.executable, str(ROOT / "scripts" / "init.py"),
+                "--organization", "sample-org", "--project", "sample-app",
+                "--engine-ref", "1" * 40, "--output", str(output),
+            ], check=True, stdout=subprocess.DEVNULL)
+            controller = first_controller(json.loads(output.read_text()))
+        self.assertNotIn("status_reporting", controller)
+
+    def test_status_reporting_requires_a_separate_engine_rollout(self) -> None:
+        previous = reference_config()
+        current = copy.deepcopy(previous)
+        first_controller(current)["engine_ref"] = "2" * 40
+        first_controller(current)["status_reporting"] = {
+            "enabled": False,
+            "config_file": "/etc/ci-fleet/monitoring.env",
+        }
+        validation = Validation()
+        validate_transition(previous, current, {}, validation)
+        self.assertTrue(any("later commit" in error for error in validation.errors), validation.errors)
+
+        staged = copy.deepcopy(current)
+        first_controller(staged).pop("status_reporting")
+        validation = Validation()
+        validate_transition(staged, current, {}, validation)
+        self.assertTrue(any("rollout evidence" in error for error in validation.errors), validation.errors)
+        validation = Validation()
+        controller_name = next(iter(staged["controllers"]))
+        validate_transition(staged, current, {
+            controller_name: {
+                "engine_ref": first_controller(staged)["engine_ref"],
+                "status_reporting_config": True,
+                "required_status_reporting": False,
+            },
+        }, validation)
+        self.assertEqual(validation.errors, [])
+        validation = Validation()
+        validate_transition(staged, current, {
+            controller_name: {
+                "engine_ref": first_controller(staged)["engine_ref"],
+                "status_reporting_config": True,
+                "required_status_reporting": False,
+            },
+        }, validation, {})
+        self.assertTrue(any("capability evidence" in error for error in validation.errors), validation.errors)
+
+    def test_retained_reporting_requires_target_engine_capabilities(self) -> None:
+        previous = reference_config()
+        first_controller(previous)["status_reporting"] = {
+            "enabled": False,
+            "config_file": "/etc/ci-fleet/monitoring.env",
+        }
+        current = copy.deepcopy(previous)
+        first_controller(current)["engine_ref"] = "2" * 40
+        controller = next(iter(current["controllers"]))
+        compatible = {
+            controller: {
+                "engine_ref": "2" * 40,
+                "status_reporting_config": True,
+                "required_status_reporting": False,
+            }
+        }
+        validation = Validation()
+        validate_transition(previous, current, compatible, validation)
+        self.assertEqual(validation.errors, [])
+
+        for evidence in (
+            {},
+            {controller: {**compatible[controller], "engine_ref": "3" * 40}},
+            {controller: {**compatible[controller], "status_reporting_config": False}},
+        ):
+            validation = Validation()
+            validate_transition(previous, current, evidence, validation)
+            self.assertTrue(any("configuration capability evidence" in error for error in validation.errors), validation.errors)
+
+        first_controller(previous)["status_reporting"]["enabled"] = True
+        first_controller(current)["status_reporting"]["enabled"] = True
+        validation = Validation()
+        validate_transition(previous, current, compatible, validation)
+        self.assertTrue(any("required status-reporting" in error for error in validation.errors), validation.errors)
+        compatible[controller]["required_status_reporting"] = True
+        validation = Validation()
+        validate_transition(previous, current, compatible, validation)
+        self.assertEqual(validation.errors, [])
+
+    def test_reporting_removal_before_engine_change_needs_no_evidence(self) -> None:
+        previous = reference_config()
+        first_controller(previous)["status_reporting"] = {
+            "enabled": True,
+            "config_file": "/etc/ci-fleet/monitoring.env",
+        }
+        removed = copy.deepcopy(previous)
+        first_controller(removed).pop("status_reporting")
+        validation = Validation()
+        validate_transition(previous, removed, {}, validation)
+        self.assertEqual(validation.errors, [])
+
+        changed = copy.deepcopy(removed)
+        first_controller(changed)["engine_ref"] = "2" * 40
+        validation = Validation()
+        validate_transition(removed, changed, {}, validation)
+        self.assertEqual(validation.errors, [])
+
+    def test_rollout_evidence_requires_explicit_capability_booleans(self) -> None:
+        base = {"engine_ref": "1" * 40, "status_reporting_config": True, "required_status_reporting": False}
+        for evidence in (
+            {key: value for key, value in base.items() if key != "status_reporting_config"},
+            {**base, "status_reporting_config": None},
+            {**base, "required_status_reporting": None},
+        ):
+            validation = Validation()
+            self.assertEqual(validate_rollout_evidence({
+                "schema_version": 1,
+                "status_reporting_engine_capabilities": {"example-ci-01": evidence},
+            }, validation), {})
+            self.assertTrue(validation.errors)
+
+    def test_new_controller_cannot_introduce_status_reporting(self) -> None:
+        previous = reference_config()
+        current = copy.deepcopy(previous)
+        controller = copy.deepcopy(first_controller(current))
+        controller["scale_set_name"] = "example-ci-02-scale"
+        controller["status_reporting"] = {
+            "enabled": False,
+            "config_file": "/etc/ci-fleet/monitoring.env",
+        }
+        current["controllers"]["example-ci-02"] = controller
+        validation = Validation()
+        validate_transition(previous, current, {}, validation)
+        self.assertTrue(any("new controller" in error for error in validation.errors), validation.errors)
+
+    def test_rollout_evidence_is_scoped_to_its_controller(self) -> None:
+        previous = reference_config()
+        second = copy.deepcopy(first_controller(previous))
+        second["scale_set_name"] = "example-ci-02-scale"
+        previous["controllers"]["example-ci-02"] = second
+        current = copy.deepcopy(previous)
+        current["controllers"]["example-ci-02"]["status_reporting"] = {
+            "enabled": False,
+            "config_file": "/etc/ci-fleet/monitoring.env",
+        }
+        validation = Validation()
+        validate_transition(previous, current, {
+            "example-ci-01": {
+                "engine_ref": second["engine_ref"],
+                "status_reporting_config": True,
+                "required_status_reporting": False,
+            },
+        }, validation)
+        self.assertTrue(any("for this controller" in error for error in validation.errors), validation.errors)
+
+    def test_rollout_evidence_requires_controller_mapping(self) -> None:
+        validation = Validation()
+        evidence = {
+            "engine_ref": "1" * 40,
+            "status_reporting_config": True,
+            "required_status_reporting": False,
+        }
+        refs = validate_rollout_evidence({
+            "schema_version": 1,
+            "status_reporting_engine_capabilities": {"example-ci-01": evidence},
+        }, validation)
+        self.assertEqual(refs, {"example-ci-01": evidence})
+        self.assertEqual(validation.errors, [])
+
+    def test_rollout_evidence_requires_previous_engine_selection(self) -> None:
+        previous = reference_config()
+        current = copy.deepcopy(previous)
+        first_controller(current)["engine_ref"] = "2" * 40
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            (root / "previous.json").write_text(json.dumps(previous), encoding="utf-8")
+            (root / "current.json").write_text(json.dumps(current), encoding="utf-8")
+            (root / "previous-evidence.json").write_text(json.dumps({
+                "schema_version": 1,
+                "status_reporting_engine_capabilities": {},
+            }), encoding="utf-8")
+            (root / "evidence.json").write_text(json.dumps({
+                "schema_version": 1,
+                "status_reporting_engine_capabilities": {
+                    next(iter(current["controllers"])): {
+                        "engine_ref": "2" * 40,
+                        "status_reporting_config": True,
+                        "required_status_reporting": False,
+                    },
+                },
+            }), encoding="utf-8")
+            result = subprocess.run([
+                sys.executable, str(ROOT / "scripts" / "validate.py"),
+                "--config", str(root / "current.json"),
+                "--previous-config", str(root / "previous.json"),
+                "--rollout-evidence", str(root / "evidence.json"),
+                "--previous-rollout-evidence", str(root / "previous-evidence.json"),
+                "--skip-path-scan",
+            ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("must already be selected", result.stderr)
+
+    def test_rollout_evidence_must_match_current_controller_ref(self) -> None:
+        previous = reference_config()
+        current = copy.deepcopy(previous)
+        controller_name = next(iter(current["controllers"]))
+        proven_ref = first_controller(previous)["engine_ref"]
+        first_controller(current)["engine_ref"] = "2" * 40
+        evidence = {
+            "schema_version": 1,
+            "status_reporting_engine_capabilities": {
+                controller_name: {
+                    "engine_ref": proven_ref,
+                    "status_reporting_config": True,
+                    "required_status_reporting": False,
+                },
+            },
+        }
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            for name, value in (
+                ("previous.json", previous),
+                ("current.json", current),
+                ("previous-evidence.json", evidence),
+                ("evidence.json", evidence),
+            ):
+                (root / name).write_text(json.dumps(value), encoding="utf-8")
+            result = subprocess.run([
+                sys.executable, str(ROOT / "scripts" / "validate.py"),
+                "--config", str(root / "current.json"),
+                "--previous-config", str(root / "previous.json"),
+                "--rollout-evidence", str(root / "evidence.json"),
+                "--previous-rollout-evidence", str(root / "previous-evidence.json"),
+                "--skip-path-scan",
+            ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("must match the current controller engine_ref", result.stderr)
+
+    def test_enabling_required_reporting_needs_required_capability_evidence(self) -> None:
+        previous = reference_config()
+        controller_name = next(iter(previous["controllers"]))
+        first_controller(previous)["status_reporting"] = {
+            "enabled": False,
+            "config_file": "/etc/ci-fleet/monitoring.env",
+        }
+        current = copy.deepcopy(previous)
+        first_controller(current)["status_reporting"]["enabled"] = True
+        evidence = {
+            "schema_version": 1,
+            "status_reporting_engine_capabilities": {
+                controller_name: {
+                    "engine_ref": first_controller(previous)["engine_ref"],
+                    "status_reporting_config": True,
+                    "required_status_reporting": False,
+                },
+            },
+        }
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            for name, value in (
+                ("previous.json", previous),
+                ("current.json", current),
+                ("previous-evidence.json", evidence),
+                ("evidence.json", evidence),
+            ):
+                (root / name).write_text(json.dumps(value), encoding="utf-8")
+            result = subprocess.run([
+                sys.executable, str(ROOT / "scripts" / "validate.py"),
+                "--config", str(root / "current.json"),
+                "--previous-config", str(root / "previous.json"),
+                "--rollout-evidence", str(root / "evidence.json"),
+                "--previous-rollout-evidence", str(root / "previous-evidence.json"),
+                "--skip-path-scan",
+            ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("requires required status-reporting rollout evidence", result.stderr)
+        validation = Validation()
+        validate_transition(previous, current, {
+            controller_name: {
+                "engine_ref": first_controller(previous)["engine_ref"],
+                "status_reporting_config": True,
+                "required_status_reporting": True,
+            },
+        }, validation)
+        self.assertEqual(validation.errors, [])
+
+    def test_alternate_config_does_not_use_fleet_rollout_evidence(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            template = Path(directory) / "template"
+            shutil.copytree(ROOT, template)
+            (template / "engine-rollout-evidence.json").write_text(json.dumps({
+                "schema_version": 1,
+                "status_reporting_engine_capabilities": {
+                    "private-ci-01": {
+                        "engine_ref": "1" * 40,
+                        "status_reporting_config": True,
+                        "required_status_reporting": False,
+                    },
+                },
+            }), encoding="utf-8")
+            result = subprocess.run([
+                sys.executable, str(template / "scripts" / "validate.py"),
+                "--config", str(template / "examples" / "multi-host" / "fleet.json"),
+                "--skip-path-scan",
+            ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+        self.assertEqual(result.returncode, 0, result.stderr)
+
+    def test_default_fleet_config_cannot_be_a_symlink(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            template = Path(directory) / "template"
+            shutil.copytree(ROOT, template)
+            fleet = template / "fleet.json"
+            fleet.rename(template / "fleet-target.json")
+            fleet.symlink_to("fleet-target.json")
+            result = subprocess.run([
+                sys.executable, str(template / "scripts" / "validate.py"),
+                "--skip-path-scan",
+            ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("symlinked JSON files are forbidden", result.stderr)
+
+    def test_rollout_evidence_cannot_be_a_symlink(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            template = Path(directory) / "template"
+            shutil.copytree(ROOT, template)
+            evidence = template / "engine-rollout-evidence.json"
+            evidence.rename(template / "evidence-target.json")
+            evidence.symlink_to("evidence-target.json")
+            result = subprocess.run([
+                sys.executable, str(template / "scripts" / "validate.py"),
+                "--skip-path-scan",
+            ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("symlinked JSON files are forbidden", result.stderr)
+
     def test_multi_host_multi_location_configuration_is_valid(self) -> None:
         config = json.loads((ROOT / "examples" / "multi-host" / "fleet.json").read_text(encoding="utf-8"))
         self.assertEqual(errors_for(config), [])
diff --git a/templates/config-repository/scripts/validate.py b/templates/config-repository/scripts/validate.py
index 77cd706f..db70032a 100755
--- a/templates/config-repository/scripts/validate.py
+++ b/templates/config-repository/scripts/validate.py
@@ -89,6 +89,9 @@ def reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
             value[key] = child
         return value
 
+    if path.is_symlink():
+        validation.errors.append(f"{path}: symlinked JSON files are forbidden")
+        return None
     try:
         return json.loads(path.read_text(encoding="utf-8"), object_pairs_hook=reject_duplicate_keys)
     except FileNotFoundError:
@@ -268,7 +271,7 @@ def validate_config(config: Any, validation: Validation, strict: bool) -> None:
     for name, controller in controllers.items():
         path = f"$.controllers.{name}"
         validation.require(isinstance(name, str) and bool(SLUG.fullmatch(name)), path, "controller ID must be a unique lowercase slug")
-        if not validation.exact_keys(controller, path, controller_keys):
+        if not validation.exact_keys(controller, path, controller_keys, {"status_reporting"}):
             continue
         pool_name = controller.get("pool")
         location = controller.get("location")
@@ -296,6 +299,10 @@ def validate_config(config: Any, validation: Validation, strict: bool) -> None:
         if type(minimum) is int and type(maximum) is int:
             validation.require(minimum <= maximum, f"{path}.min_runners", "must not exceed max_runners")
         validation.require(minimum == 0, f"{path}.min_runners", "must be zero because managed prewarmed runners are not supported")
+        status_reporting = controller.get("status_reporting")
+        if "status_reporting" in controller and validation.exact_keys(status_reporting, f"{path}.status_reporting", {"enabled", "config_file"}):
+            validation.require(type(status_reporting.get("enabled")) is bool, f"{path}.status_reporting.enabled", "must be a boolean")
+            validation.require(status_reporting.get("config_file") == "/etc/ci-fleet/monitoring.env", f"{path}.status_reporting.config_file", "must use the fixed host-local monitoring configuration")
         resources = controller.get("runner_resources")
         if validation.exact_keys(resources, f"{path}.runner_resources", {"cpu_cores", "memory_mib"}):
             cpu = resources.get("cpu_cores")
@@ -400,9 +407,126 @@ def validate_config(config: Any, validation: Validation, strict: bool) -> None:
             validation.require(repository != "example-org/example-app", f"{path}.repository", "replace the example repository before use")
 
 
+def validate_rollout_evidence(value: Any, validation: Validation) -> dict[str, dict[str, Any]]:
+    if not validation.exact_keys(
+        value,
+        "engine-rollout-evidence.json",
+        {"schema_version", "status_reporting_engine_capabilities"},
+    ):
+        return {}
+    validation.require(value.get("schema_version") == 1, "engine-rollout-evidence.json.schema_version", "must equal 1")
+    capabilities = value.get("status_reporting_engine_capabilities")
+    if not isinstance(capabilities, dict):
+        validation.errors.append("engine-rollout-evidence.json.status_reporting_engine_capabilities: must be an object mapping controller IDs to capability evidence")
+        return {}
+    valid: dict[str, dict[str, Any]] = {}
+    for controller, evidence in capabilities.items():
+        path = f"engine-rollout-evidence.json.status_reporting_engine_capabilities.{controller}"
+        controller_valid = bool(SLUG.fullmatch(controller))
+        validation.require(controller_valid, path, "controller ID must be a lowercase slug")
+        if not validation.exact_keys(evidence, path, {"engine_ref", "status_reporting_config", "required_status_reporting"}):
+            continue
+        ref = evidence.get("engine_ref")
+        configured = evidence.get("status_reporting_config")
+        required = evidence.get("required_status_reporting")
+        ref_valid = isinstance(ref, str) and bool(COMMIT_SHA.fullmatch(ref)) and ref != "0" * 40
+        validation.require(ref_valid, f"{path}.engine_ref", "must be a nonzero full lowercase commit SHA")
+        validation.require(type(configured) is bool, f"{path}.status_reporting_config", "must be a boolean")
+        validation.require(type(required) is bool, f"{path}.required_status_reporting", "must be a boolean")
+        if controller_valid and ref_valid and type(configured) is bool and type(required) is bool:
+            valid[controller] = {
+                "engine_ref": ref,
+                "status_reporting_config": configured,
+                "required_status_reporting": required,
+            }
+    return valid
+
+
+def validate_reporting_evidence(
+    name: str,
+    controller: dict[str, Any],
+    evidence: dict[str, Any],
+    validation: Validation,
+) -> None:
+    reporting = controller.get("status_reporting")
+    if not isinstance(reporting, dict):
+        return
+    validation.require(
+        evidence.get("engine_ref") == controller.get("engine_ref")
+        and evidence.get("status_reporting_config") is True,
+        f"$.controllers.{name}.status_reporting",
+        "requires status-reporting configuration capability evidence for this controller and engine_ref",
+    )
+    if reporting.get("enabled") is True:
+        validation.require(
+            evidence.get("engine_ref") == controller.get("engine_ref")
+            and evidence.get("required_status_reporting") is True,
+            f"$.controllers.{name}.status_reporting.enabled",
+            "requires required status-reporting rollout evidence for this controller and engine_ref",
+        )
+
+
+def validate_transition(
+    previous: Any,
+    current: Any,
+    compatible_engine_refs: dict[str, dict[str, Any]],
+    validation: Validation,
+    previous_compatible_engine_refs: dict[str, dict[str, Any]] | None = None,
+) -> None:
+    if not isinstance(previous, dict) or not isinstance(current, dict):
+        return
+    old_controllers = previous.get("controllers")
+    new_controllers = current.get("controllers")
+    if not isinstance(old_controllers, dict) or not isinstance(new_controllers, dict):
+        return
+    for name, new in new_controllers.items():
+        old = old_controllers.get(name)
+        if not isinstance(new, dict):
+            continue
+        if name not in old_controllers:
+            if "status_reporting" in new:
+                validation.errors.append(
+                    f"$.controllers.{name}.status_reporting: must be omitted from a new controller until its engine rollout is proven"
+                )
+            continue
+        if not isinstance(old, dict):
+            continue
+        current_evidence = compatible_engine_refs.get(name, {})
+        previous_evidence_source = (
+            compatible_engine_refs
+            if previous_compatible_engine_refs is None
+            else previous_compatible_engine_refs
+        )
+        previous_evidence = previous_evidence_source.get(name, {})
+        old_reporting = old.get("status_reporting")
+        new_reporting = new.get("status_reporting")
+        staged_capability_required = (
+            "status_reporting" not in old
+            or (
+                isinstance(new_reporting, dict)
+                and new_reporting.get("enabled") is True
+                and (not isinstance(old_reporting, dict) or old_reporting.get("enabled") is not True)
+            )
+        )
+        evidence = previous_evidence if staged_capability_required else current_evidence
+        validate_reporting_evidence(name, new, evidence, validation)
+        if "status_reporting" not in old and "status_reporting" in new:
+            validation.require(
+                old.get("engine_ref") == new.get("engine_ref"),
+                f"$.controllers.{name}.status_reporting",
+                "must be introduced in a later commit after the compatible engine_ref is active",
+            )
+            validation.require(
+                evidence.get("engine_ref") == old.get("engine_ref"),
+                f"$.controllers.{name}.status_reporting",
+                "requires reviewed rollout evidence for this controller and its already-active compatible engine_ref",
+            )
 def parse_args() -> argparse.Namespace:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--config", type=Path, default=ROOT / "fleet.json", help="configuration file to validate")
+    parser.add_argument("--previous-config", type=Path, help="previous integrated configuration for rollout validation")
+    parser.add_argument("--rollout-evidence", type=Path, help="rollout evidence file (defaults to engine-rollout-evidence.json only for the default fleet.json)")
+    parser.add_argument("--previous-rollout-evidence", type=Path, help="previous integrated rollout evidence")
     parser.add_argument("--strict", action="store_true", help="reject unchanged example values")
     parser.add_argument("--skip-path-scan", action="store_true", help="skip repository path checks (for external fixtures)")
     parser.add_argument("--tree-paths", type=Path, help="NUL-delimited committed paths to scan instead of the local template tree")
@@ -412,13 +536,71 @@ def parse_args() -> argparse.Namespace:
 def main() -> int:
     args = parse_args()
     validation = Validation()
-    config = load_json(args.config.resolve(), validation)
+    config_path = args.config.absolute()
+    config = load_json(config_path, validation)
+    evidence_path = args.rollout_evidence.absolute() if args.rollout_evidence else None
+    if evidence_path is None and config_path == ROOT / "fleet.json":
+        evidence_path = ROOT / "engine-rollout-evidence.json"
+    evidence = load_json(evidence_path, validation) if evidence_path is not None else None
+    current_compatible_engine_refs = (
+        validate_rollout_evidence(evidence, validation)
+        if evidence is not None
+        else {}
+    )
     schema = load_json(ROOT / "fleet.schema.json", validation)
     if schema is not None:
         validation.require(schema.get("$schema") == "https://json-schema.org/draft/2020-12/schema", "fleet.schema.json.$schema", "must use JSON Schema draft 2020-12")
     if config is not None:
         scan_secret_material(config, validation)
         validate_config(config, validation, args.strict)
+        current_controllers = config.get("controllers", {}) if isinstance(config, dict) else {}
+        for controller, evidence in current_compatible_engine_refs.items():
+            current_controller = current_controllers.get(controller) if isinstance(current_controllers, dict) else None
+            validation.require(
+                isinstance(current_controller, dict) and current_controller.get("engine_ref") == evidence["engine_ref"],
+                f"engine-rollout-evidence.json.status_reporting_engine_capabilities.{controller}.engine_ref",
+                "must match the current controller engine_ref; remove stale evidence before changing or removing the controller",
+            )
+        if isinstance(current_controllers, dict):
+            for controller, value in current_controllers.items():
+                if isinstance(value, dict):
+                    validate_reporting_evidence(
+                        controller,
+                        value,
+                        current_compatible_engine_refs.get(controller, {}),
+                        validation,
+                    )
+        if args.previous_config is not None:
+            previous = load_json(args.previous_config.absolute(), validation)
+            previous_evidence = (
+                load_json(args.previous_rollout_evidence.absolute(), validation)
+                if args.previous_rollout_evidence
+                else None
+            )
+            previous_compatible_engine_refs = (
+                validate_rollout_evidence(previous_evidence, validation)
+                if previous_evidence is not None
+                else {}
+            )
+            if previous is not None:
+                previous_controllers = previous.get("controllers", {}) if isinstance(previous, dict) else {}
+                for controller, evidence in current_compatible_engine_refs.items():
+                    if previous_compatible_engine_refs.get(controller) == evidence:
+                        continue
+                    ref = evidence["engine_ref"]
+                    previous_controller = previous_controllers.get(controller) if isinstance(previous_controllers, dict) else None
+                    validation.require(
+                        isinstance(previous_controller, dict) and previous_controller.get("engine_ref") == ref,
+                        f"engine-rollout-evidence.json.status_reporting_engine_capabilities.{controller}.engine_ref",
+                        f"{ref} must already be selected for this controller in the previous integrated fleet configuration",
+                    )
+                validate_transition(
+                    previous,
+                    config,
+                    current_compatible_engine_refs,
+                    validation,
+                    previous_compatible_engine_refs,
+                )
     if args.tree_paths is not None:
         scan_tree_path_list(args.tree_paths, validation)
     elif not args.skip_path_scan: