diff --git a/server/news/+durable-authority-generations.feature.md b/server/news/+durable-authority-generations.feature.md
new file mode 100644
index 0000000..74c35a3
--- /dev/null
+++ b/server/news/+durable-authority-generations.feature.md
@@ -0,0 +1,6 @@
+Authority setup publishes complete private generations with a durable active
+selector and restart reconciliation. Interrupted publication preserves the exact
+initialized IDs and trust; damaged or missing saved state never creates a new
+identity. Server startup holds the same instance lock as offline setup through
+HTTP shutdown, preventing concurrent writers. Windows publication uses native
+write-through moves. Runtime API authentication remains a later milestone.
diff --git a/server/src/arbiter_server/authority.py b/server/src/arbiter_server/authority.py
index f14aa34..9ec12fb 100644
--- a/server/src/arbiter_server/authority.py
+++ b/server/src/arbiter_server/authority.py
@@ -240,14 +240,6 @@ def unlock() -> None:
             unlock()
 
 
-def _remove_pending(root: Path) -> None:
-    for name in (_PENDING, _PENDING + "-journal"):
-        path = root / name
-        if path.exists() or path.is_symlink():
-            _check_path(path)
-            path.unlink()
-
-
 def _write_database(path: Path, record: AuthorityRecord) -> None:
     from .credentials import store_setup_credentials
 
@@ -279,7 +271,24 @@ def _write_database(path: Path, record: AuthorityRecord) -> None:
 
 
 def _publish(pending: Path, target: Path) -> None:
-    os.replace(pending, target)
+    if os.name == "nt":
+        import ctypes
+        from ctypes import wintypes
+
+        win_dll = getattr(ctypes, "WinDLL", None)
+        if win_dll is None:
+            raise OSError("Windows durable publication is unavailable")
+        move = win_dll("kernel32", use_last_error=True).MoveFileExW
+        move.argtypes = [wintypes.LPCWSTR, wintypes.LPCWSTR, wintypes.DWORD]
+        move.restype = wintypes.BOOL
+        # Same-volume moves, including directories; never permit copy-and-delete.
+        # MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH.
+        if not move(str(pending), str(target), 0x1 | 0x8):
+            raise OSError(
+                getattr(ctypes, "get_last_error")(), "MoveFileExW failed", str(target)
+            )
+    else:
+        os.replace(pending, target)
     _flush_parents(target.parent)
 
 
@@ -302,8 +311,9 @@ def initialize_authority(
     """Create one private set of IDs, refusing to replace any existing database."""
     root = _root(state_dir, create=True)
     with _locked(root):
-        target = root / DATABASE
-        if target.exists() or target.is_symlink():
+        from . import generations
+
+        if generations.reconcile(root) is not None:
             raise AuthorityError(
                 "authority database already exists; inspect authority status"
             )
@@ -317,74 +327,87 @@ def initialize_authority(
             path,
             material,
         ):
-            _remove_pending(root)
             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)
+            with generations.staging(root) as staged:
+                _write_database(staged / _PENDING, record)
+                register_and_publish(path, record, material, staged / _PENDING)
+                _publish(staged / _PENDING, staged / DATABASE)
+                generations.activate(root, staged)
             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
-
+    """Inspect authority, reconciling publication without repairing database contents."""
     root = _root(state_dir, create=False)
     if not root.exists():
         return None
     with _locked(root):
-        path = root / DATABASE
-        if not path.exists() and not path.is_symlink():
-            return None
-        _check_path(path)
-        try:
-            with closing(
-                sqlite3.connect(path.as_uri() + "?mode=ro", uri=True, timeout=0)
-            ) as db:
-                version = db.execute("PRAGMA user_version").fetchone()[0]
-                if version != SCHEMA_VERSION:
-                    raise AuthorityError(
-                        "unsupported authority database schema version"
-                    )
-                rows = db.execute(
-                    "SELECT singleton, owner_id, server_id, created_at FROM authority"
-                ).fetchall()
-                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])
+        return _current_record(root)
+
+
+def _current_record(root: Path) -> AuthorityRecord | None:
+    from .generations import reconcile
+
+    path = reconcile(root)
+    return _read_record(path) if path is not None else None
+
+
+@contextmanager
+def running_instance(state_dir: Path) -> Iterator[AuthorityRecord | None]:
+    """Own the same instance lock as stopped-instance setup for the full lifetime."""
+    root = _root(state_dir, create=True)
+    with _locked(root):
+        yield _current_record(root)
+
+
+def _read_record(path: Path) -> AuthorityRecord:
+    from .credentials import verify_setup_credentials
+    from .recovery import verify_registration
+
+    _check_path(path)
+    # Keep database verification separate from instance-lock acquisition so the
+    # owning server can inspect its live state without trying to lock itself.
+    try:
+        with closing(
+            sqlite3.connect(path.as_uri() + "?mode=ro", uri=True, timeout=0)
+        ) as db:
+            version = db.execute("PRAGMA user_version").fetchone()[0]
+            if version != SCHEMA_VERSION:
+                raise AuthorityError("unsupported authority database schema version")
+            rows = db.execute(
+                "SELECT singleton, owner_id, server_id, created_at FROM authority"
+            ).fetchall()
+            credentials = db.execute(
+                "SELECT role, certificate_der, private_key_der FROM credentials"
+            ).fetchall()
             if len(rows) != 1:
                 raise ValueError("expected one authority record")
