diff --git a/docs/guides/candidate-install.md b/docs/guides/candidate-install.md
index a5b6980c0..da63680a7 100644
--- a/docs/guides/candidate-install.md
+++ b/docs/guides/candidate-install.md
@@ -54,11 +54,31 @@ retains the existing refusal; use the channel manager's explicit upgrade command
 Shared MCP packages need the matching retained installation manager. Older channels
 keep their original `manager.py`; running a newer installer against that root does
 not update it. The newer installer refuses an MCP upgrade when its own bytes differ
-from the retained manager, before publishing a version or changing selection. Use
-the retained manager when it supports that package, or install with the new reviewed
-installer into a fresh root, such as `--root "$HOME/.hack-next-mcp"`. The old channel
-and its rollback remain intact. Automatic manager migration is not implemented.
-New channels support both old flat bundles and new MCP bundles.
+from the retained manager, before publishing a version or changing selection.
+
+To update the original flat-layout manager shipped with `5.0.0-next.1`, review a
+newer installer containing `upgrade-manager`, then run that saved file explicitly:
+
+```sh
+python3 /absolute/reviewed/install-prerelease.py --root "$HOME/.hack-next" upgrade-manager
+python3 "$HOME/.hack-next/manager.py" --root "$HOME/.hack-next" status
+```
+
+Stop every retained version's graphs and runtime first, using the commands below
+with that version's paths. Manager upgrade takes the channel's exclusive lock,
+refuses active launchers, and applies the existing quiescence checks to every
+retained version. It recognizes the exact original manager bytes and the standard
+launcher; custom or changed code, modified receipts, changed home identities, and
+aliased paths are refused. It preserves the launcher, installed bundles, home
+identities, and the selected/previous versions. The original manager and receipt
+remain in a private staging directory for inspection. Repeating the command with
+the already installed manager is a no-op.
+
+After this step, use the retained manager for ordinary bundle upgrades and rollback.
+It accepts both old flat bundles and new MCP bundles. This operation does not
+enroll MCP clients or migrate application data. For an unsupported manager, keep
+the existing channel and use a fresh root such as `--root "$HOME/.hack-next-mcp"`;
+do not edit its receipt to force adoption.
 
 Stop each candidate graph with its ordinary retained-data shutdown, then stop that
 version's owned runtime through the native executor. For the default installation
@@ -160,3 +180,16 @@ never adopted automatically. The selected prior bundle remains usable when its
 own receipt is unchanged. A later attempt refuses to overwrite the uncommitted
 version. Preserve the evidence and inspect it before an authorized cleanup; the
 manager has no automatic pruning or receipt-repair command.
+
+Manager upgrades change two files: `manager.py` and its hash in `.channel.json`.
+The manager first saves and syncs both versions and publishes an owned upgrade
+journal. While that journal exists, both the original and newer managers refuse
+ordinary launch, status, and selection commands. An interrupted upgrade can
+therefore temporarily make the channel unavailable, while its selection and data
+remain intact. Rerun `upgrade-manager` with the **same saved reviewed installer** to
+verify the journal, staged bytes, receipts, selection and stopped runtimes, then
+finish the upgrade. Recovery refuses changed or ambiguous state. Do not remove the
+journal or staging directory by hand. A failure before journal publication leaves
+the original manager usable; a completed upgrade removes the journal and retains
+the original bytes for inspection. This does not automatically downgrade the
+manager when rolling back a candidate bundle.
diff --git a/scripts/install-prerelease.py b/scripts/install-prerelease.py
index a266e7fcc..6f1123d33 100644
--- a/scripts/install-prerelease.py
+++ b/scripts/install-prerelease.py
@@ -1,8 +1,9 @@
 #!/usr/bin/env python3
 """Explicit, side-by-side native prerelease selection (Python 3.9+, standard library).
 
-The single atomic selection receipt is the commit point. Bundles and homes are
-retained by version; this manager never migrates runtime state or edits shell PATH.
+Selection changes commit through one atomic receipt. Explicit manager upgrades
+use a recoverable journal. Bundles and homes are retained by version; this manager
+never migrates runtime state or edits shell PATH.
 """
 
 import argparse
@@ -40,6 +41,12 @@
 REPOSITORY = "hack-dance/hack"
 DOWNLOAD_HOSTS = {"api.github.com", "github.com", "release-assets.githubusercontent.com",
                   "objects.githubusercontent.com"}
