From d4f3a29262cb3e7e8ef9cceaa0420f98331cc4c8 Mon Sep 17 00:00:00 2001
From: Jeff Huber <jeffhuber@users.noreply.github.com>
Date: Mon, 21 Sep 2026 00:32:41 -0700
Subject: [PATCH 1/8] Make Board replacement reconcile final state

---
 docs/board-service-lifecycle.md |  54 ++++--
 src/code_mower/board_service.py | 318 +++++++++++++++++++++++++++++++-
 tests/test_board_service.py     | 161 ++++++++++++----
 3 files changed, 484 insertions(+), 49 deletions(-)

diff --git a/docs/board-service-lifecycle.md b/docs/board-service-lifecycle.md
index 29baff03..eccdf13f 100644
--- a/docs/board-service-lifecycle.md
+++ b/docs/board-service-lifecycle.md
@@ -144,18 +144,32 @@ and the `board stop` keepalive guard could no longer see it. That case reports
 `remove_incomplete` and leaves the definition in place.
 
 `--replace` is the only way to take over an existing definition for a port, and
-the replacement is atomic: the definition file is swapped with `os.replace`, and
-a failed bootstrap restores exactly the previous definition or reports
-`rollback_failed`. The swap is also the point at which the original contents
-stop existing, so it happens only once the old job is established as unloaded --
-a `bootout` that succeeded, or the same positively-confirmed absence `remove`
-requires. Writing over a job launchd still holds would fail the bootstrap anyway
-(launchd will not accept a label its domain already holds) and the rollback
-would then preserve the replacement rather than an original that is by then
-gone, so that case refuses as `unload_failed` and changes nothing. A write that fails outright leaves the previous definition on
-disk untouched, so recovery there is to load it again rather than to restore it;
-either way the payload's `rollback` field says whether the previous service came
-back.
+the replacement is atomic: the definition file is swapped with `os.replace`,
+then the terminal result is decided from the host state rather than from a
+`launchctl` return code. The final reconciliation re-reads the definition,
+launchd's running pid and argument vector, the process checkout and supervisor,
+exclusive port ownership, repository identity, installed version and serving
+version. Success means those reads prove the new definition and new process. A
+failed apply means the same reads prove the restored definition and restored
+process. JSON carries that evidence in `reconciliation`; text prints exactly one
+verified state.
+
+This matters because `launchctl bootstrap` can register and start a job before
+reporting a timeout or permission-boundary failure. When the new binding passes
+the whole serving gate, replacement succeeded and no rollback is attempted. If
+it does not pass, rollback unloads it before restoring the backup and performs
+the same complete read against the previous binding. A result never calls both
+replacement and rollback successful. If neither binding can be verified, the
+operation reports `rollback_failed` with one recovery command and the checkout
+where that command must run.
+
+The swap is the point at which the original contents stop existing, so it
+happens only once the old job is established as unloaded -- a `bootout` that
+succeeded, or the same positively-confirmed absence `remove` requires. Writing
+over a job launchd still holds would fail the bootstrap anyway (launchd will not
+accept a label its domain already holds), so that case refuses as
+`unload_failed`. A write that fails outright leaves the previous definition on
+disk untouched; recovery loads and verifies that definition again.
 
 The backup that rollback restores is read before any of that -- before a log
 directory is created, before the old job is booted out, before a byte is
@@ -201,6 +215,22 @@ whatever the installing shell happened to have. If the package cannot be located
 on a canonical path, the request is refused before anything is applied. A
 console-script definition carries no `PYTHONPATH` at all.
 
+### Recording and the supported argument surface
+
+`board serve` is an interactive process and does not record events unless
+`--record-events` is passed. A managed `board service` is intended to preserve
+local history while it stays open, so `service render`, `service install` and
+`service restart` enable recording by default; pass `--no-record-events` to all
+three when the managed Board should remain read-only. That choice is part of the
+rendered argument vector, so changing it requires `restart --replace`.
+
+The service commands accept the durable binding options they can reproduce:
+`--repo`, `--repo-path`, `--host`, `--port`, and the recording choice. Options
+that tune one interactive `board serve` process, such as custom store paths,
+limits, refresh intervals, retention, or `--open`, are not `board service`
+arguments. Review the exact supported service argv with `service render`; do not
+copy an arbitrary `board serve` command after the `service` subcommand.
+
 ### Logs
 
 The definition sends both output streams to `<repo>/.code-mower/board/logs`.
diff --git a/src/code_mower/board_service.py b/src/code_mower/board_service.py
index 0c6b5838..b6ec46f1 100644
--- a/src/code_mower/board_service.py
+++ b/src/code_mower/board_service.py
@@ -1898,6 +1898,76 @@ def _ownership_conflict(
     )
 
 