-            singleton, owner, server, created = rows[0]
-            if singleton != 1:
-                raise ValueError("invalid authority singleton")
-            for value in (owner, server):
-                if not isinstance(value, str) or str(UUID(value, version=4)) != value:
-                    raise ValueError("invalid authority ID")
-            if owner == server or not isinstance(created, str):
-                raise ValueError("invalid authority record")
-            when = datetime.fromisoformat(created)
-            if when.tzinfo is None or when.utcoffset() != timezone.utc.utcoffset(None):
-                raise ValueError("invalid creation timestamp")
-            # Offline inspection checks the initial setup at its recorded time.
-            # Network verifiers must use the current time, independently.
-            verify_setup_credentials(credentials, owner, server, when)
-            return AuthorityRecord(owner, server, created)
-        except (sqlite3.Error, ValueError) as exc:
-            if isinstance(exc, AuthorityError):
-                raise
-            raise AuthorityError(
-                "invalid or incomplete authority database; refusing to replace it"
-            ) from exc
+            verify_registration(db, rows[0][1])
+        if len(rows) != 1:
+            raise ValueError("expected one authority record")
+        singleton, owner, server, created = rows[0]
+        if singleton != 1:
+            raise ValueError("invalid authority singleton")
+        for value in (owner, server):
+            if not isinstance(value, str) or str(UUID(value, version=4)) != value:
+                raise ValueError("invalid authority ID")
+        if owner == server or not isinstance(created, str):
+            raise ValueError("invalid authority record")
+        when = datetime.fromisoformat(created)
+        if when.tzinfo is None or when.utcoffset() != timezone.utc.utcoffset(None):
+            raise ValueError("invalid creation timestamp")
+        # Offline inspection checks the initial setup at its recorded time.
+        # Network verifiers must use the current time, independently.
+        verify_setup_credentials(credentials, owner, server, when)
+        return AuthorityRecord(owner, server, created)
+    except (sqlite3.Error, ValueError) as exc:
+        if isinstance(exc, AuthorityError):
+            raise
+        raise AuthorityError(
+            "invalid or incomplete authority database; refusing to replace it"
+        ) from exc
 
 
 def authority_cli(
diff --git a/server/src/arbiter_server/generations.py b/server/src/arbiter_server/generations.py
new file mode 100644
index 0000000..a5b271f
--- /dev/null
+++ b/server/src/arbiter_server/generations.py
@@ -0,0 +1,224 @@
+"""Private initial-state publication, always under the authority instance lock."""
+
+from __future__ import annotations
+
+from contextlib import contextmanager
+import hashlib
+from itertools import islice
+import json
+import os
+from pathlib import Path
+from typing import Iterator
+from uuid import UUID, uuid4
+
+from . import authority as a
+
+ACTIVE = "active-generation.json"
+_INTENT = ".generation-intent.json"
+_CONTROL_PENDING = ".generation-control.pending"
+_STAGING = ".generation.pending"
+
+
+def _present(path: Path) -> bool:
+    return path.exists() or path.is_symlink()
+
+
+def _base(root: Path, *, create: bool = False) -> Path:
+    return a._root(root / "generations", create=create)
+
+
+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 generation field")
+        result[key] = value
+    return result
+
+
+def _control(path: Path, *, intent: bool = False) -> dict[str, str] | None:
+    if not _present(path):
+        return None
+    a._check_path(path)
+    with path.open("rb") as handle:
+        data = handle.read(513)
+    try:
+        if len(data) > 512:
+            raise ValueError("oversize control record")
+        record = json.loads(data, object_pairs_hook=_unique)
+        keys = {"version", "active", "sha256"} if intent else {"version", "active"}
+        if not isinstance(record, dict) or set(record) != keys:
+            raise ValueError("wrong fields")
+        if type(record["version"]) is not int or record["version"] != 1:
+            raise ValueError("unsupported generation record")
+        value = record["active"]
+        if not isinstance(value, str) or str(UUID(value, version=4)) != value:
+            raise ValueError("invalid generation ID")
+        result = {"active": value}
+        if intent:
+            digest = record["sha256"]
+            if (
+                not isinstance(digest, str)
+                or len(digest) != 64
+                or any(c not in "0123456789abcdef" for c in digest)
+            ):
+                raise ValueError("invalid generation digest")
+            result["sha256"] = digest
+        return result
+    except (ValueError, TypeError, RecursionError) as exc:
+        raise a.AuthorityError(
+            "invalid active generation or publication intent"
+        ) from exc
+
+
+def _write_control(root: Path, name: str, record: dict[str, object]) -> None:
+    pending = root / _CONTROL_PENDING
+    fd = os.open(pending, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
+    try:
+        with os.fdopen(fd, "wb") as handle:
+            a._check_path(pending)
+            handle.write((json.dumps(record, sort_keys=True) + "\n").encode())
+            handle.flush()
+            os.fsync(handle.fileno())
+        a._publish(pending, root / name)
+    finally:
+        if _present(pending):
+            a._check_path(pending)
+            pending.unlink()
+
+
+def _digest(path: Path) -> str:
+    a._check_path(path)
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(65536), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _remove_control_pending(root: Path) -> None:
+    path = root / _CONTROL_PENDING
+    if _present(path):
+        a._check_path(path)
+        path.unlink()
+
+
+def _remove_staging(root: Path) -> None:
+    path = _base(root) / _STAGING
+    if not _present(path):
+        return
+    a._check_path(path, directory=True)
+    allowed = {a.DATABASE, a._PENDING, a._PENDING + "-journal"}
+    files = list(islice(path.iterdir(), len(allowed) + 1))
+    if len(files) > len(allowed) or any(p.name not in allowed for p in files):
+        raise a.AuthorityError("unexpected files in unpublished generation")
+    for file in files:
+        a._check_path(file)
+    for file in files:
+        file.unlink()
+    path.rmdir()
+
+
+def _database(root: Path, generation: str) -> Path:
+    directory = _base(root) / generation
+    a._check_path(directory, directory=True)
+    path = directory / a.DATABASE
+    a._check_path(path)
+    return path
+
+
+def _check_references(root: Path, allowed: set[str], *, selected: bool) -> None:
+    base = _base(root)
+    if not base.exists():
+        return
+    entries = list(islice(base.iterdir(), len(allowed) + 1))
+    if len(entries) > len(allowed) or any(p.name not in allowed for p in entries):
+        reason = (
+            "unreferenced generation"
+            if selected
+            else "saved generation has no active selector"
+        )
+        raise a.AuthorityError(f"{reason}; preserve existing state")
+
+
+def reconcile(root: Path) -> Path | None:
+    """Finish only our durable initial publication; never choose another identity."""
+    for name in (a.DATABASE, a._PENDING, a._PENDING + "-journal"):
+        if _present(root / name):
+            a._check_path(root / name)
+            raise a.AuthorityError(
+                "unsupported flat authority layout; preserve existing state"
+            )
+    active = _control(root / ACTIVE)
+    intent = _control(root / _INTENT, intent=True)
+    allowed = {_STAGING}
+    for reference in (active, intent):
+        if reference is not None:
+            allowed.add(reference["active"])
+    # Check the bounded reference set before publishing or removing anything.
+    _check_references(root, allowed, selected=active is not None or intent is not None)
+    _remove_control_pending(root)
+    if intent is not None:
+        if active is not None and active["active"] != intent["active"]:
+            raise a.AuthorityError(
+                "active generation conflicts with publication intent"
+            )
+        base = _base(root)
+        staged, final = base / _STAGING, base / intent["active"]
+        if active is not None and not _present(final):
+            raise a.AuthorityError(
+                "active generation is missing; preserve existing state"
+            )
+        source = final if _present(final) else staged
+        if _present(final) and _present(staged):
+            raise a.AuthorityError("ambiguous pending generation")
+        a._check_path(source, directory=True)
+        database = source / a.DATABASE
+        if active is None:
+            if [p.name for p in islice(source.iterdir(), 2)] != [a.DATABASE]:
+                raise a.AuthorityError("unexpected files in initial generation")
+            if _digest(database) != intent["sha256"]:
+                raise a.AuthorityError("pending generation digest mismatch")
+        # An already durable selector is authoritative. Its live DB can change
+        # after activation; a stale intent must never roll those changes back.
+        a._read_record(database)
+        if source == staged:
+            a._publish(staged, final)
+        if active is None:
+            active = {"active": intent["active"]}
+            _write_control(root, ACTIVE, {"version": 1, **active})
+        (root / _INTENT).unlink()
+        a._flush_parents(root)
+    _remove_staging(root)
+    if active is None:
+        base = _base(root)
+        if base.exists() and next(base.iterdir(), None) is not None:
+            raise a.AuthorityError(
+                "saved generation has no active selector; preserve existing state"
+            )
+        return None
+    return _database(root, active["active"])
+
+
+@contextmanager
+def staging(root: Path) -> Iterator[Path]:
+    """Reserve the one unpublished directory; retain durable intent on failure."""
+    directory = _base(root, create=True) / _STAGING
+    directory.mkdir(mode=0o777 if os.name == "nt" else 0o700)
+    a._check_path(directory, directory=True)
+    try:
+        yield directory
+    finally:
+        if not _present(root / _INTENT):
+            _remove_staging(root)
+
+
+def activate(root: Path, staged: Path) -> None:
+    database = staged / a.DATABASE
+    a._read_record(database)
+    _write_control(
+        root,
+        _INTENT,
+        {"version": 1, "active": str(uuid4()), "sha256": _digest(database)},
+    )
+    reconcile(root)
diff --git a/server/src/arbiter_server/main.py b/server/src/arbiter_server/main.py
index 08635d3..2a9fbb0 100644
--- a/server/src/arbiter_server/main.py
+++ b/server/src/arbiter_server/main.py
@@ -2272,10 +2272,13 @@ def _create_native_http_app(
 
     @asynccontextmanager
     async def lifespan(_app: object) -> AsyncGenerator[None, None]:
-        try:
-            yield
-        finally:
-            http_work.close()
+        from .authority import running_instance
+
+        with running_instance(_server_data_root(_storage_config(cfg)) / "authority"):
+            try:
+                yield
+            finally:
+                http_work.close()
 
     async def health(_request: Request) -> object:
         return _native_response({"status": "ok"})
diff --git a/server/tests/unit/test_artifact_access_logs.py b/server/tests/unit/test_artifact_access_logs.py
index b4a0855..5f6e1b6 100644
--- a/server/tests/unit/test_artifact_access_logs.py
+++ b/server/tests/unit/test_artifact_access_logs.py
@@ -26,6 +26,7 @@
 
 
 @pytest.mark.parametrize("encoded_route", [False, True])
+@pytest.mark.usefixtures("private_windows_test_parent")
 def test_live_artifact_tokens_are_absent_from_native_access_logs(
     encoded_route: bool,
     tmp_path: Path,
diff --git a/server/tests/unit/test_artifact_http_admission.py b/server/tests/unit/test_artifact_http_admission.py
index 289c42a..b3242db 100644
--- a/server/tests/unit/test_artifact_http_admission.py
+++ b/server/tests/unit/test_artifact_http_admission.py
@@ -379,6 +379,7 @@ def wire_request(port, method, path):
     "method,content", [("GET", False), ("HEAD", True), ("GET", True)]
 )
 @pytest.mark.parametrize("valid_nonce", [False, True])
+@pytest.mark.usefixtures("private_windows_test_parent")
 def test_real_socket_artifact_contention_does_not_delay_health(
     tmp_path, monkeypatch, method, content, valid_nonce
 ):
diff --git a/server/tests/unit/test_authority.py b/server/tests/unit/test_authority.py
index fb0ed71..0175ee5 100644
--- a/server/tests/unit/test_authority.py
+++ b/server/tests/unit/test_authority.py
@@ -20,6 +20,16 @@
 pytestmark = pytest.mark.usefixtures("private_windows_test_parent")
 
 
+def dbpath(root):
+    selector = root / "active-generation.json"
+    name = (
+        json.loads(selector.read_text())["active"]
+        if selector.exists()
+        else ".generation.pending"
+    )
+    return root / "generations" / name / authority.DATABASE
+
+
 def initialize(root):
     parent = next(
         p
@@ -42,12 +52,12 @@ def test_initialize_reopen_and_independent_roots(tmp_path: Path) -> None:
     assert {record.owner_id, record.server_id}.isdisjoint(
         {other.owner_id, other.server_id}
     )
-    with closing(sqlite3.connect(root / authority.DATABASE)) as db:
+    with closing(sqlite3.connect(dbpath(root))) as db:
         assert db.execute("PRAGMA integrity_check").fetchone() == ("ok",)
         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
+        assert (dbpath(root)).stat().st_mode & 0o777 == 0o600
 
 
 def test_database_flush_has_write_access(tmp_path: Path, monkeypatch) -> None:
@@ -71,17 +81,17 @@ def test_status_missing_is_not_initialization(tmp_path: Path) -> None:
     assert not root.exists()
     root.mkdir(mode=0o777 if os.name == "nt" else 0o700)
     assert authority.authority_status(root) is None
-    assert not (root / authority.DATABASE).exists()
+    assert not (dbpath(root)).exists()
 
 
 def test_second_initialization_preserves_database(tmp_path: Path) -> None:
     record = initialize(tmp_path / "state")
-    path = tmp_path / "state" / authority.DATABASE
+    path = dbpath(tmp_path / "state")
     original = path.read_bytes()
     with pytest.raises(authority.AuthorityError, match="already exists"):
-        initialize(path.parent)
+        initialize(tmp_path / "state")
     assert path.read_bytes() == original
-    assert authority.authority_status(path.parent) == record
+    assert authority.authority_status(tmp_path / "state") == record
     assert path.read_bytes() == original
 
 
@@ -135,7 +145,7 @@ def no_composition(*args, **kwargs):
 def test_corrupt_state_never_reinitialized(tmp_path: Path, damage: str) -> None:
     root = tmp_path / "state"
     initialize(root)
-    path = root / authority.DATABASE
+    path = dbpath(root)
     if damage == "garbage":
         path.write_bytes(b"not a database")
     else:
@@ -168,7 +178,7 @@ def test_broad_permissions_rejected(tmp_path: Path, target: str) -> None:
     initialize(root)
     path = {
         "directory": root,
-        "database": root / authority.DATABASE,
+        "database": dbpath(root),
         "lock": root / ".authority.lock",
     }[target]
     path.chmod(0o755 if target == "directory" else 0o644)
@@ -207,7 +217,7 @@ def test_directory_alias_and_hardlinked_database_rejected(tmp_path: Path) -> Non
     alias.symlink_to(root, target_is_directory=True)
     with pytest.raises(authority.AuthorityError, match="symlink"):
         authority.authority_status(alias)
-    os.link(root / authority.DATABASE, tmp_path / "linked-db")
+    os.link(dbpath(root), tmp_path / "linked-db")
     with pytest.raises(authority.AuthorityError, match="exactly one link"):
         authority.authority_status(root)
 
@@ -237,7 +247,7 @@ def test_nested_missing_parent_rejected_after_creation(
     assert root.exists()
     assert root.stat().st_mode & 0o777 == 0o700
     assert root.parent.stat().st_mode & 0o022
-    assert not (root / authority.DATABASE).exists()
+    assert not (dbpath(root)).exists()
     assert not (root / authority._LOCK).exists()
 
 
@@ -280,7 +290,7 @@ def fail_write(path, record):
         raise OSError("injected write failure")
 
     def fail_publish(pending, target):
-        if target.name != authority.DATABASE:
+        if target.name != "active-generation.json":
             return real_publish(pending, target)
         if failure == "after_publish":
             real_publish(pending, target)
@@ -292,11 +302,18 @@ def fail_publish(pending, target):
     monkeypatch.setattr(authority, "_publish", fail_publish)
     with pytest.raises(OSError, match="injected"):
         initialize(root)
-    assert not (root / ".authority.pending.sqlite3").exists()
-    assert not (root / ".authority.pending.sqlite3-journal").exists()
+    assert not (
+        root / "generations" / ".generation.pending" / ".authority.pending.sqlite3"
+    ).exists()
+    assert not (
+        root
+        / "generations"
+        / ".generation.pending"
+        / ".authority.pending.sqlite3-journal"
+    ).exists()
     monkeypatch.setattr(authority, "_write_database", real_write)
     monkeypatch.setattr(authority, "_publish", real_publish)
-    if failure == "after_publish":
+    if failure != "write":
         assert authority.authority_status(root) is not None
         with pytest.raises(authority.AuthorityError, match="already exists"):
             initialize(root)
@@ -352,22 +369,30 @@ def partial(path, record):
     )
     try:
         _wait_for_file(ready, child)
-        assert not (root / authority.DATABASE).exists()
+        assert not (dbpath(root)).exists()
         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()
+    assert (
+        root / "generations" / ".generation.pending" / ".authority.pending.sqlite3"
+    ).exists()
     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"}
+    assert not (
+        root / "generations" / ".generation.pending" / ".authority.pending.sqlite3"
+    ).exists()
+    assert {p.name for p in root.iterdir()} == {
+        "generations",
+        "active-generation.json",
+        ".authority.lock",
+    }
     if recovery_before:
         assert (
             json.loads(recovery_path.read_text())["recovery_signer_pkcs8"]
@@ -590,7 +615,7 @@ def test_windows_broad_directory_acl_rejected_without_repair(tmp_path: Path) ->
 def test_windows_broad_database_acl_rejected(tmp_path: Path) -> None:
     root = tmp_path / "state"
     initialize(root)
-    path = root / authority.DATABASE
+    path = dbpath(root)
     subprocess.run(
         ["icacls", str(path), "/grant", "*S-1-5-32-545:R"],
         check=True,
diff --git a/server/tests/unit/test_credentials.py b/server/tests/unit/test_credentials.py
index 3922bab..e079996 100644
--- a/server/tests/unit/test_credentials.py
+++ b/server/tests/unit/test_credentials.py
@@ -19,6 +19,16 @@
 pytestmark = pytest.mark.usefixtures("private_windows_test_parent")
 
 
+def dbpath(root):
+    selector = root / "active-generation.json"
+    name = (
+        json.loads(selector.read_text())["active"]
+        if selector.exists()
+        else ".generation.pending"
+    )
+    return root / "generations" / name / authority.DATABASE
+
+
 def initialize(root):
     parent = next(
         p
@@ -286,7 +296,7 @@ def test_setup_is_real_atomic_private_and_restartable(tmp_path, capsys):
     assert (
         record is not None and record.owner_id in output and record.server_id in output
     )
-    with closing(sqlite3.connect(root / authority.DATABASE)) as db:
+    with closing(sqlite3.connect(dbpath(root))) as db:
         rows = db.execute(
             "SELECT role, certificate_der, private_key_der FROM credentials"
         ).fetchall()
@@ -299,12 +309,16 @@ def test_setup_is_real_atomic_private_and_restartable(tmp_path, capsys):
     )
     for _, _, private in rows:
         assert base64.b64encode(private).decode() not in output
-    before = (root / authority.DATABASE).read_bytes()
+    before = (dbpath(root)).read_bytes()
     with pytest.raises(authority.AuthorityError, match="already exists"):
         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}
+    assert (dbpath(root)).read_bytes() == before
+    assert {p.name for p in root.iterdir()} == {
+        "generations",
+        "active-generation.json",
+        authority._LOCK,
+    }
 
 
 @pytest.mark.parametrize(
@@ -314,7 +328,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"
     initialize(root)
-    with closing(sqlite3.connect(root / authority.DATABASE)) as db:
+    with closing(sqlite3.connect(dbpath(root))) as db:
         if damage == "old-schema":
             db.execute("PRAGMA user_version=1")
         elif damage == "missing":
@@ -336,10 +350,10 @@ def test_damaged_setup_never_repairs_or_replaces_keys(tmp_path, damage, capsys):
                 f"UPDATE credentials SET {column}=? WHERE role='owner'", (value,)
             )
         db.commit()
-    before = (root / authority.DATABASE).read_bytes()
+    before = (dbpath(root)).read_bytes()
     with pytest.raises(authority.AuthorityError):
         authority.authority_status(root)
-    assert (root / authority.DATABASE).read_bytes() == before
+    assert (dbpath(root)).read_bytes() == before
     if damage == "unsupported-key":
         assert authority.authority_cli("status", root, as_json=True) == 1
         captured = capsys.readouterr()
@@ -358,7 +372,7 @@ def fail(db, owner, server, now):
     monkeypatch.setattr(c, "store_setup_credentials", fail)
     with pytest.raises(OSError, match="injected"):
         initialize(root)
-    assert {p.name for p in root.iterdir()} == {authority._LOCK}
+    assert {p.name for p in root.iterdir()} == {authority._LOCK, "generations"}
     monkeypatch.setattr(c, "store_setup_credentials", real)
     assert initialize(root) == authority.authority_status(root)
 
diff --git a/server/tests/unit/test_generations.py b/server/tests/unit/test_generations.py
new file mode 100644
index 0000000..d9e9b10
--- /dev/null
+++ b/server/tests/unit/test_generations.py
@@ -0,0 +1,370 @@
+from __future__ import annotations
+
+import asyncio
+from contextlib import closing
+import json
+import os
+from pathlib import Path
+import shutil
+import sqlite3
+import subprocess
+import sys
+import time
+from typing import Any, cast
+from uuid import uuid4
+
+import pytest
+
+from arbiter_server import authority as a, generations as g
+from arbiter_server.config import AppConfig
+from arbiter_server.main import build_server
+from server.tests.unit.test_native_http_protocol import _asgi_request
+
+pytestmark = pytest.mark.usefixtures("private_windows_test_parent")
+
+
+def initialize(tmp_path):
+    return a.initialize_authority(
+        tmp_path / "state", recovery_file=tmp_path / "operator.json"
+    )
+
+
+def test_generation_restart_preserves_complete_identity_and_trust(tmp_path):
+    record = initialize(tmp_path)
+    root = tmp_path / "state"
+    active = root / g.ACTIVE
+    selection = active.read_bytes()
+    generation = json.loads(selection)["active"]
+    directory = root / "generations" / generation
+    assert set(p.name for p in directory.iterdir()) == {a.DATABASE}
+    database = directory / a.DATABASE
+    original = database.read_bytes()
+    assert a.authority_status(root) == record
+    with a.running_instance(root) as current:
+        assert current == record
+        with pytest.raises(a.AuthorityError, match="busy"):
+            a.authority_status(root)
+    assert database.read_bytes() == original and active.read_bytes() == selection
+    database.unlink()
+    with pytest.raises(OSError):
+        a.authority_status(root)
+    with pytest.raises(OSError):
+        initialize(tmp_path)
+    assert active.read_bytes() == selection
+
+
+def test_stale_intent_cannot_roll_back_live_database_changes(tmp_path, monkeypatch):
+    real = a._publish
+    saved = []
+
+    def retain_intent(source, target):
+        real(source, target)
+        if target.name == g._INTENT:
+            saved.append(target.read_bytes())
+
+    monkeypatch.setattr(a, "_publish", retain_intent)
+    record = initialize(tmp_path)
+    root = tmp_path / "state"
+    database = (
+        root
+        / "generations"
+        / json.loads((root / g.ACTIVE).read_bytes())["active"]
+        / a.DATABASE
+    )
+    with closing(sqlite3.connect(database)) as db:
+        db.execute("CREATE TABLE synthetic_runtime(value TEXT)")
+        db.execute("INSERT INTO synthetic_runtime VALUES ('preserve')")
+        db.commit()
+    before = database.read_bytes()
+    intent = root / g._INTENT
+    intent.write_bytes(saved[0])
+    intent.chmod(0o600)
+    assert a.authority_status(root) == record
+    assert database.read_bytes() == before and not intent.exists()
+
+
+@pytest.mark.parametrize("phase", ["intent", "generation", "active"])
+def test_real_kill_reconciles_exact_durable_publication(tmp_path, phase):
+    root, ready = tmp_path / "state", tmp_path / "ready"
+    script = """
+import sys, time
+from pathlib import Path
+from arbiter_server import authority as a, generations as g
+root, ready, phase = Path(sys.argv[1]), Path(sys.argv[2]), sys.argv[3]
+real = a._publish
+def publish(source, target):
+    real(source, target)
+    if ((phase == "intent" and target.name == g._INTENT)
+        or (phase == "generation" and source.name == g._STAGING)
+        or (phase == "active" and target.name == g.ACTIVE)):
+        ready.write_text("ready")
+        time.sleep(60)
+a._publish = publish
+a.initialize_authority(root, recovery_file=root.parent / "operator.json")
+"""
+    child = subprocess.Popen(
+        [sys.executable, "-c", script, str(root), str(ready), phase],
+        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(), "initializer did not reach durable publication boundary"
+        paths = list((root / "generations").rglob(a.DATABASE))
+        assert len(paths) == 1
+        expected = a._read_record(paths[0])
+        custody = (tmp_path / "operator.json").read_bytes()
+    finally:
+        if child.poll() is None:
+            child.kill()
+        child.communicate(timeout=15)
+    assert a.authority_status(root) == expected
+    assert a.authority_status(root) == expected
+    assert (tmp_path / "operator.json").read_bytes() == custody
+    assert not (root / g._INTENT).exists()
+    assert not (root / g._CONTROL_PENDING).exists()
+    assert len(list((root / "generations").iterdir())) == 1
+    with pytest.raises(a.AuthorityError, match="already exists"):
+        initialize(tmp_path)
+
+
+@pytest.mark.parametrize(
+    "damage", ["digest", "conflict", "unknown-staging", "extra-with-intent"]
+)
+def test_corrupt_publication_never_selects_or_deletes_unproved_state(
+    tmp_path, monkeypatch, damage
+):
+    real = a._publish
+
+    def stop_after_intent(source, target):
+        real(source, target)
+        if target.name == g._INTENT:
+            raise OSError("injected after intent")
+
+    with monkeypatch.context() as patch:
+        patch.setattr(a, "_publish", stop_after_intent)
+        with pytest.raises(OSError, match="injected"):
+            initialize(tmp_path)
+    root = tmp_path / "state"
+    intent = root / g._INTENT
+    staged = root / "generations" / g._STAGING
+    before = (staged / a.DATABASE).read_bytes()
+    if damage == "digest":
+        record = json.loads(intent.read_bytes())
+        record["sha256"] = "0" * 64
+        intent.write_text(json.dumps(record))
+    elif damage == "conflict":
+        (root / g.ACTIVE).write_text(json.dumps({"version": 1, "active": str(uuid4())}))
+        (root / g.ACTIVE).chmod(0o600)
+    else:
+        if damage == "unknown-staging":
+            intent.unlink()
+        (staged / "unexpected").write_bytes(b"keep")
+    selection = (root / g.ACTIVE).read_bytes() if (root / g.ACTIVE).exists() else None
+    with pytest.raises(a.AuthorityError):
+        a.authority_status(root)
+    assert (staged / a.DATABASE).read_bytes() == before
+    assert (
+        (root / g.ACTIVE).read_bytes() if (root / g.ACTIVE).exists() else None
+    ) == selection
+
+
+@pytest.mark.parametrize(
+    "record",
+    [
+        b'{"version":1,"active":"../outside"}',
+        b'{"version":true,"active":"00000000-0000-4000-8000-000000000001"}',
+        b'{"version":1,"active":"x","active":"y"}',
+        b'{"version":1,"active":null}',
+        b"[" * 600,
+    ],
+)
+def test_malformed_selector_is_not_repaired(tmp_path, record):
+    initialize(tmp_path)
+    path = tmp_path / "state" / g.ACTIVE
+    path.write_bytes(record)
+    with pytest.raises(a.AuthorityError):
+        a.authority_status(path.parent)
+    assert path.read_bytes() == record
+
+
+def test_missing_selector_never_creates_a_replacement_identity(tmp_path):
+    initialize(tmp_path)
+    root = tmp_path / "state"
+    selection = (root / g.ACTIVE).read_bytes()
+    (root / g.ACTIVE).unlink()
+    custody = (tmp_path / "operator.json").read_bytes()
+    with pytest.raises(a.AuthorityError, match="no active selector"):
+        a.authority_status(root)
+    with pytest.raises(a.AuthorityError, match="no active selector"):
+        initialize(tmp_path)
+    assert len(list((root / "generations").iterdir())) == 1
+    assert (tmp_path / "operator.json").read_bytes() == custody
+    (root / g.ACTIVE).write_bytes(selection)
+    (root / g.ACTIVE).chmod(0o600)
+    assert a.authority_status(root) is not None
+
+
+@pytest.mark.parametrize("pending", [False, True])
+def test_unreferenced_generation_blocks_status_init_and_server_start(
+    tmp_path, monkeypatch, pending
+):
+    root = tmp_path / "server" / "authority"
+    recovery = tmp_path / "operator.json"
+    real = a._publish
+
+    def stop_after_intent(source, target):
+        real(source, target)
+        if target.name == g._INTENT:
+            raise OSError("injected after intent")
+
+    with monkeypatch.context() as patch:
+        if pending:
+            patch.setattr(a, "_publish", stop_after_intent)
+            with pytest.raises(OSError, match="injected"):
+                a.initialize_authority(root, recovery_file=recovery)
+        else:
+            a.initialize_authority(root, recovery_file=recovery)
+    base = root / "generations"
+    source = next(base.iterdir())
+    shutil.copytree(source, base / str(uuid4()))
+    before = {
+        p.relative_to(root): p.read_bytes() for p in root.rglob("*") if p.is_file()
+    }
+    custody = recovery.read_bytes()
+    with pytest.raises(a.AuthorityError, match="unreferenced generation"):
+        a.authority_status(root)
+    with pytest.raises(a.AuthorityError, match="unreferenced generation"):
+        a.initialize_authority(root, recovery_file=recovery)
+    cfg = AppConfig()
+    cfg.arbiter.storage.server_data_dir = str(root.parent)
+    cfg.arbiter.storage.plugin_data_dir = str(tmp_path / "plugins")
+    app = cast(Any, build_server(cfg, service_plugins=[], prepare_tls=False).app)
+
+    async def start():
+        with pytest.raises(a.AuthorityError, match="unreferenced generation"):
+            async with app.router.lifespan_context(app):
+                pytest.fail("server accepted an unreferenced generation")
+
+    asyncio.run(start())
+    assert {
+        p.relative_to(root): p.read_bytes() for p in root.rglob("*") if p.is_file()
+    } == before
+    assert recovery.read_bytes() == custody
+
+
+@pytest.mark.skipif(os.name == "nt", reason="POSIX modes and unprivileged links")
+@pytest.mark.parametrize(
+    "damage", ["selector-mode", "selector-link", "generation-link"]
+)
+def test_selector_and_generation_permissions_and_aliases_fail(tmp_path, damage):
+    initialize(tmp_path)
+    root = tmp_path / "state"
+    selector = root / g.ACTIVE
+    original = selector.read_bytes()
+    if damage == "selector-mode":
+        selector.chmod(0o644)
+    elif damage == "selector-link":
+        os.link(selector, tmp_path / "alias")
+    else:
+        directory = root / "generations" / json.loads(original)["active"]
+        other = root / "generations" / "original"
+        directory.rename(other)
+        directory.symlink_to(other, target_is_directory=True)
+    with pytest.raises(a.AuthorityError):
+        a.authority_status(root)
+    assert selector.read_bytes() == original
+
+
+@pytest.mark.skipif(os.name != "nt", reason="native Windows generation ACLs")
+@pytest.mark.parametrize("target", ["selector", "generation"])
+def test_windows_generation_acl_is_not_repaired(tmp_path, target):
+    initialize(tmp_path)
+    root = tmp_path / "state"
+    selector = root / g.ACTIVE
+    path = (
+        selector
+        if target == "selector"
+        else root / "generations" / json.loads(selector.read_bytes())["active"]
+    )
+    subprocess.run(
+        ["icacls", str(path), "/grant", "*S-1-5-32-545:M"],
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+    before = subprocess.check_output(["icacls", str(path)], text=True)
+    with pytest.raises(a.AuthorityError):
+        a.authority_status(root)
+    assert subprocess.check_output(["icacls", str(path)], text=True) == before
+
+
+@pytest.mark.parametrize("initialized", [False, True])
+def test_live_instance_process_excludes_setup_and_second_start(tmp_path, initialized):
+    root = tmp_path / "state"
+    record = initialize(tmp_path) if initialized else None
+    ready = tmp_path / "ready"
+    script = """
+import sys, time
+from pathlib import Path
+from arbiter_server.authority import running_instance
+with running_instance(Path(sys.argv[1])):
+    Path(sys.argv[2]).write_text("ready")
+    time.sleep(60)
+"""
+    child = subprocess.Popen([sys.executable, "-c", script, str(root), str(ready)])
+    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()
+        with pytest.raises(a.AuthorityError, match="busy"):
+            initialize(tmp_path)
+        with pytest.raises(a.AuthorityError, match="busy"):
+            with a.running_instance(root):
+                pytest.fail("second writer entered")
+    finally:
+        if child.poll() is None:
+            child.kill()
+        child.communicate(timeout=15)
+    assert a.authority_status(root) == record
+
+
+def test_actual_asgi_lifetime_owns_instance_lock_and_releases_it(tmp_path, monkeypatch):
+    cfg = AppConfig()
+    cfg.arbiter.storage.server_data_dir = str(tmp_path / "server")
+    cfg.arbiter.storage.plugin_data_dir = str(tmp_path / "plugins")
+    server = build_server(cfg, service_plugins=[], prepare_tls=False)
+    app = cast(Any, server.app)
+    root = tmp_path / "server" / "authority"
+    from arbiter_server.http_work import HTTPWork
+
+    real_close = HTTPWork.close
+    closed = []
+
+    def close_under_lock(work):
+        with pytest.raises(a.AuthorityError, match="busy"):
+            with a.running_instance(root):
+                pytest.fail("instance lock released before HTTP cleanup")
+        closed.append(True)
+        real_close(work)
+
+    monkeypatch.setattr(HTTPWork, "close", close_under_lock)
+
+    async def exercise():
+        async with app.router.lifespan_context(app):
+            with pytest.raises(a.AuthorityError, match="busy"):
+                a.initialize_authority(root, recovery_file=tmp_path / "operator.json")
+            assert (await _asgi_request(app, "GET", "/_health_")).status_code == 200
+
+    asyncio.run(exercise())
+    assert closed == [True]
+    assert a.initialize_authority(
+        root, recovery_file=tmp_path / "operator.json"
+    ) == a.authority_status(root)
diff --git a/server/tests/unit/test_native_http_protocol.py b/server/tests/unit/test_native_http_protocol.py
index f98a289..7608290 100644
--- a/server/tests/unit/test_native_http_protocol.py
+++ b/server/tests/unit/test_native_http_protocol.py
@@ -344,6 +344,7 @@ def _client(
         }
     }
     cfg.arbiter.storage.plugin_data_dir = str(tmp_path)
+    cfg.arbiter.storage.server_data_dir = str(tmp_path / "server")
     server = build_server(
         OmegaConf.structured(cfg),
         service_plugins=[cast(ServicePlugin, plugin or FakePlugin())],
@@ -505,6 +506,7 @@ def invoke_operation(self, operation, arguments, context):
     assert calls == 2
 
 
+@pytest.mark.usefixtures("private_windows_test_parent")
 def test_server_shutdown_does_not_block_the_loop_on_running_worker(tmp_path):
     entered = threading.Event()
     release = threading.Event()
diff --git a/server/tests/unit/test_operation_body_admission.py b/server/tests/unit/test_operation_body_admission.py
index 6e5662c..ea73d93 100644
--- a/server/tests/unit/test_operation_body_admission.py
+++ b/server/tests/unit/test_operation_body_admission.py
@@ -185,6 +185,7 @@ async def exercise():
 
 @pytest.mark.parametrize("chunked", [False, True])
 @pytest.mark.parametrize("finish", ["complete", "disconnect"])
+@pytest.mark.usefixtures("private_windows_test_parent")
 def test_real_socket_slow_body_admission_health_and_recovery(tmp_path, chunked, finish):
     import socket
     import time
@@ -250,6 +251,7 @@ def test_real_socket_slow_body_admission_health_and_recovery(tmp_path, chunked,
 
 @pytest.mark.parametrize("chunked", [False, True])
 @pytest.mark.parametrize("size,status", [(63, 200), (64, 200), (65, 413)])
+@pytest.mark.usefixtures("private_windows_test_parent")
 def test_real_socket_body_limit_boundaries_and_recovery(
     tmp_path, chunked, size, status
 ):
@@ -277,6 +279,7 @@ def test_real_socket_body_limit_boundaries_and_recovery(
 
 
 @pytest.mark.parametrize("busy", [False, True])
+@pytest.mark.usefixtures("private_windows_test_parent")
 def test_real_socket_oversized_declared_body_is_rejected_before_continue(
     tmp_path, busy
 ):
diff --git a/server/tests/unit/test_recovery.py b/server/tests/unit/test_recovery.py
index 2ebce9c..82cb95e 100644
--- a/server/tests/unit/test_recovery.py
+++ b/server/tests/unit/test_recovery.py
@@ -20,6 +20,16 @@
 pytestmark = pytest.mark.usefixtures("private_windows_test_parent")
 
 
+def dbpath(root):
+    selector = root / "active-generation.json"
+    name = (
+        json.loads(selector.read_text())["active"]
+        if selector.exists()
+        else ".generation.pending"
+    )
+    return root / "generations" / name / a.DATABASE
+
+
 def initialize(tmp_path, name="state", *, reuse=False):
     return a.initialize_authority(
         tmp_path / name,
@@ -57,7 +67,7 @@ def test_real_setup_separates_private_custody_and_public_trust(tmp_path, capsys)
     )
     assert isinstance(signer, ec.EllipticCurvePrivateKey)
     assert isinstance(decryptor, x25519.X25519PrivateKey)
-    with closing(sqlite3.connect(tmp_path / "state" / a.DATABASE)) as db:
+    with closing(sqlite3.connect(dbpath(tmp_path / "state"))) as db:
         _, owner, verifier, recipient, origin_verifier, origin_private = db.execute(
             "SELECT * FROM recovery_trust"
         ).fetchone()
@@ -82,11 +92,15 @@ def test_real_setup_separates_private_custody_and_public_trust(tmp_path, capsys)
     )
     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()
+    database_bytes = (dbpath(tmp_path / "state")).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}
+    assert set(p.name for p in (tmp_path / "state").iterdir()) == {
+        "generations",
+        "active-generation.json",
+        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.
@@ -100,7 +114,7 @@ def test_explicit_reuse_preserves_owner_keys_and_all_origin_trust(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()
+    assert not (dbpath(tmp_path / "other")).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
@@ -197,7 +211,7 @@ def test_invalid_recovery_file_never_rewrites_or_initializes(tmp_path, damage):
     with pytest.raises(a.AuthorityError):
         initialize(tmp_path, "other", reuse=True)
     assert path.read_bytes() == data
-    assert not (tmp_path / "other" / a.DATABASE).exists()
+    assert not (dbpath(tmp_path / "other")).exists()
 
 
 @pytest.mark.parametrize(
@@ -206,13 +220,13 @@ def test_invalid_recovery_file_never_rewrites_or_initializes(tmp_path, damage):
 )
 def test_damaged_registered_trust_is_not_repaired(tmp_path, column):
     initialize(tmp_path)
-    path = tmp_path / "state" / a.DATABASE
+    path = dbpath(tmp_path / "state")
     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)
+        a.authority_status(tmp_path / "state")
     assert path.read_bytes() == before
 
 
@@ -228,7 +242,7 @@ def test_setup_requires_explicit_safe_external_destination(tmp_path, destination
         a.initialize_authority(
             root, recovery_file=file, reuse_recovery_file=destination == "missing-reuse"
         )
-    assert not (root / a.DATABASE).exists()
+    assert not (dbpath(root)).exists()
     assert not file.exists()
 
 
@@ -364,7 +378,7 @@ def test_reconciliation_never_accepts_unrelated_hardlinks(tmp_path, state):
     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()
+    assert not (dbpath(tmp_path / "other")).exists()
 
 
 @pytest.mark.skipif(os.name != "nt", reason="native Windows recovery custody ACL")
diff --git a/website/docs/operate/credentials.md b/website/docs/operate/credentials.md
index 01e5828..ba212e6 100644
--- a/website/docs/operate/credentials.md
+++ b/website/docs/operate/credentials.md
@@ -68,6 +68,21 @@ there is no implicit upgrade
 or reinitialization. Preserve old state rather than deleting identity records to
 work around a version error.
 
+Initialized authority now lives in a private `generations/UUID/authority.sqlite3`
+directory, selected by `active-generation.json`. IDs, credentials and registered
+trust are published together. A bounded publication intent lets stopped-instance
+status or the next start finish an interrupted publication of those exact bytes.
+It does not invent a replacement identity or repair damaged database contents.
+Only reserved incomplete staging is removed; unexpected files cause an error.
+
+Missing or damaged selectors and generations require operator investigation.
+Preserve every saved generation and the recovery file. Earlier experimental flat
+database layouts are rejected without conversion or deletion. A failed setup can
+have published durable intent; inspect status before retrying initialization.
+POSIX publication flushes files and parent directories. Windows uses same-volume
+[write-through moves](https://learn.microsoft.com/en-us/windows/win32/fileio/moving-directories)
+for files and generation directories; filesystem errors are reported as failures.
+
 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.
diff --git a/website/docs/operate/security.md b/website/docs/operate/security.md
index 433469a..995a634 100644
--- a/website/docs/operate/security.md
+++ b/website/docs/operate/security.md
@@ -167,9 +167,9 @@ 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/server-data/authority \
   --recovery-file /operator-only/private/operator-recovery.json
-arbiter-server authority status --state-dir /private/path/authority --json
+arbiter-server authority status --state-dir /private/server-data/authority --json
 ```
 
 Use an operator-owned private directory. The command creates persistent owner
@@ -178,6 +178,19 @@ 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.
 
+Use `authority` beneath the configured `arbiter.storage.server_data_dir` when
+preparing state for the running server. With no override, this is beneath
+Arbiter's existing platform-specific server data directory. Startup takes the
+same persistent instance lock as offline setup and holds it through HTTP cleanup;
+a second start, offline initialization or offline status fails while it is held.
+Stop the instance before using these offline commands. This enforces state
+ownership; runtime caller authentication remains a later milestone.
+
+State is selected from a private generation. Stopped-instance status and startup
+can finish a valid interrupted initial publication, preserving its IDs and keys.
+They refuse malformed selectors, missing saved state and earlier flat layouts.
+Do not delete a selector or generation to bypass an error.
+
 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