+# Original flat-layout manager, shipped before optional shared-MCP bundles.
+# Keep this an explicit allowlist; a matching user-written receipt is not provenance.
+MANAGER_PREDECESSORS = frozenset({
+    "b7c49e3fec6b06790e833db1d2dcb441d2223c283b792713be46826aa2eef877",
+})
+MANAGER_UPGRADE = ".manager-upgrade.json"
 
 
 class Refusal(Exception):
@@ -135,10 +142,10 @@ def write_file(path, contents, mode=0o600):
         os.fsync(target.fileno())
 
 
-def atomic_json(path, value):
+def atomic_file(path, contents):
     temporary = path.parent / (".selection-" + uuid.uuid4().hex)
     try:
-        write_file(temporary, json_bytes(value))
+        write_file(temporary, contents)
         os.replace(str(temporary), str(path))
         sync_directory(path.parent)
     finally:
@@ -146,6 +153,15 @@ def atomic_json(path, value):
             temporary.unlink()
 
 
+def atomic_json(path, value):
+    atomic_file(path, json_bytes(value))
+
+
+def launcher_bytes():
+    return ("#!/bin/sh\nset -eu\nroot=$(CDPATH= cd -- \"$(dirname -- \"$0\")/..\" && pwd -P)\n"
+            "exec /usr/bin/python3 -I -S \"$root/manager.py\" --root \"$root\" run -- \"$@\"\n").encode()
+
+
 def checksums(raw, expected):
     try:
         lines = raw.decode("ascii").splitlines()
@@ -428,9 +444,7 @@ def initialize(self):
         (self.root / "bin").mkdir(mode=0o700)
         (self.root / "versions").mkdir(mode=0o700)
         write_file(self.root / "manager.py", Path(__file__).read_bytes())