+def _recovery_instruction(spec: ServiceSpec, *, show_local_paths: bool) -> dict[str, Any]:
+    """One copyable command that converges an unresolved binding on `spec`.
+
+    The command uses ``--repo-path .`` so the payload can remain publishable.
+    ``cwd`` names the exact checkout locally (or carries the normal redaction),
+    and is part of the instruction rather than being interpolated into shell
+    text where a private path would leak through JSON or text output.
+    """
+
+    command = (
+        "code-mower board service restart "
+        f"--repo {spec.repo} --repo-path . --host {spec.host} --port {spec.port} --replace"
+    )
+    if "--record-events" not in spec.arguments:
+        command += " --no-record-events"
+    return {
+        "cwd": redact_path(str(spec.repo_path), show_local_paths=show_local_paths),
+        "cwd_redacted": not show_local_paths,
+        "command": command,
+    }
+
+
+def _final_binding_read(
+    spec: ServiceSpec,
+    *,
+    provider: Any,
+    command_runner: lane_status.CommandRunner,
+    identity_probe: Callable[[str, int], Mapping[str, Any]] | None,
+    show_local_paths: bool,
+    expected_digest: str | None = None,
+    expected_arguments: Sequence[str] | None = None,
+) -> dict[str, Any]:
+    """Re-read every fact that can make an applied service truthful.
+
+    This deliberately does not reuse a delayed-health result. A terminal
+    result is based on one fresh read of the installed bytes, launchd's current
+    pid and argv, the pid's cwd and supervisor, every listener on the port, and
+    the Board identity/version response.
+    """
+
+    return validate_binding(
+        spec,
+        provider=provider,
+        command_runner=command_runner,
+        identity_probe=identity_probe,
+        show_local_paths=show_local_paths,
+        expected_digest=expected_digest,
+        expected_arguments=expected_arguments,
+    )
+
+
+def _reconciliation(
+    target: str,
+    binding: Mapping[str, Any],
+    *,
+    spec: ServiceSpec,
+    show_local_paths: bool,
+) -> dict[str, Any]:
+    verified = binding.get("status") == "pass"
+    result: dict[str, Any] = {
+        "state": target if verified else "unresolved",
+        "verified": verified,
+        "target": target,
+        "binding": dict(binding),
+    }
+    if not verified:
+        result["recovery"] = _recovery_instruction(spec, show_local_paths=show_local_paths)
+    return result
+
+
 def _apply(
     spec: ServiceSpec,
     *,
@@ -1974,6 +2044,32 @@ def _apply(
         if not unloaded:
             still_loaded, _pid = _job_load_state(provider, spec.label)
             if still_loaded:
+                if previous.readable:
+                    previous_spec = spec_from_service(previous)
+                    previous_binding = _final_binding_read(
+                        previous_spec,
+                        provider=provider,
+                        command_runner=command_runner,
+                        identity_probe=identity_probe,
+                        show_local_paths=show_local_paths,
+                        expected_digest=previous.digest,
+                        expected_arguments=previous.arguments,
+                    )
+                    reconciled = _reconciliation(
+                        "previous",
+                        previous_binding,
+                        spec=spec,
+                        show_local_paths=show_local_paths,
+                    )
+                else:
+                    reconciled = {
+                        "state": "unresolved",
+                        "verified": False,
+                        "target": "previous",
+                        "recovery": _recovery_instruction(
+                            spec, show_local_paths=show_local_paths
+                        ),
+                    }
                 return _operation_payload(
                     "unload_failed",
                     (
@@ -1986,6 +2082,7 @@ def _apply(
                     expected,
                     show_local_paths=show_local_paths,
                     known_paths=_provider_known_paths(provider, spec.label),
+                    reconciliation=reconciled,
                 )
     try:
         provider.write_definition(spec.label, rendered)
@@ -1998,6 +2095,55 @@ def _apply(
         # The previous service was already booted out, so failing here without
         # reloading it would leave a Board that was working stopped.
         restore = _restore_unwritten(provider, spec)
+        if not previous.readable:
+            reconciled = {
+                "state": "unresolved",
+                "verified": False,
+                "target": "previous",
+                "recovery": _recovery_instruction(spec, show_local_paths=show_local_paths),
+            }
+            return _operation_payload(
+                "rollback_failed",
+                f"{detail}; the unreadable previous binding could not be verified after reload",
+                spec,
+                expected,
+                show_local_paths=show_local_paths,
+                known_paths=_provider_known_paths(provider, spec.label),
+                rollback=restore,
+                reconciliation=reconciled,
+            )
+        previous_spec = spec_from_service(previous)
+        restore_health = delayed_health(
+            previous_spec,
+            provider=provider,
+            command_runner=command_runner,
+            identity_probe=identity_probe,
+            settle_seconds=settle_seconds,
+            refresh_seconds=refresh_seconds,
+            timeout_seconds=timeout_seconds,
+            show_local_paths=show_local_paths,
+            sleeper=sleeper,
+            clock=clock,
+        )
+        previous_binding = _final_binding_read(
+            previous_spec,
+            provider=provider,
+            command_runner=command_runner,
+            identity_probe=identity_probe,
+            show_local_paths=show_local_paths,
+            expected_digest=previous.digest,
+            expected_arguments=previous.arguments,
+        )
+        reconciled = _reconciliation(
+            "previous", previous_binding, spec=spec, show_local_paths=show_local_paths
+        )
+        restore["ok"] = bool(reconciled["verified"])
+        restore["restored"] = bool(reconciled["verified"])
+        restore["delayed_health"] = restore_health
+        if reconciled["verified"]:
+            restore["detail"] = (
+                "reloaded and independently verified the untouched previous definition"
+            )
         return _operation_payload(
             "apply_failed" if restore["ok"] else "rollback_failed",
             detail if restore["ok"] else f"{detail}; rollback also failed: {restore['detail']}",
@@ -2006,12 +2152,108 @@ def _apply(
             show_local_paths=show_local_paths,
             known_paths=_provider_known_paths(provider, spec.label),
             rollback=restore,
+            reconciliation=reconciled,
         )
     ok, detail = provider.bootstrap(spec.label)
     if not ok:
+        # `launchctl bootstrap` can register and start the job, then return a
+        # failure. The provider result is therefore not the outcome. If the
+        # replacement clears the complete serving gate, keep it instead of
+        # beginning a rollback that can create a definition/process split.
+        load_state, _pid = provider.runtime_state(spec.label)
+        if load_state == JOB_LOADED:
+            applied_health = delayed_health(
+                spec,
+                provider=provider,
+                command_runner=command_runner,
+                identity_probe=identity_probe,
+                settle_seconds=settle_seconds,
+                refresh_seconds=refresh_seconds,
+                timeout_seconds=timeout_seconds,
+                show_local_paths=show_local_paths,
+                sleeper=sleeper,
+                clock=clock,
+            )
+            applied_binding = _final_binding_read(
+                spec,
+                provider=provider,
+                command_runner=command_runner,
+                identity_probe=identity_probe,
+                show_local_paths=show_local_paths,
+                expected_digest=expected,
+                expected_arguments=spec.arguments,
+            )
+            applied = _reconciliation(
+                "new", applied_binding, spec=spec, show_local_paths=show_local_paths
+            )
+            if applied["verified"]:
+                return _operation_payload(
+                    succeeded_status,
+                    (
+                        f"{succeeded_message}; launchd reported bootstrap failure, but the "
+                        "replacement definition and serving process were independently verified"
+                    ),
+                    spec,
+                    expected,
+                    delayed=applied_health if applied_health["state"] == "pass" else None,
+                    show_local_paths=show_local_paths,
+                    reconciliation=applied,
+                    provider_detail=detail,
+                )
         rollback = _rollback(provider, spec, previous_text)
-        status = "rollback_failed" if not rollback["ok"] else "apply_failed"
-        message = detail if rollback["ok"] else f"{detail}; rollback also failed: {rollback['detail']}"
+        if previous is not None and previous.readable:
+            previous_spec = spec_from_service(previous)
+            rollback_health = delayed_health(
+                previous_spec,
+                provider=provider,
+                command_runner=command_runner,
+                identity_probe=identity_probe,
+                settle_seconds=settle_seconds,
+                refresh_seconds=refresh_seconds,
+                timeout_seconds=timeout_seconds,
+                show_local_paths=show_local_paths,
+                sleeper=sleeper,
+                clock=clock,
+            )
+            previous_binding = _final_binding_read(
+                previous_spec,
+                provider=provider,
+                command_runner=command_runner,
+                identity_probe=identity_probe,
+                show_local_paths=show_local_paths,
+                expected_digest=previous.digest,
+                expected_arguments=previous.arguments,
+            )
+            restored = _reconciliation(
+                "previous", previous_binding, spec=spec, show_local_paths=show_local_paths
+            )
+            # A restoring bootstrap can report failure after loading the old
+            # job. State wins over that return code too.
+            rollback["ok"] = bool(restored["verified"])
+            rollback["restored"] = bool(restored["verified"])
+            rollback["delayed_health"] = rollback_health
+            rollback["detail"] = (
+                "restored and independently verified the previous definition and serving process"
+                if restored["verified"]
+                else rollback.get("detail") or "the previous binding could not be verified"
+            )
+        else:
+            absent = provider.read_service(spec.label) is None
+            restored = {
+                "state": "absent" if absent and rollback.get("ok") else "unresolved",
+                "verified": bool(absent and rollback.get("ok")),
+                "target": "absent",
+            }
+            if not restored["verified"]:
+                restored["recovery"] = _recovery_instruction(
+                    spec, show_local_paths=show_local_paths
+                )
+        status = "apply_failed" if restored["verified"] else "rollback_failed"
+        message = (
+            f"{detail}; the previous service was restored and verified"
+            if restored["verified"]
+            else f"{detail}; rollback also failed to reach a verified state"
+        )
         return _operation_payload(
             status,
             message,
@@ -2020,6 +2262,7 @@ def _apply(
             show_local_paths=show_local_paths,
             known_paths=_provider_known_paths(provider, spec.label),
             rollback=rollback,
+            reconciliation=restored,
         )
 
     health = delayed_health(
@@ -2035,6 +2278,12 @@ def _apply(
         clock=clock,
     )
     if health["state"] != "pass":
+        reconciled = _reconciliation(
+            "new",
+            health.get("binding") if isinstance(health.get("binding"), Mapping) else {},
+            spec=spec,
+            show_local_paths=show_local_paths,
+        )
         return _operation_payload(
             "delayed_health_failed",
             "the service applied but its binding did not validate within the delayed health window",
@@ -2042,6 +2291,27 @@ def _apply(
             expected,
             delayed=health,
             show_local_paths=show_local_paths,
+            reconciliation=reconciled,
+        )
+    final_binding = _final_binding_read(
+        spec,
+        provider=provider,
+        command_runner=command_runner,
+        identity_probe=identity_probe,
+        show_local_paths=show_local_paths,
+        expected_digest=expected,
+        expected_arguments=spec.arguments,
+    )
+    reconciled = _reconciliation("new", final_binding, spec=spec, show_local_paths=show_local_paths)
+    if not reconciled["verified"]:
+        return _operation_payload(
+            "delayed_health_failed",
+            "the service passed delayed health but its final binding re-read did not validate",
+            spec,
+            expected,
+            delayed=health,
+            show_local_paths=show_local_paths,
+            reconciliation=reconciled,
         )
     return _operation_payload(
         succeeded_status,
@@ -2050,6 +2320,7 @@ def _apply(
         expected,
         delayed=health,
         show_local_paths=show_local_paths,
+        reconciliation=reconciled,
     )
 
 
@@ -2291,7 +2562,30 @@ def restart_service(
             if health["state"] == "pass"
             else "the service restarted but its binding did not validate within the delayed health window"
         )
-        return _operation_payload(status, message, spec, expected, delayed=health, show_local_paths=show_local_paths)
+        final_binding = _final_binding_read(
+            spec,
+            provider=provider,
+            command_runner=command_runner,
+            identity_probe=identity_probe,
+            show_local_paths=show_local_paths,
+            expected_digest=expected,
+            expected_arguments=spec.arguments,
+        )
+        reconciled = _reconciliation(
+            "new", final_binding, spec=spec, show_local_paths=show_local_paths
+        )
+        if status == "restarted" and not reconciled["verified"]:
+            status = "delayed_health_failed"
+            message = "the service passed delayed health but its final binding re-read did not validate"
+        return _operation_payload(
+            status,
+            message,
+            spec,
+            expected,
+            delayed=health,
+            show_local_paths=show_local_paths,
+            reconciliation=reconciled,
+        )
     return _apply(
         spec,
         provider=provider,
@@ -2735,6 +3029,24 @@ def render_operation_text(payload: Mapping[str, Any]) -> str:
     rollback = payload.get("rollback") if isinstance(payload.get("rollback"), Mapping) else {}
     if rollback:
         lines.append(f"Rollback: {'ok' if rollback.get('ok') else 'failed'} - {rollback.get('detail')}")
+    reconciliation = (
+        payload.get("reconciliation")
+        if isinstance(payload.get("reconciliation"), Mapping)
+        else {}
+    )
+    if reconciliation:
+        lines.append(
+            f"Reconciliation: {reconciliation.get('state') or 'unresolved'} "
+            f"({'verified' if reconciliation.get('verified') else 'not verified'})"
+        )
+        recovery = (
+            reconciliation.get("recovery")
+            if isinstance(reconciliation.get("recovery"), Mapping)
+            else {}
+        )
+        if recovery:
+            lines.append(f"Recovery cwd: {recovery.get('cwd')}")
+            lines.append(f"Recovery command: {recovery.get('command')}")
     return "\n".join(line for line in lines if line) + "\n"
 
 
diff --git a/tests/test_board_service.py b/tests/test_board_service.py
index d11f291e..fb498a8d 100644
--- a/tests/test_board_service.py
+++ b/tests/test_board_service.py
@@ -943,7 +943,15 @@ def delete_definition(self, label: str) -> bool:
         self.assertFalse(payload["rollback"]["deleted"])
         self.assertIn("next login", payload["rollback"]["detail"])
         self.assertTrue((self.root / "ai.codemower.board.5332.plist").exists())
+        recovery = payload["reconciliation"]["recovery"]
+        self.assertEqual(recovery["cwd"], lane_status.LOCAL_PATH_REDACTION)
+        self.assertEqual(
+            recovery["command"],
+            "code-mower board service restart --repo codemower-ai/code-mower "
+            "--repo-path . --host 127.0.0.1 --port 5332 --replace",
+        )
         self.assertIn("Rollback: failed", board_service.render_operation_text(payload))
+        self.assertIn("Recovery command:", board_service.render_operation_text(payload))
 
     def test_a_failed_apply_reports_why_without_publishing_the_definition_path(self) -> None:
         # The failure reason is the actionable part and stays; the definition
@@ -1717,12 +1725,11 @@ def test_a_first_install_needs_no_backup_and_still_applies(self) -> None:
         self.assertIn("write_definition", mutations)
         self.assertIn("ai.codemower.board.5332", self.host.loaded)
 
-    def test_a_partially_registered_replacement_is_unloaded_before_restoring(self) -> None:
-        # A bootstrap can register the job and *then* give up waiting for it.
-        # Restoring the previous definition on top of a replacement launchd
-        # still holds would leave launchd supervising the replacement while the
-        # definition on disk describes the service it replaced, and the
-        # restoring bootstrap would fail because the label is already loaded.
+    def test_a_bootstrap_failure_that_loaded_the_replacement_is_reconciled_as_new(self) -> None:
+        # A bootstrap can register the job and *then* report a permission or
+        # timeout failure. The provider result is ambiguous; the installed
+        # bytes, pid, argv, port owner, repository identity and served versions
+        # prove that the replacement actually won.
         self.install(self.spec())
         before = (self.root / "ai.codemower.board.5332.plist").read_text(encoding="utf-8")
         drifted = self.spec(repo="codemower-ai/private-repo", repo_path=self.other_checkout, port=5332)
@@ -1734,7 +1741,7 @@ def bootstrap(self, label: str) -> tuple[bool, str]:
                 ok, detail = super().bootstrap(label)
                 if not self.gave_up:
                     self.gave_up = True
-                    return False, "timed out waiting for the job to answer"
+                    return False, "Bootstrap failed: 5: Operation not permitted"
                 return ok, detail
 
         payload = self._restart_with(
@@ -1745,20 +1752,109 @@ def bootstrap(self, label: str) -> tuple[bool, str]:
             replace=True,
         )
 
+        self.assertEqual(payload["status"], "restarted")
+        self.assertEqual(payload["reconciliation"]["state"], "new")
+        self.assertTrue(payload["reconciliation"]["verified"])
+        self.assertNotIn("rollback", payload)
+        self.assertNotEqual(
+            (self.root / "ai.codemower.board.5332.plist").read_text(encoding="utf-8"), before
+        )
+        self.assertEqual(self.host.identities[5332]["repo"], "codemower-ai/private-repo")
+        self.assertEqual(payload["reconciliation"]["binding"]["failing_checks"], [])
+        self.assertEqual(
+            tuple(check["id"] for check in payload["reconciliation"]["binding"]["checks"]),
+            board_service.BINDING_CHECK_IDS,
+        )
+
+    def test_a_restoring_bootstrap_failure_is_reconciled_from_the_old_binding(self) -> None:
+        # The replacement never loads, while the restoring bootstrap loads the
+        # previous job and then returns a failure. The final state read makes
+        # this one verified rollback, without claiming replacement success.
+        original = self.spec()
+        self.install(original)
+        before = (self.root / f"{original.label}.plist").read_text(encoding="utf-8")
+        drifted = self.spec(
+            repo="codemower-ai/private-repo", repo_path=self.other_checkout, port=5332
+        )
+
+        class RestoreLoadsThenReportsFailure(board_service.LaunchdProvider):
+            bootstraps = 0
+
+            def bootstrap(self, label: str) -> tuple[bool, str]:
+                self.bootstraps += 1
+                if self.bootstraps == 1:
+                    return False, "Bootstrap failed: 5: Operation not permitted"
+                super().bootstrap(label)
+                return False, "Bootstrap failed after registering the restored job"
+
+        payload = self._restart_with(
+            RestoreLoadsThenReportsFailure(
+                command_runner=self.host.run,
+                root=self.root,
+                uid=self.host.uid,
+                platform="darwin",
+            ),
+            drifted,
+            replace=True,
+        )
+
         self.assertEqual(payload["status"], "apply_failed")
+        self.assertEqual(payload["reconciliation"]["state"], "previous")
         self.assertTrue(payload["rollback"]["ok"])
         self.assertTrue(payload["rollback"]["restored"])
-        # On disk and in launchd, what is left is the previous service -- not a
-        # replacement running against a definition that no longer describes it.
-        self.assertEqual((self.root / "ai.codemower.board.5332.plist").read_text(encoding="utf-8"), before)
-        self.assertIn("ai.codemower.board.5332", self.host.loaded)
+        self.assertEqual(
+            (self.root / f"{original.label}.plist").read_text(encoding="utf-8"), before
+        )
         self.assertEqual(self.host.identities[5332]["repo"], "codemower-ai/code-mower")
 
-    def test_a_replacement_that_cannot_be_unloaded_leaves_the_definition_alone(self) -> None:
-        # The other half: if the replacement cannot be confirmed unloaded, the
-        # previous definition is not written under it. Overwriting the
-        # definition of a job launchd still supervises would leave neither the
-        # replacement nor the previous service described by what is on disk.
+    def test_an_unreconciled_partial_replacement_has_one_exact_recovery_command(self) -> None:
+        self.install(self.spec())
+        drifted = self.spec(
+            repo="codemower-ai/private-repo", repo_path=self.other_checkout, port=5332
+        )
+        host = self.host
+
+        class LoadsWrongIdentityAndCannotRollBack(board_service.LaunchdProvider):
+            booted_out = False
+
+            def bootstrap(self, label: str) -> tuple[bool, str]:
+                super().bootstrap(label)
+                host.identities[5332]["repo"] = "codemower-ai/code-mower"
+                return False, "Bootstrap failed: 5: Operation not permitted"
+
+            def bootout(self, label: str) -> tuple[bool, str]:
+                if self.booted_out:
+                    return False, "Bootout failed: 125: Operation not permitted"
+                self.booted_out = True
+                return super().bootout(label)
+
+        payload = self._restart_with(
+            LoadsWrongIdentityAndCannotRollBack(
+                command_runner=self.host.run,
+                root=self.root,
+                uid=self.host.uid,
+                platform="darwin",
+            ),
+            drifted,
+            replace=True,
+        )
+
+        self.assertEqual(payload["status"], "rollback_failed")
+        self.assertEqual(payload["reconciliation"]["state"], "unresolved")
+        recovery = payload["reconciliation"]["recovery"]
+        self.assertEqual(
+            recovery["command"],
+            "code-mower board service restart --repo codemower-ai/private-repo "
+            "--repo-path . --host 127.0.0.1 --port 5332 --replace",
+        )
+        rendered = board_service.render_operation_text(payload)
+        self.assertEqual(rendered.count("Recovery command:"), 1)
+        self.assertNotIn("Reconciliation: new (verified)", rendered)
+
+    def test_a_verified_replacement_is_kept_without_attempting_a_failing_rollback(self) -> None:
+        # Reconciliation happens before rollback. Even a provider that would
+        # refuse the rollback bootout cannot turn a verified new binding into a
+        # split old-definition/new-process state.
         self.install(self.spec())
         drifted = self.spec(repo="codemower-ai/private-repo", repo_path=self.other_checkout, port=5332)
         replacement_text = board_service.render_definition(drifted)
@@ -1786,24 +1882,19 @@ def bootout(self, label: str) -> tuple[bool, str]:
             replace=True,
         )
 
-        self.assertEqual(payload["status"], "rollback_failed")
-        self.assertFalse(payload["rollback"]["ok"])
-        self.assertFalse(payload["rollback"]["restored"])
-        self.assertIn("could not be unloaded", payload["rollback"]["detail"])
+        self.assertEqual(payload["status"], "restarted")
+        self.assertTrue(payload["reconciliation"]["verified"])
+        self.assertNotIn("rollback", payload)
         self.assertEqual(
             (self.root / "ai.codemower.board.5332.plist").read_text(encoding="utf-8"),
             replacement_text,
         )
-        self.assertIn("Rollback: failed", board_service.render_operation_text(payload))
+        self.assertIn("Reconciliation: new (verified)", board_service.render_operation_text(payload))
 
-    def test_a_rollback_that_cannot_unload_the_job_keeps_its_definition(self) -> None:
-        # launchd registered the job and then bootstrap gave up waiting for it,
-        # so the apply failed with a job still supervised -- and the rollback's
-        # bootout failed too. The definition is the only handle `board service
-        # status`, `remove` and the `board stop` keepalive guard have on that
-        # job, because all three discover services by scanning definitions.
-        # Deleting it would strand a running, self-restarting Board outside the
-        # inventory entirely, so it is kept until the job is confirmed gone.
+    def test_a_first_install_bootstrap_failure_is_truthful_when_the_job_is_serving(self) -> None:
+        # launchd registered the job and then bootstrap gave up waiting for it.
+        # A complete binding read proves the install succeeded before the
+        # rollback path can mistake the provider error for the machine state.
         class StrandsTheJob(board_service.LaunchdProvider):
             def bootstrap(self, label: str) -> tuple[bool, str]:
                 super().bootstrap(label)
@@ -1825,10 +1916,10 @@ def bootout(self, label: str) -> tuple[bool, str]:
             sleeper=self.sleeper,
         )
 
-        self.assertEqual(payload["status"], "rollback_failed")
-        self.assertFalse(payload["rollback"]["ok"])
-        self.assertFalse(payload["rollback"]["deleted"])
-        self.assertIn("still-loaded job stays discoverable", payload["rollback"]["detail"])
+        self.assertEqual(payload["status"], "installed")
+        self.assertEqual(payload["reconciliation"]["state"], "new")
+        self.assertTrue(payload["reconciliation"]["verified"])
+        self.assertNotIn("rollback", payload)
         self.assertIn("ai.codemower.board.5332", self.host.loaded)
         self.assertTrue((self.root / "ai.codemower.board.5332.plist").exists())
         # The point of keeping it: the stranded job is still manageable.
@@ -1836,7 +1927,9 @@ def bootout(self, label: str) -> tuple[bool, str]:
             [item.label for item in self.host.provider().list_services()],
             ["ai.codemower.board.5332"],
         )
-        self.assertIn("Rollback: failed", board_service.render_operation_text(payload))
+        rendered = board_service.render_operation_text(payload)
+        self.assertIn("Reconciliation: new (verified)", rendered)
+        self.assertNotIn("Rollback:", rendered)
 
     def test_an_identity_probe_brackets_an_ipv6_address(self) -> None:
         # `http://::1:5332/api/identity` names something other than the service

From 89178ca9fe50bab5ac3d1bb79bc863ea1a9d2c15 Mon Sep 17 00:00:00 2001
From: Jeff Huber <jeffhuber@users.noreply.github.com>
Date: Mon, 21 Sep 2026 00:43:16 -0700
Subject: [PATCH 2/8] Reconcile Board health failures and prior versions

---
 docs/board-service-lifecycle.md |  23 ++-
 src/code_mower/board_service.py | 355 ++++++++++++++++++++++++--------
 tests/test_board_service.py     | 102 +++++++++
 3 files changed, 382 insertions(+), 98 deletions(-)

diff --git a/docs/board-service-lifecycle.md b/docs/board-service-lifecycle.md
index eccdf13f..e6bb5d1f 100644
--- a/docs/board-service-lifecycle.md
+++ b/docs/board-service-lifecycle.md
@@ -62,14 +62,13 @@ comes back with stale arguments, fails the gate instead of passing on a single
 early probe. The payload reports `delayed_health` with its state, the settle and
 refresh windows, and how many refreshes it took.
 
-When that window closes without a validating binding, the operation reports
-`delayed_health_failed`: the service applied but its binding never validated.
-This is the one failure that is *not* in the fail-closed table below, and the
-difference matters. The refusals below are decided before anything is applied
-and leave no local state behind; `delayed_health_failed` is decided after
-launchd already holds the job, so the definition stays installed and
-`board service status` keeps reporting the failing binding until you repair or
-remove it.
+When that window closes without a validating binding, an in-place restart or an
+unchanged install reports `delayed_health_failed`, because no definition was
+replaced. A new apply is different: a fresh install is removed again, while a
+replacement restores and verifies the previous binding. Those operations report
+`apply_failed` when the pre-operation state is verified and `rollback_failed`
+with a recovery command when neither state can be verified. The payload keeps
+the failed `delayed_health` evidence alongside the reconciliation result.
 
 ## Fail-closed refusals
 
@@ -163,6 +162,14 @@ replacement and rollback successful. If neither binding can be verified, the
 operation reports `rollback_failed` with one recovery command and the checkout
 where that command must run.
 
+Rollback version checks use the identity/version evidence captured from the old
+Board before it is stopped. A byte-for-byte restored older Board therefore
+validates as the previous state even when the command performing the replacement
+comes from a newer installation. If the prior definition was unreadable,
+`--replace` knowingly discarded bytes that could not be backed up; removing a
+failed replacement is then reported as unresolved cleanup, never as a restored
+previous service.
+
 The swap is the point at which the original contents stop existing, so it
 happens only once the old job is established as unloaded -- a `bootout` that
 succeeded, or the same positively-confirmed absence `remove` requires. Writing
diff --git a/src/code_mower/board_service.py b/src/code_mower/board_service.py
index b6ec46f1..4b214c1e 100644
--- a/src/code_mower/board_service.py
+++ b/src/code_mower/board_service.py
@@ -1332,6 +1332,7 @@ def validate_binding(
     show_local_paths: bool = False,
     expected_digest: str | None = None,
     expected_arguments: Sequence[str] | None = None,
+    expected_version: Mapping[str, Any] | None = None,
 ) -> dict[str, Any]:
     """The serving gate.
 
@@ -1519,16 +1520,27 @@ def validate_binding(
     # distribution metadata on either side, so the Board reports an empty
     # installed version and the honest expectation is that it stays empty --
     # demanding the imported version there would fail a healthy service forever.
-    expected_installed = installed_distribution_version()
     reported_installed = _text(version.get("installed_version"))
-    if expected_installed is None:
+    if expected_version is not None:
+        expected_installed = _text(expected_version.get("installed_version"))
+        installed_ok = reported_installed == expected_installed
+        installed_message = (
+            "the served installed version matches the pre-replacement binding"
+            if installed_ok
+            else "the served installed version does not match the pre-replacement binding"
+        )
+        source_checkout = expected_installed == ""
+    else:
+        expected_installed = installed_distribution_version()
+        source_checkout = expected_installed is None
+    if expected_version is None and expected_installed is None:
         installed_ok = reported_installed == ""
         installed_message = (
             "neither side has distribution metadata, as expected for a source checkout"
             if installed_ok
             else "the served Board reports an installed distribution this checkout does not have"
         )
-    else:
+    elif expected_version is None:
         installed_ok = reported_installed == expected_installed
         installed_message = (
             "the served installed version matches this installation"
@@ -1542,7 +1554,7 @@ def validate_binding(
             installed_message,
             installed_version=reported_installed,
             expected_installed_version=expected_installed or "",
-            source_checkout=expected_installed is None,
+            source_checkout=source_checkout,
         )
     )
     # The serving version is the code the process is actually running, and both
@@ -1551,16 +1563,33 @@ def validate_binding(
     # that means the same thing in both modes, and `restart_recommended` still
     # carries the Board's own verdict that it is running behind its install.
     reported_serving = _text(version.get("serving_version"))
-    serving_ok = reported_serving == CODE_MOWER_VERSION and not version.get("restart_recommended")
+    if expected_version is not None:
+        expected_serving = _text(expected_version.get("serving_version"))
+        expected_restart = bool(expected_version.get("restart_recommended"))
+        serving_ok = (
+            reported_serving == expected_serving
+            and bool(version.get("restart_recommended")) == expected_restart
+        )
+        serving_message = (
+            "the serving version matches the pre-replacement binding"
+            if serving_ok
+            else "the serving version does not match the pre-replacement binding"
+        )
+    else:
+        expected_serving = CODE_MOWER_VERSION
+        serving_ok = reported_serving == expected_serving and not version.get("restart_recommended")
+        serving_message = (
+            "the serving version matches this Code Mower"
+            if serving_ok
+            else "the serving version is stale against this Code Mower"
+        )
     checks.append(
         _check(
             "binding.serving_version",
             "pass" if serving_ok else "fail",
-            "the serving version matches this Code Mower"
-            if serving_ok
-            else "the serving version is stale against this Code Mower",
+            serving_message,
             serving_version=reported_serving,
-            expected_serving_version=CODE_MOWER_VERSION,
+            expected_serving_version=expected_serving,
         )
     )
     return _binding_payload(spec, checks, service, show_local_paths=show_local_paths, expected_digest=expected)
@@ -1620,6 +1649,9 @@ def delayed_health(
     refresh_seconds: float = DEFAULT_REFRESH_SECONDS,
     timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
     show_local_paths: bool = False,
+    expected_digest: str | None = None,
+    expected_arguments: Sequence[str] | None = None,
+    expected_version: Mapping[str, Any] | None = None,
     sleeper: Sleeper = time.sleep,
     clock: Clock = time.monotonic,
 ) -> dict[str, Any]:
@@ -1645,6 +1677,9 @@ def delayed_health(
             command_runner=command_runner,
             identity_probe=identity_probe,
             show_local_paths=show_local_paths,
+            expected_digest=expected_digest,
+            expected_arguments=expected_arguments,
+            expected_version=expected_version,
         )
         if binding.get("status") == "pass" or clock() >= deadline:
             break
@@ -1929,6 +1964,7 @@ def _final_binding_read(
     show_local_paths: bool,
     expected_digest: str | None = None,
     expected_arguments: Sequence[str] | None = None,
+    expected_version: Mapping[str, Any] | None = None,
 ) -> dict[str, Any]:
     """Re-read every fact that can make an applied service truthful.
 
@@ -1946,6 +1982,7 @@ def _final_binding_read(
         show_local_paths=show_local_paths,
         expected_digest=expected_digest,
         expected_arguments=expected_arguments,
+        expected_version=expected_version,
     )
 
 
@@ -1968,6 +2005,158 @@ def _reconciliation(
     return result
 
 
+def _previous_version_evidence(
+    previous: ManagedService | None,
+    identity_probe: Callable[[str, int], Mapping[str, Any]] | None,
+) -> dict[str, Any] | None:
+    """Capture the old Board's version contract before its process is stopped."""
+
+    if previous is None or not previous.readable or not previous.port:
+        return None
+    probe = identity_probe or (lambda host, port: probe_identity(host, port))
+    identity = probe(previous.host, previous.port)
+    board = identity.get("board") if isinstance(identity.get("board"), Mapping) else {}
+    version = board.get("version") if isinstance(board.get("version"), Mapping) else {}
+    if not _text(version.get("serving_version")):
+        return None
+    return {
+        "installed_version": _text(version.get("installed_version")),
+        "serving_version": _text(version.get("serving_version")),
+        "restart_recommended": bool(version.get("restart_recommended")),
+    }
+
+
+def _rollback_and_reconcile(
+    spec: ServiceSpec,
+    *,
+    provider: Any,
+    previous: ManagedService | None,
+    previous_text: str,
+    previous_version: Mapping[str, Any] | None,
+    expected: str,
+    failure_detail: str,
+    command_runner: lane_status.CommandRunner,
+    identity_probe: Callable[[str, int], Mapping[str, Any]] | None,
+    settle_seconds: float,
+    refresh_seconds: float,
+    timeout_seconds: float,
+    show_local_paths: bool,
+    sleeper: Sleeper,
+    clock: Clock,
+    attempted_health: Mapping[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Roll a failed replacement back, then prove the exact state left behind."""
+
+    rollback = _rollback(provider, spec, previous_text)
+    if previous is not None and previous.readable:
+        previous_spec = spec_from_service(previous)
+        rollback_health = delayed_health(
+            previous_spec,
+            provider=provider,
+            command_runner=command_runner,
+            identity_probe=identity_probe,
+            settle_seconds=settle_seconds,
+            refresh_seconds=refresh_seconds,
+            timeout_seconds=timeout_seconds,
+            show_local_paths=show_local_paths,
+            expected_digest=previous.digest,
+            expected_arguments=previous.arguments,
+            expected_version=previous_version,
+            sleeper=sleeper,
+            clock=clock,
+        )
+        previous_binding = _final_binding_read(
+            previous_spec,
+            provider=provider,
+            command_runner=command_runner,
+            identity_probe=identity_probe,
+            show_local_paths=show_local_paths,
+            expected_digest=previous.digest,
+            expected_arguments=previous.arguments,
+            expected_version=previous_version,
+        )
+        restored = _reconciliation(
+            "previous", previous_binding, spec=spec, show_local_paths=show_local_paths
+        )
+        if previous_version is None:
+            restored.update(
+                state="unresolved",
+                verified=False,
+                version_evidence_captured=False,
+                recovery=_recovery_instruction(spec, show_local_paths=show_local_paths),
+            )
+        else:
+            restored["version_evidence_captured"] = True
+        rollback["ok"] = bool(restored["verified"])
+        rollback["restored"] = bool(restored["verified"])
+        rollback["delayed_health"] = rollback_health
+        rollback["detail"] = (
+            "restored and independently verified the previous definition and serving process"
+            if restored["verified"]
+            else rollback.get("detail") or "the previous binding could not be verified"
+        )
+        status = "apply_failed" if restored["verified"] else "rollback_failed"
+        message = (
+            f"{failure_detail}; the previous service was restored and verified"
+            if restored["verified"]
+            else f"{failure_detail}; rollback also failed to reach a verified state"
+        )
+    elif previous is not None:
+        # `--replace` explicitly allowed the unreadable definition to be
+        # discarded, but it never made those bytes recoverable. Deleting the
+        # failed replacement is cleanup, not restoration of the prior state.
+        rollback["ok"] = False
+        rollback["restored"] = False
+        rollback["detail"] = (
+            "the failed replacement was removed, but the unreadable prior definition had no "
+            "recoverable backup"
+        )
+        restored = {
+            "state": "unresolved",
+            "verified": False,
+            "target": "previous",
+            "recovery": _recovery_instruction(spec, show_local_paths=show_local_paths),
+        }
+        status = "rollback_failed"
+        message = (
+            f"{failure_detail}; the unreadable prior definition was discarded by the explicit "
+            "takeover and could not be restored"
+        )
+    else:
+        load_state, _pid = provider.runtime_state(spec.label)
+        absent = (
+            provider.read_service(spec.label) is None
+            and load_state == JOB_ABSENT
+            and bool(rollback.get("ok"))
+        )
+        restored = {
+            "state": "absent" if absent else "unresolved",
+            "verified": absent,
+            "target": "absent",
+        }
+        if not absent:
+            restored["recovery"] = _recovery_instruction(
+                spec, show_local_paths=show_local_paths
+            )
+        status = "apply_failed" if absent else "rollback_failed"
+        message = (
+            f"{failure_detail}; the pre-operation absence was restored and verified"
+            if absent
+            else f"{failure_detail}; rollback also failed to reach a verified state"
+        )
+    return _operation_payload(
+        status,
+        message,
+        spec,
+        expected,
+        delayed=attempted_health,
+        show_local_paths=show_local_paths,
+        known_paths=_provider_known_paths(provider, spec.label),
+        rollback=rollback,
+        reconciliation=restored,
+    )
+
+
 def _apply(
     spec: ServiceSpec,
     *,
@@ -2019,6 +2208,7 @@ def _apply(
     # the two cases where empty is the truth: no definition was installed, or
     # `--replace` is knowingly taking over one that could not be read at all,
     # whose contents were never recoverable and whose refusal said so.
+    previous_version = _previous_version_evidence(previous, identity_probe)
 
     # Before anything is booted out: launchd cannot start a job whose log files
     # it cannot open, and a filesystem failure must not be discovered after the
@@ -2054,6 +2244,7 @@ def _apply(
                         show_local_paths=show_local_paths,
                         expected_digest=previous.digest,
                         expected_arguments=previous.arguments,
+                        expected_version=previous_version,
                     )
                     reconciled = _reconciliation(
                         "previous",
@@ -2122,6 +2313,9 @@ def _apply(
             refresh_seconds=refresh_seconds,
             timeout_seconds=timeout_seconds,
             show_local_paths=show_local_paths,
+            expected_digest=previous.digest,
+            expected_arguments=previous.arguments,
+            expected_version=previous_version,
             sleeper=sleeper,
             clock=clock,
         )
@@ -2133,10 +2327,20 @@ def _apply(
             show_local_paths=show_local_paths,
             expected_digest=previous.digest,
             expected_arguments=previous.arguments,
+            expected_version=previous_version,
         )
         reconciled = _reconciliation(
             "previous", previous_binding, spec=spec, show_local_paths=show_local_paths
         )
+        if previous_version is None:
+            reconciled.update(
+                state="unresolved",
+                verified=False,
+                version_evidence_captured=False,
+                recovery=_recovery_instruction(spec, show_local_paths=show_local_paths),
+            )
+        else:
+            reconciled["version_evidence_captured"] = True
         restore["ok"] = bool(reconciled["verified"])
         restore["restored"] = bool(reconciled["verified"])
         restore["delayed_health"] = restore_health
@@ -2161,6 +2365,7 @@ def _apply(
         # replacement clears the complete serving gate, keep it instead of
         # beginning a rollback that can create a definition/process split.
         load_state, _pid = provider.runtime_state(spec.label)
+        attempted_health: Mapping[str, Any] | None = None
         if load_state == JOB_LOADED:
             applied_health = delayed_health(
                 spec,
@@ -2174,6 +2379,7 @@ def _apply(
                 sleeper=sleeper,
                 clock=clock,
             )
+            attempted_health = applied_health
             applied_binding = _final_binding_read(
                 spec,
                 provider=provider,
@@ -2186,7 +2392,7 @@ def _apply(
             applied = _reconciliation(
                 "new", applied_binding, spec=spec, show_local_paths=show_local_paths
             )
-            if applied["verified"]:
+            if applied_health["state"] == "pass" and applied["verified"]:
                 return _operation_payload(
                     succeeded_status,
                     (
@@ -2195,74 +2401,28 @@ def _apply(
                     ),
                     spec,
                     expected,
-                    delayed=applied_health if applied_health["state"] == "pass" else None,
+                    delayed=applied_health,
                     show_local_paths=show_local_paths,
                     reconciliation=applied,
                     provider_detail=detail,
                 )
-        rollback = _rollback(provider, spec, previous_text)
-        if previous is not None and previous.readable:
-            previous_spec = spec_from_service(previous)
-            rollback_health = delayed_health(
-                previous_spec,
-                provider=provider,
-                command_runner=command_runner,
-                identity_probe=identity_probe,
-                settle_seconds=settle_seconds,
-                refresh_seconds=refresh_seconds,
-                timeout_seconds=timeout_seconds,
-                show_local_paths=show_local_paths,
-                sleeper=sleeper,
-                clock=clock,
-            )
-            previous_binding = _final_binding_read(
-                previous_spec,
-                provider=provider,
-                command_runner=command_runner,
-                identity_probe=identity_probe,
-                show_local_paths=show_local_paths,
-                expected_digest=previous.digest,
-                expected_arguments=previous.arguments,
-            )
-            restored = _reconciliation(
-                "previous", previous_binding, spec=spec, show_local_paths=show_local_paths
-            )
-            # A restoring bootstrap can report failure after loading the old
-            # job. State wins over that return code too.
-            rollback["ok"] = bool(restored["verified"])
-            rollback["restored"] = bool(restored["verified"])
-            rollback["delayed_health"] = rollback_health
-            rollback["detail"] = (
-                "restored and independently verified the previous definition and serving process"
-                if restored["verified"]
-                else rollback.get("detail") or "the previous binding could not be verified"
-            )
-        else:
-            absent = provider.read_service(spec.label) is None
-            restored = {
-                "state": "absent" if absent and rollback.get("ok") else "unresolved",
-                "verified": bool(absent and rollback.get("ok")),
-                "target": "absent",
-            }
-            if not restored["verified"]:
-                restored["recovery"] = _recovery_instruction(
-                    spec, show_local_paths=show_local_paths
-                )
-        status = "apply_failed" if restored["verified"] else "rollback_failed"
-        message = (
-            f"{detail}; the previous service was restored and verified"
-            if restored["verified"]
-            else f"{detail}; rollback also failed to reach a verified state"
-        )
-        return _operation_payload(
-            status,
-            message,
+        return _rollback_and_reconcile(
             spec,
-            expected,
+            provider=provider,
+            previous=previous,
+            previous_text=previous_text,
+            previous_version=previous_version,
+            expected=expected,
+            failure_detail=detail,
+            command_runner=command_runner,
+            identity_probe=identity_probe,
+            settle_seconds=settle_seconds,
+            refresh_seconds=refresh_seconds,
+            timeout_seconds=timeout_seconds,
             show_local_paths=show_local_paths,
-            known_paths=_provider_known_paths(provider, spec.label),
-            rollback=rollback,
-            reconciliation=restored,
+            sleeper=sleeper,
+            clock=clock,
+            attempted_health=attempted_health,
         )
 
     health = delayed_health(
@@ -2278,20 +2438,26 @@ def _apply(
         clock=clock,
     )
     if health["state"] != "pass":
-        reconciled = _reconciliation(
-            "new",
-            health.get("binding") if isinstance(health.get("binding"), Mapping) else {},
-            spec=spec,
-            show_local_paths=show_local_paths,
-        )
-        return _operation_payload(
-            "delayed_health_failed",
-            "the service applied but its binding did not validate within the delayed health window",
+        return _rollback_and_reconcile(
             spec,
-            expected,
-            delayed=health,
+            provider=provider,
+            previous=previous,
+            previous_text=previous_text,
+            previous_version=previous_version,
+            expected=expected,
+            failure_detail=(
+                "the replacement loaded but its binding did not validate within the delayed "
+                "health window"
+            ),
+            command_runner=command_runner,
+            identity_probe=identity_probe,
+            settle_seconds=settle_seconds,
+            refresh_seconds=refresh_seconds,
+            timeout_seconds=timeout_seconds,
             show_local_paths=show_local_paths,
-            reconciliation=reconciled,
+            sleeper=sleeper,
+            clock=clock,
+            attempted_health=health,
         )
     final_binding = _final_binding_read(
         spec,
@@ -2304,14 +2470,23 @@ def _apply(
     )
     reconciled = _reconciliation("new", final_binding, spec=spec, show_local_paths=show_local_paths)
     if not reconciled["verified"]:
-        return _operation_payload(
-            "delayed_health_failed",
-            "the service passed delayed health but its final binding re-read did not validate",
+        return _rollback_and_reconcile(
             spec,
-            expected,
-            delayed=health,
+            provider=provider,
+            previous=previous,
+            previous_text=previous_text,
+            previous_version=previous_version,
+            expected=expected,
+            failure_detail="the replacement failed its final binding re-read",
+            command_runner=command_runner,
+            identity_probe=identity_probe,
+            settle_seconds=settle_seconds,
+            refresh_seconds=refresh_seconds,
+            timeout_seconds=timeout_seconds,
             show_local_paths=show_local_paths,
-            reconciliation=reconciled,
+            sleeper=sleeper,
+            clock=clock,
+            attempted_health=health,
         )
     return _operation_payload(
         succeeded_status,
diff --git a/tests/test_board_service.py b/tests/test_board_service.py
index fb498a8d..bb06b5e1 100644
--- a/tests/test_board_service.py
+++ b/tests/test_board_service.py
@@ -1807,6 +1807,108 @@ def bootstrap(self, label: str) -> tuple[bool, str]:
         )
         self.assertEqual(self.host.identities[5332]["repo"], "codemower-ai/code-mower")
 
+    def test_a_successful_bootstrap_with_failed_health_restores_the_old_binding(self) -> None:
+        original = self.spec()
+        self.install(original)
+        before = (self.root / f"{original.label}.plist").read_text(encoding="utf-8")
+        drifted = self.spec(
+            repo="codemower-ai/private-repo", repo_path=self.other_checkout, port=5332
+        )
+
+        def wrong_replacement_identity(host: str, port: int) -> dict[str, object]:
+            identity = self.host.identity_probe(host, port)
+            if identity.get("repo") == "codemower-ai/private-repo":
+                identity["repo"] = "codemower-ai/code-mower"
+            return identity
+
+        payload = board_service.restart_service(
+            drifted,
+            provider=self.host.provider(),
+            replace=True,
+            command_runner=self.host.run,
+            identity_probe=wrong_replacement_identity,
+            settle_seconds=0.0,
+            refresh_seconds=0.1,
+            timeout_seconds=0.0,
+            sleeper=self.sleeper,
+        )
+
+        self.assertEqual(payload["status"], "apply_failed")
+        self.assertEqual(payload["delayed_health"]["state"], "fail")
+        self.assertEqual(payload["reconciliation"]["state"], "previous")
+        self.assertTrue(payload["rollback"]["ok"])
+        self.assertEqual(
+            (self.root / f"{original.label}.plist").read_text(encoding="utf-8"), before
+        )
+        self.assertEqual(self.host.identities[5332]["repo"], "codemower-ai/code-mower")
+
+    def test_an_unreadable_takeover_failure_never_claims_the_prior_state_was_restored(self) -> None:
+        original = self.spec()
+        self.install(original)
+        path = self.root / f"{original.label}.plist"
+        path.write_bytes(b"bplist00\xff\xfe\x00")
+        self.host.bootstrap_failures.add(original.label)
+        drifted = self.spec(
+            repo="codemower-ai/private-repo", repo_path=self.other_checkout, port=5332
+        )
+
+        payload = self.restart(drifted, replace=True)
+
+        self.assertEqual(payload["status"], "rollback_failed")
+        self.assertEqual(payload["reconciliation"]["state"], "unresolved")
+        self.assertFalse(payload["rollback"]["ok"])
+        self.assertFalse(payload["rollback"]["restored"])
+        self.assertIn("unreadable prior definition", payload["message"])
+        self.assertNotIn("restored and verified", payload["message"])
+        self.assertIn("recovery", payload["reconciliation"])
+        self.assertFalse(path.exists())
+
+    def test_an_older_restored_board_is_verified_against_its_pre_mutation_version(self) -> None:
+        original = self.spec()
+        self.install(original)
+        old_version = {
+            "installed_version": "1.5.1",
+            "serving_version": "1.5.1",
+            "restart_recommended": False,
+        }
+        self.host.identities[5332]["board"]["version"] = dict(old_version)
+        drifted = self.spec(
+            repo="codemower-ai/private-repo", repo_path=self.other_checkout, port=5332
+        )
+        host = self.host
+
+        class ReplacementFailsThenRestoresOldVersion(board_service.LaunchdProvider):
+            bootstraps = 0
+
+            def bootstrap(self, label: str) -> tuple[bool, str]:
+                self.bootstraps += 1
+                if self.bootstraps == 1:
+                    return False, "Bootstrap failed: 5: Operation not permitted"
+                ok, detail = super().bootstrap(label)
+                host.identities[5332]["board"]["version"] = dict(old_version)
+                return ok, detail
+
+        payload = self._restart_with(
+            ReplacementFailsThenRestoresOldVersion(
+                command_runner=self.host.run,
+                root=self.root,
+                uid=self.host.uid,
+                platform="darwin",
+            ),
+            drifted,
+            replace=True,
+        )
+
+        self.assertEqual(payload["status"], "apply_failed")
+        self.assertEqual(payload["reconciliation"]["state"], "previous")
+        self.assertTrue(payload["reconciliation"]["version_evidence_captured"])
+        checks = {
+            check["id"]: check for check in payload["reconciliation"]["binding"]["checks"]
+        }
+        self.assertEqual(checks["binding.installed_version"]["expected_installed_version"], "1.5.1")
+        self.assertEqual(checks["binding.serving_version"]["expected_serving_version"], "1.5.1")
+        self.assertTrue(payload["rollback"]["ok"])
+
     def test_an_unreconciled_partial_replacement_has_one_exact_recovery_command(self) -> None:
         self.install(self.spec())
         drifted = self.spec(

From fe35555d7c241f68041aaf95d03b55bad77bbdd9 Mon Sep 17 00:00:00 2001
From: Jeff Huber <jeffhuber@users.noreply.github.com>
Date: Mon, 21 Sep 2026 00:50:09 -0700
Subject: [PATCH 3/8] Verify port absence after Board rollback

---
 docs/board-service-lifecycle.md |  6 +++
 src/code_mower/board_service.py | 29 ++++++++++++-
 tests/test_board_service.py     | 75 +++++++++++++++++++++++++++++++++
 3 files changed, 108 insertions(+), 2 deletions(-)

diff --git a/docs/board-service-lifecycle.md b/docs/board-service-lifecycle.md
index e6bb5d1f..76d80d8e 100644
--- a/docs/board-service-lifecycle.md
+++ b/docs/board-service-lifecycle.md
@@ -162,6 +162,12 @@ replacement and rollback successful. If neither binding can be verified, the
 operation reports `rollback_failed` with one recovery command and the checkout
 where that command must run.
 
+A failed first install has absence as its rollback target. That result re-reads
+the definition, launchd job and unfiltered listener inventory after cleanup;
+only no definition, no job and an available inventory with zero listeners on
+the target port is verified absence. An unavailable inventory or any surviving
+listener is unresolved `rollback_failed`, with the same exact recovery command.
+
 Rollback version checks use the identity/version evidence captured from the old
 Board before it is stopped. A byte-for-byte restored older Board therefore
 validates as the previous state even when the command performing the replacement
diff --git a/src/code_mower/board_service.py b/src/code_mower/board_service.py
index 4b214c1e..b3bb3d3b 100644
--- a/src/code_mower/board_service.py
+++ b/src/code_mower/board_service.py
@@ -2124,20 +2124,45 @@ def _rollback_and_reconcile(
         )
     else:
         load_state, _pid = provider.runtime_state(spec.label)
+        installed = provider.read_service(spec.label)
+        inventory = port_listener_inventory(spec.port, command_runner)
+        definition_absent = installed is None
+        job_absent = load_state == JOB_ABSENT
+        port_absent = bool(inventory["available"]) and not inventory["listeners"]
         absent = (
-            provider.read_service(spec.label) is None
-            and load_state == JOB_ABSENT
+            definition_absent
+            and job_absent
+            and port_absent
             and bool(rollback.get("ok"))
         )
         restored = {
             "state": "absent" if absent else "unresolved",
             "verified": absent,
             "target": "absent",
+            "definition_absent": definition_absent,
+            "job_absent": job_absent,
+            "listener_inventory_available": bool(inventory["available"]),
+            "listener_count": len(inventory["listeners"]),
         }
         if not absent:
             restored["recovery"] = _recovery_instruction(
                 spec, show_local_paths=show_local_paths
             )
+            if not inventory["available"]:
+                restored["detail"] = (
+                    "the local listener inventory could not be read, so target-port absence "
+                    "could not be verified"
+                )
+            elif inventory["listeners"]:
+                restored["detail"] = (
+                    f"port {spec.port} still has {len(inventory['listeners'])} listener(s) "
+                    "after rollback"
+                )
+            else:
+                restored["detail"] = rollback.get("detail") or (
+                    "the definition or launchd job was still present after rollback"
+                )
+        rollback["ok"] = absent
         status = "apply_failed" if absent else "rollback_failed"
         message = (
             f"{failure_detail}; the pre-operation absence was restored and verified"
diff --git a/tests/test_board_service.py b/tests/test_board_service.py
index bb06b5e1..13003c3b 100644
--- a/tests/test_board_service.py
+++ b/tests/test_board_service.py
@@ -1842,6 +1842,81 @@ def wrong_replacement_identity(host: str, port: int) -> dict[str, object]:
         )
         self.assertEqual(self.host.identities[5332]["repo"], "codemower-ai/code-mower")
 
+    def test_a_first_install_rollback_with_a_detached_listener_is_unresolved(self) -> None:
+        host = self.host
+
+        class LoadsWrongIdentityAndLeavesDetachedListener(board_service.LaunchdProvider):
+            def bootstrap(self, label: str) -> tuple[bool, str]:
+                super().bootstrap(label)
+                host.identities[5332]["repo"] = "codemower-ai/private-repo"
+                return False, "Bootstrap failed after registering the new job"
+
+            def bootout(self, label: str) -> tuple[bool, str]:
+                ok, detail = super().bootout(label)
+                host.add_foreign_listener(
+                    5332,
+                    command="/usr/local/bin/code-mower board serve --repo codemower-ai/private-repo",
+                    ppid=1,
+                )
+                return ok, detail
+
+        payload = self._restart_with(
+            LoadsWrongIdentityAndLeavesDetachedListener(
+                command_runner=self.host.run,
+                root=self.root,
+                uid=self.host.uid,
+                platform="darwin",
+            ),
+            self.spec(),
+        )
+
+        self.assertEqual(payload["status"], "rollback_failed")
+        self.assertEqual(payload["reconciliation"]["state"], "unresolved")
+        self.assertFalse(payload["reconciliation"]["verified"])
+        self.assertTrue(payload["reconciliation"]["definition_absent"])
+        self.assertTrue(payload["reconciliation"]["job_absent"])
+        self.assertTrue(payload["reconciliation"]["listener_inventory_available"])
+        self.assertEqual(payload["reconciliation"]["listener_count"], 1)
+        self.assertFalse(payload["rollback"]["ok"])
+        self.assertEqual(
+            payload["reconciliation"]["recovery"]["command"],
+            "code-mower board service restart --repo codemower-ai/code-mower "
+            "--repo-path . --host 127.0.0.1 --port 5332 --replace",
+        )
+        self.assertFalse((self.root / "ai.codemower.board.5332.plist").exists())
+        self.assertNotIn("ai.codemower.board.5332", self.host.loaded)
+        self.assertIn(5332, self.host.listeners)
+
+    def test_a_first_install_rollback_with_no_listener_inventory_is_unresolved(self) -> None:
+        host = self.host
+
+        class LosesListenerInventoryDuringRollback(board_service.LaunchdProvider):
+            def bootstrap(self, label: str) -> tuple[bool, str]:
+                super().bootstrap(label)
+                host.identities[5332]["repo"] = "codemower-ai/private-repo"
+                return False, "Bootstrap failed after registering the new job"
+
+            def bootout(self, label: str) -> tuple[bool, str]:
+                ok, detail = super().bootout(label)
+                host.listener_inventory_available = False
+                return ok, detail
+
+        payload = self._restart_with(
+            LosesListenerInventoryDuringRollback(
+                command_runner=self.host.run,
+                root=self.root,
+                uid=self.host.uid,
+                platform="darwin",
+            ),
+            self.spec(),
+        )
+
+        self.assertEqual(payload["status"], "rollback_failed")
+        self.assertEqual(payload["reconciliation"]["state"], "unresolved")
+        self.assertFalse(payload["reconciliation"]["listener_inventory_available"])
+        self.assertIn("recovery", payload["reconciliation"])
+        self.assertFalse(payload["rollback"]["ok"])
+
     def test_an_unreadable_takeover_failure_never_claims_the_prior_state_was_restored(self) -> None:
         original = self.spec()
         self.install(original)

From 9573efd05f79f2e6aabb5ce6ba7c143db0595012 Mon Sep 17 00:00:00 2001
From: Jeff Huber <jeffhuber@users.noreply.github.com>
Date: Mon, 21 Sep 2026 00:57:06 -0700
Subject: [PATCH 4/8] Reconcile ambiguous Board restarts

---
 src/code_mower/board_service.py | 54 +++++++++++++++-------
 tests/test_board_service.py     | 82 +++++++++++++++++++++++++++++++++
 2 files changed, 120 insertions(+), 16 deletions(-)

diff --git a/src/code_mower/board_service.py b/src/code_mower/board_service.py
index b3bb3d3b..8e4246e0 100644
--- a/src/code_mower/board_service.py
+++ b/src/code_mower/board_service.py
@@ -2731,19 +2731,16 @@ def restart_service(
         # bootstrapping it, which is what the runbook's restart has to do.
         if existing.loaded:
             ok, detail = provider.kickstart(spec.label)
+            action = "kickstart"
             restarted_message = "restarted the managed Board service in place and validated its binding"
         else:
             ok, detail = provider.bootstrap(spec.label)
+            action = "bootstrap"
             restarted_message = "loaded the installed definition, which was not running, and validated its binding"
-        if not ok:
-            return _operation_payload(
-                "apply_failed",
-                detail,
-                spec,
-                expected,
-                show_local_paths=show_local_paths,
-                known_paths=_provider_known_paths(provider, spec.label),
-            )
+        # Both launchd operations can complete before their command reports a
+        # timeout or permission-boundary failure. The provider return is not a
+        # terminal host state: re-read the complete binding before deciding
+        # whether this unchanged definition is healthy or unresolved.
         health = delayed_health(
             spec,
             provider=provider,
@@ -2756,12 +2753,6 @@ def restart_service(
             sleeper=sleeper,
             clock=clock,
         )
-        status = "restarted" if health["state"] == "pass" else "delayed_health_failed"
-        message = (
-            restarted_message
-            if health["state"] == "pass"
-            else "the service restarted but its binding did not validate within the delayed health window"
-        )
         final_binding = _final_binding_read(
             spec,
             provider=provider,
@@ -2774,9 +2765,39 @@ def restart_service(
         reconciled = _reconciliation(
             "new", final_binding, spec=spec, show_local_paths=show_local_paths
         )
-        if status == "restarted" and not reconciled["verified"]:
+        healthy = health["state"] == "pass" and bool(reconciled["verified"])
+        reconciled["delayed_health_verified"] = health["state"] == "pass"
+        if health["state"] != "pass" and reconciled["verified"]:
+            reconciled.update(
+                state="unresolved",
+                verified=False,
+                detail="the binding did not remain valid throughout the delayed health window",
+                recovery=_recovery_instruction(spec, show_local_paths=show_local_paths),
+            )
+        if healthy:
+            status = "restarted"
+            message = restarted_message
+            if not ok:
+                message += (
+                    f"; launchd reported {action} failure, but the installed definition and "
+                    "serving process were independently verified"
+                )
+        elif not ok:
+            status = "apply_failed"
+            message = (
+                (detail or f"launchd {action} failed")
+                + "; the installed binding could not be independently verified"
+            )
+        elif health["state"] != "pass":
+            status = "delayed_health_failed"
+            message = (
+                "the service restarted but its binding did not validate within the delayed "
+                "health window"
+            )
+        else:
             status = "delayed_health_failed"
             message = "the service passed delayed health but its final binding re-read did not validate"
+        provider_result = {"provider_detail": detail} if not ok else {}
         return _operation_payload(
             status,
             message,
@@ -2785,6 +2806,7 @@ def restart_service(
             delayed=health,
             show_local_paths=show_local_paths,
             reconciliation=reconciled,
+            **provider_result,
         )
     return _apply(
         spec,
diff --git a/tests/test_board_service.py b/tests/test_board_service.py
index 13003c3b..9c5954c8 100644
--- a/tests/test_board_service.py
+++ b/tests/test_board_service.py
@@ -2183,6 +2183,88 @@ def test_restart_loads_a_valid_definition_whose_job_was_unloaded(self) -> None:
             self.host.calls,
         )
 
+    def test_restart_reconciles_an_ambiguous_bootstrap_of_an_unloaded_definition(self) -> None:
+        spec = self.spec()
+        self.install(spec)
+        self.host.provider().bootout(spec.label)
+        definition = (self.root / f"{spec.label}.plist").read_text(encoding="utf-8")
+
+        class LoadsThenReportsFailure(board_service.LaunchdProvider):
+            def bootstrap(self, label: str) -> tuple[bool, str]:
+                super().bootstrap(label)
+                return False, "Bootstrap failed after registering the installed definition"
+
+        payload = self._restart_with(
+            LoadsThenReportsFailure(
+                command_runner=self.host.run,
+                root=self.root,
+                uid=self.host.uid,
+                platform="darwin",
+            ),
+            spec,
+        )
+
+        self.assertEqual(payload["status"], "restarted")
+        self.assertEqual(payload["delayed_health"]["state"], "pass")
+        self.assertEqual(payload["reconciliation"]["state"], "new")
+        self.assertTrue(payload["reconciliation"]["verified"])
+        self.assertTrue(payload["reconciliation"]["delayed_health_verified"])
+        self.assertIn("reported bootstrap failure", payload["message"])
+        self.assertIn("Bootstrap failed", payload["provider_detail"])
+        self.assertEqual(
+            payload["reconciliation"]["binding"]["failing_checks"], []
+        )
+        self.assertIn(spec.label, self.host.loaded)
+        self.assertEqual(
+            (self.root / f"{spec.label}.plist").read_text(encoding="utf-8"), definition
+        )
+
+    def test_restart_reports_an_unresolved_bootstrap_that_left_no_job(self) -> None:
+        spec = self.spec()
+        self.install(spec)
+        self.host.provider().bootout(spec.label)
+        self.host.bootstrap_failures.add(spec.label)
+
+        payload = self.restart(spec)
+
+        self.assertEqual(payload["status"], "apply_failed")
+        self.assertEqual(payload["delayed_health"]["state"], "fail")
+        self.assertEqual(payload["reconciliation"]["state"], "unresolved")
+        self.assertFalse(payload["reconciliation"]["verified"])
+        self.assertFalse(payload["reconciliation"]["delayed_health_verified"])
+        self.assertEqual(
+            payload["reconciliation"]["recovery"]["command"],
+            "code-mower board service restart --repo codemower-ai/code-mower "
+            "--repo-path . --host 127.0.0.1 --port 5332 --replace",
+        )
+        self.assertNotIn(spec.label, self.host.loaded)
+
+    def test_restart_reconciles_an_ambiguous_kickstart_as_healthy(self) -> None:
+        spec = self.spec()
+        self.install(spec)
+        first_pid = self.host.loaded[spec.label]
+
+        class RestartsThenReportsFailure(board_service.LaunchdProvider):
+            def kickstart(self, label: str) -> tuple[bool, str]:
+                super().kickstart(label)
+                return False, "Kickstart failed after restarting the installed job"
+
+        payload = self._restart_with(
+            RestartsThenReportsFailure(
+                command_runner=self.host.run,
+                root=self.root,
+                uid=self.host.uid,
+                platform="darwin",
+            ),
+            spec,
+        )
+
+        self.assertEqual(payload["status"], "restarted")
+        self.assertEqual(payload["reconciliation"]["state"], "new")
+        self.assertTrue(payload["reconciliation"]["verified"])
+        self.assertIn("reported kickstart failure", payload["message"])
+        self.assertNotEqual(self.host.loaded[spec.label], first_pid)
+
     def test_removal_that_cannot_delete_the_definition_is_not_reported_as_removed(self) -> None:
         self.install(self.spec())
 

From 156dfc06a97299a18834eccf7d84d6bcbdd8f896 Mon Sep 17 00:00:00 2001
From: Jeff Huber <jeffhuber@users.noreply.github.com>
Date: Mon, 21 Sep 2026 01:03:56 -0700
Subject: [PATCH 5/8] Require PID change for ambiguous kickstart

---
 src/code_mower/board_service.py | 44 ++++++++++++++++++++++++++++-----
 tests/test_board_service.py     | 41 ++++++++++++++++++++++++++++++
 2 files changed, 79 insertions(+), 6 deletions(-)

diff --git a/src/code_mower/board_service.py b/src/code_mower/board_service.py
index 8e4246e0..9861da31 100644
--- a/src/code_mower/board_service.py
+++ b/src/code_mower/board_service.py
@@ -2730,10 +2730,12 @@ def restart_service(
         # after a logout, or a manual `launchctl bootout` -- is recovered by
         # bootstrapping it, which is what the runbook's restart has to do.
         if existing.loaded:
+            pre_action_pid = existing.pid
             ok, detail = provider.kickstart(spec.label)
             action = "kickstart"
             restarted_message = "restarted the managed Board service in place and validated its binding"
         else:
+            pre_action_pid = None
             ok, detail = provider.bootstrap(spec.label)
             action = "bootstrap"
             restarted_message = "loaded the installed definition, which was not running, and validated its binding"
@@ -2765,8 +2767,18 @@ def restart_service(
         reconciled = _reconciliation(
             "new", final_binding, spec=spec, show_local_paths=show_local_paths
         )
-        healthy = health["state"] == "pass" and bool(reconciled["verified"])
+        final_pid = final_binding.get("pid")
+        pid_transition_required = action == "kickstart" and not ok
+        pid_transition_verified = bool(
+            pre_action_pid and final_pid and final_pid != pre_action_pid
+        )
+        binding_healthy = health["state"] == "pass" and bool(reconciled["verified"])
         reconciled["delayed_health_verified"] = health["state"] == "pass"
+        reconciled["command_succeeded"] = ok
+        reconciled["pre_action_pid"] = pre_action_pid
+        reconciled["final_pid"] = final_pid
+        reconciled["pid_transition_required"] = pid_transition_required
+        reconciled["pid_transition_verified"] = pid_transition_verified
         if health["state"] != "pass" and reconciled["verified"]:
             reconciled.update(
                 state="unresolved",
@@ -2774,7 +2786,20 @@ def restart_service(
                 detail="the binding did not remain valid throughout the delayed health window",
                 recovery=_recovery_instruction(spec, show_local_paths=show_local_paths),
             )
-        if healthy:
+        if pid_transition_required and not pid_transition_verified:
+            reconciled.update(
+                state="unresolved",
+                verified=False,
+                detail=(
+                    "kickstart reported failure and the final supervised pid did not prove that "
+                    "the process restarted"
+                ),
+                recovery=_recovery_instruction(spec, show_local_paths=show_local_paths),
+            )
+        restart_verified = binding_healthy and (
+            not pid_transition_required or pid_transition_verified
+        )
+        if restart_verified:
             status = "restarted"
             message = restarted_message
             if not ok:
@@ -2784,10 +2809,17 @@ def restart_service(
                 )
         elif not ok:
             status = "apply_failed"
-            message = (
-                (detail or f"launchd {action} failed")
-                + "; the installed binding could not be independently verified"
-            )
+            if binding_healthy and pid_transition_required and not pid_transition_verified:
+                message = (
+                    (detail or "launchd kickstart failed")
+                    + "; the service is healthy, but its supervised pid did not change, so the "
+                    "requested restart could not be verified"
+                )
+            else:
+                message = (
+                    (detail or f"launchd {action} failed")
+                    + "; the installed binding could not be independently verified"
+                )
         elif health["state"] != "pass":
             status = "delayed_health_failed"
             message = (
diff --git a/tests/test_board_service.py b/tests/test_board_service.py
index 9c5954c8..1772859a 100644
--- a/tests/test_board_service.py
+++ b/tests/test_board_service.py
@@ -2264,6 +2264,47 @@ def kickstart(self, label: str) -> tuple[bool, str]:
         self.assertTrue(payload["reconciliation"]["verified"])
         self.assertIn("reported kickstart failure", payload["message"])
         self.assertNotEqual(self.host.loaded[spec.label], first_pid)
+        self.assertEqual(payload["reconciliation"]["pre_action_pid"], first_pid)
+        self.assertEqual(
+            payload["reconciliation"]["final_pid"], self.host.loaded[spec.label]
+        )
+        self.assertTrue(payload["reconciliation"]["pid_transition_required"])
+        self.assertTrue(payload["reconciliation"]["pid_transition_verified"])
+
+    def test_restart_rejects_a_failed_kickstart_that_left_the_old_pid_running(self) -> None:
+        spec = self.spec()
+        self.install(spec)
+        first_pid = self.host.loaded[spec.label]
+
+        class RefusesKickstartWithoutChangingTheJob(board_service.LaunchdProvider):
+            def kickstart(self, label: str) -> tuple[bool, str]:
+                return False, "Kickstart failed: 1: Operation not permitted"
+
+        payload = self._restart_with(
+            RefusesKickstartWithoutChangingTheJob(
+                command_runner=self.host.run,
+                root=self.root,
+                uid=self.host.uid,
+                platform="darwin",
+            ),
+            spec,
+        )
+
+        self.assertEqual(payload["status"], "apply_failed")
+        self.assertEqual(payload["delayed_health"]["state"], "pass")
+        self.assertEqual(payload["reconciliation"]["state"], "unresolved")
+        self.assertFalse(payload["reconciliation"]["verified"])
+        self.assertTrue(payload["reconciliation"]["pid_transition_required"])
+        self.assertFalse(payload["reconciliation"]["pid_transition_verified"])
+        self.assertEqual(payload["reconciliation"]["pre_action_pid"], first_pid)
+        self.assertEqual(payload["reconciliation"]["final_pid"], first_pid)
+        self.assertEqual(self.host.loaded[spec.label], first_pid)
+        self.assertEqual(
+            payload["reconciliation"]["recovery"]["command"],
+            "code-mower board service restart --repo codemower-ai/code-mower "
+            "--repo-path . --host 127.0.0.1 --port 5332 --replace",
+        )
+        self.assertIn("pid did not change", payload["message"])
 
     def test_removal_that_cannot_delete_the_definition_is_not_reported_as_removed(self) -> None:
         self.install(self.spec())

From a564cff2d7ff140f885db3e4cbe867b8e16cf35a Mon Sep 17 00:00:00 2001
From: Jeff Huber <jeffhuber@users.noreply.github.com>
Date: Mon, 21 Sep 2026 01:38:04 -0700
Subject: [PATCH 6/8] Reuse settled Board binding evidence

---
 src/code_mower/board_service.py | 74 +++++++++++----------------------
 tests/test_board_service.py     | 64 ++++++++++++++++++++++++++++
 2 files changed, 88 insertions(+), 50 deletions(-)

diff --git a/src/code_mower/board_service.py b/src/code_mower/board_service.py
index 9861da31..55cc7f8e 100644
--- a/src/code_mower/board_service.py
+++ b/src/code_mower/board_service.py
@@ -27,6 +27,7 @@
 import os
 import plistlib
 import re
+import shlex
 import shutil
 import subprocess
 import sys
@@ -1695,6 +1696,20 @@ def delayed_health(
     }
 
 
+def _health_binding(health: Mapping[str, Any]) -> dict[str, Any]:
+    """Return the final full binding sample from one settled health window.
+
+    ``delayed_health`` exits only after a complete ``validate_binding`` sample
+    passes or its bounded window expires. Reusing that last sample keeps the
+    terminal decision tied to the settled window. A separate unretired probe
+    immediately afterward could overturn a verified pass on one transient
+    listener or HTTP-read failure and cause an unnecessary rollback.
+    """
+
+    binding = health.get("binding")
+    return dict(binding) if isinstance(binding, Mapping) else {}
+
+
 # -- operations -------------------------------------------------------------
 
 
@@ -1944,7 +1959,8 @@ def _recovery_instruction(spec: ServiceSpec, *, show_local_paths: bool) -> dict[
 
     command = (
         "code-mower board service restart "
-        f"--repo {spec.repo} --repo-path . --host {spec.host} --port {spec.port} --replace"
+        f"--repo {shlex.quote(spec.repo)} --repo-path . "
+        f"--host {shlex.quote(spec.host)} --port {spec.port} --replace"
     )
     if "--record-events" not in spec.arguments:
         command += " --no-record-events"
@@ -2065,16 +2081,7 @@ def _rollback_and_reconcile(
             sleeper=sleeper,
             clock=clock,
         )
-        previous_binding = _final_binding_read(
-            previous_spec,
-            provider=provider,
-            command_runner=command_runner,
-            identity_probe=identity_probe,
-            show_local_paths=show_local_paths,
-            expected_digest=previous.digest,
-            expected_arguments=previous.arguments,
-            expected_version=previous_version,
-        )
+        previous_binding = _health_binding(rollback_health)
         restored = _reconciliation(
             "previous", previous_binding, spec=spec, show_local_paths=show_local_paths
         )
@@ -2344,16 +2351,7 @@ def _apply(
             sleeper=sleeper,
             clock=clock,
         )
-        previous_binding = _final_binding_read(
-            previous_spec,
-            provider=provider,
-            command_runner=command_runner,
-            identity_probe=identity_probe,
-            show_local_paths=show_local_paths,
-            expected_digest=previous.digest,
-            expected_arguments=previous.arguments,
-            expected_version=previous_version,
-        )
+        previous_binding = _health_binding(restore_health)
         reconciled = _reconciliation(
             "previous", previous_binding, spec=spec, show_local_paths=show_local_paths
         )
@@ -2405,15 +2403,7 @@ def _apply(
                 clock=clock,
             )
             attempted_health = applied_health
-            applied_binding = _final_binding_read(
-                spec,
-                provider=provider,
-                command_runner=command_runner,
-                identity_probe=identity_probe,
-                show_local_paths=show_local_paths,
-                expected_digest=expected,
-                expected_arguments=spec.arguments,
-            )
+            applied_binding = _health_binding(applied_health)
             applied = _reconciliation(
                 "new", applied_binding, spec=spec, show_local_paths=show_local_paths
             )
@@ -2484,15 +2474,7 @@ def _apply(
             clock=clock,
             attempted_health=health,
         )
-    final_binding = _final_binding_read(
-        spec,
-        provider=provider,
-        command_runner=command_runner,
-        identity_probe=identity_probe,
-        show_local_paths=show_local_paths,
-        expected_digest=expected,
-        expected_arguments=spec.arguments,
-    )
+    final_binding = _health_binding(health)
     reconciled = _reconciliation("new", final_binding, spec=spec, show_local_paths=show_local_paths)
     if not reconciled["verified"]:
         return _rollback_and_reconcile(
@@ -2502,7 +2484,7 @@ def _apply(
             previous_text=previous_text,
             previous_version=previous_version,
             expected=expected,
-            failure_detail="the replacement failed its final binding re-read",
+            failure_detail="the replacement health window did not retain a verified binding",
             command_runner=command_runner,
             identity_probe=identity_probe,
             settle_seconds=settle_seconds,
@@ -2755,15 +2737,7 @@ def restart_service(
             sleeper=sleeper,
             clock=clock,
         )
-        final_binding = _final_binding_read(
-            spec,
-            provider=provider,
-            command_runner=command_runner,
-            identity_probe=identity_probe,
-            show_local_paths=show_local_paths,
-            expected_digest=expected,
-            expected_arguments=spec.arguments,
-        )
+        final_binding = _health_binding(health)
         reconciled = _reconciliation(
             "new", final_binding, spec=spec, show_local_paths=show_local_paths
         )
@@ -2828,7 +2802,7 @@ def restart_service(
             )
         else:
             status = "delayed_health_failed"
-            message = "the service passed delayed health but its final binding re-read did not validate"
+            message = "the delayed health window did not retain a verified final binding"
         provider_result = {"provider_detail": detail} if not ok else {}
         return _operation_payload(
             status,
diff --git a/tests/test_board_service.py b/tests/test_board_service.py
index 1772859a..302e80e7 100644
--- a/tests/test_board_service.py
+++ b/tests/test_board_service.py
@@ -1842,6 +1842,42 @@ def wrong_replacement_identity(host: str, port: int) -> dict[str, object]:
         )
         self.assertEqual(self.host.identities[5332]["repo"], "codemower-ai/code-mower")
 
+    def test_a_transient_probe_after_settled_replacement_health_cannot_trigger_rollback(self) -> None:
+        original = self.spec()
+        self.install(original)
+        drifted = self.spec(
+            repo="codemower-ai/private-repo", repo_path=self.other_checkout, port=5332
+        )
+        replacement_probes = 0
+
+        def healthy_once(host: str, port: int) -> dict[str, object]:
+            nonlocal replacement_probes
+            identity = self.host.identity_probe(host, port)
+            if identity.get("repo") == "codemower-ai/private-repo":
+                replacement_probes += 1
+                if replacement_probes > 1:
+                    return {"available": False, "error": "transient probe failure"}
+            return identity
+
+        payload = board_service.restart_service(
+            drifted,
+            provider=self.host.provider(),
+            replace=True,
+            command_runner=self.host.run,
+            identity_probe=healthy_once,
+            settle_seconds=0.0,
+            refresh_seconds=0.1,
+            timeout_seconds=0.0,
+            sleeper=self.sleeper,
+        )
+
+        self.assertEqual(payload["status"], "restarted")
+        self.assertEqual(payload["reconciliation"]["state"], "new")
+        self.assertTrue(payload["reconciliation"]["verified"])
+        self.assertEqual(replacement_probes, 1)
+        self.assertNotIn("rollback", payload)
+        self.assertEqual(self.host.identities[5332]["repo"], "codemower-ai/private-repo")
+
     def test_a_first_install_rollback_with_a_detached_listener_is_unresolved(self) -> None:
         host = self.host
 
@@ -2271,6 +2307,34 @@ def kickstart(self, label: str) -> tuple[bool, str]:
         self.assertTrue(payload["reconciliation"]["pid_transition_required"])
         self.assertTrue(payload["reconciliation"]["pid_transition_verified"])
 
+    def test_restart_uses_the_settled_health_sample_as_its_terminal_binding(self) -> None:
+        spec = self.spec()
+        self.install(spec)
+        identity_probes = 0
+
+        def healthy_once(host: str, port: int) -> dict[str, object]:
+            nonlocal identity_probes
+            identity_probes += 1
+            if identity_probes > 1:
+                return {"available": False, "error": "transient probe failure"}
+            return self.host.identity_probe(host, port)
+
+        payload = board_service.restart_service(
+            spec,
+            provider=self.host.provider(),
+            command_runner=self.host.run,
+            identity_probe=healthy_once,
+            settle_seconds=0.0,
+            refresh_seconds=0.1,
+            timeout_seconds=0.0,
+            sleeper=self.sleeper,
+        )
+
+        self.assertEqual(payload["status"], "restarted")
+        self.assertEqual(payload["reconciliation"]["state"], "new")
+        self.assertTrue(payload["reconciliation"]["verified"])
+        self.assertEqual(identity_probes, 1)
+
     def test_restart_rejects_a_failed_kickstart_that_left_the_old_pid_running(self) -> None:
         spec = self.spec()
         self.install(spec)

From fad91dbd36ea8808a3d318678e370ec5f488b76b Mon Sep 17 00:00:00 2001
From: Jeff Huber <jeffhuber@users.noreply.github.com>
Date: Mon, 21 Sep 2026 08:34:12 -0700
Subject: [PATCH 7/8] Preserve verified unloaded Board rollback

---
 docs/board-service-lifecycle.md |   6 +-
 src/code_mower/board_service.py |  21 ++++--
 tests/test_board_service.py     | 118 ++++++++++++++++++++++++++++++++
 3 files changed, 139 insertions(+), 6 deletions(-)

diff --git a/docs/board-service-lifecycle.md b/docs/board-service-lifecycle.md
index 76d80d8e..d95bed87 100644
--- a/docs/board-service-lifecycle.md
+++ b/docs/board-service-lifecycle.md
@@ -171,7 +171,11 @@ listener is unresolved `rollback_failed`, with the same exact recovery command.
 Rollback version checks use the identity/version evidence captured from the old
 Board before it is stopped. A byte-for-byte restored older Board therefore
 validates as the previous state even when the command performing the replacement
-comes from a newer installation. If the prior definition was unreadable,
+comes from a newer installation. A readable definition whose job was already
+unloaded has no running version to capture; after restoration, its complete
+binding is checked against the normal current-version contract. Missing version
+evidence remains unresolved when a prior job was loaded or its state was unknown.
+If the prior definition was unreadable,
 `--replace` knowingly discarded bytes that could not be backed up; removing a
 failed replacement is then reported as unresolved cleanup, never as a restored
 previous service.
diff --git a/src/code_mower/board_service.py b/src/code_mower/board_service.py
index 55cc7f8e..d8fbff1f 100644
--- a/src/code_mower/board_service.py
+++ b/src/code_mower/board_service.py
@@ -2027,7 +2027,12 @@ def _previous_version_evidence(
 ) -> dict[str, Any] | None:
     """Capture the old Board's version contract before its process is stopped."""
 
-    if previous is None or not previous.readable or not previous.port:
+    if (
+        previous is None
+        or not previous.readable
+        or not previous.port
+        or previous.load_state == JOB_ABSENT
+    ):
         return None
     probe = identity_probe or (lambda host, port: probe_identity(host, port))
     identity = probe(previous.host, previous.port)
@@ -2085,15 +2090,18 @@ def _rollback_and_reconcile(
         restored = _reconciliation(
             "previous", previous_binding, spec=spec, show_local_paths=show_local_paths
         )
-        if previous_version is None:
+        version_evidence_required = previous.load_state != JOB_ABSENT
+        if previous_version is None and version_evidence_required:
             restored.update(
                 state="unresolved",
                 verified=False,
                 version_evidence_captured=False,
+                version_evidence_required=True,
                 recovery=_recovery_instruction(spec, show_local_paths=show_local_paths),
             )
         else:
-            restored["version_evidence_captured"] = True
+            restored["version_evidence_captured"] = previous_version is not None
+            restored["version_evidence_required"] = version_evidence_required
         rollback["ok"] = bool(restored["verified"])
         rollback["restored"] = bool(restored["verified"])
         rollback["delayed_health"] = rollback_health
@@ -2355,15 +2363,18 @@ def _apply(
         reconciled = _reconciliation(
             "previous", previous_binding, spec=spec, show_local_paths=show_local_paths
         )
-        if previous_version is None:
+        version_evidence_required = previous.load_state != JOB_ABSENT
+        if previous_version is None and version_evidence_required:
             reconciled.update(
                 state="unresolved",
                 verified=False,
                 version_evidence_captured=False,
+                version_evidence_required=True,
                 recovery=_recovery_instruction(spec, show_local_paths=show_local_paths),
             )
         else:
-            reconciled["version_evidence_captured"] = True
+            reconciled["version_evidence_captured"] = previous_version is not None
+            reconciled["version_evidence_required"] = version_evidence_required
         restore["ok"] = bool(reconciled["verified"])
         restore["restored"] = bool(reconciled["verified"])
         restore["delayed_health"] = restore_health
diff --git a/tests/test_board_service.py b/tests/test_board_service.py
index 302e80e7..45521fb3 100644
--- a/tests/test_board_service.py
+++ b/tests/test_board_service.py
@@ -1608,6 +1608,30 @@ def test_a_failed_replacement_write_reloads_the_service_it_stopped(self) -> None
         self.assertIn("ai.codemower.board.5332", self.host.loaded)
         self.assertEqual(self.host.identities[5332]["repo"], "codemower-ai/code-mower")
 
+    def test_a_failed_write_restores_a_readable_definition_that_was_already_unloaded(self) -> None:
+        original = self.spec()
+        self.install(original)
+        before = (self.root / f"{original.label}.plist").read_text(encoding="utf-8")
+        self.host.provider().bootout(original.label)
+        drifted = self.spec(
+            repo="codemower-ai/private-repo", repo_path=self.other_checkout, port=5332
+        )
+
+        payload = self._restart_with(self._refusing_writes(), drifted, replace=True)
+
+        self.assertEqual(payload["status"], "apply_failed")
+        self.assertEqual(payload["reconciliation"]["state"], "previous")
+        self.assertTrue(payload["reconciliation"]["verified"])
+        self.assertFalse(payload["reconciliation"]["version_evidence_captured"])
+        self.assertFalse(payload["reconciliation"]["version_evidence_required"])
+        self.assertTrue(payload["rollback"]["ok"])
+        self.assertTrue(payload["rollback"]["restored"])
+        self.assertEqual(
+            (self.root / f"{original.label}.plist").read_text(encoding="utf-8"), before
+        )
+        self.assertEqual(payload["reconciliation"]["binding"]["failing_checks"], [])
+        self.assertEqual(self.host.identities[5332]["repo"], "codemower-ai/code-mower")
+
     def test_a_failed_write_that_cannot_be_reloaded_says_so(self) -> None:
         self.install(self.spec())
         self.host.bootstrap_failures.add("ai.codemower.board.5332")
@@ -1807,6 +1831,50 @@ def bootstrap(self, label: str) -> tuple[bool, str]:
         )
         self.assertEqual(self.host.identities[5332]["repo"], "codemower-ai/code-mower")
 
+    def test_a_failed_replacement_restores_a_readable_definition_that_was_already_unloaded(
+        self,
+    ) -> None:
+        original = self.spec()
+        self.install(original)
+        before = (self.root / f"{original.label}.plist").read_text(encoding="utf-8")
+        self.host.provider().bootout(original.label)
+        drifted = self.spec(
+            repo="codemower-ai/private-repo", repo_path=self.other_checkout, port=5332
+        )
+
+        class ReplacementFailsBeforeLoading(board_service.LaunchdProvider):
+            bootstraps = 0
+
+            def bootstrap(self, label: str) -> tuple[bool, str]:
+                self.bootstraps += 1
+                if self.bootstraps == 1:
+                    return False, "Bootstrap failed: 5: Operation not permitted"
+                return super().bootstrap(label)
+
+        payload = self._restart_with(
+            ReplacementFailsBeforeLoading(
+                command_runner=self.host.run,
+                root=self.root,
+                uid=self.host.uid,
+                platform="darwin",
+            ),
+            drifted,
+            replace=True,
+        )
+
+        self.assertEqual(payload["status"], "apply_failed")
+        self.assertEqual(payload["reconciliation"]["state"], "previous")
+        self.assertTrue(payload["reconciliation"]["verified"])
+        self.assertFalse(payload["reconciliation"]["version_evidence_captured"])
+        self.assertFalse(payload["reconciliation"]["version_evidence_required"])
+        self.assertTrue(payload["rollback"]["ok"])
+        self.assertTrue(payload["rollback"]["restored"])
+        self.assertEqual(
+            (self.root / f"{original.label}.plist").read_text(encoding="utf-8"), before
+        )
+        self.assertEqual(payload["reconciliation"]["binding"]["failing_checks"], [])
+        self.assertEqual(self.host.identities[5332]["repo"], "codemower-ai/code-mower")
+
     def test_a_successful_bootstrap_with_failed_health_restores_the_old_binding(self) -> None:
         original = self.spec()
         self.install(original)
@@ -2020,6 +2088,56 @@ def bootstrap(self, label: str) -> tuple[bool, str]:
         self.assertEqual(checks["binding.serving_version"]["expected_serving_version"], "1.5.1")
         self.assertTrue(payload["rollback"]["ok"])
 
+    def test_a_running_previous_board_with_missing_version_evidence_stays_unresolved(self) -> None:
+        original = self.spec()
+        self.install(original)
+        drifted = self.spec(
+            repo="codemower-ai/private-repo", repo_path=self.other_checkout, port=5332
+        )
+        identity_probes = 0
+
+        def miss_only_the_pre_mutation_probe(host: str, port: int) -> dict[str, object]:
+            nonlocal identity_probes
+            identity_probes += 1
+            if identity_probes == 1:
+                return {"available": False, "error": "transient probe failure"}
+            return self.host.identity_probe(host, port)
+
+        class ReplacementFailsBeforeLoading(board_service.LaunchdProvider):
+            bootstraps = 0
+
+            def bootstrap(self, label: str) -> tuple[bool, str]:
+                self.bootstraps += 1
+                if self.bootstraps == 1:
+                    return False, "Bootstrap failed: 5: Operation not permitted"
+                return super().bootstrap(label)
+
+        payload = board_service.restart_service(
+            drifted,
+            provider=ReplacementFailsBeforeLoading(
+                command_runner=self.host.run,
+                root=self.root,
+                uid=self.host.uid,
+                platform="darwin",
+            ),
+            replace=True,
+            command_runner=self.host.run,
+            identity_probe=miss_only_the_pre_mutation_probe,
+            settle_seconds=0.0,
+            refresh_seconds=0.1,
+            timeout_seconds=0.0,
+            sleeper=self.sleeper,
+        )
+
+        self.assertEqual(payload["status"], "rollback_failed")
+        self.assertEqual(payload["reconciliation"]["state"], "unresolved")
+        self.assertFalse(payload["reconciliation"]["verified"])
+        self.assertFalse(payload["reconciliation"]["version_evidence_captured"])
+        self.assertTrue(payload["reconciliation"]["version_evidence_required"])
+        self.assertEqual(payload["reconciliation"]["binding"]["failing_checks"], [])
+        self.assertFalse(payload["rollback"]["ok"])
+        self.assertIn("recovery", payload["reconciliation"])
+
     def test_an_unreconciled_partial_replacement_has_one_exact_recovery_command(self) -> None:
         self.install(self.spec())
         drifted = self.spec(

From 446442d3c22864dd88fb1da2325bb66bf593e668 Mon Sep 17 00:00:00 2001
From: Jeff Huber <jeffhuber@users.noreply.github.com>
Date: Mon, 21 Sep 2026 10:27:07 -0700
Subject: [PATCH 8/8] Fail closed on ambiguous Board provider state

---
 src/code_mower/board_service.py |  39 +++++++---
 tests/test_board_service.py     | 133 +++++++++++++++++++++++++++++++-
 2 files changed, 161 insertions(+), 11 deletions(-)

diff --git a/src/code_mower/board_service.py b/src/code_mower/board_service.py
index d8fbff1f..50c49729 100644
--- a/src/code_mower/board_service.py
+++ b/src/code_mower/board_service.py
@@ -2047,6 +2047,24 @@ def _previous_version_evidence(
     }
 
 
+def _runtime_state_or_unknown(provider: Any, label: str) -> tuple[str, int | None]:
+    """Read provider runtime state without turning uncertainty into absence.
+
+    Reconciliation runs after a provider operation has already returned an
+    ambiguous result. If the follow-up query itself raises, the only truthful
+    answer is `JOB_UNKNOWN`: a missing answer must never prove that a job is
+    absent or let an exception bypass the rollback result.
+    """
+
+    runtime_state = getattr(provider, "runtime_state", None)
+    if runtime_state is None:
+        return JOB_UNKNOWN, None
+    try:
+        return runtime_state(label)
+    except (OSError, subprocess.SubprocessError):
+        return JOB_UNKNOWN, None
+
+
 def _rollback_and_reconcile(
     spec: ServiceSpec,
     *,
@@ -2138,7 +2156,7 @@ def _rollback_and_reconcile(
             "takeover and could not be restored"
         )
     else:
-        load_state, _pid = provider.runtime_state(spec.label)
+        load_state, _pid = _runtime_state_or_unknown(provider, spec.label)
         installed = provider.read_service(spec.label)
         inventory = port_listener_inventory(spec.port, command_runner)
         definition_absent = installed is None
@@ -2154,6 +2172,7 @@ def _rollback_and_reconcile(
             "state": "absent" if absent else "unresolved",
             "verified": absent,
             "target": "absent",
+            "job_state": load_state,
             "definition_absent": definition_absent,
             "job_absent": job_absent,
             "listener_inventory_available": bool(inventory["available"]),
@@ -2163,7 +2182,11 @@ def _rollback_and_reconcile(
             restored["recovery"] = _recovery_instruction(
                 spec, show_local_paths=show_local_paths
             )
-            if not inventory["available"]:
+            if load_state == JOB_UNKNOWN:
+                restored["detail"] = (
+                    "the launchd job state could not be read, so job absence could not be verified"
+                )
+            elif not inventory["available"]:
                 restored["detail"] = (
                     "the local listener inventory could not be read, so target-port absence "
                     "could not be verified"
@@ -2398,7 +2421,7 @@ def _apply(
         # failure. The provider result is therefore not the outcome. If the
         # replacement clears the complete serving gate, keep it instead of
         # beginning a rollback that can create a definition/process split.
-        load_state, _pid = provider.runtime_state(spec.label)
+        load_state, _pid = _runtime_state_or_unknown(provider, spec.label)
         attempted_health: Mapping[str, Any] | None = None
         if load_state == JOB_LOADED:
             applied_health = delayed_health(
@@ -2429,6 +2452,7 @@ def _apply(
                     expected,
                     delayed=applied_health,
                     show_local_paths=show_local_paths,
+                    known_paths=_provider_known_paths(provider, spec.label),
                     reconciliation=applied,
                     provider_detail=detail,
                 )
@@ -2822,6 +2846,7 @@ def restart_service(
             expected,
             delayed=health,
             show_local_paths=show_local_paths,
+            known_paths=_provider_known_paths(provider, spec.label),
             reconciliation=reconciled,
             **provider_result,
         )
@@ -2893,12 +2918,8 @@ def _job_load_state(provider: Any, label: str) -> tuple[bool, int | None]:
     characterises as a missing job, releases the definition.
     """
 
-    runtime_state = getattr(provider, "runtime_state", None)
-    if runtime_state is not None:
-        try:
-            state, pid = runtime_state(label)
-        except (OSError, subprocess.SubprocessError):
-            return True, None
+    if getattr(provider, "runtime_state", None) is not None:
+        state, pid = _runtime_state_or_unknown(provider, label)
         return state != JOB_ABSENT, pid
     runtime = getattr(provider, "runtime", None)
     if runtime is None:
diff --git a/tests/test_board_service.py b/tests/test_board_service.py
index 45521fb3..e67ac8a8 100644
--- a/tests/test_board_service.py
+++ b/tests/test_board_service.py
@@ -914,6 +914,72 @@ def test_a_failed_first_install_leaves_no_definition_behind(self) -> None:
         self.assertEqual(payload["status"], "apply_failed")
         self.assertFalse((self.root / "ai.codemower.board.5332.plist").exists())
 
+    def test_first_install_rollback_fails_closed_when_runtime_state_raises(self) -> None:
+        class RuntimeQueryFailsAfterCleanup(board_service.LaunchdProvider):
+            fail_next_runtime_read = False
+
+            def bootstrap(self, label: str) -> tuple[bool, str]:
+                return False, "Bootstrap failed before loading the job"
+
+            def delete_definition(self, label: str) -> bool:
+                deleted = super().delete_definition(label)
+                self.fail_next_runtime_read = True
+                return deleted
+
+            def runtime_state(self, label: str) -> tuple[str, int | None]:
+                if self.fail_next_runtime_read:
+                    self.fail_next_runtime_read = False
+                    raise PermissionError("launchd state became unreadable")
+                return super().runtime_state(label)
+
+        payload = self._restart_with(
+            RuntimeQueryFailsAfterCleanup(
+                command_runner=self.host.run,
+                root=self.root,
+                uid=self.host.uid,
+                platform="darwin",
+            ),
+            self.spec(),
+        )
+
+        self.assertEqual(payload["status"], "rollback_failed")
+        self.assertEqual(payload["reconciliation"]["state"], "unresolved")
+        self.assertEqual(payload["reconciliation"]["job_state"], board_service.JOB_UNKNOWN)
+        self.assertFalse(payload["reconciliation"]["job_absent"])
+        self.assertTrue(payload["reconciliation"]["definition_absent"])
+        self.assertIn("job state could not be read", payload["reconciliation"]["detail"])
+
+    def test_ambiguous_bootstrap_runtime_error_rolls_back_instead_of_raising(self) -> None:
+        class LoadsThenRuntimeQueryTimesOut(board_service.LaunchdProvider):
+            fail_next_runtime_read = False
+
+            def bootstrap(self, label: str) -> tuple[bool, str]:
+                super().bootstrap(label)
+                self.fail_next_runtime_read = True
+                return False, "Bootstrap timed out after launchd accepted the definition"
+
+            def runtime_state(self, label: str) -> tuple[str, int | None]:
+                if self.fail_next_runtime_read:
+                    self.fail_next_runtime_read = False
+                    raise subprocess.TimeoutExpired(["launchctl", "print"], 1.0)
+                return super().runtime_state(label)
+
+        payload = self._restart_with(
+            LoadsThenRuntimeQueryTimesOut(
+                command_runner=self.host.run,
+                root=self.root,
+                uid=self.host.uid,
+                platform="darwin",
+            ),
+            self.spec(),
+        )
+
+        self.assertEqual(payload["status"], "apply_failed")
+        self.assertEqual(payload["reconciliation"]["state"], "absent")
+        self.assertTrue(payload["reconciliation"]["verified"])
+        self.assertNotIn(self.spec().label, self.host.loaded)
+        self.assertFalse((self.root / f"{self.spec().label}.plist").exists())
+
     def test_a_failed_first_install_that_cannot_be_cleaned_up_is_a_failed_rollback(self) -> None:
         # Nothing was installed before, so rolling back means leaving nothing
         # behind. A definition that survives its failed apply starts the service
@@ -2346,7 +2412,11 @@ def test_restart_reconciles_an_ambiguous_bootstrap_of_an_unloaded_definition(sel
         class LoadsThenReportsFailure(board_service.LaunchdProvider):
             def bootstrap(self, label: str) -> tuple[bool, str]:
                 super().bootstrap(label)
-                return False, "Bootstrap failed after registering the installed definition"
+                return (
+                    False,
+                    "Bootstrap failed after registering "
+                    f"{self.definition_path(label)}: Operation not permitted",
+                )
 
         payload = self._restart_with(
             LoadsThenReportsFailure(
@@ -2365,6 +2435,10 @@ def bootstrap(self, label: str) -> tuple[bool, str]:
         self.assertTrue(payload["reconciliation"]["delayed_health_verified"])
         self.assertIn("reported bootstrap failure", payload["message"])
         self.assertIn("Bootstrap failed", payload["provider_detail"])
+        self.assertIn(lane_status.LOCAL_PATH_REDACTION, payload["provider_detail"])
+        self.assertIn("Operation not permitted", payload["provider_detail"])
+        published = json.dumps(payload) + board_service.render_operation_text(payload)
+        self.assertNotIn(str(self.root), published)
         self.assertEqual(
             payload["reconciliation"]["binding"]["failing_checks"], []
         )
@@ -2373,6 +2447,30 @@ def bootstrap(self, label: str) -> tuple[bool, str]:
             (self.root / f"{spec.label}.plist").read_text(encoding="utf-8"), definition
         )
 
+    def test_ambiguous_bootstrap_provider_detail_honors_show_local_paths(self) -> None:
+        spec = self.spec()
+        self.install(spec)
+        self.host.provider().bootout(spec.label)
+
+        class LoadsThenReportsPath(board_service.LaunchdProvider):
+            def bootstrap(self, label: str) -> tuple[bool, str]:
+                super().bootstrap(label)
+                return False, f"Bootstrap failed after registering {self.definition_path(label)}"
+
+        payload = self._restart_with(
+            LoadsThenReportsPath(
+                command_runner=self.host.run,
+                root=self.root,
+                uid=self.host.uid,
+                platform="darwin",
+            ),
+            spec,
+            show_local_paths=True,
+        )
+
+        self.assertEqual(payload["status"], "restarted")
+        self.assertIn(str(self.root / f"{spec.label}.plist"), payload["provider_detail"])
+
     def test_restart_reports_an_unresolved_bootstrap_that_left_no_job(self) -> None:
         spec = self.spec()
         self.install(spec)
@@ -2401,7 +2499,11 @@ def test_restart_reconciles_an_ambiguous_kickstart_as_healthy(self) -> None:
         class RestartsThenReportsFailure(board_service.LaunchdProvider):
             def kickstart(self, label: str) -> tuple[bool, str]:
                 super().kickstart(label)
-                return False, "Kickstart failed after restarting the installed job"
+                return (
+                    False,
+                    "Kickstart failed after restarting the installed job from "
+                    f"{self.definition_path(label)}: Operation not permitted",
+                )
 
         payload = self._restart_with(
             RestartsThenReportsFailure(
@@ -2424,6 +2526,33 @@ def kickstart(self, label: str) -> tuple[bool, str]:
         )
         self.assertTrue(payload["reconciliation"]["pid_transition_required"])
         self.assertTrue(payload["reconciliation"]["pid_transition_verified"])
+        self.assertIn(lane_status.LOCAL_PATH_REDACTION, payload["provider_detail"])
+        self.assertIn("Operation not permitted", payload["provider_detail"])
+        published = json.dumps(payload) + board_service.render_operation_text(payload)
+        self.assertNotIn(str(self.root), published)
+
+    def test_ambiguous_kickstart_provider_detail_honors_show_local_paths(self) -> None:
+        spec = self.spec()
+        self.install(spec)
+
+        class RestartsThenReportsPath(board_service.LaunchdProvider):
+            def kickstart(self, label: str) -> tuple[bool, str]:
+                super().kickstart(label)
+                return False, f"Kickstart failed after reading {self.definition_path(label)}"
+
+        payload = self._restart_with(
+            RestartsThenReportsPath(
+                command_runner=self.host.run,
+                root=self.root,
+                uid=self.host.uid,
+                platform="darwin",
+            ),
+            spec,
+            show_local_paths=True,
+        )
+
+        self.assertEqual(payload["status"], "restarted")
+        self.assertIn(str(self.root / f"{spec.label}.plist"), payload["provider_detail"])
 
     def test_restart_uses_the_settled_health_sample_as_its_terminal_binding(self) -> None:
         spec = self.spec()