diff --git a/server/news/+operator-recovery-custody.feature.md b/server/news/+operator-recovery-custody.feature.md
new file mode 100644
index 0000000..fab50e4
--- /dev/null
+++ b/server/news/+operator-recovery-custody.feature.md
@@ -0,0 +1,7 @@
+Offline authority setup now requires an external private operator recovery file.
+It creates separate recovery-signing and backup-decryption keys, registers only
+their public trust in server state, and retains trusted backup-origin public keys
+outside the server. Explicit reuse preserves recovery keys across setup retries
+and additional servers. Setup refuses unsafe custody, unexpected overwrites and
+incomplete recovery trust. Runtime recovery and encrypted backup export remain
+later authentication milestones.
diff --git a/server/src/arbiter_server/authority.py b/server/src/arbiter_server/authority.py
index 5c3843b..f14aa34 100644
--- a/server/src/arbiter_server/authority.py
+++ b/server/src/arbiter_server/authority.py
@@ -16,7 +16,7 @@
 DATABASE = "authority.sqlite3"
 _PENDING = ".authority.pending.sqlite3"
 _LOCK = ".authority.lock"
-SCHEMA_VERSION = 2
+SCHEMA_VERSION = 3
 # Windows Modules Installer is part of the trusted host OS. This exception is
 # for ancestors only; authority-root and file ACL allowlists remain unchanged.
 _TRUSTED_INSTALLER = "S-1-5-80-956008885-3418522649-1831038044-1853292631-2271478464"
@@ -36,7 +36,9 @@ def to_dict(self) -> dict[str, object]:
         return {"initialized": True, "schema_version": SCHEMA_VERSION, **asdict(self)}
 
 
-def _check_path(path: Path, *, directory: bool = False) -> None:
+def _check_path(
+    path: Path, *, directory: bool = False, expected_links: int = 1
+) -> None:
     info = path.lstat()
     expected = stat.S_ISDIR if directory else stat.S_ISREG
     if not expected(info.st_mode) or (
@@ -53,8 +55,9 @@ def _check_path(path: Path, *, directory: bool = False) -> None:
             )
     else:
         _check_windows_permissions(path, directory=directory)
-    if not directory and info.st_nlink != 1:
-        raise AuthorityError(f"authority file must have exactly one link: {path}")
+    if not directory and info.st_nlink != expected_links:
+        count = "one link" if expected_links == 1 else f"{expected_links} links"
+        raise AuthorityError(f"authority file must have exactly {count}: {path}")
 
 
 def _windows_caller_sid() -> str:
@@ -277,8 +280,12 @@ def _write_database(path: Path, record: AuthorityRecord) -> None:
 
 def _publish(pending: Path, target: Path) -> None:
     os.replace(pending, target)
+    _flush_parents(target.parent)
+
+
+def _flush_parents(parent: Path) -> None:
     if os.name != "nt":
-        for directory in (target.parent, *target.parent.parents):
+        for directory in (parent, *parent.parents):
             fd = os.open(directory, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
             try:
                 os.fsync(fd)
@@ -286,7 +293,12 @@ def _publish(pending: Path, target: Path) -> None:
                 os.close(fd)
 
 
-def initialize_authority(state_dir: Path) -> AuthorityRecord:
+def initialize_authority(
+    state_dir: Path,
+    *,
+    recovery_file: Path | None = None,
+    reuse_recovery_file: bool = False,
+) -> AuthorityRecord:
     """Create one private set of IDs, refusing to replace any existing database."""
     root = _root(state_dir, create=True)
     with _locked(root):
@@ -295,23 +307,34 @@ def initialize_authority(state_dir: Path) -> AuthorityRecord:
             raise AuthorityError(
                 "authority database already exists; inspect authority status"
             )
-        _remove_pending(root)
-        record = AuthorityRecord(
-            str(uuid4()), str(uuid4()), datetime.now(timezone.utc).isoformat()
-        )
-        try:
-            _write_database(root / _PENDING, record)
-            _publish(root / _PENDING, target)
-        finally:
-            # A killed process may leave only these bounded, unpublished files.
-            # Never remove the target after a possibly successful publication.
+        if recovery_file is None:
+            raise AuthorityError(
+                "select an external private --recovery-file destination"
+            )
+        from .recovery import register_and_publish, setup_file
+
+        with setup_file(recovery_file, root, reuse=reuse_recovery_file) as (
+            path,
+            material,
+        ):
             _remove_pending(root)
-        return record
+            record = AuthorityRecord(
+                material.owner_id, str(uuid4()), datetime.now(timezone.utc).isoformat()
+            )
+            try:
+                _write_database(root / _PENDING, record)
+                register_and_publish(path, record, material, root / _PENDING)
+                _publish(root / _PENDING, target)
+            finally:
+                # Keep external recovery custody even if database publication fails.
+                _remove_pending(root)
+            return record
 
 
 def authority_status(state_dir: Path) -> AuthorityRecord | None:
     """Inspect committed records without creating or repairing a database."""
     from .credentials import verify_setup_credentials
+    from .recovery import verify_registration
 
     root = _root(state_dir, create=False)
     if not root.exists():
@@ -336,6 +359,9 @@ def authority_status(state_dir: Path) -> AuthorityRecord | None:
                 credentials = db.execute(
                     "SELECT role, certificate_der, private_key_der FROM credentials"
                 ).fetchall()
+                if len(rows) != 1:
+                    raise ValueError("expected one authority record")
+                verify_registration(db, rows[0][1])
             if len(rows) != 1:
                 raise ValueError("expected one authority record")
             singleton, owner, server, created = rows[0]
@@ -361,13 +387,24 @@ def authority_status(state_dir: Path) -> AuthorityRecord | None:
             ) from exc
 
 
-def authority_cli(action: str, state_dir: Path, *, as_json: bool) -> int:
+def authority_cli(
+    action: str,
+    state_dir: Path,
+    *,
+    as_json: bool,
+    recovery_file: Path | None = None,
+    reuse_recovery_file: bool = False,
+) -> int:
     from .cli_errors import print_cli_error
     import json
 
     try:
         record = (
-            initialize_authority(state_dir)
+            initialize_authority(
+                state_dir,
+                recovery_file=recovery_file,
+                reuse_recovery_file=reuse_recovery_file,
+            )
             if action == "init"
             else authority_status(state_dir)
         )
diff --git a/server/src/arbiter_server/main.py b/server/src/arbiter_server/main.py
index fb23238..08635d3 100644
--- a/server/src/arbiter_server/main.py
+++ b/server/src/arbiter_server/main.py
@@ -4427,6 +4427,18 @@ def _build_parser() -> argparse.ArgumentParser:
         command = authority_commands.add_parser(action)
         command.add_argument("--state-dir", type=Path, required=True)
         command.add_argument("--json", action="store_true")