-        launcher = ("#!/bin/sh\nset -eu\nroot=$(CDPATH= cd -- \"$(dirname -- \"$0\")/..\" && pwd -P)\n"
-                    "exec /usr/bin/python3 -I -S \"$root/manager.py\" --root \"$root\" run -- \"$@\"\n")
-        write_file(self.root / "bin/hack-next", launcher.encode(), 0o755)
+        write_file(self.root / "bin/hack-next", launcher_bytes(), 0o755)
         root_info = self.root.stat()
         write_file(self.root / ".channel.json", json_bytes({
             "schema": "hack.prerelease-install/v1", "root": str(self.root), "uid": os.getuid(),
@@ -444,7 +458,7 @@ def initialize(self):
         sync_directory(self.root)
 
     @contextlib.contextmanager
-    def lock(self, shared=False):
+    def lock(self, shared=False, manager_upgrade=False):
         owned(self.root, directory=True, mode=0o700)
         owned(self.root / ".manager.lock", mode=0o600)
         descriptor = os.open(str(self.root / ".manager.lock"), os.O_RDWR | os.O_NOFOLLOW)
@@ -453,11 +467,15 @@ def lock(self, shared=False):
                 fcntl.flock(handle, (fcntl.LOCK_SH if shared else fcntl.LOCK_EX) | fcntl.LOCK_NB)
             except BlockingIOError as error:
                 raise Refusal("Another candidate manager or launcher is active.") from error
-            self.validate()
+            self.validate(manager_upgrade=manager_upgrade)
             yield
 
-    def validate(self):
+    def validate(self, manager_upgrade=False):
         root_info = owned(self.root, directory=True, mode=0o700)
+        pending = os.path.lexists(self.root / MANAGER_UPGRADE)
+        require(not pending or manager_upgrade,
+                "Manager upgrade is pending; rerun upgrade-manager with the same reviewed installer.")
+        transition = self.read_manager_upgrade() if pending else None
         marker = private_json(self.root / ".channel.json")
         require(set(marker) == {"schema", "root", "uid", "device", "inode",
                                 "manager_sha256", "launcher_sha256"}
@@ -468,7 +486,10 @@ def validate(self):
         owned(self.root / "manager.py", mode=0o600)
         owned(self.root / "bin", directory=True, mode=0o700)
         owned(self.root / "bin/hack-next", mode=0o755)
-        require(digest(self.root / "manager.py") == marker["manager_sha256"]
+        manager_hash = digest(self.root / "manager.py")
+        require((manager_hash == marker["manager_sha256"] or transition is not None
+                 and manager_hash == transition["to_sha256"]
+                 and marker["manager_sha256"] == transition["from_sha256"])
                 and digest(self.root / "bin/hack-next") == marker["launcher_sha256"],
                 "Installation manager or launcher changed.")
         require({p.name for p in (self.root / "bin").iterdir()} == {"hack-next"},
@@ -497,6 +518,8 @@ def validate(self):
                 private_json(entry / ".receipt.json")
                 self.verify_version(entry.name, digest(entry / ".receipt.json"))
         allowed = {".channel.json", ".selection.json", ".manager.lock", "manager.py", "bin", "versions"}
+        if transition is not None:
+            allowed.add(MANAGER_UPGRADE)
         for entry in self.root.iterdir():
             if entry.name in allowed:
                 continue
@@ -509,6 +532,112 @@ def validate(self):
             raise Refusal("Foreign installation entry: " + entry.name)
         self.state = state
 
+    def read_manager_upgrade(self):
+        """Verify both sides of the only supported two-file transition before recovery.
+
+        The journal closes ordinary launch admission before either file changes.
+        Recovery may finish this exact transition, never adopt changed receipts or
+        select software. The staged originals remain available for inspection.
+        """
+        plan = private_json(self.root / MANAGER_UPGRADE)
+        require(set(plan) == {"schema", "stage", "from_sha256", "to_sha256",
+                              "channel_sha256", "selection_sha256"}
+                and plan["schema"] == "hack.prerelease-manager-upgrade/v1"
+                and isinstance(plan["stage"], str)
+                and re.fullmatch(r"\.stage-[0-9a-f]{32}", plan["stage"]),
+                "Malformed manager upgrade journal.")
+        require(all(isinstance(plan[key], str) and SHA256.fullmatch(plan[key])
+                    for key in ("from_sha256", "to_sha256", "channel_sha256", "selection_sha256")),
+                "Malformed manager upgrade digest.")
+        require(plan["from_sha256"] in MANAGER_PREDECESSORS
+                and plan["to_sha256"] == digest(Path(__file__)),
+                "Manager recovery requires the same reviewed installer and a known predecessor.")
+        stage = self.root / plan["stage"]
+        owned(stage, directory=True, mode=0o700)
+        require({p.name for p in stage.iterdir()} == {
+            "manager-before.py", "manager-after.py", "channel-before.json", "channel-after.json"},
+            "Foreign manager upgrade staging entry.")
+        for entry in stage.iterdir():
+            owned(entry, mode=0o600)
+        before = private_json(stage / "channel-before.json")
+        after = private_json(stage / "channel-after.json")
+        require(digest(stage / "manager-before.py") == plan["from_sha256"]
+                and digest(stage / "manager-after.py") == plan["to_sha256"]
+                and digest(stage / "channel-before.json") == plan["channel_sha256"]
+                and before.get("manager_sha256") == plan["from_sha256"]
+                and after == dict(before, manager_sha256=plan["to_sha256"]),
+                "Manager upgrade staging changed.")
+        owned(self.root / "manager.py", mode=0o600)
+        private_json(self.root / ".channel.json")
+        private_json(self.root / ".selection.json")
+        old_receipt = (stage / "channel-before.json").read_bytes()
+        new_receipt = (stage / "channel-after.json").read_bytes()
+        current_receipt = (self.root / ".channel.json").read_bytes()
+        current_hash = digest(self.root / "manager.py")
+        require((current_hash == plan["from_sha256"] and current_receipt == old_receipt)
+                or (current_hash == plan["to_sha256"] and current_receipt in (old_receipt, new_receipt)),
+                "Manager upgrade publication state changed.")
+        require(digest(self.root / ".selection.json") == plan["selection_sha256"],
+                "Selection changed during manager upgrade; inspection required.")
+        owned(self.root / "bin", directory=True, mode=0o700)
+        owned(self.root / "bin/hack-next", mode=0o755)
+        require((self.root / "bin/hack-next").read_bytes() == launcher_bytes(),
+                "Custom launcher cannot be upgraded.")
+        return plan
+
+    def upgrade_manager(self):
+        """Explicitly replace a known manager, with recoverable fail-closed publication."""
+        self.validate(manager_upgrade=True)
+        pending = os.path.lexists(self.root / MANAGER_UPGRADE)
+        source = canonical(Path(__file__).absolute())
+        owned(source)
+        target = source.read_bytes()
+        target_hash = hashlib.sha256(target).hexdigest()
+        if not pending:
+            current_hash = digest(self.root / "manager.py")
+            if current_hash == target_hash:
+                return
+            require(current_hash in MANAGER_PREDECESSORS,
+                    "Unknown or customized manager; explicit upgrade supports only the reviewed flat-layout predecessor.")
+            require((self.root / "bin/hack-next").read_bytes() == launcher_bytes(),
+                    "Custom launcher cannot be upgraded.")
+        # All retained versions share this manager. Keep the existing executor
+        # ownership checks, including status/down/status, for each of them.
+        for version in self.state["installed"]:
+            self.require_quiescent(version, manager_upgrade=True)
+        self.validate(manager_upgrade=True)
+        if not pending:
+            stage = self.root / (".stage-" + uuid.uuid4().hex)
+            stage.mkdir(mode=0o700)
+            before = (self.root / ".channel.json").read_bytes()
+            after = dict(parse_json(before), manager_sha256=target_hash)
+            write_file(stage / "manager-before.py", (self.root / "manager.py").read_bytes())
+            write_file(stage / "manager-after.py", target)
+            write_file(stage / "channel-before.json", before)
+            write_file(stage / "channel-after.json", json_bytes(after))
+            plan = {"schema": "hack.prerelease-manager-upgrade/v1", "stage": stage.name,
+                    "from_sha256": current_hash, "to_sha256": target_hash,
+                    "channel_sha256": hashlib.sha256(before).hexdigest(),
+                    "selection_sha256": digest(self.root / ".selection.json")}
+            write_file(stage / "journal.json", json_bytes(plan))
+            sync_directory(stage)
+            self.validate()
+            os.rename(str(stage / "journal.json"), str(self.root / MANAGER_UPGRADE))
+            sync_directory(stage)
+            sync_directory(self.root)
+        self.validate(manager_upgrade=True)
+        plan = self.read_manager_upgrade()
+        stage = self.root / plan["stage"]
+        if digest(self.root / "manager.py") != target_hash:
+            atomic_file(self.root / "manager.py", (stage / "manager-after.py").read_bytes())
+        self.validate(manager_upgrade=True)
+        if (self.root / ".channel.json").read_bytes() != (stage / "channel-after.json").read_bytes():
+            atomic_file(self.root / ".channel.json", (stage / "channel-after.json").read_bytes())
+        self.validate(manager_upgrade=True)
+        (self.root / MANAGER_UPGRADE).unlink()
+        sync_directory(self.root)
+        self.validate()
+
     def verify_version(self, version, checksum):
         directory = self.root / "versions" / version
         owned(directory, directory=True, mode=0o700)
@@ -542,10 +671,10 @@ def environment(self, version):
                             "HACK_HOME": str(directory / "cli-home")})
         return environment
 
-    def require_quiescent(self, version):
+    def require_quiescent(self, version, manager_upgrade=False):
         if version is None:
             return
-        self.validate()
+        self.validate(manager_upgrade=manager_upgrade)
         directory = self.verify_version(version, self.state["installed"][version])
         verify_signatures(directory / "bundle")
         for action in ("status", "down", "status"):
@@ -566,7 +695,7 @@ def require_quiescent(self, version):
             if phase == "uninitialized":
                 require(not list((directory / "native-home").iterdir()),
                         "Uninitialized home contains state; inspect it through its owning CLI.")
-            self.validate()
+            self.validate(manager_upgrade=manager_upgrade)
 
     def select(self, version, installed=None):
         self.require_quiescent(self.state["selected"])
@@ -612,11 +741,11 @@ def install(self, version, archive=None, checksum=None, upgrade=False):
                                              release_metadata)
         # The retained launcher always executes its recorded manager. Never select
         # a new layout that an older manager cannot subsequently validate/rollback.
-        # Automatic manager replacement needs a separate atomic migration protocol.
+        # Manager replacement is a separate explicit, recoverable operation.
         require(set(manifest) <= PAYLOAD
                 or digest(self.root / "manager.py") == digest(Path(__file__)),
                 "Shared MCP requires this channel's retained manager to match the installer. "
-                "Use the retained manager.py, or install with this installer into a fresh --root; "
+                "Run upgrade-manager with this reviewed installer, or use a fresh --root; "
                 "the existing channel and its selection are unchanged.")
         homes = {}
         for name in ("native-home", "cli-home"):
@@ -678,6 +807,7 @@ def parser():
     rollback.add_argument("--version", help="retained version; default previous selection")
     commands.add_parser("stable", help="deselect candidate; retain all bundles and homes")
     commands.add_parser("status")
+    commands.add_parser("upgrade-manager", help="explicitly upgrade a known channel manager, or finish its interrupted upgrade")
     run = commands.add_parser("run")
     run.add_argument("arguments", nargs=argparse.REMAINDER)
     return arguments
@@ -693,9 +823,11 @@ def main(argv=None):
         version_number(args.version)
         require(bool(args.archive) == bool(args.sha256), "--archive and --sha256 must be supplied together.")
         channel.initialize()
-    with channel.lock(shared=args.command == "run"):
+    with channel.lock(shared=args.command == "run", manager_upgrade=args.command == "upgrade-manager"):
         if args.command in ("install", "upgrade"):
             channel.install(args.version, args.archive, args.sha256, args.command == "upgrade")
+        elif args.command == "upgrade-manager":
+            channel.upgrade_manager()
         elif args.command == "rollback":
             version = args.version or channel.state["previous"]
             require(version is not None and version in channel.state["installed"], "No retained rollback version.")
diff --git a/tests/python/test_prerelease_install.py b/tests/python/test_prerelease_install.py
index 33a2385c0..5cb254665 100644
--- a/tests/python/test_prerelease_install.py
+++ b/tests/python/test_prerelease_install.py
@@ -8,6 +8,7 @@
 import os
 from pathlib import Path
 import subprocess
+import sys
 import tarfile
 import tempfile
 import unittest
@@ -18,6 +19,7 @@
 SPEC = importlib.util.spec_from_file_location("install_prerelease", SOURCE)
 installer = importlib.util.module_from_spec(SPEC)
 SPEC.loader.exec_module(installer)
+REAL_PROCESS_RUN = subprocess.run
 
 
 def sha(contents):
@@ -140,6 +142,61 @@ def install(self, version="5.0.0-next.1", upgrade=False):
     def selection(self):
         return installer.private_json(self.channel.root / ".selection.json")
 
+    def legacy_channel(self, name="legacy-channel"):
+        source = SOURCE.parent.parent / "tests/fixtures/prerelease-manager-flat-v1.py"
+        self.assertEqual(installer.digest(source),
+                         "b7c49e3fec6b06790e833db1d2dcb441d2223c283b792713be46826aa2eef877")
+        spec = importlib.util.spec_from_file_location("legacy_prerelease_manager", source)
+        legacy = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(legacy)
+        self.channel = legacy.Channel(self.root / name)
+        self.channel.initialize()
+        self.assertEqual((self.channel.root / "bin/hack-next").read_bytes(), installer.launcher_bytes())
+        return legacy
+
+    def retained_snapshot(self):
+        paths = [self.channel.root / ".selection.json", self.channel.root / "bin/hack-next",
+                 *sorted((self.channel.root / "versions").rglob("*"))]
+        return {str(path.relative_to(self.channel.root)): (
+                    path.stat().st_dev, path.stat().st_ino, path.stat().st_mode,
+                    path.read_bytes() if path.is_file() else None)
+                for path in paths}
+
+    def upgrade_manager(self):
+        current = installer.Channel(self.channel.root)
+        with current.lock(manager_upgrade=True):
+            current.upgrade_manager()
+        return current
+
+    def assert_pending_refuses_installed_readers(self, legacy):
+        # Exercise the real predecessor and the exact manager bytes addressed by
+        # the unchanged legacy launcher. Only platform detection is substituted;
+        # refusal must occur before any candidate subprocess can be reached.
+        source = self.channel.root / "manager.py"
+        spec = importlib.util.spec_from_file_location("retained_prerelease_manager", source)
+        retained = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(retained)
+        for module in (legacy, retained, installer):
+            for command in ("status", "run"):
+                with self.subTest(reader=module.__name__, command=command), \
+                        mock.patch.object(module.platform, "system", return_value="Darwin"), \
+                        mock.patch.object(module.platform, "machine", return_value="arm64"), \
+                        mock.patch.object(module.subprocess, "Popen") as spawn:
+                    with self.assertRaisesRegex(module.Refusal, "pending|Foreign installation entry|manager or launcher changed"):
+                        module.main(["--root", str(self.channel.root), command])
+                    spawn.assert_not_called()
+        self.assertEqual((self.channel.root / "bin/hack-next").read_bytes(), installer.launcher_bytes())
+
+        if sys.platform == "darwin" and installer.platform.machine() == "arm64":
+            # On the supported host, run the unchanged shell launcher itself.
+            # A pending transition must fail before codesign or candidate startup.
+            result = REAL_PROCESS_RUN([str(self.channel.root / "bin/hack-next"), "--version"],
+                                      env={"HOME": str(self.stable_home), "PATH": "/usr/bin:/bin"},
+                                      capture_output=True, text=True, timeout=10)
+            self.assertEqual(result.returncode, 1)
+            self.assertEqual(result.stdout, "")
+            self.assertRegex(result.stderr, "pending|Foreign installation entry|manager or launcher changed")
+
     def rejects_install(self, mutate, message):
         if not self.selection()["installed"]:
             self.install()
@@ -276,6 +333,220 @@ def test_actual_legacy_manager_refuses_new_layout_without_losing_status_or_rollb
                 self.assertEqual(legacy.main(["--root", str(self.channel.root), "rollback"]), 0)
             self.assertEqual(json.loads(output.getvalue())["selected"], "5.0.0-next.1")
 
+    def test_explicit_manager_upgrade_preserves_retained_data_and_enables_mcp_rollback(self):
+        self.legacy_channel()
+        self.install()
+        self.install("5.0.0-next.2", True)
+        for version in ("5.0.0-next.1", "5.0.0-next.2"):
+            for home in ("native-home", "cli-home"):
+                (self.channel.root / "versions" / version / home / "marker").write_text(version + home)
+        before = self.retained_snapshot()
+        original_manager = (self.channel.root / "manager.py").read_bytes()
+        self.calls = []
+        with mock.patch.object(installer.platform, "system", return_value="Darwin"), \
+                mock.patch.object(installer.platform, "machine", return_value="arm64"), \
+                contextlib.redirect_stdout(io.StringIO()) as output:
+            self.assertEqual(installer.main(["--root", str(self.channel.root), "upgrade-manager"]), 0)
+        self.assertEqual(json.loads(output.getvalue())["selected"], "5.0.0-next.2")
+        self.assertEqual(self.retained_snapshot(), before)
+        self.assertEqual((self.channel.root / "manager.py").read_bytes(), SOURCE.read_bytes())
+        self.assertEqual(installer.private_json(self.channel.root / ".channel.json")["manager_sha256"],
+                         installer.digest(SOURCE))
+        stage = next(self.channel.root.glob(".stage-*/manager-before.py"))
+        self.assertEqual(stage.read_bytes(), original_manager)
+        self.assertFalse((self.channel.root / installer.MANAGER_UPGRADE).exists())
+        self.assertEqual([args[4] for args, _ in self.calls if args[0] != "/usr/bin/codesign"],
+                         ["status", "down", "status"] * 2)
+        manager_stat = (self.channel.root / "manager.py").stat()
+        self.upgrade_manager()
+        self.assertEqual((self.channel.root / "manager.py").stat(), manager_stat)
+        self.assertEqual(self.retained_snapshot(), before)
+        self.channel = installer.Channel(self.channel.root)
+        archive, checksum = self.archive("5.0.0-next.3", with_mcp)
+        with self.channel.lock():
+            self.channel.install("5.0.0-next.3", archive, checksum, True)
+            self.channel.select("5.0.0-next.1")
+            self.channel.select("5.0.0-next.3")
+        self.assertEqual(self.selection()["selected"], "5.0.0-next.3")
+        for version in ("5.0.0-next.1", "5.0.0-next.2"):
+            for home in ("native-home", "cli-home"):
+                self.assertEqual((self.channel.root / "versions" / version / home / "marker").read_text(), version + home)
+
+    def test_manager_upgrade_refuses_custom_changed_and_aliased_inputs_without_writes(self):
+        for case in ("custom-manager", "changed-manager", "custom-launcher", "manager-symlink",
+                     "manager-hardlink", "receipt-symlink", "unsafe-root"):
+            with self.subTest(case=case):
+                self.legacy_channel(case)
+                self.install()
+                manager = self.channel.root / "manager.py"
+                marker_path = self.channel.root / ".channel.json"
+                marker = installer.private_json(marker_path)
+                if case in ("custom-manager", "changed-manager"):
+                    manager.write_bytes(manager.read_bytes() + b"# custom\n")
+                    if case == "custom-manager":
+                        marker["manager_sha256"] = installer.digest(manager)
+                        marker_path.write_bytes(installer.json_bytes(marker))
+                elif case == "custom-launcher":
+                    launcher = self.channel.root / "bin/hack-next"
+                    launcher.write_bytes(launcher.read_bytes() + b"# custom\n")
+                    marker["launcher_sha256"] = installer.digest(launcher)
+                    marker_path.write_bytes(installer.json_bytes(marker))
+                elif case in ("manager-symlink", "receipt-symlink"):
+                    path = manager if case == "manager-symlink" else marker_path
+                    saved = self.root / (case + "-saved")
+                    path.rename(saved)
+                    path.symlink_to(saved)
+                elif case == "manager-hardlink":
+                    os.link(manager, self.root / "manager-link")
+                else:
+                    self.channel.root.chmod(0o755)
+                before = self.retained_snapshot()
+                entries = sorted(path.name for path in self.channel.root.iterdir())
+                self.calls = []
+                with self.assertRaises(installer.Refusal):
+                    self.upgrade_manager()
+                self.assertEqual(self.retained_snapshot(), before)
+                self.assertEqual(sorted(path.name for path in self.channel.root.iterdir()), entries)
+                self.assertEqual(self.calls, [])
+
+    def test_manager_upgrade_keeps_launcher_and_runtime_quiescence_gates(self):
+        self.legacy_channel()
+        self.install()
+        self.install("5.0.0-next.2", True)
+        manager = (self.channel.root / "manager.py").read_bytes()
+        before = self.retained_snapshot()
+        with self.channel.lock(shared=True):
+            with self.assertRaisesRegex(installer.Refusal, "active"):
+                self.upgrade_manager()
+        stopped = {"phase": "uninitialized", "process_alive": False}
+        for response in ({"phase": "running", "process_alive": True},
+                         {"phase": "unknown", "process_alive": False}, (1, b""), (0, b"malformed"),
+                         subprocess.TimeoutExpired(["status"], 30)):
+            self.runtime_responses = [stopped] * 3 + [response]
+            with self.assertRaises(installer.Refusal):
+                self.upgrade_manager()
+            self.assertEqual((self.channel.root / "manager.py").read_bytes(), manager)
+            self.assertEqual(self.retained_snapshot(), before)
+            self.assertFalse((self.channel.root / installer.MANAGER_UPGRADE).exists())
+
+    def test_real_process_exit_at_each_manager_publication_boundary_is_recoverable(self):
+        for boundary in ("journal", "manager", "receipt", "complete"):
+            with self.subTest(boundary=boundary):
+                legacy = self.legacy_channel("interrupt-" + boundary)
+                self.install()
+                marker = self.channel.root / "versions/5.0.0-next.1/native-home/marker"
+                marker.write_bytes(b"retained data")
+                before = self.retained_snapshot()
+                pid = os.fork()
+                if pid == 0:
+                    original_rename, original_replace, original_unlink = os.rename, os.replace, Path.unlink
+                    def rename(source, destination):
+                        original_rename(source, destination)
+                        if boundary == "journal" and Path(destination).name == installer.MANAGER_UPGRADE:
+                            os._exit(86)
+                    def replace(source, destination):
+                        original_replace(source, destination)
+                        if Path(destination).name == {"manager": "manager.py", "receipt": ".channel.json"}.get(boundary):
+                            os._exit(86)
+                    def unlink(path, *args, **kwargs):
+                        original_unlink(path, *args, **kwargs)
+                        if boundary == "complete" and path.name == installer.MANAGER_UPGRADE:
+                            os._exit(86)
+                    try:
+                        with mock.patch.object(os, "rename", side_effect=rename), \
+                                mock.patch.object(os, "replace", side_effect=replace), \
+                                mock.patch.object(Path, "unlink", new=unlink):
+                            self.upgrade_manager()
+                    finally:
+                        os._exit(87)
+                _, status = os.waitpid(pid, 0)
+                self.assertEqual(os.waitstatus_to_exitcode(status), 86)
+                self.assertEqual(self.retained_snapshot(), before)
+                if boundary != "complete":
+                    self.assert_pending_refuses_installed_readers(legacy)
+                else:
+                    with installer.Channel(self.channel.root).lock():
+                        pass
+                self.upgrade_manager()
+                self.assertEqual(self.retained_snapshot(), before)
+                self.assertEqual((self.channel.root / "manager.py").read_bytes(), SOURCE.read_bytes())
+                self.assertFalse((self.channel.root / installer.MANAGER_UPGRADE).exists())
+                with installer.Channel(self.channel.root).lock():
+                    pass
+
+    def test_manager_staging_failure_leaves_actual_legacy_status_usable(self):
+        legacy = self.legacy_channel()
+        self.install()
+        before = self.retained_snapshot()
+        manager = (self.channel.root / "manager.py").read_bytes()
+        original = installer.write_file
+        def fail(path, contents, mode=0o600):
+            if path.name == "manager-after.py":
+                raise OSError("injected staging failure")
+            return original(path, contents, mode)
+        with mock.patch.object(installer, "write_file", side_effect=fail):
+            with self.assertRaisesRegex(OSError, "staging failure"):
+                self.upgrade_manager()
+        self.assertEqual(self.retained_snapshot(), before)
+        self.assertEqual((self.channel.root / "manager.py").read_bytes(), manager)
+        self.assertFalse((self.channel.root / installer.MANAGER_UPGRADE).exists())
+        with mock.patch.object(legacy.platform, "system", return_value="Darwin"), \
+                mock.patch.object(legacy.platform, "machine", return_value="arm64"), \
+                contextlib.redirect_stdout(io.StringIO()) as output:
+            self.assertEqual(legacy.main(["--root", str(self.channel.root), "status"]), 0)
+        self.assertEqual(json.loads(output.getvalue())["selected"], "5.0.0-next.1")
+        self.upgrade_manager()
+        self.assertEqual(self.retained_snapshot(), before)
+
+    def test_manager_recovery_refuses_changed_journal_stage_selection_and_unordered_state(self):
+        for case in ("new-bytes", "old-bytes", "other-installer", "bad-digest-type",
+                     "stage-symlink", "stage-extra", "journal-symlink", "journal-hardlink",
+                     "selection", "unordered-receipt", "changed-bundle"):
+            with self.subTest(case=case):
+                self.legacy_channel("recovery-" + case)
+                self.install()
+                with mock.patch.object(installer, "atomic_file", side_effect=OSError("before manager publication")):
+                    with self.assertRaisesRegex(OSError, "before manager publication"):
+                        self.upgrade_manager()
+                journal = self.channel.root / installer.MANAGER_UPGRADE
+                plan = installer.private_json(journal)
+                stage = self.channel.root / plan["stage"]
+                if case in ("new-bytes", "old-bytes"):
+                    path = stage / ("manager-after.py" if case == "new-bytes" else "manager-before.py")
+                    path.write_bytes(path.read_bytes() + b"# changed\n")
+                elif case in ("other-installer", "bad-digest-type"):
+                    plan["to_sha256"] = "0" * 64 if case == "other-installer" else {}
+                    journal.write_bytes(installer.json_bytes(plan))
+                elif case in ("stage-symlink", "journal-symlink"):
+                    path = stage if case == "stage-symlink" else journal
+                    saved = self.root / (case + "-saved")
+                    path.rename(saved)
+                    path.symlink_to(saved)
+                elif case == "stage-extra":
+                    (stage / "foreign").write_bytes(b"not owned")
+                elif case == "journal-hardlink":
+                    os.link(journal, self.root / "journal-alias")
+                elif case == "selection":
+                    state = self.selection()
+                    state.update(selected=None, previous=state["selected"])
+                    (self.channel.root / ".selection.json").write_bytes(installer.json_bytes(state))
+                elif case == "unordered-receipt":
+                    (self.channel.root / ".channel.json").write_bytes((stage / "channel-after.json").read_bytes())
+                else:
+                    (self.channel.root / "versions/5.0.0-next.1/bundle/hack-cli").chmod(0o700)
+                    (self.channel.root / "versions/5.0.0-next.1/bundle/hack-cli").write_bytes(b"changed")
+                before = self.retained_snapshot()
+                manager_before = (self.channel.root / "manager.py").read_bytes()
+                channel_before = (self.channel.root / ".channel.json").read_bytes()
+                self.calls = []
+                with self.assertRaises(installer.Refusal):
+                    self.upgrade_manager()
+                self.assertEqual(self.calls, [])
+                self.assertEqual(self.retained_snapshot(), before)
+                self.assertEqual((self.channel.root / "manager.py").read_bytes(), manager_before)
+                self.assertEqual((self.channel.root / ".channel.json").read_bytes(), channel_before)
+                self.assertTrue(os.path.lexists(journal))
+
     def test_active_unknown_failed_and_timed_out_status_never_trigger_down(self):
         self.install()
         cases = [{"phase": "running", "process_alive": True},