diff --git a/tools/gear-ownership.py b/tools/gear-ownership.py
index fa7ec6d..5ce0ff8 100644
--- a/tools/gear-ownership.py
+++ b/tools/gear-ownership.py
@@ -3,6 +3,7 @@
 import argparse
 import base64
 import copy
+from datetime import datetime, timezone
 from decimal import Decimal
 import hashlib
 import json
@@ -16,6 +17,11 @@
 
 import yaml
 
+LEGACY_EMPTY_PRODUCERS = {"d864c30aa6c01453e38bc2539d0ccea889b846aa47e9050982026e894a2e8247": "Magic 0.5.2"}
+LEGACY_EMPTY_CANONICAL = b"stations: {}\n"
+LEGACY_EMPTY_STAGING = ".gear-legacy-empty-"
+LEGACY_EMPTY_HISTORY = ("gear-ownership", "gear-returns")
+
 
 class GearLoader(yaml.SafeLoader):
     def construct_mapping(self, node, deep=False):
@@ -472,6 +478,155 @@ def backup_gear(root, backup):
     return original
 
 
+def _legacy_empty_read(path, label):
+    info = os.lstat(path)
+    if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode):
+        raise ValueError(label + " must be a regular nonsymlink file")
+    descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
+    try:
+        before = os.fstat(descriptor)
+        if not stat.S_ISREG(before.st_mode):
+            raise ValueError(label + " must be a regular file")
+        chunks = []
+        while chunk := os.read(descriptor, 65536):
+            chunks.append(chunk)
+        after = os.fstat(descriptor)
+        fields = ("st_dev", "st_ino", "st_size", "st_mode", "st_uid", "st_gid", "st_mtime_ns", "st_ctime_ns", "st_nlink")
+        if any(getattr(before, field) != getattr(after, field) for field in fields):
+            raise ValueError(label + " changed while being read")
+        return after, b"".join(chunks)
+    finally:
+        os.close(descriptor)
+
+
+def _legacy_empty_state(root, legacy_jar, own_staging=None):
+    root = Path(root)
+    if not root.is_absolute() or Path(os.path.realpath(root)) != root or not stat.S_ISDIR(os.lstat(root).st_mode):
+        raise ValueError("--data must be an absolute, symlink-free directory")
+    if legacy_jar is None:
+        raise ValueError("legacy-empty import requires --legacy-jar")
+    _, jar_bytes = _legacy_empty_read(Path(legacy_jar), "legacy JAR")
+    producer = digest(jar_bytes)
+    if producer not in LEGACY_EMPTY_PRODUCERS:
+        raise ValueError("legacy JAR is not a reviewed empty-save producer")
+    for name in sorted(os.listdir(root)):
+        if own_staging is not None and name == own_staging.name:
+            continue
+        if name.startswith("gear-stations") and name != "gear-stations.yml":
+            raise ValueError("station sibling needs staff review")
+        if name.startswith("magic-") and name.endswith(".tmp") or name.startswith(LEGACY_EMPTY_STAGING):
+            raise ValueError("staging evidence needs staff review")
+    for name in LEGACY_EMPTY_HISTORY:
+        try:
+            os.lstat(root / name)
+        except FileNotFoundError:
+            continue
+        raise ValueError("ownership/return history exists (NOFOLLOW): " + name)
+    station = root / "gear-stations.yml"
+    try:
+        info, contents = _legacy_empty_read(station, "station")
+    except FileNotFoundError as error:
+        raise ValueError("station file is absent; legacy-empty import does not apply") from error
+    if info.st_nlink != 1:
+        raise ValueError("station file has other hard links")
+    record = {
+        "data": str(root), "producer": LEGACY_EMPTY_PRODUCERS[producer], "producer-sha256": producer,
+        "station": {"dev": info.st_dev, "ino": info.st_ino, "size": info.st_size, "sha256": digest(contents),
+                    "mode": oct(stat.S_IMODE(info.st_mode)), "uid": info.st_uid, "gid": info.st_gid,
+                    "mtime-ns": info.st_mtime_ns, "ctime-ns": info.st_ctime_ns, "nlink": info.st_nlink},
+        "history-absent": list(LEGACY_EMPTY_HISTORY), "target-sha256": digest(LEGACY_EMPTY_CANONICAL),
+    }
+    if contents == LEGACY_EMPTY_CANONICAL and info.st_size == len(LEGACY_EMPTY_CANONICAL):
+        return record, None, "ALREADY_CANONICAL"
+    if contents != b"" or info.st_size != 0:
+        raise ValueError("station file is not exactly zero bytes; ordinary strict recovery applies")
+    return record, digest(json.dumps(record, sort_keys=True).encode()), "ELIGIBLE_REQUIRES_LINEAGE_ATTESTATION"
+
+
+def legacy_empty_import(root, legacy_jar, apply=False, expect_baseline=None, attest=None, backup=None):
+    """Explicit stopped import; file state cannot establish the required historical lineage."""
+    report, staged, staged_inode, replace_started = {}, None, None, False
+    try:
+        root = Path(root)
+        record, baseline, verdict = _legacy_empty_state(root, legacy_jar)
+        report.update(verdict=verdict, baseline=baseline, record=record)
+        if not apply or verdict == "ALREADY_CANONICAL":
+            return report
+        if attest is None or backup is None or not isinstance(expect_baseline, str) or not re.fullmatch(r"[0-9a-f]{64}", expect_baseline):
+            raise ValueError("apply requires --attest FILE, --expect-baseline HEX and a new --backup")
+        if baseline != expect_baseline:
+            raise ValueError("current state differs from the attested baseline")
+        _, attestation = _legacy_empty_read(Path(attest), "attestation")
+        if not attestation.strip() or not re.search(rb"(?<![0-9a-fA-F])" + baseline.encode() + rb"(?![0-9a-fA-F])", attestation):
+            raise ValueError("attestation must contain the exact baseline digest and lineage evidence")
+        backup_path = Path(backup)
+        if os.path.lexists(backup_path):
+            raise ValueError("recovery backup must be a new directory")
+        backup_path = backup_path.resolve()
+        if backup_path == root or root in backup_path.parents:
+            raise ValueError("recovery backup must be outside Magic data")
+        force(root)
+        original = backup_gear(root, backup_path)
+        if original != {"gear-stations.yml": digest(b"")} or _legacy_empty_state(root, legacy_jar)[1] != baseline:
+            raise ValueError("gear changed during backup; publication refused")
+        durable(backup_path / "LINEAGE-ATTESTATION", attestation)
+        plan = {"baseline": baseline, "record": record, "attestation-sha256": digest(attestation),
+                "attestation-file": "LINEAGE-ATTESTATION", "tool-sha256": digest(Path(__file__).read_bytes()),
+                "created-at-utc": datetime.now(timezone.utc).isoformat()}
+        durable(backup_path / "IMPORT-PLAN.json", json.dumps(plan, indent=2, sort_keys=True).encode())
+        staged = root / (LEGACY_EMPTY_STAGING + baseline[:16] + ".tmp")
+        descriptor = os.open(staged, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
+        try:
+            info = os.fstat(descriptor)
+            staged_inode = (info.st_dev, info.st_ino)
+            pending = LEGACY_EMPTY_CANONICAL
+            while pending:
+                written = os.write(descriptor, pending)
+                if written <= 0:
+                    raise OSError("incomplete legacy-empty staging write")
+                pending = pending[written:]
+            info = os.fstat(descriptor)
+            if (info.st_uid, info.st_gid) != (record["station"]["uid"], record["station"]["gid"]):
+                os.fchown(descriptor, record["station"]["uid"] if info.st_uid != record["station"]["uid"] else -1,
+                          record["station"]["gid"] if info.st_gid != record["station"]["gid"] else -1)
+            os.fchmod(descriptor, int(record["station"]["mode"], 8))
+            os.fsync(descriptor)
+        finally:
+            os.close(descriptor)
+        if _legacy_empty_state(root, legacy_jar, staged)[1] != baseline:
+            raise ValueError("station or history changed before publication")
+        info, contents = _legacy_empty_read(staged, "staging")
+        if (info.st_dev, info.st_ino) != staged_inode or info.st_nlink != 1 or contents != LEGACY_EMPTY_CANONICAL or stat.S_IMODE(info.st_mode) != int(record["station"]["mode"], 8) or (info.st_uid, info.st_gid) != (record["station"]["uid"], record["station"]["gid"]):
+            raise ValueError("staging changed before publication")
+        # The final recheck/rename interval requires the stopped server and deployment lock.
+        replace_started = True
+        os.replace(staged, root / "gear-stations.yml")
+        force(root)
+        final, _, verdict = _legacy_empty_state(root, legacy_jar)
+        if verdict != "ALREADY_CANONICAL" or any(final["station"][field] != record["station"][field] for field in ("mode", "uid", "gid")):
+            raise ValueError("post-publication verification failed")
+        report.update(verdict="IMPORTED", published=final["station"])
+        durable(backup_path / "IMPORT-RESULT.json", json.dumps(report, indent=2, sort_keys=True).encode())
+    except (OSError, ValueError) as error:
+        report.update(verdict="PUBLISHED_UNCONFIRMED" if replace_started else "REFUSED", error=str(error))
+        if replace_started:
+            report["recovery"] = "Keep Paper stopped, preserve the backup and rerun preview"
+    finally:
+        if staged_inode is not None:
+            try:
+                info = os.lstat(staged)
+                if (info.st_dev, info.st_ino) != staged_inode:
+                    raise ValueError("staging path changed; preserve it for staff review")
+                os.unlink(staged)
+            except FileNotFoundError:
+                pass
+            except (OSError, ValueError) as error:
+                report["verdict"] = "PUBLISHED_UNCONFIRMED" if replace_started else "REFUSED"
+                report["staging-cleanup-error"] = str(error)
+                report["recovery"] = "Keep Paper stopped and preserve staging evidence for staff review"
+    return report
+
+
 def archive(path, destination):
     destination.parent.mkdir(parents=True, exist_ok=True)
     force(destination.parent)
@@ -549,12 +704,27 @@ def inspect(root, kind="estates", selected=None, apply=False, backup=None):
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--data", type=Path, required=True, help="plugins/Magic/data; Paper must be stopped")
-    parser.add_argument("--kind", choices=("migration", "estates"), default="estates")
+    parser.add_argument("--kind", choices=("migration", "estates"))
     parser.add_argument("--journal", help="one migration run or estate death UUID; required with apply")
     parser.add_argument("--apply", action="store_true")
     parser.add_argument("--backup", type=Path, help="new external gear recovery directory")
+    parser.add_argument("--legacy-empty-import", action="store_true", help="stopped, attested import of an exact old empty save")
+    parser.add_argument("--legacy-jar", type=Path, help="reviewed pre-ownership Magic JAR")
+    parser.add_argument("--expect-baseline", help="exact digest from legacy-empty preview")
+    parser.add_argument("--attest", type=Path, help="file with the baseline digest and checkpoint/lineage evidence")
     args = parser.parse_args()
-    print(json.dumps(inspect(args.data, args.kind, args.journal, args.apply, args.backup), indent=2))
+    if args.legacy_empty_import:
+        if args.kind is not None or args.journal is not None:
+            report = {"verdict": "REFUSED", "error": "--legacy-empty-import cannot be combined with --kind or --journal"}
+        else:
+            report = legacy_empty_import(args.data, args.legacy_jar, args.apply, args.expect_baseline, args.attest, args.backup)
+        print(json.dumps(report, indent=2))
+        if report["verdict"] in ("REFUSED", "PUBLISHED_UNCONFIRMED"):
+            raise SystemExit(2)
+    else:
+        if args.legacy_jar is not None or args.expect_baseline is not None or args.attest is not None:
+            parser.error("--legacy-jar, --expect-baseline and --attest require --legacy-empty-import")
+        print(json.dumps(inspect(args.data, args.kind or "estates", args.journal, args.apply, args.backup), indent=2))
 
 
 if __name__ == "__main__":
diff --git a/tools/test_gear_ownership.py b/tools/test_gear_ownership.py
index a3e73c1..7c13d39 100644
--- a/tools/test_gear_ownership.py
+++ b/tools/test_gear_ownership.py
@@ -1,9 +1,16 @@
 import base64
 import copy
+import contextlib
 import importlib.util
+import io
 import json
+import os
 from pathlib import Path
 import tempfile
+import time
+import stat
+import sys
+from types import SimpleNamespace
 import unittest
 from unittest.mock import patch
 import uuid
@@ -369,5 +376,530 @@ def test_native_world_and_java_integer_selector_constraints_are_mirrored(self):
             self.assertFalse(self.backup.exists())
 
 
+class LegacyEmptyProducerPolicyTests(unittest.TestCase):
+    def test_shipped_allowlist_is_exact_reviewed_preownership_producer(self):
+        unpatched_spec = importlib.util.spec_from_file_location("shipped_gear_ownership", Path(__file__).with_name("gear-ownership.py"))
+        shipped = importlib.util.module_from_spec(unpatched_spec)
+        unpatched_spec.loader.exec_module(shipped)
+        self.assertEqual({"d864c30aa6c01453e38bc2539d0ccea889b846aa47e9050982026e894a2e8247": "Magic 0.5.2"}, shipped.LEGACY_EMPTY_PRODUCERS)
+        self.assertNotIn("fb40f229b4b79fd1a6cd10c196d9a9a7d6dc40db4e97d4a7d6806b1f645002af", shipped.LEGACY_EMPTY_PRODUCERS)
+
+
+class LegacyEmptyImportTests(unittest.TestCase):
+    # Exact named output of the retained actual 0.5.2 save probe; no external JAR needed by this suite.
+    OLD_0_5_2_EMPTY_SAVE = b""
+
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.base = Path(self.temp.name)
+        self.root = self.base / "plugins/Magic/data"
+        self.root.mkdir(parents=True)
+        (self.root / "characters").mkdir()
+        self.station = self.root / "gear-stations.yml"
+        self.station.write_bytes(self.OLD_0_5_2_EMPTY_SAVE)
+        self.station.chmod(0o640)
+        self.profile = self.root / "characters/unchanged.yml"
+        self.profile.write_bytes(b"focus: keep unchanged\n")
+        self.jar = self.base / "legacy.jar"
+        self.jar.write_bytes(b"isolated test producer; actual release proved separately")
+        # Isolate file-hash validation without licensing/external artifacts in normal CI.
+        producer = patch.dict(gear.LEGACY_EMPTY_PRODUCERS, {gear.digest(self.jar.read_bytes()): "Magic 0.5.2"}, clear=True)
+        producer.start()
+        self.addCleanup(producer.stop)
+        self.backup = self.base / "external/recovery"
+        self.attest = self.base / "lineage.txt"
+
+    def snapshot(self):
+        rows = {}
+        for path in sorted(self.root.rglob("*")):
+            info = os.lstat(path)
+            data = path.read_bytes() if stat.S_ISREG(info.st_mode) else os.readlink(path) if stat.S_ISLNK(info.st_mode) else None
+            stamp = (info.st_mtime_ns, info.st_ctime_ns) if stat.S_ISREG(info.st_mode) else None
+            rows[str(path.relative_to(self.root))] = (info.st_dev, info.st_ino, info.st_mode, info.st_uid, info.st_gid, info.st_nlink, stamp, data)
+        return rows
+
+    def preview(self, **kwargs):
+        return gear.legacy_empty_import(self.root, self.jar, **kwargs)
+
+    def prepared(self):
+        baseline = self.preview()["baseline"]
+        self.attest.write_text("Synthetic stopped checkpoint and old-only lineage evidence\nBaseline: " + baseline + "\n")
+        return dict(apply=True, expect_baseline=baseline, attest=self.attest, backup=self.backup)
+
+    def apply(self, **overrides):
+        args = self.prepared()
+        args.update(overrides)
+        return self.preview(**args)
+
+    def refused(self, call, reason=None):
+        before = self.snapshot()
+        report = call()
+        self.assertEqual("REFUSED", report["verdict"], report)
+        if reason:
+            self.assertIn(reason, report["error"])
+        self.assertEqual(before, self.snapshot(), "refusal must preserve source bytes and inodes")
+        return report
+
+    def cli(self, *args):
+        output, error = io.StringIO(), io.StringIO()
+        code = 0
+        with patch.object(sys, "argv", ["gear-ownership.py", "--data", str(self.root), *map(str, args)]), contextlib.redirect_stdout(output), contextlib.redirect_stderr(error):
+            try:
+                gear.main()
+            except SystemExit as stopped:
+                code = stopped.code
+        self.assertNotIn("Traceback", output.getvalue() + error.getvalue())
+        return code, json.loads(output.getvalue()) if output.getvalue() else None
+
+    def test_actual_old_empty_fixture_preview_is_inert_and_binds_metadata(self):
+        before = self.snapshot()
+        report = self.preview()
+        self.assertEqual("ELIGIBLE_REQUIRES_LINEAGE_ATTESTATION", report["verdict"])
+        self.assertEqual(gear.digest(b""), report["record"]["station"]["sha256"])
+        self.assertEqual({"dev", "ino", "size", "sha256", "mode", "uid", "gid", "mtime-ns", "ctime-ns", "nlink"}, set(report["record"]["station"]))
+        self.assertEqual(["gear-ownership", "gear-returns"], report["record"]["history-absent"])
+        self.assertEqual("8a4c1c4742d8bc8be8c84c2f91a2b5337a275b0caa0acf9f4ac88c7628fb5619", report["record"]["target-sha256"])
+        self.assertEqual(before, self.snapshot())
+        self.assertFalse(self.backup.exists())
+
+    def test_apply_publishes_only_canonical_empty_and_durable_evidence(self):
+        args = self.prepared()
+        before = self.snapshot()
+        with patch.object(gear.os, "open", wraps=os.open) as opens:
+            report = self.preview(**args)
+        self.assertEqual("IMPORTED", report["verdict"], report)
+        self.assertEqual(b"stations: {}\n", self.station.read_bytes())
+        self.assertEqual(0o640, stat.S_IMODE(self.station.stat().st_mode))
+        self.assertEqual(before["characters/unchanged.yml"], self.snapshot()["characters/unchanged.yml"])
+        self.assertEqual({"characters", "gear-stations.yml"}, set(p.name for p in self.root.iterdir()))
+        self.assertEqual(b"", (self.backup / "data/gear-stations.yml").read_bytes())
+        self.assertEqual({"gear-stations.yml": gear.digest(b"")}, json.loads((self.backup / "SHA256.json").read_text()))
+        self.assertEqual(self.attest.read_bytes(), (self.backup / "LINEAGE-ATTESTATION").read_bytes())
+        plan = json.loads((self.backup / "IMPORT-PLAN.json").read_text())
+        self.assertEqual(args["expect_baseline"], plan["baseline"])
+        self.assertEqual(gear.digest(self.attest.read_bytes()), plan["attestation-sha256"])
+        self.assertEqual(gear.digest(Path(gear.__file__).read_bytes()), plan["tool-sha256"])
+        self.assertTrue(plan["created-at-utc"].endswith("+00:00"))
+        self.assertEqual(report, json.loads((self.backup / "IMPORT-RESULT.json").read_text()))
+        flags = next(call.args[1] for call in opens.call_args_list if Path(call.args[0]).name.startswith(gear.LEGACY_EMPTY_STAGING))
+        self.assertEqual(os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, flags & (os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW))
+
+    def test_already_canonical_replay_creates_no_backup_or_new_publication(self):
+        self.assertEqual("IMPORTED", self.apply()["verdict"])
+        before = self.snapshot()
+        second = self.base / "second-recovery"
+        report = self.preview(apply=True, expect_baseline="stale", backup=second)
+        self.assertEqual("ALREADY_CANONICAL", report["verdict"])
+        self.assertEqual(before, self.snapshot())
+        self.assertFalse(second.exists())
+        for legacy_args in [(), ("--kind", "migration"), ("--kind", "estates")]:
+            code, report = self.cli(*legacy_args)
+            self.assertEqual(0, code)
+            self.assertEqual({"journals": []}, report)
+
+    @staticmethod
+    def ownership_view(info, **changed):
+        fields = {name: getattr(info, name) for name in dir(info) if name.startswith("st_")}
+        fields.update(changed)
+        return SimpleNamespace(**fields)
+
+    def test_matching_ownership_never_calls_chown_including_mirrored_views(self):
+        owner = self.station.stat()
+        with patch.object(gear.os, "fchown", side_effect=PermissionError("matching bindfs ownership must not be changed")) as chown:
+            report = self.apply()
+        self.assertEqual("IMPORTED", report["verdict"], report)
+        chown.assert_not_called()
+        self.assertEqual((owner.st_uid, owner.st_gid), (report["published"]["uid"], report["published"]["gid"]))
+        self.assertEqual((owner.st_uid, owner.st_gid), (self.station.stat().st_uid, self.station.stat().st_gid))
+
+    def test_different_staging_owner_is_restored_before_mode_and_fsync(self):
+        real_open, real_stat, real_chown, real_chmod, real_fsync = os.open, os.fstat, os.fchown, os.fchmod, os.fsync
+        for uid_differs, gid_differs in [(True, False), (False, True), (True, True)]:
+            with self.subTest(uid_differs=uid_differs, gid_differs=gid_differs):
+                self.station.write_bytes(b"")
+                self.backup = self.base / ("recovery-" + str(uid_differs) + "-" + str(gid_differs))
+                owner = self.station.stat()
+                inode, observed, events = [], [], []
+                def opened(path, flags, *args):
+                    fd = real_open(path, flags, *args)
+                    if Path(path).name.startswith(gear.LEGACY_EMPTY_STAGING) and flags & os.O_CREAT:
+                        info = real_stat(fd)
+                        inode.append((info.st_dev, info.st_ino))
+                    return fd
+                def statted(fd):
+                    info = real_stat(fd)
+                    if inode and (info.st_dev, info.st_ino) == inode[0] and info.st_size == len(gear.LEGACY_EMPTY_CANONICAL) and not observed:
+                        observed.append(True)
+                        return self.ownership_view(info, st_uid=owner.st_uid + int(uid_differs), st_gid=owner.st_gid + int(gid_differs))
+                    return info
+                def chowned(fd, uid, gid):
+                    self.assertEqual((owner.st_uid if uid_differs else -1, owner.st_gid if gid_differs else -1), (uid, gid))
+                    events.append("chown")
+                    return real_chown(fd, uid, gid)
+                def chmodded(fd, mode):
+                    events.append("chmod")
+                    return real_chmod(fd, mode)
+                def synced(fd):
+                    info = real_stat(fd)
+                    if inode and (info.st_dev, info.st_ino) == inode[0]: events.append("fsync")
+                    return real_fsync(fd)
+                with patch.object(gear.os, "open", side_effect=opened), patch.object(gear.os, "fstat", side_effect=statted), patch.object(gear.os, "fchown", side_effect=chowned), patch.object(gear.os, "fchmod", side_effect=chmodded), patch.object(gear.os, "fsync", side_effect=synced):
+                    report = self.apply()
+                self.assertEqual("IMPORTED", report["verdict"], report)
+                self.assertEqual(["chown", "chmod", "fsync"], events)
+                self.assertEqual((owner.st_uid, owner.st_gid), (report["published"]["uid"], report["published"]["gid"]))
+                self.assertEqual(stat.S_IMODE(owner.st_mode), stat.S_IMODE(self.station.stat().st_mode))
+
+    def test_silently_ignored_chown_cannot_publish_disagreeing_ownership(self):
+        real_stat = os.fstat
+        for field in ["st_uid", "st_gid"]:
+            with self.subTest(field=field):
+                self.backup = self.base / ("ignored-" + field)
+                owner = self.station.stat()
+                good = self.prepared()
+                def statted(fd):
+                    info = real_stat(fd)
+                    if stat.S_ISREG(info.st_mode) and info.st_size == len(gear.LEGACY_EMPTY_CANONICAL):
+                        return self.ownership_view(info, **{field: getattr(info, field) + 1})
+                    return info
+                with patch.object(gear.os, "fstat", side_effect=statted), patch.object(gear.os, "fchown", return_value=None) as chown:
+                    self.refused(lambda: self.preview(**good), "staging changed")
+                chown.assert_called_once()
+                self.assertEqual((owner.st_uid if field == "st_uid" else -1, owner.st_gid if field == "st_gid" else -1), chown.call_args.args[1:])
+                self.assertTrue((self.backup / "IMPORT-PLAN.json").exists())
+                self.assertFalse((self.backup / "IMPORT-RESULT.json").exists())
+                self.assertFalse(list(self.root.glob(gear.LEGACY_EMPTY_STAGING + "*")))
+
+    def test_unavailable_chown_refuses_without_replacing_original(self):
+        good = self.prepared()
+        real_stat = os.fstat
+        def statted(fd):
+            info = real_stat(fd)
+            if stat.S_ISREG(info.st_mode) and info.st_size == len(gear.LEGACY_EMPTY_CANONICAL) and stat.S_IMODE(info.st_mode) == 0o600:
+                return self.ownership_view(info, st_uid=info.st_uid + 1)
+            return info
+        with patch.object(gear.os, "fstat", side_effect=statted), patch.object(gear.os, "fchown", side_effect=PermissionError("ownership is not permitted")) as chown:
+            self.refused(lambda: self.preview(**good), "ownership is not permitted")
+        chown.assert_called_once()
+        self.assertTrue((self.backup / "IMPORT-PLAN.json").is_file())
+        self.assertFalse((self.backup / "IMPORT-RESULT.json").exists())
+        self.assertFalse(list(self.root.glob(gear.LEGACY_EMPTY_STAGING + "*")))
+
+    def test_staged_owner_or_group_disagreement_refuses_before_rename(self):
+        real_read = gear._legacy_empty_read
+        for field in ["st_uid", "st_gid"]:
+            with self.subTest(field=field):
+                import shutil
+                if self.backup.exists(): shutil.rmtree(self.backup)
+                self.station.write_bytes(b"")
+                good = self.prepared()
+                def read(path, label):
+                    info, data = real_read(path, label)
+                    if label == "staging": info = self.ownership_view(info, **{field: getattr(info, field) + 1})
+                    return info, data
+                with patch.object(gear, "_legacy_empty_read", side_effect=read), patch.object(gear.os, "replace", wraps=os.replace) as replace:
+                    self.refused(lambda: self.preview(**good), "staging changed")
+                self.assertFalse(any(call.args[1] == self.station for call in replace.call_args_list))
+                self.assertTrue((self.backup / "IMPORT-PLAN.json").is_file())
+                self.assertFalse((self.backup / "IMPORT-RESULT.json").exists())
+                self.assertFalse(list(self.root.glob(gear.LEGACY_EMPTY_STAGING + "*")))
+                import shutil
+                shutil.rmtree(self.backup)
+
+    def test_published_owner_or_group_disagreement_is_unconfirmed(self):
+        real_read, real_replace = gear._legacy_empty_read, os.replace
+        for field in ["st_uid", "st_gid"]:
+            with self.subTest(field=field):
+                import shutil
+                if self.backup.exists(): shutil.rmtree(self.backup)
+                self.station.write_bytes(b"")
+                good = self.prepared()
+                replaced = []
+                def replace(source, target):
+                    result = real_replace(source, target)
+                    if target == self.station: replaced.append(True)
+                    return result
+                def read(path, label):
+                    info, data = real_read(path, label)
+                    if replaced and label == "station": info = self.ownership_view(info, **{field: getattr(info, field) + 1})
+                    return info, data
+                with patch.object(gear, "_legacy_empty_read", side_effect=read), patch.object(gear.os, "replace", side_effect=replace):
+                    report = self.preview(**good)
+                self.assertEqual("PUBLISHED_UNCONFIRMED", report["verdict"], report)
+                self.assertIn("post-publication verification failed", report["error"])
+                self.assertIn("Keep Paper stopped", report["recovery"])
+                self.assertEqual(gear.LEGACY_EMPTY_CANONICAL, self.station.read_bytes())
+                self.assertTrue((self.backup / "IMPORT-PLAN.json").is_file())
+                self.assertFalse((self.backup / "IMPORT-RESULT.json").exists())
+                self.assertFalse(list(self.root.glob(gear.LEGACY_EMPTY_STAGING + "*")))
+                import shutil
+                shutil.rmtree(self.backup)
+                self.station.write_bytes(b"")
+
+    def test_nonempty_or_noncanonical_documents_refuse_without_loosening_yaml(self):
+        for data in [b"\n", b" \n\t\n", b"~\n", b"stations:\n", b"{}\n", b"stations: {}", b"stations: {}\nstations: {}\n", b"\xff"]:
+            with self.subTest(data=data):
+                self.station.write_bytes(data)
+                self.refused(self.preview, "not exactly zero bytes")
+        for data in [b"", b"~\n", b"[]\n", b"stations: {}\nstations: {}\n"]:
+            with self.assertRaises((ValueError, yaml.YAMLError)):
+                gear.raw_yaml(data)
+
+    def test_station_and_data_path_shapes_refuse_without_following_aliases(self):
+        self.station.unlink()
+        self.refused(self.preview, "absent")
+        target = self.base / "empty-target"
+        target.write_bytes(b"")
+        for linked in [target, self.base / "missing-target"]:
+            self.station.symlink_to(linked)
+            self.refused(self.preview, "nonsymlink")
+            self.station.unlink()
+        for content in [b"", gear.LEGACY_EMPTY_CANONICAL]:
+            target.write_bytes(content)
+            os.link(target, self.station)
+            self.refused(self.preview, "hard links")
+            self.station.unlink()
+        self.station.mkdir()
+        self.refused(self.preview, "regular")
+        self.station.rmdir()
+        os.mkfifo(self.station)
+        self.refused(self.preview, "regular")
+        self.station.unlink()
+        self.station.write_bytes(b"")
+        alias = self.base / "alias"
+        alias.symlink_to(self.root.parent.parent, target_is_directory=True)
+        for path in [Path("relative/data"), alias / "Magic/data", self.station]:
+            self.refused(lambda: gear.legacy_empty_import(path, self.jar), "directory")
+
+    def test_history_containers_hold_empty_malformed_and_dangling_states(self):
+        for name in gear.LEGACY_EMPTY_HISTORY:
+            path = self.root / name
+            for kind in ["dir", "file", "dangling", "link-dir"]:
+                with self.subTest(name=name, kind=kind):
+                    if kind == "dir": path.mkdir()
+                    elif kind == "file": path.write_bytes(b"malformed history")
+                    else: path.symlink_to(self.base / "missing" if kind == "dangling" else self.base, target_is_directory=True)
+                    self.refused(self.preview, "history exists")
+                    path.rmdir() if kind == "dir" else path.unlink()
+
+    def test_station_siblings_and_all_staging_shapes_hold(self):
+        for name in ["gear-stations.yml.rejected-anything", "gear-stations.yml.bak", "gear-stations-other", "magic-1.tmp", gear.LEGACY_EMPTY_STAGING + "old.tmp"]:
+            for kind in ["file", "dir", "dangling"]:
+                with self.subTest(name=name, kind=kind):
+                    path = self.root / name
+                    if kind == "file": path.write_bytes(b"unknown evidence")
+                    elif kind == "dir": path.mkdir()
+                    else: path.symlink_to(self.base / "missing")
+                    self.refused(self.preview, "staff review")
+                    path.rmdir() if kind == "dir" else path.unlink()
+
+    def test_producer_is_required_allowlisted_and_not_a_link_or_special_file(self):
+        other = self.base / "other.jar"
+        other.write_bytes(b"nonallowlisted ownership build")
+        self.refused(lambda: gear.legacy_empty_import(self.root, other), "reviewed")
+        self.refused(lambda: gear.legacy_empty_import(self.root, None), "--legacy-jar")
+        link = self.base / "linked.jar"
+        link.symlink_to(self.jar)
+        self.refused(lambda: gear.legacy_empty_import(self.root, link), "nonsymlink")
+        code, report = self.cli("--legacy-empty-import")
+        self.assertEqual(2, code)
+        self.assertEqual("REFUSED", report["verdict"])
+
+    def test_apply_needs_exact_baseline_and_a_nonempty_attestation_file(self):
+        good = self.prepared()
+        for changed in [{"attest": None}, {"backup": None}, {"expect_baseline": None}, {"expect_baseline": "x" * 64}, {"expect_baseline": "0" * 64}]:
+            self.refused(lambda: self.preview(**(good | changed)))
+        for data in [b"", b" \n", b"unrelated old checkpoint", b"f" + good["expect_baseline"].encode() + b"f"]:
+            self.attest.write_bytes(data)
+            self.refused(lambda: self.preview(**good), "exact baseline")
+        self.attest.unlink()
+        self.refused(lambda: self.preview(**good))
+        self.assertFalse(self.backup.exists())
+
+    def test_touch_and_same_bytes_old_writer_restarts_break_the_baseline(self):
+        for mutation in [lambda: os.utime(self.station, ns=(1, 1)), lambda: self.station.write_bytes(b""), lambda: (self.station.unlink(), self.station.write_bytes(b""), self.station.chmod(0o640))]:
+            os.utime(self.station, ns=(2, 2))
+            good = self.prepared()
+            mutation()
+            self.assertNotEqual(2, self.station.stat().st_mtime_ns)
+            self.assertEqual(0o640, stat.S_IMODE(self.station.stat().st_mode))
+            self.refused(lambda: self.preview(**good), "attested baseline")
+            self.assertFalse(self.backup.exists())
+
+    def test_backup_must_be_new_outside_and_failure_retains_source_evidence(self):
+        self.backup.mkdir(parents=True)
+        self.refused(lambda: self.apply(), "new directory")
+        self.backup.rmdir()
+        self.backup.symlink_to(self.base / "missing-backup")
+        self.refused(lambda: self.apply(), "new directory")
+        self.backup.unlink()
+        self.refused(lambda: self.apply(backup=self.root / "inside"), "outside Magic")
+        real = gear.durable
+        def partial(path, data):
+            if path.name == "SHA256.json": raise OSError("backup manifest failure")
+            return real(path, data)
+        with patch.object(gear, "durable", side_effect=partial):
+            self.refused(lambda: self.apply())
+        self.assertEqual(b"", (self.backup / "data/gear-stations.yml").read_bytes())
+
+    def test_new_history_or_source_changes_during_backup_block_publication(self):
+        real = gear.backup_gear
+        for mutation in [lambda: (self.root / "gear-returns").mkdir(), lambda: self.station.write_bytes(b"new source"), lambda: (self.root / "gear-stations.yml.bak").write_bytes(b"other evidence")]:
+            good = self.prepared()
+            changed = []
+            def racing(root, backup):
+                result = real(root, backup)
+                mutation()
+                changed.append(self.snapshot())
+                return result
+            with patch.object(gear, "backup_gear", side_effect=racing):
+                report = self.preview(**good)
+            self.assertEqual("REFUSED", report["verdict"])
+            self.assertEqual(changed[0], self.snapshot())
+            import shutil
+            shutil.rmtree(self.backup)
+            for path in [self.root / "gear-returns", self.root / "gear-stations.yml.bak"]:
+                if path.exists(): path.rmdir() if path.is_dir() else path.unlink()
+            self.station.write_bytes(b"")
+
+    def test_last_recheck_keeps_a_new_source_and_cleans_only_own_staging(self):
+        real = os.fsync
+        good = self.prepared()
+        changed = []
+        def racing(fd):
+            result = real(fd)
+            info = os.fstat(fd)
+            if stat.S_ISREG(info.st_mode) and info.st_size == len(gear.LEGACY_EMPTY_CANONICAL):
+                self.station.unlink()
+                self.station.write_bytes(b"newer unknown station bytes")
+                changed.append(self.station.read_bytes())
+            return result
+        with patch.object(gear.os, "fsync", side_effect=racing):
+            report = self.preview(**good)
+        self.assertEqual("REFUSED", report["verdict"])
+        self.assertEqual([b"newer unknown station bytes"], changed)
+        self.assertEqual(changed[0], self.station.read_bytes())
+        self.assertFalse(list(self.root.glob(gear.LEGACY_EMPTY_STAGING + "*")))
+
+    def test_prepublication_io_cuts_refuse_without_changing_source_inode(self):
+        real_force, real_durable, real_open, real_write, real_fsync = gear.force, gear.durable, os.open, os.write, os.fsync
+        for cut in ["preflight", "attestation", "plan", "open", "write", "chmod", "fsync"]:
+            with self.subTest(cut=cut):
+                good = self.prepared()
+                def force(path):
+                    if cut == "preflight" and path == self.root: raise OSError("preflight failure")
+                    return real_force(path)
+                def durable(path, data):
+                    if (cut == "attestation" and path.name == "LINEAGE-ATTESTATION") or (cut == "plan" and path.name == "IMPORT-PLAN.json"): raise OSError("plan evidence failure")
+                    return real_durable(path, data)
+                def opened(path, flags, *rest):
+                    if cut == "open" and Path(path).name.startswith(gear.LEGACY_EMPTY_STAGING): raise OSError("exclusive staging open failed")
+                    return real_open(path, flags, *rest)
+                def written(fd, data):
+                    if cut == "write":
+                        real_write(fd, data[:2])
+                        raise OSError("partial staging write")
+                    return real_write(fd, data)
+                def synced(fd):
+                    if cut == "fsync" and stat.S_ISREG(os.fstat(fd).st_mode) and os.fstat(fd).st_size == 13: raise OSError("staging fsync failed")
+                    return real_fsync(fd)
+                with patch.object(gear, "force", side_effect=force), patch.object(gear, "durable", side_effect=durable), patch.object(gear.os, "open", side_effect=opened), patch.object(gear.os, "write", side_effect=written), patch.object(gear.os, "fsync", side_effect=synced), patch.object(gear.os, "fchmod", side_effect=OSError("mode failed") if cut == "chmod" else os.fchmod):
+                    self.refused(lambda: self.preview(**good))
+                self.assertFalse(list(self.root.glob(gear.LEGACY_EMPTY_STAGING + "*")))
+                if self.backup.exists():
+                    import shutil
+                    shutil.rmtree(self.backup)
+        # Failed cleanup keeps explicit evidence and the original source; preview cannot ignore it.
+        good = self.prepared()
+        original = self.snapshot()["gear-stations.yml"]
+        real_unlink = os.unlink
+        def failed_write(fd, data):
+            real_write(fd, data)
+            raise OSError("staging write acknowledgment failed")
+        def failed_cleanup(path, *args, **kwargs):
+            if Path(path).name.startswith(gear.LEGACY_EMPTY_STAGING): raise OSError("cleanup storage failed")
+            return real_unlink(path, *args, **kwargs)
+        with patch.object(gear.os, "write", side_effect=failed_write), patch.object(gear.os, "unlink", side_effect=failed_cleanup):
+            report = self.preview(**good)
+        self.assertEqual("REFUSED", report["verdict"])
+        self.assertIn("staging-cleanup-error", report)
+        self.assertEqual(original, self.snapshot()["gear-stations.yml"])
+        leftover = next(self.root.glob(gear.LEGACY_EMPTY_STAGING + "*"))
+        self.assertEqual(gear.LEGACY_EMPTY_CANONICAL, leftover.read_bytes())
+        self.refused(self.preview, "staging evidence")
+
+    def test_replace_and_postpublication_io_failures_are_unconfirmed(self):
+        real_replace, real_force, real_durable, real_read = os.replace, gear.force, gear.durable, gear._legacy_empty_read
+        for cut in ["replace", "dirsync", "read", "result", "new-history"]:
+            with self.subTest(cut=cut):
+                good = self.prepared()
+                original_inode = self.station.stat().st_ino
+                renamed = []
+                def replace(source, target):
+                    if target == self.station:
+                        if cut == "replace": raise OSError("rename failed")
+                        real_replace(source, target)
+                        renamed.append(True)
+                        if cut == "new-history": (self.root / "gear-returns").mkdir()
+                        return
+                    return real_replace(source, target)
+                def force(path):
+                    if renamed and path == self.root and cut == "dirsync": raise OSError("publication directory fsync failed")
+                    return real_force(path)
+                def durable(path, data):
+                    if cut == "result" and path.name == "IMPORT-RESULT.json": raise OSError("result receipt failure")
+                    return real_durable(path, data)
+                def read(path, label):
+                    if renamed and cut == "read" and path == self.station: raise OSError("post-publication read failed")
+                    return real_read(path, label)
+                with patch.object(gear.os, "replace", side_effect=replace), patch.object(gear, "force", side_effect=force), patch.object(gear, "durable", side_effect=durable), patch.object(gear, "_legacy_empty_read", side_effect=read):
+                    report = self.preview(**good)
+                self.assertEqual("PUBLISHED_UNCONFIRMED", report["verdict"], report)
+                self.assertIn("Keep Paper stopped", report["recovery"])
+                self.assertFalse(list(self.root.glob(gear.LEGACY_EMPTY_STAGING + "*")))
+                self.assertTrue((self.backup / "IMPORT-PLAN.json").exists())
+                if cut == "replace":
+                    self.assertEqual(original_inode, self.station.stat().st_ino)
+                    self.assertEqual(b"", self.station.read_bytes())
+                else:
+                    self.assertEqual(gear.LEGACY_EMPTY_CANONICAL, self.station.read_bytes())
+                import shutil
+                shutil.rmtree(self.backup)
+                if (self.root / "gear-returns").exists(): (self.root / "gear-returns").rmdir()
+                self.station.write_bytes(b"")
+
+    def test_same_mode_chmod_and_truncated_first_deposit_break_prior_baselines(self):
+        good = self.prepared()
+        time.sleep(0.02)  # Advance the local inode timestamp clock before same-mode chmod.
+        self.station.chmod(0o640)
+        self.refused(lambda: self.preview(**good), "attested baseline")
+        good = self.prepared()
+        staged = self.base / "synthetic-first-deposit.yml"
+        staged.write_bytes(b"stations:\n  synthetic: {effect-state: depositing, item-bytes: AQ==}\n")
+        os.replace(staged, self.station)  # Actual 0.6.0 uses atomic replacement; no history is needed.
+        with self.station.open("r+b") as stream: stream.truncate(0)
+        report = self.preview()
+        self.assertEqual("ELIGIBLE_REQUIRES_LINEAGE_ATTESTATION", report["verdict"])
+        self.assertNotEqual(good["expect_baseline"], report["baseline"])
+        self.refused(lambda: self.preview(**good), "attested baseline")
+
+    def test_cli_keeps_import_scope_and_explicit_attestation_distinct(self):
+        for conflicting in [("--kind", "migration"), ("--journal", str(uuid.uuid4()))]:
+            before = self.snapshot()
+            code, report = self.cli("--legacy-empty-import", "--legacy-jar", self.jar, *conflicting)
+            self.assertEqual(2, code)
+            self.assertEqual("REFUSED", report["verdict"])
+            self.assertEqual(before, self.snapshot())
+        code, report = self.cli("--legacy-empty-import", "--legacy-jar", self.jar)
+        self.assertEqual(0, code)
+        self.assertEqual("ELIGIBLE_REQUIRES_LINEAGE_ATTESTATION", report["verdict"])
+        code, report = self.cli("--legacy-empty-import", "--legacy-jar", self.jar, "--apply")
+        self.assertEqual(2, code)
+        self.assertEqual("REFUSED", report["verdict"])
+        code, _ = self.cli("--legacy-jar", self.jar)
+        self.assertEqual(2, code)
+
+
 if __name__ == "__main__":
     unittest.main()