+        if action == "init":
+            command.add_argument(
+                "--recovery-file",
+                type=Path,
+                required=True,
+                help="external operator-only recovery file in an existing private directory",
+            )
+            command.add_argument(
+                "--reuse-recovery-file",
+                action="store_true",
+                help="reuse existing recovery keys and register this server's backup verifier",
+            )
 
     serve = subcommands.add_parser("serve", help="run the Arbiter HTTP server")
     _add_override_arguments(
@@ -4634,7 +4646,11 @@ def main(argv: Sequence[str] | None = None) -> int:
         from .authority import authority_cli
 
         return authority_cli(
-            namespace.authority_command, namespace.state_dir, as_json=namespace.json
+            namespace.authority_command,
+            namespace.state_dir,
+            as_json=namespace.json,
+            recovery_file=getattr(namespace, "recovery_file", None),
+            reuse_recovery_file=getattr(namespace, "reuse_recovery_file", False),
         )
     if namespace.command == "serve":
         return _run_serve(
diff --git a/server/src/arbiter_server/recovery.py b/server/src/arbiter_server/recovery.py
new file mode 100644
index 0000000..11dd306
--- /dev/null
+++ b/server/src/arbiter_server/recovery.py
@@ -0,0 +1,274 @@
+"""Offline recovery custody. Running servers retain only recovery public trust."""
+
+from __future__ import annotations
+
+import base64
+from contextlib import closing, contextmanager
+from dataclasses import dataclass, field
+import hashlib
+import json
+import os
+from pathlib import Path
+import sqlite3
+from typing import Iterator
+from uuid import UUID, uuid4
+
+from cryptography.exceptions import UnsupportedAlgorithm
+from cryptography.hazmat.primitives import serialization as s
+from cryptography.hazmat.primitives.asymmetric import ec, x25519
+
+from . import authority as a
+
+WARNING = (
+    "NEVER SHARE THIS. This file decrypts backups AND authorizes live identity "
+    "recovery. Keep it outside server, agent, export-job and synced bundle storage."
+)
+MAX_FILE_BYTES = 65536
+
+
+def _encode(data: bytes) -> str:
+    if not isinstance(data, bytes):
+        raise a.AuthorityError("invalid recovery key storage")
+    return base64.b64encode(data).decode("ascii")
+
+
+def _decode(value: object, limit: int = 512) -> bytes:
+    if not isinstance(value, str) or len(value) > 4 * ((limit + 2) // 3):
+        raise a.AuthorityError("invalid recovery key encoding")
+    try:
+        data = base64.b64decode(value, validate=True)
+    except ValueError as exc:
+        raise a.AuthorityError("invalid recovery key encoding") from exc
+    if not data or len(data) > limit or _encode(data) != value:
+        raise a.AuthorityError("invalid recovery key encoding")
+    return data
+
+
+def _public(key: object) -> bytes:
+    if not isinstance(key, (ec.EllipticCurvePublicKey, x25519.X25519PublicKey)):
+        raise a.AuthorityError("unsupported recovery public key")
+    return key.public_bytes(s.Encoding.DER, s.PublicFormat.SubjectPublicKeyInfo)
+
+
+def _private(key: object) -> bytes:
+    if not isinstance(key, (ec.EllipticCurvePrivateKey, x25519.X25519PrivateKey)):
+        raise a.AuthorityError("unsupported recovery private key")
+    return key.private_bytes(s.Encoding.DER, s.PrivateFormat.PKCS8, s.NoEncryption())
+
+
+def _load(value: object, *, private: bool, signing: bool) -> object:
+    data = _decode(value)
+    try:
+        key = (
+            s.load_der_private_key(data, password=None)
+            if private
+            else s.load_der_public_key(data)
+        )
+    except (ValueError, TypeError, UnsupportedAlgorithm) as exc:
+        raise a.AuthorityError("invalid recovery key") from exc
+    if signing:
+        expected = ec.EllipticCurvePrivateKey if private else ec.EllipticCurvePublicKey
+        valid = isinstance(key, expected) and isinstance(key.curve, ec.SECP256R1)
+    else:
+        expected = x25519.X25519PrivateKey if private else x25519.X25519PublicKey
+        valid = isinstance(key, expected)
+    if not valid or (_private(key) if private else _public(key)) != data:
+        raise a.AuthorityError("wrong recovery key purpose or noncanonical DER")
+    return key
+
+
+def _uuid(value: object) -> str:
+    if not isinstance(value, str) or str(UUID(value, version=4)) != value:
+        raise a.AuthorityError("invalid recovery identity")
+    return value
+
+
+def _unique(pairs: list[tuple[str, object]]) -> dict[str, object]:
+    result: dict[str, object] = {}
+    for key, value in pairs:
+        if key in result:
+            raise a.AuthorityError("duplicate recovery field")
+        result[key] = value
+    return result
+
+
+@dataclass
+class RecoveryMaterial:
+    owner_id: str
+    signer: ec.EllipticCurvePrivateKey = field(repr=False)
+    decryptor: x25519.X25519PrivateKey = field(repr=False)
+    origins: dict[str, str] = field(default_factory=dict)
+    original_sha256: str | None = field(default=None, repr=False)
+
+    def document(self) -> dict[str, object]:
+        return {
+            "warning": WARNING,
+            "schema": "arbiter-operator-recovery-v1",
+            "owner_id": self.owner_id,
+            "recovery_signer_pkcs8": _encode(_private(self.signer)),
+            "backup_x25519_pkcs8": _encode(_private(self.decryptor)),
+            "backup_origins_spki": self.origins,
+        }
+
+
+def _read(path: Path) -> RecoveryMaterial:
+    a._check_path(path)
+    with path.open("rb") as handle:
+        info = os.fstat(handle.fileno())
+        current = path.lstat()
+        if (info.st_dev, info.st_ino) != (current.st_dev, current.st_ino):
+            raise a.AuthorityError("recovery file changed while opening")
+        data = handle.read(MAX_FILE_BYTES + 1)
+    if len(data) > MAX_FILE_BYTES:
+        raise a.AuthorityError("recovery file exceeds size bound")
+    try:
+        doc = json.loads(data, object_pairs_hook=_unique)
+        if not isinstance(doc, dict) or set(doc) != {
+            "warning",
+            "schema",
+            "owner_id",
+            "recovery_signer_pkcs8",
+            "backup_x25519_pkcs8",
+            "backup_origins_spki",
+        }:
+            raise ValueError("wrong fields")
+        if doc["schema"] != "arbiter-operator-recovery-v1" or doc["warning"] != WARNING:
+            raise ValueError("wrong recovery schema")
+        signer = _load(doc["recovery_signer_pkcs8"], private=True, signing=True)
+        decryptor = _load(doc["backup_x25519_pkcs8"], private=True, signing=False)
+        assert isinstance(signer, ec.EllipticCurvePrivateKey)
+        assert isinstance(decryptor, x25519.X25519PrivateKey)
+        origins = doc["backup_origins_spki"]
+        if not isinstance(origins, dict):
+            raise ValueError("invalid origins")
+        for server, value in origins.items():
+            _uuid(server)
+            _load(value, private=False, signing=True)
+        return RecoveryMaterial(
+            _uuid(doc["owner_id"]),
+            signer,
+            decryptor,
+            origins,
+            hashlib.sha256(data).hexdigest(),
+        )
+    except (ValueError, TypeError, RecursionError) as exc:
+        raise a.AuthorityError("invalid operator recovery file") from exc
+
+
+@contextmanager
+def setup_file(
+    path: Path, state_root: Path, *, reuse: bool
+) -> Iterator[tuple[Path, RecoveryMaterial]]:
+    # A caller explicitly selects an existing private parent, never live state.
+    path = path.expanduser().absolute()
+    parent = a._root(path.parent, create=False)
+    if not parent.is_dir():
+        raise a.AuthorityError("prepare a private recovery directory first")
+    target = parent / path.name
+    if target.is_relative_to(state_root):
+        raise a.AuthorityError("operator recovery file must be outside server state")
+    with a._locked(parent):
+        pending = target.with_name(f".{target.name}.arbiter-pending")
+        if pending.exists() or pending.is_symlink():
+            links = 1
+            if target.exists() or target.is_symlink():
+                staged, published = pending.lstat(), target.lstat()
+                if (staged.st_dev, staged.st_ino) == (
+                    published.st_dev,
+                    published.st_ino,
+                ):
+                    # Only our exact no-overwrite publication pair may have two
+                    # names. Both still need the complete private/native ACL check.
+                    links = 2
+                    a._check_path(target, expected_links=links)
+            a._check_path(pending, expected_links=links)
+            pending.unlink()
+        exists = target.exists() or target.is_symlink()
+        if exists and not reuse:
+            raise a.AuthorityError("recovery file exists; explicit reuse is required")
+        if reuse and not exists:
+            raise a.AuthorityError("recovery file to reuse does not exist")
+        material = (
+            _read(target)
+            if exists
+            else RecoveryMaterial(
+                str(uuid4()),
+                ec.generate_private_key(ec.SECP256R1()),
+                x25519.X25519PrivateKey.generate(),
+            )
+        )
+        yield target, material
+
+
+def register_and_publish(
+    path: Path, record: a.AuthorityRecord, material: RecoveryMaterial, pending_db: Path
+) -> None:
+    """Public trust commits in unpublished DB; external custody publishes first."""
+    origin = ec.generate_private_key(ec.SECP256R1())
+    origin_public = _public(origin.public_key())
+    material.origins[record.server_id] = _encode(origin_public)
+    data = (json.dumps(material.document(), indent=2) + "\n").encode()
+    if len(data) > MAX_FILE_BYTES:
+        raise a.AuthorityError("recovery file exceeds size bound")
+    with closing(sqlite3.connect(pending_db, timeout=0)) as db:
+        db.execute("PRAGMA synchronous=FULL")
+        db.execute(
+            "CREATE TABLE recovery_trust (singleton INTEGER PRIMARY KEY CHECK(singleton=1), owner_id TEXT NOT NULL, verifier BLOB NOT NULL, backup_recipient BLOB NOT NULL, origin_verifier BLOB NOT NULL, origin_private BLOB NOT NULL)"
+        )
+        db.execute(
+            "INSERT INTO recovery_trust VALUES (1, ?, ?, ?, ?, ?)",
+            (
+                record.owner_id,
+                _public(material.signer.public_key()),
+                _public(material.decryptor.public_key()),
+                origin_public,
+                _private(origin),
+            ),
+        )
+        db.commit()
+    with pending_db.open("r+b") as handle:
+        os.fsync(handle.fileno())
+    # Private parent lock serializes explicit reuse, including separate servers.
+    # Keep published recovery keys after any later failure; never unlink custody.
+    staged = path.with_name(f".{path.name}.arbiter-pending")
+    try:
+        fd = os.open(staged, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
+        with os.fdopen(fd, "wb") as handle:
+            a._check_path(staged)
+            handle.write(data)
+            handle.flush()
+            os.fsync(handle.fileno())
+        if material.original_sha256 is None:
+            # A new destination can never replace a file that appeared meanwhile.
+            os.link(staged, path)
+            staged.unlink()
+            a._flush_parents(path.parent)
+        else:
+            a._check_path(path)
+            with path.open("rb") as handle:
+                current = handle.read(MAX_FILE_BYTES + 1)
+            if hashlib.sha256(current).hexdigest() != material.original_sha256:
+                raise a.AuthorityError("recovery file changed before publication")
+            a._publish(staged, path)
+    finally:
+        if staged.exists():
+            a._check_path(staged)
+            staged.unlink()
+
+
+def verify_registration(db: sqlite3.Connection, owner_id: str) -> None:
+    rows = db.execute(
+        "SELECT singleton, owner_id, verifier, backup_recipient, origin_verifier, origin_private FROM recovery_trust"
+    ).fetchall()
+    if len(rows) != 1 or rows[0][:2] != (1, owner_id):
+        raise a.AuthorityError("missing or mismatched recovery trust")
+    _, _, verifier, recipient, origin_verifier, origin_private = rows[0]
+    recovery = _load(_encode(verifier), private=False, signing=True)
+    _load(_encode(recipient), private=False, signing=False)
+    origin_public = _load(_encode(origin_verifier), private=False, signing=True)
+    origin = _load(_encode(origin_private), private=True, signing=True)
+    assert isinstance(origin, ec.EllipticCurvePrivateKey)
+    if _public(origin.public_key()) != _public(origin_public) or _public(
+        recovery
+    ) == _public(origin_public):
+        raise a.AuthorityError("backup origin key mismatch or purpose reuse")
diff --git a/server/tests/unit/test_authority.py b/server/tests/unit/test_authority.py
index 6a20f76..fb0ed71 100644
--- a/server/tests/unit/test_authority.py
+++ b/server/tests/unit/test_authority.py
@@ -20,19 +20,31 @@
 pytestmark = pytest.mark.usefixtures("private_windows_test_parent")
 
 
+def initialize(root):
+    parent = next(
+        p
+        for p in root.parents
+        if p.is_dir() and (os.name == "nt" or not p.stat().st_mode & 0o077)
+    )
+    recovery = parent / (root.name + ".recovery.json")
+    return authority.initialize_authority(
+        root, recovery_file=recovery, reuse_recovery_file=recovery.exists()
+    )
+
+
 def test_initialize_reopen_and_independent_roots(tmp_path: Path) -> None:
     root = tmp_path / "one"
-    record = authority.initialize_authority(root)
+    record = initialize(root)
     assert authority.authority_status(root) == record
     assert UUID(record.owner_id).version == UUID(record.server_id).version == 4
     assert record.owner_id != record.server_id
-    other = authority.initialize_authority(tmp_path / "two")
+    other = initialize(tmp_path / "two")
     assert {record.owner_id, record.server_id}.isdisjoint(
         {other.owner_id, other.server_id}
     )
     with closing(sqlite3.connect(root / authority.DATABASE)) as db:
         assert db.execute("PRAGMA integrity_check").fetchone() == ("ok",)
-        assert db.execute("PRAGMA user_version").fetchone() == (2,)
+        assert db.execute("PRAGMA user_version").fetchone() == (3,)
     if os.name != "nt":
         assert root.stat().st_mode & 0o777 == 0o700
         assert (root / authority.DATABASE).stat().st_mode & 0o777 == 0o600
@@ -49,7 +61,7 @@ def writable_flush(fd):
         real_fsync(fd)
 
     monkeypatch.setattr(authority.os, "fsync", writable_flush)
-    record = authority.initialize_authority(tmp_path / "state")
+    record = initialize(tmp_path / "state")
     assert authority.authority_status(tmp_path / "state") == record
 
 
@@ -63,11 +75,11 @@ def test_status_missing_is_not_initialization(tmp_path: Path) -> None:
 
 
 def test_second_initialization_preserves_database(tmp_path: Path) -> None:
-    record = authority.initialize_authority(tmp_path / "state")
+    record = initialize(tmp_path / "state")
     path = tmp_path / "state" / authority.DATABASE
     original = path.read_bytes()
     with pytest.raises(authority.AuthorityError, match="already exists"):
-        authority.initialize_authority(path.parent)
+        initialize(path.parent)
     assert path.read_bytes() == original
     assert authority.authority_status(path.parent) == record
     assert path.read_bytes() == original
@@ -83,13 +95,35 @@ def no_composition(*args, **kwargs):
     state = ["--state-dir", str(tmp_path / "state"), "--json"]
     assert main([*args, "status", *state]) == 0
     assert json.loads(capsys.readouterr().out) == {"initialized": False}
-    assert main([*args, "init", *state]) == 0
+    assert (
+        main(
+            [
+                *args,
+                "init",
+                *state,
+                "--recovery-file",
+                str(tmp_path / "cli.recovery.json"),
+            ]
+        )
+        == 0
+    )
     output = json.loads(capsys.readouterr().out)
     assert output["initialized"] is True
-    assert output["schema_version"] == 2
+    assert output["schema_version"] == 3
     assert main([*args, "status", *state]) == 0
     assert json.loads(capsys.readouterr().out) == output
-    assert main([*args, "init", *state]) == 1
+    assert (
+        main(
+            [
+                *args,
+                "init",
+                *state,
+                "--recovery-file",
+                str(tmp_path / "cli.recovery.json"),
+            ]
+        )
+        == 1
+    )
     captured = capsys.readouterr()
     assert not captured.out
     assert "already exists" in captured.err
@@ -100,7 +134,7 @@ def no_composition(*args, **kwargs):
 )
 def test_corrupt_state_never_reinitialized(tmp_path: Path, damage: str) -> None:
     root = tmp_path / "state"
-    authority.initialize_authority(root)
+    initialize(root)
     path = root / authority.DATABASE
     if damage == "garbage":
         path.write_bytes(b"not a database")
@@ -123,7 +157,7 @@ def test_corrupt_state_never_reinitialized(tmp_path: Path, damage: str) -> None:
     with pytest.raises(authority.AuthorityError):
         authority.authority_status(root)
     with pytest.raises(authority.AuthorityError, match="already exists"):
-        authority.initialize_authority(root)
+        initialize(root)
     assert path.read_bytes() == original
 
 
@@ -131,7 +165,7 @@ def test_corrupt_state_never_reinitialized(tmp_path: Path, damage: str) -> None:
 @pytest.mark.parametrize("target", ["directory", "database", "lock"])
 def test_broad_permissions_rejected(tmp_path: Path, target: str) -> None:
     root = tmp_path / "state"
-    authority.initialize_authority(root)
+    initialize(root)
     path = {
         "directory": root,
         "database": root / authority.DATABASE,
@@ -156,9 +190,7 @@ def test_symlink_files_rejected_without_touching_destination(
     outside.chmod(0o600)
     (root / name).symlink_to(outside)
     operation = (
-        authority.initialize_authority
-        if name.endswith("pending.sqlite3")
-        else authority.authority_status
+        initialize if name.endswith("pending.sqlite3") else authority.authority_status
     )
     expected: tuple[type[Exception], ...] = (ValueError, OSError)
     with pytest.raises(expected):
@@ -170,7 +202,7 @@ def test_symlink_files_rejected_without_touching_destination(
 @pytest.mark.skipif(os.name == "nt", reason="POSIX symlink and hardlink semantics")
 def test_directory_alias_and_hardlinked_database_rejected(tmp_path: Path) -> None:
     root = tmp_path / "state"
-    authority.initialize_authority(root)
+    initialize(root)
     alias = tmp_path / "alias"
     alias.symlink_to(root, target_is_directory=True)
     with pytest.raises(authority.AuthorityError, match="symlink"):
@@ -186,7 +218,7 @@ def test_writable_parent_rejected(tmp_path: Path) -> None:
     parent.mkdir(mode=0o777)
     parent.chmod(0o777)
     with pytest.raises(authority.AuthorityError, match="another user"):
-        authority.initialize_authority(parent / "state")
+        initialize(parent / "state")
     assert not (parent / "state").exists()
 
 
@@ -199,7 +231,7 @@ def test_nested_missing_parent_rejected_after_creation(
     previous_umask = os.umask(umask)
     try:
         with pytest.raises(authority.AuthorityError, match="another user"):
-            authority.initialize_authority(root)
+            initialize(root)
     finally:
         os.umask(previous_umask)
     assert root.exists()
@@ -224,12 +256,12 @@ def track(fd):
     monkeypatch.setattr(authority.os, "fsync", track)
     previous_umask = os.umask(0o022)
     try:
-        record = authority.initialize_authority(root)
+        record = initialize(root)
     finally:
         os.umask(previous_umask)
     assert authority.authority_status(root) == record
     assert root.parent.stat().st_mode & 0o777 == 0o755
-    assert flushed == [
+    assert flushed[-len((root, *root.parents)) :] == [
         (p.stat().st_dev, p.stat().st_ino) for p in (root, *root.parents)
     ]
 
@@ -248,6 +280,8 @@ def fail_write(path, record):
         raise OSError("injected write failure")
 
     def fail_publish(pending, target):
+        if target.name != authority.DATABASE:
+            return real_publish(pending, target)
         if failure == "after_publish":
             real_publish(pending, target)
         raise OSError("injected publication failure")
@@ -257,7 +291,7 @@ def fail_publish(pending, target):
     )
     monkeypatch.setattr(authority, "_publish", fail_publish)
     with pytest.raises(OSError, match="injected"):
-        authority.initialize_authority(root)
+        initialize(root)
     assert not (root / ".authority.pending.sqlite3").exists()
     assert not (root / ".authority.pending.sqlite3-journal").exists()
     monkeypatch.setattr(authority, "_write_database", real_write)
@@ -265,10 +299,10 @@ def fail_publish(pending, target):
     if failure == "after_publish":
         assert authority.authority_status(root) is not None
         with pytest.raises(authority.AuthorityError, match="already exists"):
-            authority.initialize_authority(root)
+            initialize(root)
     else:
         assert authority.authority_status(root) is None
-        assert authority.initialize_authority(root) == authority.authority_status(root)
+        assert initialize(root) == authority.authority_status(root)
 
 
 def _wait_for_file(path: Path, child: subprocess.Popen) -> None:
@@ -308,7 +342,7 @@ def partial(path, record):
     authority._write_database = partial
 else:
     authority._publish = wait
-authority.initialize_authority(Path(sys.argv[1]))
+authority.initialize_authority(Path(sys.argv[1]), recovery_file=Path(sys.argv[1]).with_name(Path(sys.argv[1]).name + ".recovery.json"))
 """
     child = subprocess.Popen(
         [sys.executable, "-c", script, str(root), str(ready), phase],
@@ -319,17 +353,26 @@ def partial(path, record):
     try:
         _wait_for_file(ready, child)
         assert not (root / authority.DATABASE).exists()
-        for operation in (authority.initialize_authority, authority.authority_status):
+        for operation in (initialize, authority.authority_status):
             with pytest.raises(authority.AuthorityError, match="busy"):
                 operation(root)
     finally:
         child.kill()
         child.communicate(timeout=15)
     assert (root / ".authority.pending.sqlite3").exists()
-    record = authority.initialize_authority(root)
+    recovery_path = root.with_name(root.name + ".recovery.json")
+    recovery_before = (
+        json.loads(recovery_path.read_text()) if recovery_path.exists() else None
+    )
+    record = initialize(root)
     assert authority.authority_status(root) == record
     assert not (root / ".authority.pending.sqlite3").exists()
     assert {p.name for p in root.iterdir()} == {authority.DATABASE, ".authority.lock"}
+    if recovery_before:
+        assert (
+            json.loads(recovery_path.read_text())["recovery_signer_pkcs8"]
+            == recovery_before["recovery_signer_pkcs8"]
+        )
 
 
 def test_concurrent_process_initializers_publish_one_identity(tmp_path: Path) -> None:
@@ -339,7 +382,7 @@ def test_concurrent_process_initializers_publish_one_identity(tmp_path: Path) ->
 from pathlib import Path
 from arbiter_server.authority import initialize_authority, AuthorityError
 try:
-    print(json.dumps(initialize_authority(Path(sys.argv[1])).to_dict()))
+    print(json.dumps(initialize_authority(Path(sys.argv[1]), recovery_file=Path(sys.argv[1]).parent / "concurrent.recovery.json").to_dict()))
 except AuthorityError as exc:
     print(str(exc), file=sys.stderr)
     sys.exit(1)
@@ -375,7 +418,7 @@ def test_concurrent_process_initializers_publish_one_identity(tmp_path: Path) ->
 @pytest.mark.skipif(os.name == "nt", reason="POSIX ownership")
 def test_different_uid_is_not_authorized(tmp_path: Path, monkeypatch) -> None:
     root = tmp_path / "state"
-    authority.initialize_authority(root)
+    initialize(root)
     current = os.geteuid()
     monkeypatch.setattr(authority.os, "geteuid", lambda: current + 1)
     with pytest.raises(authority.AuthorityError):
@@ -393,7 +436,7 @@ def test_windows_non_inheritable_parent_acl_rejected(tmp_path: Path) -> None:
         text=True,
     )
     with pytest.raises(authority.AuthorityError, match="another user"):
-        authority.initialize_authority(parent / "state")
+        initialize(parent / "state")
     assert not (parent / "state").exists()
 
 
@@ -464,11 +507,7 @@ def test_windows_foreign_state_owner_rejected(tmp_path: Path, name: str) -> None
         capture_output=True,
         text=True,
     )
-    operation = (
-        authority.authority_status
-        if name == authority.DATABASE
-        else authority.initialize_authority
-    )
+    operation = authority.authority_status if name == authority.DATABASE else initialize
     with pytest.raises(authority.AuthorityError, match="owned by the caller"):
         operation(root)
     assert _windows_file_security(path).owner_sid == foreign
@@ -542,7 +581,7 @@ def test_windows_broad_directory_acl_rejected_without_repair(tmp_path: Path) ->
     )
     before = subprocess.check_output(["icacls", str(root)], text=True)
     with pytest.raises(ValueError, match="unsafe config directory permissions"):
-        authority.initialize_authority(root)
+        initialize(root)
     assert not list(root.iterdir())
     assert subprocess.check_output(["icacls", str(root)], text=True) == before
 
@@ -550,7 +589,7 @@ def test_windows_broad_directory_acl_rejected_without_repair(tmp_path: Path) ->
 @pytest.mark.skipif(os.name != "nt", reason="native Windows ACL boundary")
 def test_windows_broad_database_acl_rejected(tmp_path: Path) -> None:
     root = tmp_path / "state"
-    authority.initialize_authority(root)
+    initialize(root)
     path = root / authority.DATABASE
     subprocess.run(
         ["icacls", str(path), "/grant", "*S-1-5-32-545:R"],
diff --git a/server/tests/unit/test_credentials.py b/server/tests/unit/test_credentials.py
index 102b696..3922bab 100644
--- a/server/tests/unit/test_credentials.py
+++ b/server/tests/unit/test_credentials.py
@@ -4,6 +4,7 @@
 from contextlib import closing
 from datetime import datetime, timedelta, timezone
 import json
+import os
 from pathlib import Path
 import sqlite3
 
@@ -17,6 +18,19 @@
 
 pytestmark = pytest.mark.usefixtures("private_windows_test_parent")
 
+
+def initialize(root):
+    parent = next(
+        p
+        for p in root.parents
+        if p.is_dir() and (os.name == "nt" or not p.stat().st_mode & 0o077)
+    )
+    recovery = parent / (root.name + ".recovery.json")
+    return authority.initialize_authority(
+        root, recovery_file=recovery, reuse_recovery_file=recovery.exists()
+    )
+
+
 NOW = datetime(2026, 10, 21, tzinfo=timezone.utc)
 OWNER = "00000000-0000-4000-8000-000000000001"
 SERVER = "00000000-0000-4000-8000-000000000002"
@@ -261,7 +275,12 @@ def test_der_boundaries_and_signature(issuers, change):
 
 def test_setup_is_real_atomic_private_and_restartable(tmp_path, capsys):
     root = tmp_path / "state"
-    assert authority.authority_cli("init", root, as_json=True) == 0
+    assert (
+        authority.authority_cli(
+            "init", root, as_json=True, recovery_file=tmp_path / "cli.recovery.json"
+        )
+        == 0
+    )
     output = capsys.readouterr().out
     record = authority.authority_status(root)
     assert (
@@ -282,7 +301,7 @@ def test_setup_is_real_atomic_private_and_restartable(tmp_path, capsys):
         assert base64.b64encode(private).decode() not in output
     before = (root / authority.DATABASE).read_bytes()
     with pytest.raises(authority.AuthorityError, match="already exists"):
-        authority.initialize_authority(root)
+        initialize(root)
     assert authority.authority_status(root) == record
     assert (root / authority.DATABASE).read_bytes() == before
     assert {p.name for p in root.iterdir()} == {authority.DATABASE, authority._LOCK}
@@ -294,7 +313,7 @@ def test_setup_is_real_atomic_private_and_restartable(tmp_path, capsys):
 )
 def test_damaged_setup_never_repairs_or_replaces_keys(tmp_path, damage, capsys):
     root = tmp_path / "state"
-    authority.initialize_authority(root)
+    initialize(root)
     with closing(sqlite3.connect(root / authority.DATABASE)) as db:
         if damage == "old-schema":
             db.execute("PRAGMA user_version=1")
@@ -338,10 +357,10 @@ def fail(db, owner, server, now):
 
     monkeypatch.setattr(c, "store_setup_credentials", fail)
     with pytest.raises(OSError, match="injected"):
-        authority.initialize_authority(root)
+        initialize(root)
     assert {p.name for p in root.iterdir()} == {authority._LOCK}
     monkeypatch.setattr(c, "store_setup_credentials", real)
-    assert authority.initialize_authority(root) == authority.authority_status(root)
+    assert initialize(root) == authority.authority_status(root)
 
 
 def test_shared_profile_fixtures():
diff --git a/server/tests/unit/test_recovery.py b/server/tests/unit/test_recovery.py
new file mode 100644
index 0000000..2ebce9c
--- /dev/null
+++ b/server/tests/unit/test_recovery.py
@@ -0,0 +1,386 @@
+from __future__ import annotations
+
+import base64
+from contextlib import closing
+import json
+import os
+from pathlib import Path
+import sqlite3
+import subprocess
+import sys
+import time
+
+import pytest
+from cryptography.hazmat.primitives import hashes, serialization as s
+from cryptography.hazmat.primitives.asymmetric import ec, x25519
+
+from arbiter_server import authority as a, recovery as r
+from arbiter_server.main import main
+
+pytestmark = pytest.mark.usefixtures("private_windows_test_parent")
+
+
+def initialize(tmp_path, name="state", *, reuse=False):
+    return a.initialize_authority(
+        tmp_path / name,
+        recovery_file=tmp_path / "operator.json",
+        reuse_recovery_file=reuse,
+    )
+
+
+def read(tmp_path):
+    return json.loads((tmp_path / "operator.json").read_bytes())
+
+
+def test_real_setup_separates_private_custody_and_public_trust(tmp_path, capsys):
+    assert (
+        a.authority_cli(
+            "init",
+            tmp_path / "state",
+            as_json=True,
+            recovery_file=tmp_path / "operator.json",
+        )
+        == 0
+    )
+    output = capsys.readouterr().out
+    doc = read(tmp_path)
+    assert next(iter(doc)) == "warning" and doc["warning"].startswith(
+        "NEVER SHARE THIS"
+    )
+    record = a.authority_status(tmp_path / "state")
+    assert record is not None and doc["owner_id"] == record.owner_id
+    signer = s.load_der_private_key(
+        base64.b64decode(doc["recovery_signer_pkcs8"]), None
+    )
+    decryptor = s.load_der_private_key(
+        base64.b64decode(doc["backup_x25519_pkcs8"]), None
+    )
+    assert isinstance(signer, ec.EllipticCurvePrivateKey)
+    assert isinstance(decryptor, x25519.X25519PrivateKey)
+    with closing(sqlite3.connect(tmp_path / "state" / a.DATABASE)) as db:
+        _, owner, verifier, recipient, origin_verifier, origin_private = db.execute(
+            "SELECT * FROM recovery_trust"
+        ).fetchone()
+    assert owner == record.owner_id
+    assert verifier == r._public(signer.public_key())
+    assert recipient == r._public(decryptor.public_key())
+    statement = b"synthetic recovery proof"
+    recovery_public = s.load_der_public_key(verifier)
+    assert isinstance(recovery_public, ec.EllipticCurvePublicKey)
+    recovery_public.verify(
+        signer.sign(statement, ec.ECDSA(hashes.SHA256())),
+        statement,
+        ec.ECDSA(hashes.SHA256()),
+    )
+    peer = x25519.X25519PrivateKey.generate()
+    backup_public = s.load_der_public_key(recipient)
+    assert isinstance(backup_public, x25519.X25519PublicKey)
+    assert peer.exchange(backup_public) == decryptor.exchange(peer.public_key())
+    assert (
+        base64.b64decode(doc["backup_origins_spki"][record.server_id])
+        == origin_verifier
+    )
+    origin = s.load_der_private_key(origin_private, None)
+    assert r._public(origin.public_key()) == origin_verifier != verifier
+    database_bytes = (tmp_path / "state" / a.DATABASE).read_bytes()
+    for name in ("recovery_signer_pkcs8", "backup_x25519_pkcs8"):
+        assert doc[name] not in output
+        assert base64.b64decode(doc[name]) not in database_bytes
+    assert set(p.name for p in (tmp_path / "state").iterdir()) == {a.DATABASE, a._LOCK}
+    if os.name != "nt":
+        assert (tmp_path / "operator.json").stat().st_mode & 0o777 == 0o600
+    # Runtime inspection requires public trust, never the external private file.
+    (tmp_path / "operator.json").unlink()
+    assert a.authority_status(tmp_path / "state") == record
+
+
+def test_explicit_reuse_preserves_owner_keys_and_all_origin_trust(tmp_path):
+    first = initialize(tmp_path)
+    before = read(tmp_path)
+    with pytest.raises(a.AuthorityError, match="explicit reuse"):
+        initialize(tmp_path, "other")
+    assert read(tmp_path) == before
+    assert not (tmp_path / "other" / a.DATABASE).exists()
+    second = initialize(tmp_path, "other", reuse=True)
+    after = read(tmp_path)
+    assert first.owner_id == second.owner_id and first.server_id != second.server_id
+    assert before["recovery_signer_pkcs8"] == after["recovery_signer_pkcs8"]
+    assert before["backup_x25519_pkcs8"] == after["backup_x25519_pkcs8"]
+    assert before["backup_origins_spki"].items() <= after["backup_origins_spki"].items()
+    assert set(after["backup_origins_spki"]) == {first.server_id, second.server_id}
+
+
+@pytest.mark.parametrize("phase", ["write", "external", "database"])
+def test_failed_setup_preserves_custody_without_claiming_success(
+    tmp_path, monkeypatch, phase
+):
+    real = a._publish
+    real_flush = a._flush_parents
+
+    def publish(pending, target):
+        if phase == "database" and target.name == a.DATABASE:
+            raise OSError("injected database publication failure")
+        return real(pending, target)
+
+    def flush(parent):
+        if phase == "external" and parent == tmp_path:
+            raise OSError("injected external publication flush failure")
+        real_flush(parent)
+
+    with monkeypatch.context() as patch:
+        patch.setattr(a, "_publish", publish)
+        patch.setattr(a, "_flush_parents", flush)
+        if phase == "write":
+            patch.setattr(
+                r,
+                "_private",
+                lambda key: (_ for _ in ()).throw(
+                    OSError("injected key write failure")
+                ),
+            )
+        with pytest.raises(OSError, match="injected"):
+            initialize(tmp_path)
+    assert a.authority_status(tmp_path / "state") is None
+    assert not (tmp_path / "state" / a._PENDING).exists()
+    assert not (tmp_path / ".operator.json.arbiter-pending").exists()
+    exists = (tmp_path / "operator.json").exists()
+    assert exists == (phase != "write")
+    previous = read(tmp_path) if exists else None
+    record = initialize(tmp_path, reuse=exists)
+    assert a.authority_status(tmp_path / "state") == record
+    if previous:
+        assert (
+            read(tmp_path)["recovery_signer_pkcs8"] == previous["recovery_signer_pkcs8"]
+        )
+
+
+@pytest.mark.parametrize(
+    "damage",
+    [
+        "unknown",
+        "duplicate",
+        "trailing",
+        "large",
+        "nested",
+        "curve",
+        "origin",
+        "owner",
+        "base64",
+    ],
+)
+def test_invalid_recovery_file_never_rewrites_or_initializes(tmp_path, damage):
+    initialize(tmp_path)
+    doc = read(tmp_path)
+    if damage == "unknown":
+        doc["extra_authority"] = True
+    elif damage == "curve":
+        doc["recovery_signer_pkcs8"] = r._encode(
+            r._private(x25519.X25519PrivateKey.generate())
+        )
+    elif damage == "origin":
+        doc["backup_origins_spki"] = {"not-an-id": "garbage"}
+    elif damage == "owner":
+        doc["owner_id"] = "not-an-id"
+    elif damage == "base64":
+        doc["backup_x25519_pkcs8"] += "="
+    data = json.dumps(doc).encode()
+    if damage == "duplicate":
+        data = data[:-1] + b', "owner_id": "duplicate"}'
+    elif damage == "trailing":
+        data += b"junk"
+    elif damage == "large":
+        data = b"x" * (r.MAX_FILE_BYTES + 1)
+    elif damage == "nested":
+        data = b"[" * 2000 + b"]" * 2000
+    path = tmp_path / "operator.json"
+    path.write_bytes(data)
+    with pytest.raises(a.AuthorityError):
+        initialize(tmp_path, "other", reuse=True)
+    assert path.read_bytes() == data
+    assert not (tmp_path / "other" / a.DATABASE).exists()
+
+
+@pytest.mark.parametrize(
+    "column",
+    ["verifier", "backup_recipient", "origin_verifier", "origin_private", "owner_id"],
+)
+def test_damaged_registered_trust_is_not_repaired(tmp_path, column):
+    initialize(tmp_path)
+    path = tmp_path / "state" / a.DATABASE
+    with closing(sqlite3.connect(path)) as db:
+        db.execute(f"UPDATE recovery_trust SET {column}=?", ("invalid",))
+        db.commit()
+    before = path.read_bytes()
+    with pytest.raises(a.AuthorityError):
+        a.authority_status(path.parent)
+    assert path.read_bytes() == before
+
+
+@pytest.mark.parametrize("destination", ["inside", "missing-parent", "missing-reuse"])
+def test_setup_requires_explicit_safe_external_destination(tmp_path, destination):
+    root = tmp_path / "state"
+    file = {
+        "inside": root / "operator.json",
+        "missing-parent": tmp_path / "absent" / "operator.json",
+        "missing-reuse": tmp_path / "operator.json",
+    }[destination]
+    with pytest.raises(a.AuthorityError):
+        a.initialize_authority(
+            root, recovery_file=file, reuse_recovery_file=destination == "missing-reuse"
+        )
+    assert not (root / a.DATABASE).exists()
+    assert not file.exists()
+
+
+@pytest.mark.skipif(os.name == "nt", reason="POSIX aliases and modes")
+@pytest.mark.parametrize("damage", ["symlink", "hardlink", "mode", "parent"])
+def test_external_custody_rejects_unsafe_paths(tmp_path, damage):
+    initialize(tmp_path)
+    path = tmp_path / "operator.json"
+    before = path.read_bytes()
+    if damage == "symlink":
+        other = tmp_path / "original"
+        path.rename(other)
+        path.symlink_to(other)
+    elif damage == "hardlink":
+        os.link(path, tmp_path / "alias")
+    elif damage == "mode":
+        path.chmod(0o644)
+    else:
+        tmp_path.chmod(0o755)
+    with pytest.raises(a.AuthorityError):
+        initialize(tmp_path, "other", reuse=True)
+    assert path.read_bytes() == before
+
+
+def test_new_file_that_appears_during_setup_is_never_overwritten(tmp_path, monkeypatch):
+    real = r.register_and_publish
+
+    def publish(path, *args):
+        path.write_bytes(b"preserve unexpected file")
+        path.chmod(0o600)
+        return real(path, *args)
+
+    monkeypatch.setattr(r, "register_and_publish", publish)
+    with pytest.raises(FileExistsError):
+        initialize(tmp_path)
+    assert (tmp_path / "operator.json").read_bytes() == b"preserve unexpected file"
+    assert a.authority_status(tmp_path / "state") is None
+
+
+def test_cli_requires_destination_and_never_prints_private_material(tmp_path, capsys):
+    with pytest.raises(SystemExit):
+        main(["authority", "init", "--state-dir", str(tmp_path / "state")])
+    capsys.readouterr()
+    assert (
+        main(
+            [
+                "authority",
+                "init",
+                "--state-dir",
+                str(tmp_path / "state"),
+                "--recovery-file",
+                str(tmp_path / "operator.json"),
+                "--json",
+            ]
+        )
+        == 0
+    )
+    output = capsys.readouterr()
+    doc = read(tmp_path)
+    assert not output.err
+    assert doc["recovery_signer_pkcs8"] not in output.out
+    assert doc["backup_x25519_pkcs8"] not in output.out
+
+
+def test_interrupted_external_staging_is_bounded_and_reconciled(tmp_path):
+    # This is the deterministic artifact left by a killed offline initializer.
+    pending = tmp_path / ".operator.json.arbiter-pending"
+    pending.write_bytes(b"interrupted private staging")
+    pending.chmod(0o600)
+    record = initialize(tmp_path)
+    assert a.authority_status(tmp_path / "state") == record
+    assert not pending.exists()
+    assert not list(tmp_path.glob("*.pending"))
+
+
+def test_real_kill_after_link_preserves_keys_and_retry_completes(tmp_path):
+    ready = tmp_path / "linked"
+    script = """
+import sys, time
+from pathlib import Path
+from arbiter_server import authority, recovery
+root = Path(sys.argv[1])
+real = recovery.os.link
+def pause(source, target):
+    real(source, target)
+    (root / "linked").write_text("ready")
+    time.sleep(60)
+recovery.os.link = pause
+authority.initialize_authority(root / "state", recovery_file=root / "operator.json")
+"""
+    child = subprocess.Popen(
+        [sys.executable, "-c", script, str(tmp_path)],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+    try:
+        deadline = time.monotonic() + 15
+        while (
+            not ready.exists() and child.poll() is None and time.monotonic() < deadline
+        ):
+            time.sleep(0.02)
+        assert ready.exists(), "child did not reach the hard-link publication boundary"
+    finally:
+        if child.poll() is None:
+            child.kill()
+        child.communicate(timeout=15)
+    file = tmp_path / "operator.json"
+    pending = tmp_path / ".operator.json.arbiter-pending"
+    assert file.stat().st_nlink == pending.stat().st_nlink == 2
+    before = read(tmp_path)
+    record = initialize(tmp_path, reuse=True)
+    assert a.authority_status(tmp_path / "state") == record
+    assert not pending.exists() and file.stat().st_nlink == 1
+    after = read(tmp_path)
+    for name in ("recovery_signer_pkcs8", "backup_x25519_pkcs8", "owner_id"):
+        assert before[name] == after[name]
+    assert before["backup_origins_spki"].items() <= after["backup_origins_spki"].items()
+
+
+@pytest.mark.parametrize("state", ["third-link", "mismatched-inode"])
+def test_reconciliation_never_accepts_unrelated_hardlinks(tmp_path, state):
+    initialize(tmp_path)
+    target = tmp_path / "operator.json"
+    pending = tmp_path / ".operator.json.arbiter-pending"
+    before = target.read_bytes()
+    if state == "third-link":
+        os.link(target, pending)
+        os.link(target, tmp_path / "extra")
+    else:
+        pending.write_bytes(before)
+        pending.chmod(0o600)
+        os.link(pending, tmp_path / "extra")
+    with pytest.raises(a.AuthorityError, match="exactly"):
+        initialize(tmp_path, "other", reuse=True)
+    assert target.read_bytes() == before and pending.read_bytes() == before
+    assert not (tmp_path / "other" / a.DATABASE).exists()
+
+
+@pytest.mark.skipif(os.name != "nt", reason="native Windows recovery custody ACL")
+@pytest.mark.parametrize("damage", ["readable", "foreign-owner"])
+def test_windows_recovery_file_acl_is_not_repaired(tmp_path, damage):
+    initialize(tmp_path)
+    file = tmp_path / "operator.json"
+    original = file.read_bytes()
+    command = (
+        ["icacls", str(file), "/grant", "*S-1-5-32-545:R"]
+        if damage == "readable"
+        else ["icacls", str(file), "/setowner", "*S-1-5-32-545"]
+    )
+    subprocess.run(command, check=True, capture_output=True, text=True)
+    before_acl = subprocess.check_output(["icacls", str(file)], text=True)
+    with pytest.raises(a.AuthorityError):
+        initialize(tmp_path, "other", reuse=True)
+    assert file.read_bytes() == original
+    assert subprocess.check_output(["icacls", str(file)], text=True) == before_acl
diff --git a/website/docs/operate/credentials.md b/website/docs/operate/credentials.md
index 6718f74..01e5828 100644
--- a/website/docs/operate/credentials.md
+++ b/website/docs/operate/credentials.md
@@ -62,8 +62,9 @@ Certificate status output does not export private material. The setup transactio
 must succeed completely before publication; failure does not replace existing
 IDs or keys. Never copy the authority database into an agent workspace.
 
-Authority schema 2 requires the complete credential set. The earlier experimental
-IDs-only schema 1 is rejected without modification; there is no implicit upgrade
+Authority schema 3 requires the complete credential set and registered recovery
+trust. Earlier experimental schemas 1 and 2 are rejected without modification;
+there is no implicit upgrade
 or reinitialization. Preserve old state rather than deleting identity records to
 work around a version error.
 
@@ -71,3 +72,46 @@ Shared public-only verification vectors are in
 `tests/fixtures/authentication/credential-profiles-v1.json`. They are synthetic
 test trust, consumed by Python now and intended for Go's slice 2.1 verifier.
 They must never become product trust or default credentials.
+
+## Operator recovery custody
+
+Setup requires `--recovery-file PATH` in an existing operator-owned private
+directory outside server state. POSIX directories must be private (normally
+`0700`); files use `0600`. Windows uses the same operator/SYSTEM/Administrators
+ownership and inheritable ACL boundary as authority state. Permissions and
+aliases are validated, never repaired. The parent retains a private
+`.authority.lock` to serialize publication and explicit reuse. Do not delete a
+lock while setup is running.
+
+The version-one JSON file starts with `NEVER SHARE THIS`. It holds a separate
+P-256 recovery signer and X25519 backup decryptor as base64 canonical PKCS8 DER,
+the owner ID, and trusted per-server backup-origin SPKI public keys. Keys are
+unencrypted within this private file. Its combined powers are intentional:
+anyone who steals it can decrypt backups and authorize live identity recovery.
+Keep a safe copy outside the server, agents, export jobs and synced bundles.
+Loss of every copy makes future encrypted backups unrecoverable.
+
+Server state contains only the recovery verifier and backup recipient SPKI,
+plus a separate server-held backup-origin signing key. Private external recovery
+keys never enter that state or status output. The X25519 public/private records
+use standard DER; the future age export adapter will encode them as age
+identities/recipients. This setup does not implement export, restore or live
+credential recovery yet.
+
+Initialization refuses an existing recovery file unless you select
+`--reuse-recovery-file`. Explicit reuse preserves both private keys and the owner
+ID and adds this server's public backup-origin verifier without removing earlier
+verifiers. It does not share server identities or grant access between servers.
+Reuse never regenerates existing private keys; there is no force-reset option.
+
+Custody publishes before the complete authority database. If setup fails before
+custody publication, retry normally. If it publishes the recovery file but not
+the database, retain the file and retry with explicit reuse. An unused public
+origin entry can remain from that failed attempt; it grants no live access.
+A killed initializer can leave one bounded private staging file, named
+`.FILENAME.arbiter-pending`; the next setup reconciles it under the parent lock.
+If the database already exists, inspect `authority status`; never reinitialize
+or delete durable identity records to bypass an error. Status needs only the
+registered public trust, so the recovery file can be moved to safe external
+custody after setup. Filesystem flush failures are reported as failures even if
+publication already occurred; inspect both destinations before retrying.
diff --git a/website/docs/operate/security.md b/website/docs/operate/security.md
index 88c715e..433469a 100644
--- a/website/docs/operate/security.md
+++ b/website/docs/operate/security.md
@@ -167,7 +167,8 @@ explicit.
 The first authentication milestone provides stopped-instance identity setup:
 
 ```bash
-arbiter-server authority init --state-dir /private/path/authority
+arbiter-server authority init --state-dir /private/path/authority \
+  --recovery-file /operator-only/private/operator-recovery.json
 arbiter-server authority status --state-dir /private/path/authority --json
 ```
 
@@ -177,6 +178,12 @@ configuration, load service plugins, or contact upstream accounts. A second
 initialization fails instead of replacing existing identity records. Status
 reports absent or damaged state without creating a replacement database.
 
+Prepare the recovery file's private parent directory separately. Keep the file
+outside the authority directory, server/agent storage and synced backup storage.
+It holds both backup-decryption and live-recovery authority; theft compromises
+both. See [operator recovery custody](credentials.md#operator-recovery-custody)
+for reuse and interrupted setup.
+
 Stop the instance before initializing. Concurrent initialization attempts use
 one persistent process lock. A failed unpublished initialization can be retried;
 a database already published must be inspected rather than overwritten. Keep