From 1c7a22c0a551d6e2dceb95aa66ad0506556fec04 Mon Sep 17 00:00:00 2001
From: Omry Yadan <omry@yadan.net>
Date: Wed, 21 Oct 2026 03:59:15 +0800
Subject: [PATCH] [authentication/1.4] Publish durable authority generations
 and own the instance lock

Initialize complete private authority generations behind a durable active selector. Reconcile interrupted initial publication with exact IDs and trust, and preserve/refuse malformed or unreferenced saved state. Use native write-through publication on Windows.

Hold the existing persistent authority lock for the actual ASGI lifetime through HTTP cleanup so offline setup and competing starts cannot write concurrently. Keep runtime API authentication and later restore/undo behavior in their planned slices.

Validation: 1280 Python unit tests passed on native Python and Python 3.10, 13 platform skips each; Black/Pyrefly, Go tests/vet and website build passed. Real killed-process, orphan preservation and directory-flush failure probes cover publication recovery. DBB code review and independent full local review are clean. Native Windows x64 and ARM64 CI remains required on the published head.
---
 .../+durable-authority-generations.feature.md |   6 +
 server/src/arbiter_server/authority.py        | 155 ++++----
 server/src/arbiter_server/generations.py      | 224 +++++++++++
 server/src/arbiter_server/main.py             |  11 +-
 .../tests/unit/test_artifact_access_logs.py   |   1 +
 .../unit/test_artifact_http_admission.py      |   1 +
 server/tests/unit/test_authority.py           |  63 ++-
 server/tests/unit/test_credentials.py         |  30 +-
 server/tests/unit/test_generations.py         | 370 ++++++++++++++++++
 .../tests/unit/test_native_http_protocol.py   |   2 +
 .../unit/test_operation_body_admission.py     |   3 +
 server/tests/unit/test_recovery.py            |  32 +-
 website/docs/operate/credentials.md           |  15 +
 website/docs/operate/security.md              |  17 +-
 14 files changed, 822 insertions(+), 108 deletions(-)
 create mode 100644 server/news/+durable-authority-generations.feature.md
 create mode 100644 server/src/arbiter_server/generations.py
 create mode 100644 server/tests/unit/test_generations.py

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