diff --git a/devtools/schema_commit.py b/devtools/schema_commit.py
index e31f5393a1..83f58958d6 100644
--- a/devtools/schema_commit.py
+++ b/devtools/schema_commit.py
@@ -18,6 +18,10 @@
 
 from polylogue.cli.shared.schema_command_support import build_schema_privacy_config
 from polylogue.config import get_config
+from polylogue.maintenance.schema_inference_gate import (
+    authorize_schema_generation,
+    resolve_schema_inference_archive_root,
+)
 from polylogue.schemas.operator.commit import commit_provider_schema
 from polylogue.schemas.operator.models import SchemaCommitRequest
 
@@ -64,6 +68,11 @@ def _build_parser() -> argparse.ArgumentParser:
         help="Preview what a commit would change without writing to --output-dir.",
     )
     parser.add_argument("--json", action="store_true", help="Output as JSON.")
+    parser.add_argument(
+        "--schema-inference-receipt",
+        type=Path,
+        help="Fresh authoritative PASS receipt from devtools verify schema-inference-gate.",
+    )
     return parser
 
 
@@ -81,17 +90,25 @@ def main(argv: list[str] | None = None) -> int:
         return 1
     output_dir = args.output_dir if args.output_dir is not None else DEFAULT_OUTPUT_DIR
 
-    result = commit_provider_schema(
-        SchemaCommitRequest(
-            provider=str(args.provider),
-            output_dir=output_dir,
-            db_path=get_config().db_path,
-            max_samples=args.max_samples,
-            privacy_config=privacy_config,
-            full_corpus=bool(args.full_corpus),
-            dry_run=bool(args.dry_run),
-        )
+    config = get_config()
+    if not args.dry_run and args.schema_inference_receipt is None:
+        print("schema-commit: --schema-inference-receipt is required when persisting schema packages", file=sys.stderr)
+        return 1
+    request = SchemaCommitRequest(
+        provider=str(args.provider),
+        output_dir=output_dir,
+        db_path=config.db_path,
+        max_samples=args.max_samples,
+        privacy_config=privacy_config,
+        full_corpus=bool(args.full_corpus),
+        dry_run=bool(args.dry_run),
     )
+    if args.dry_run:
+        result = commit_provider_schema(request)
+    else:
+        archive_root = resolve_schema_inference_archive_root(config, fallback_db_path=config.db_path)
+        with authorize_schema_generation(archive_root, args.schema_inference_receipt):
+            result = commit_provider_schema(request)
 
     if not result.success:
         error = result.generation.error or "Schema generation failed"
diff --git a/devtools/schema_generate.py b/devtools/schema_generate.py
index a64e51b6e9..c2cefc90dc 100644
--- a/devtools/schema_generate.py
+++ b/devtools/schema_generate.py
@@ -19,6 +19,10 @@
 from polylogue.cli.shared.schema_rendering import render_schema_generate_result
 from polylogue.config import get_config
 from polylogue.core.json import JSONDocument
+from polylogue.maintenance.schema_inference_gate import (
+    authorize_schema_generation,
+    resolve_schema_inference_archive_root,
+)
 from polylogue.schemas.operator.models import SchemaInferRequest
 from polylogue.schemas.operator.workflow import infer_schema
 from polylogue.storage.sqlite.connection_profile import open_readonly_connection
@@ -89,6 +93,12 @@ def _build_parser() -> argparse.ArgumentParser:
     parser.add_argument(
         "--receipt", type=Path, default=None, help="Write an aggregate-only generation receipt as JSON."
     )
+    parser.add_argument(
+        "--schema-inference-receipt",
+        type=Path,
+        required=True,
+        help="Fresh authoritative PASS receipt from devtools verify schema-inference-gate.",
+    )
     return parser
 
 
@@ -104,21 +114,24 @@ def on_progress(_phase: str, payload: JSONDocument) -> None:
             print(f"schema-generate: {json.dumps(event, sort_keys=True)}", file=sys.stderr, flush=True)
 
     try:
+        config = get_config()
         privacy_config = build_schema_privacy_config(
             privacy=args.privacy,
             privacy_config_path=args.privacy_config,
         )
-        result = infer_schema(
-            SchemaInferRequest(
-                provider=str(args.provider),
-                db_path=get_config().db_path,
-                max_samples=args.max_samples,
-                privacy_config=privacy_config,
-                cluster=bool(args.cluster),
-                full_corpus=bool(args.full_corpus),
-                progress_callback=on_progress if args.progress or args.receipt is not None else None,
+        archive_root = resolve_schema_inference_archive_root(config, fallback_db_path=config.db_path)
+        with authorize_schema_generation(archive_root, args.schema_inference_receipt):
+            result = infer_schema(
+                SchemaInferRequest(
+                    provider=str(args.provider),
+                    db_path=config.db_path,
+                    max_samples=args.max_samples,
+                    privacy_config=privacy_config,
+                    cluster=bool(args.cluster),
+                    full_corpus=bool(args.full_corpus),
+                    progress_callback=on_progress if args.progress or args.receipt is not None else None,
+                )
             )
-        )
     except ValueError as exc:
         print(f"schema-generate: {exc}", file=sys.stderr)
         return 1
diff --git a/polylogue/maintenance/schema_inference_gate.py b/polylogue/maintenance/schema_inference_gate.py
index 54354160cc..5e74b53a9a 100644
--- a/polylogue/maintenance/schema_inference_gate.py
+++ b/polylogue/maintenance/schema_inference_gate.py
@@ -11,29 +11,41 @@
 
 import hashlib
 import json
+import os
 import platform
 import sqlite3
 import sys
 from collections import Counter
-from collections.abc import Iterable, Mapping, Sequence
+from collections.abc import Iterable, Iterator, Mapping, Sequence
+from contextlib import contextmanager
 from dataclasses import dataclass
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 from pathlib import Path
 from typing import Any, Literal, NotRequired, TypeAlias, TypedDict, cast
+from uuid import uuid4
 
 from polylogue.maintenance.archive_verification import CORPUS_FIDELITY_CHECKS, verify_archive
+from polylogue.maintenance.offline_guard import running_daemon_pid
 from polylogue.storage.archive_identity import ArchiveIdentity, ArchiveLocation
+from polylogue.storage.backup_attestation import (
+    BackupAttestationError,
+    sign_verification_receipt,
+    verify_verification_receipt,
+)
 from polylogue.storage.blob_store import BlobStore
+from polylogue.storage.index_generation import RebuildLease, RebuildLeaseUnavailableError
 from polylogue.storage.introspection import table_exists
 from polylogue.storage.sqlite.archive_tiers.bootstrap import ARCHIVE_TIER_SPECS
 from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier
 from polylogue.storage.sqlite.connection_profile import open_readonly_connection
 from polylogue.version import POLYLOGUE_VERSION
 
-RECEIPT_SCHEMA = "polylogue.schema-inference-gate.v1"
-GATE_VERSION = "2"
+RECEIPT_SCHEMA = "polylogue.schema-inference-gate.v2"
+GATE_VERSION = "3"
 DEFAULT_SAMPLE_LIMIT = 10
 RECEIPT_FILENAME = "schema-inference-gate-receipt.json"
+RECEIPT_TTL = timedelta(hours=24)
+RECEIPT_CLOCK_SKEW = timedelta(minutes=5)
 
 _ALLOWED_RESIDUAL_EXPLANATIONS = frozenset(
     {"materialized", "superseded-duplicate", "legitimately-excluded-non-conversation"}
@@ -474,26 +486,98 @@ def _failed_source_gates(reason: str) -> dict[str, object]:
     }
 
 
+def _file_sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        while chunk := handle.read(8 * 1024 * 1024):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _sqlite_schema_digest(conn: sqlite3.Connection) -> str:
+    rows = conn.execute(
+        "SELECT type, name, tbl_name, sql FROM sqlite_master "
+        "WHERE sql IS NOT NULL AND name != 'sqlite_stat1' ORDER BY type, name, tbl_name"
+    ).fetchall()
+    encoded = json.dumps(
+        [[str(value) if value is not None else None for value in row] for row in rows],
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
+
+
+def _canonical_schema_digest(tier: ArchiveTier) -> str | None:
+    """Return the schema identity of a fresh canonical tier, when loadable."""
+
+    from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_tier
+    from polylogue.storage.sqlite.sqlite_vec_extension import try_load_sqlite_vec
+
+    conn = sqlite3.connect(":memory:")
+    try:
+        if tier is ArchiveTier.EMBEDDINGS:
+            loaded, _error = try_load_sqlite_vec(conn)
+            if not loaded:
+                return None
+        initialize_archive_tier(conn, tier)
+        return _sqlite_schema_digest(conn)
+    except sqlite3.Error:
+        return None
+    finally:
+        conn.close()
+
+
 def _tier_schema_identity(archive_root: Path, location: ArchiveLocation) -> dict[str, object]:
     tiers: dict[str, object] = {}
     for tier, spec in ARCHIVE_TIER_SPECS.items():
-        path = location.active_index_path if tier is ArchiveTier.INDEX else archive_root / spec.filename
+        identity = (
+            location.active_index if tier is ArchiveTier.INDEX else location.configured_tier(cast(Any, tier.value))
+        )
+        path = identity.resolved_path
+        expected_schema_sha256 = _canonical_schema_digest(tier)
         entry: dict[str, object] = {
             "path": str(path),
+            "stable_id": identity.stable_id,
             "expected_user_version": spec.version,
             "durability": spec.durability,
             "exists": path.exists(),
             "actual_user_version": None,
+            "content_sha256": None,
+            "schema_sha256": None,
+            "expected_schema_sha256": expected_schema_sha256,
         }
         if path.exists():
             try:
                 with open_readonly_connection(path) as conn:
                     entry["actual_user_version"] = int(conn.execute("PRAGMA user_version").fetchone()[0])
+                    entry["schema_sha256"] = _sqlite_schema_digest(conn)
             except sqlite3.Error as exc:
                 entry["error"] = str(exc)
+            try:
+                entry["content_sha256"] = _file_sha256(path)
+            except OSError as exc:
+                entry["error"] = str(exc)
         entry["matches_expected"] = entry["actual_user_version"] == spec.version
+        if expected_schema_sha256 is not None:
+            entry["matches_expected"] = bool(entry["matches_expected"]) and (
+                entry["schema_sha256"] == expected_schema_sha256
+            )
         tiers[tier.value] = entry
-    return {"archive": ArchiveIdentity.resolve_location(location).as_dict(), "tiers": tiers}
+    archive_identity = ArchiveIdentity.resolve_location(location)
+    runtime_archive_payload = archive_identity.as_dict()
+    archive_payload = {
+        key: runtime_archive_payload[key]
+        for key in (
+            "configured_root",
+            "durable_id",
+            "active_generation",
+            "generation_owner",
+            "generation_state",
+            "tiers",
+        )
+    }
+    archive_payload["authority_identity_digest"] = archive_identity.authority_identity_digest
+    return {"archive": archive_payload, "tiers": tiers}
 
 
 def _resolve_receipt_path(receipt_path: Path, *, archive_root: Path) -> Path:
@@ -621,6 +705,22 @@ def _external_inventory(roots: Sequence[Path]) -> list[_ExternalGroundTruthFile]
     return inventory
 
 
+def _validate_external_ground_truth_roots(archive_root: Path, roots: Sequence[Path]) -> None:
+    """Reject archive-owned bytes from posing as independent source evidence."""
+
+    protected = (archive_root.resolve(), (archive_root / "blob").resolve())
+    for root in roots:
+        candidate = root.resolve()
+        for owned in protected:
+            try:
+                candidate.relative_to(owned)
+            except ValueError:
+                continue
+            raise SchemaInferenceGateError(
+                f"ground-truth root must be external to the archive and blob namespace: {candidate}"
+            )
+
+
 def _external_receipt(
     item: _ExternalGroundTruthFile,
     disposition: ExternalDisposition,
@@ -743,6 +843,16 @@ def _ground_truth_evidence(
                 evidence[origin] = {"exempt": True, "reason": declared.get("reason")}
                 continue
             declared_roots = tuple(Path(path).expanduser().resolve() for path in root_map.get(origin, ()))
+            try:
+                _validate_external_ground_truth_roots(archive_root, declared_roots)
+            except SchemaInferenceGateError as exc:
+                errors.append(f"ground truth for {origin} is invalid: {exc}")
+                evidence[origin] = {
+                    "exempt": False,
+                    "declared_roots": [str(path) for path in declared_roots],
+                    "passed": False,
+                }
+                continue
             unavailable = [str(path) for path in declared_roots if not path.exists()]
             if not declared_roots or unavailable:
                 errors.append(f"ground truth for {origin} is unavailable or undeclared")
@@ -1018,26 +1128,161 @@ def _int_or_zero(value: object) -> int:
     return value if isinstance(value, int) and not isinstance(value, bool) else 0
 
 
-def _write_json(path: Path, payload: dict[str, object]) -> None:
+def schema_inference_gate_receipt_digest(payload: Mapping[str, object]) -> str:
+    """Digest every receipt field except the self-authenticating digest."""
+
+    body = {key: value for key, value in payload.items() if key not in {"receipt_sha256", "attestations"}}
+    encoded = json.dumps(body, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False)
+    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
+
+
+def _write_json(path: Path, payload: dict[str, object], *, authority_paths: Mapping[str, Path]) -> None:
+    payload["receipt_sha256"] = schema_inference_gate_receipt_digest(payload)
+    payload = sign_verification_receipt(payload, authority_paths=authority_paths)
     path.parent.mkdir(parents=True, exist_ok=True)
-    temporary = path.with_name(f".{path.name}.tmp")
-    temporary.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
-    temporary.replace(path)
+    encoded = (json.dumps(payload, indent=2, sort_keys=True) + "\n").encode("utf-8")
+    try:
+        descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+    except FileExistsError as exc:
+        raise SchemaInferenceGateError(f"immutable schema-inference gate receipt already exists: {path}") from exc
+    try:
+        with os.fdopen(descriptor, "wb") as handle:
+            descriptor = -1
+            handle.write(encoded)
+            handle.flush()
+            os.fsync(handle.fileno())
+    finally:
+        if descriptor != -1:
+            os.close(descriptor)
+    directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
+    try:
+        os.fsync(directory)
+    finally:
+        os.close(directory)
 
 
-def run_schema_inference_gate(
+def _archive_config(root: Path) -> Any:
+    from polylogue.config import Config
+
+    return Config(archive_root=root, render_root=root / "render", sources=[])
+
+
+def resolve_schema_inference_archive_root(config: object, *, fallback_db_path: Path) -> Path:
+    """Resolve the archive root consistently across privileged schema routes."""
+
+    configured_root = getattr(config, "archive_root", None)
+    return Path(configured_root) if configured_root is not None else fallback_db_path.parent
+
+
+@contextmanager
+def schema_inference_quiescence(archive_root: Path) -> Iterator[None]:
+    """Hold the archive-wide offline lease for the complete evidence window."""
+
+    root = Path(archive_root).absolute()
+    daemon_pid = running_daemon_pid(_archive_config(root))
+    if daemon_pid is not None:
+        raise SchemaInferenceGateError(
+            f"schema-inference gate requires a quiesced archive; polylogued PID {daemon_pid} is running"
+        )
+    try:
+        with RebuildLease(root):
+            yield
+    except RebuildLeaseUnavailableError as exc:
+        raise SchemaInferenceGateError(
+            f"schema-inference gate requires exclusive offline archive ownership: {exc}"
+        ) from exc
+
+
+def _parse_receipt_time(value: object) -> datetime:
+    if not isinstance(value, str):
+        raise SchemaInferenceGateError("schema-inference gate receipt generated_at is missing")
+    try:
+        parsed = datetime.fromisoformat(value)
+    except ValueError as exc:
+        raise SchemaInferenceGateError("schema-inference gate receipt generated_at is invalid") from exc
+    if parsed.tzinfo is None:
+        raise SchemaInferenceGateError("schema-inference gate receipt generated_at must include a timezone")
+    return parsed.astimezone(UTC)
+
+
+def validate_schema_inference_gate_receipt(
+    receipt_path: Path,
+    *,
+    archive_root: Path,
+    now: datetime | None = None,
+) -> dict[str, object]:
+    """Validate a fresh, immutable PASS receipt against the live archive."""
+
+    root = Path(archive_root).absolute()
+    safe_path = _resolve_receipt_path(Path(receipt_path), archive_root=root)
+    try:
+        raw = safe_path.read_bytes()
+        loaded = json.loads(raw.decode("utf-8"))
+    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise SchemaInferenceGateError(f"unable to read schema-inference gate receipt: {exc}") from exc
+    if not isinstance(loaded, dict):
+        raise SchemaInferenceGateError("schema-inference gate receipt must be a JSON object")
+    payload = cast(dict[str, object], loaded)
+    if payload.get("schema") != RECEIPT_SCHEMA or payload.get("gate_version") != GATE_VERSION:
+        raise SchemaInferenceGateError("schema-inference gate receipt schema version is unsupported")
+    if payload.get("verdict") != "PASS":
+        raise SchemaInferenceGateError("schema-inference gate receipt is not a PASS")
+    if payload.get("receipt_sha256") != schema_inference_gate_receipt_digest(payload):
+        raise SchemaInferenceGateError("schema-inference gate receipt content digest mismatch")
+    try:
+        verify_verification_receipt(payload, tier="source", live_tier_path=root / "source.db")
+        verify_verification_receipt(payload, tier="user", live_tier_path=root / "user.db")
+    except BackupAttestationError as exc:
+        raise SchemaInferenceGateError(f"schema-inference gate receipt attestation is invalid: {exc}") from exc
+    generated_at = _parse_receipt_time(payload.get("generated_at"))
+    current_time = (now or datetime.now(UTC)).astimezone(UTC)
+    age = current_time - generated_at
+    if age < -RECEIPT_CLOCK_SKEW or age > RECEIPT_TTL:
+        raise SchemaInferenceGateError("schema-inference gate receipt is stale or from the future")
+    location = ArchiveLocation.resolve(root)
+    if payload.get("archive_root") != str(root):
+        raise SchemaInferenceGateError("schema-inference gate receipt targets a different archive")
+    live_identity = _tier_schema_identity(root, location)
+    if payload.get("schema_identity") != live_identity:
+        raise SchemaInferenceGateError("schema-inference gate receipt archive or tier identity is stale")
+    archive_payload = _as_dict(live_identity.get("archive"))
+    if payload.get("archive_identity_digest") != archive_payload.get("authority_identity_digest"):
+        raise SchemaInferenceGateError("schema-inference gate receipt archive identity is stale")
+    if payload.get("source_schema_identity") != _as_dict(_as_dict(live_identity.get("tiers")).get("source")):
+        raise SchemaInferenceGateError("schema-inference gate receipt source schema identity is stale")
+    if not all(bool(_as_dict(value).get("matches_expected")) for value in _as_dict(live_identity["tiers"]).values()):
+        raise SchemaInferenceGateError("live archive has a stale durable or derived tier schema identity")
+    input_paths = _as_dict(payload.get("input_paths"))
+    if input_paths.get("receipt") != str(safe_path):
+        raise SchemaInferenceGateError("schema-inference gate receipt is bound to a different receipt path")
+    reasons = payload.get("pass_fail_reasons")
+    if not isinstance(reasons, list) or reasons:
+        raise SchemaInferenceGateError("schema-inference gate receipt contains failure reasons")
+    return payload
+
+
+@contextmanager
+def authorize_schema_generation(archive_root: Path, receipt_path: Path) -> Iterator[dict[str, object]]:
+    """Hold quiescence for one fresh schema operation or compatible short sequence."""
+
+    with schema_inference_quiescence(archive_root):
+        yield validate_schema_inference_gate_receipt(receipt_path, archive_root=archive_root)
+
+
+def _run_schema_inference_gate_locked(
     archive_root: Path,
     *,
     receipt_path: Path,
     ground_truth_roots: Mapping[str, Sequence[Path]] | None = None,
     sample_limit: int = DEFAULT_SAMPLE_LIMIT,
 ) -> SchemaInferenceGateResult:
-    """Run and persist the schema-inference prerequisite receipt."""
+    """Run and persist the schema-inference prerequisite while already locked."""
 
     if sample_limit <= 0:
         raise SchemaInferenceGateError("sample_limit must be positive")
     root = Path(archive_root).absolute()
     safe_receipt_path = _resolve_receipt_path(Path(receipt_path), archive_root=root)
+    location: ArchiveLocation | None = None
     try:
         location = ArchiveLocation.resolve(root)
         index_path = location.active_index_path
@@ -1091,6 +1336,8 @@ def run_schema_inference_gate(
     reasons = [
         str(_as_dict(result).get("reason")) for result in gate_results.values() if _as_dict(result).get("reason")
     ]
+    if not source_gates.get("source_counts"):
+        reasons.append("schema inference requires at least one reconciled source raw")
     if not source_schema_ok:
         reasons.append("source.db schema identity is missing or does not match the packaged schema")
     if not bool(fidelity.get("passed")):
@@ -1104,14 +1351,26 @@ def run_schema_inference_gate(
         if isinstance(ground_truth_reasons, list):
             reasons.extend(str(reason) for reason in ground_truth_reasons)
 
+    final_schema_identity = _tier_schema_identity(root, location) if location is not None else schema_identity
+    tier_entries = _as_dict(final_schema_identity.get("tiers"))
+    schema_identity_ok = len(tier_entries) == len(ARCHIVE_TIER_SPECS) and all(
+        bool(_as_dict(value).get("matches_expected")) for value in tier_entries.values()
+    )
+    if not schema_identity_ok:
+        for tier_name, entry in tier_entries.items():
+            if not bool(_as_dict(entry).get("matches_expected")):
+                reasons.append(f"{tier_name}.db schema identity is stale or does not match the packaged schema")
+    archive_payload = _as_dict(final_schema_identity.get("archive"))
     payload: dict[str, object] = {
         "schema": RECEIPT_SCHEMA,
         "gate_version": GATE_VERSION,
         "generated_at": datetime.now(UTC).isoformat(),
-        "verdict": "PASS" if not reasons and passed_hard_gates else "FAIL",
+        "receipt_nonce": uuid4().hex,
+        "verdict": "PASS" if not reasons and passed_hard_gates and schema_identity_ok else "FAIL",
         "archive_root": str(root),
-        "schema_identity": schema_identity,
-        "source_schema_identity": source_entry,
+        "archive_identity_digest": archive_payload.get("authority_identity_digest"),
+        "schema_identity": final_schema_identity,
+        "source_schema_identity": _as_dict(tier_entries.get("source")),
         "query_results": gate_results,
         "source_denominators": source_gates.get("source_counts", {}),
         "blob_denominators": blob_denominators,
@@ -1126,6 +1385,7 @@ def run_schema_inference_gate(
             "active_index_db": str(index_path),
             "receipt": str(safe_receipt_path),
         },
+        "quiescence": {"lease": "index-rebuild-exclusive", "daemon_pid": None},
         "tool_versions": {
             "polylogue": POLYLOGUE_VERSION,
             "gate": GATE_VERSION,
@@ -1135,10 +1395,32 @@ def run_schema_inference_gate(
         },
         "pass_fail_reasons": reasons,
     }
-    _write_json(safe_receipt_path, payload)
+    _write_json(
+        safe_receipt_path,
+        payload,
+        authority_paths={"source": root / "source.db", "user": root / "user.db"},
+    )
     return SchemaInferenceGateResult(payload)
 
 
+def run_schema_inference_gate(
+    archive_root: Path,
+    *,
+    receipt_path: Path,
+    ground_truth_roots: Mapping[str, Sequence[Path]] | None = None,
+    sample_limit: int = DEFAULT_SAMPLE_LIMIT,
+) -> SchemaInferenceGateResult:
+    """Run the prerequisite under exclusive offline ownership."""
+
+    with schema_inference_quiescence(Path(archive_root).absolute()):
+        return _run_schema_inference_gate_locked(
+            archive_root,
+            receipt_path=receipt_path,
+            ground_truth_roots=ground_truth_roots,
+            sample_limit=sample_limit,
+        )
+
+
 __all__ = [
     "DEFAULT_SAMPLE_LIMIT",
     "GROUND_TRUTH_INPUTS",
@@ -1146,5 +1428,9 @@ def run_schema_inference_gate(
     "RECEIPT_SCHEMA",
     "SchemaInferenceGateError",
     "SchemaInferenceGateResult",
+    "authorize_schema_generation",
+    "schema_inference_gate_receipt_digest",
+    "schema_inference_quiescence",
+    "validate_schema_inference_gate_receipt",
     "run_schema_inference_gate",
 ]
diff --git a/polylogue/schemas/operator/schema_inference.py b/polylogue/schemas/operator/schema_inference.py
index 69221a49ef..edebb61583 100644
--- a/polylogue/schemas/operator/schema_inference.py
+++ b/polylogue/schemas/operator/schema_inference.py
@@ -22,6 +22,12 @@
 
 from pathlib import Path
 
+from polylogue.config import get_config
+from polylogue.maintenance.schema_inference_gate import (
+    authorize_schema_generation,
+    resolve_schema_inference_archive_root,
+)
+
 # Re-export the full public API so callers don't need to know the split.
 from polylogue.schemas.field_stats.stats import (
     UUID_PATTERN,
@@ -121,17 +127,27 @@ def cli_main(args: list[str] | None = None) -> int:
         action="store_true",
         help="Also write an aggregate archive composition profile beside staged provider packages",
     )
+    parser.add_argument(
+        "--schema-inference-receipt",
+        type=Path,
+        required=True,
+        help="Fresh authoritative PASS receipt required before writing schema packages.",
+    )
 
     parsed = parser.parse_args(args)
 
     providers = None if parsed.provider == "all" else [parsed.provider]
-    results = generate_all_schemas(
-        output_dir=parsed.output_dir,
-        db_path=parsed.db_path,
-        providers=providers,
-        max_samples=parsed.max_samples,
-        include_archive_workload_profile=parsed.archive_workload_profile,
-    )
+    config = get_config()
+    db_path = parsed.db_path or config.db_path
+    archive_root = resolve_schema_inference_archive_root(config, fallback_db_path=db_path)
+    with authorize_schema_generation(archive_root, parsed.schema_inference_receipt):
+        results = generate_all_schemas(
+            output_dir=parsed.output_dir,
+            db_path=db_path,
+            providers=providers,
+            max_samples=parsed.max_samples,
+            include_archive_workload_profile=parsed.archive_workload_profile,
+        )
 
     success = []
     failed = []
diff --git a/tests/infra/schema_inference.py b/tests/infra/schema_inference.py
new file mode 100644
index 0000000000..e7d264b59c
--- /dev/null
+++ b/tests/infra/schema_inference.py
@@ -0,0 +1,66 @@
+"""Shared archive fixtures for schema-inference gate routes."""
+
+from __future__ import annotations
+
+import sqlite3
+from pathlib import Path
+
+from polylogue.storage.blob_store import BlobStore
+from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root
+
+
+def seed_schema_inference_archive(root: Path) -> Path:
+    """Create one source raw whose actual external file and blob agree."""
+
+    initialize_active_archive_root(root)
+    ground_truth = root.parent / f"{root.name}-codex-ground-truth"
+    ground_truth.mkdir()
+    payload = b"actual external codex raw"
+    source_file = ground_truth / "session.jsonl"
+    source_file.write_bytes(payload)
+    blob_hash, blob_size = BlobStore(root / "blob").write_from_bytes(payload)
+    with sqlite3.connect(root / "source.db") as conn:
+        conn.execute(
+            """
+            INSERT INTO raw_sessions(
+                raw_id, origin, native_id, source_path, blob_hash, blob_size,
+                acquired_at_ms, logical_source_key, revision_authority
+            ) VALUES ('raw-1', 'codex-session', 'session', ?, ?, ?, 100,
+                      'codex:session', 'byte_proven')
+            """,
+            (str(source_file), bytes.fromhex(blob_hash), blob_size),
+        )
+        conn.execute(
+            """
+            INSERT INTO raw_session_memberships(
+                raw_id, logical_source_key, provider_session_id, source_revision,
+                normalized_content_hash, message_count, decision, decided_at_ms
+            ) VALUES ('raw-1', 'codex:session', 'session', 'rev-1', ?, 1, 'applied', 100)
+            """,
+            (b"m" * 32,),
+        )
+    with sqlite3.connect(root / "index.db") as conn:
+        conn.execute(
+            """
+            INSERT INTO sessions(native_id, origin, raw_id, content_hash, message_count)
+            VALUES ('session', 'codex-session', 'raw-1', ?, 1)
+            """,
+            (b"s" * 32,),
+        )
+        conn.execute(
+            """
+            INSERT INTO messages(session_id, position, role, material_origin, content_hash)
+            VALUES ('codex-session:session', 0, 'user', 'human_authored', ?)
+            """,
+            (b"n" * 32,),
+        )
+        conn.execute(
+            """
+            INSERT INTO blocks(message_id, session_id, position, block_type, text)
+            VALUES ('codex-session:session:0.0', 'codex-session:session', 0, 'text', 'hello')
+            """
+        )
+        conn.execute("ANALYZE blocks")
+        conn.execute("ANALYZE messages")
+        conn.execute("ANALYZE action_pairs")
+    return ground_truth
diff --git a/tests/unit/core/test_schema_generation.py b/tests/unit/core/test_schema_generation.py
index 66c3c53bb1..624014c656 100644
--- a/tests/unit/core/test_schema_generation.py
+++ b/tests/unit/core/test_schema_generation.py
@@ -751,14 +751,14 @@ class TestCliMain:
     """CLI entry point behavior."""
 
     def test_cli_with_no_db(self, tmp_path: Path) -> None:
-        exit_code = cli_main(
-            [
-                "--provider",
-                "chatgpt",
-                "--output-dir",
-                str(tmp_path / "out"),
-                "--db-path",
-                str(tmp_path / "missing.db"),
-            ]
-        )
-        assert isinstance(exit_code, int)
+        with pytest.raises(SystemExit, match="2"):
+            cli_main(
+                [
+                    "--provider",
+                    "chatgpt",
+                    "--output-dir",
+                    str(tmp_path / "out"),
+                    "--db-path",
+                    str(tmp_path / "missing.db"),
+                ]
+            )
diff --git a/tests/unit/devtools/test_schema_commit_command.py b/tests/unit/devtools/test_schema_commit_command.py
index 892fa3473e..f68c657aee 100644
--- a/tests/unit/devtools/test_schema_commit_command.py
+++ b/tests/unit/devtools/test_schema_commit_command.py
@@ -8,6 +8,8 @@
 from __future__ import annotations
 
 import json
+from collections.abc import Iterator
+from contextlib import contextmanager
 from dataclasses import dataclass
 from pathlib import Path
 
@@ -23,6 +25,11 @@ class _ConfigStub:
     db_path: Path
 
 
+@contextmanager
+def _allow_schema_generation(*_args: object, **_kwargs: object) -> Iterator[dict[str, object]]:
+    yield {}
+
+
 def test_schema_commit_forwards_request_and_defaults_output_dir(
     monkeypatch: pytest.MonkeyPatch, tmp_path: Path
 ) -> None:
@@ -42,8 +49,18 @@ def fake_commit(request: SchemaCommitRequest) -> SchemaCommitResult:
 
     monkeypatch.setattr(schema_commit, "get_config", fake_get_config)
     monkeypatch.setattr(schema_commit, "commit_provider_schema", fake_commit)
+    authorization_calls: list[tuple[object, ...]] = []
 
-    assert schema_commit.main(["--provider", "chatgpt"]) == 0
+    @contextmanager
+    def allow_schema_generation(*args: object, **_kwargs: object) -> Iterator[dict[str, object]]:
+        authorization_calls.append(args)
+        yield {}
+
+    monkeypatch.setattr(schema_commit, "authorize_schema_generation", allow_schema_generation)
+
+    assert (
+        schema_commit.main(["--provider", "chatgpt", "--schema-inference-receipt", str(tmp_path / "receipt.json")]) == 0
+    )
 
     assert len(captured) == 1
     request = captured[0]
@@ -52,6 +69,17 @@ def fake_commit(request: SchemaCommitRequest) -> SchemaCommitResult:
     assert request.db_path == tmp_path / "archive.db"
     assert request.full_corpus is True
     assert request.dry_run is False
+    assert authorization_calls == [(tmp_path, tmp_path / "receipt.json")]
+
+
+def test_schema_commit_refuses_persistence_without_authoritative_receipt(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    monkeypatch.setattr(schema_commit, "get_config", lambda: _ConfigStub(db_path=tmp_path / "archive.db"))
+    monkeypatch.setattr(schema_commit, "commit_provider_schema", pytest.fail)
+
+    assert schema_commit.main(["--provider", "chatgpt"]) == 1
+    assert "schema-inference-receipt is required" in capsys.readouterr().err
 
 
 def test_schema_commit_honors_output_dir_and_dry_run_overrides(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
@@ -87,6 +115,7 @@ def test_schema_commit_json_output_reports_success(
     monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str]
 ) -> None:
     monkeypatch.setattr(schema_commit, "get_config", lambda: _ConfigStub(db_path=tmp_path / "archive.db"))
+    monkeypatch.setattr(schema_commit, "authorize_schema_generation", _allow_schema_generation)
     monkeypatch.setattr(
         schema_commit,
         "commit_provider_schema",
@@ -102,7 +131,12 @@ def test_schema_commit_json_output_reports_success(
         ),
     )
 
-    assert schema_commit.main(["--provider", "chatgpt", "--json"]) == 0
+    assert (
+        schema_commit.main(
+            ["--provider", "chatgpt", "--json", "--schema-inference-receipt", str(tmp_path / "receipt.json")]
+        )
+        == 0
+    )
 
     payload = json.loads(capsys.readouterr().out)
     assert payload["provider"] == "chatgpt"
@@ -117,6 +151,7 @@ def test_schema_commit_exits_nonzero_on_generation_failure(
     monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str]
 ) -> None:
     monkeypatch.setattr(schema_commit, "get_config", lambda: _ConfigStub(db_path=tmp_path / "archive.db"))
+    monkeypatch.setattr(schema_commit, "authorize_schema_generation", _allow_schema_generation)
     monkeypatch.setattr(
         schema_commit,
         "commit_provider_schema",
@@ -128,7 +163,12 @@ def test_schema_commit_exits_nonzero_on_generation_failure(
         ),
     )
 
-    assert schema_commit.main(["--provider", "broken-provider", "--json"]) == 1
+    assert (
+        schema_commit.main(
+            ["--provider", "broken-provider", "--json", "--schema-inference-receipt", str(tmp_path / "receipt.json")]
+        )
+        == 1
+    )
     payload = json.loads(capsys.readouterr().out)
     assert payload["success"] is False
     assert payload["error"] == "No samples"
@@ -139,6 +179,7 @@ def test_schema_commit_exits_nonzero_when_narrowed(monkeypatch: pytest.MonkeyPat
     not report a clean exit code -- the whole point of the report is that a
     bad promotion can't land unnoticed."""
     monkeypatch.setattr(schema_commit, "get_config", lambda: _ConfigStub(db_path=tmp_path / "archive.db"))
+    monkeypatch.setattr(schema_commit, "authorize_schema_generation", _allow_schema_generation)
     monkeypatch.setattr(
         schema_commit,
         "commit_provider_schema",
@@ -154,4 +195,6 @@ def test_schema_commit_exits_nonzero_when_narrowed(monkeypatch: pytest.MonkeyPat
         ),
     )
 
-    assert schema_commit.main(["--provider", "chatgpt"]) == 1
+    assert (
+        schema_commit.main(["--provider", "chatgpt", "--schema-inference-receipt", str(tmp_path / "receipt.json")]) == 1
+    )
diff --git a/tests/unit/devtools/test_schema_inference_gate.py b/tests/unit/devtools/test_schema_inference_gate.py
index 28857491bd..7e967db8af 100644
--- a/tests/unit/devtools/test_schema_inference_gate.py
+++ b/tests/unit/devtools/test_schema_inference_gate.py
@@ -7,7 +7,7 @@
 
 from devtools import schema_inference_gate
 from polylogue.maintenance.schema_inference_gate import RECEIPT_FILENAME
-from tests.unit.maintenance.test_schema_inference_gate import _seed_archive
+from tests.infra.schema_inference import seed_schema_inference_archive as _seed_archive
 
 
 def test_devtools_command_requires_caller_root_and_persists_receipt(tmp_path: Path) -> None:
diff --git a/tests/unit/devtools/test_schema_lab_commands.py b/tests/unit/devtools/test_schema_lab_commands.py
index 4d6920cd61..d1a51d1c64 100644
--- a/tests/unit/devtools/test_schema_lab_commands.py
+++ b/tests/unit/devtools/test_schema_lab_commands.py
@@ -1,13 +1,21 @@
 from __future__ import annotations
 
 import json
+import sqlite3
+from collections.abc import Iterator
+from contextlib import contextmanager
 from dataclasses import dataclass
+from datetime import UTC, datetime
 from pathlib import Path
 
 import pytest
 
 from devtools import schema_audit, schema_generate, schema_inspect, schema_promote
 from polylogue.core.outcomes import OutcomeCheck, OutcomeStatus
+from polylogue.maintenance.schema_inference_gate import (
+    run_schema_inference_gate,
+    schema_inference_gate_receipt_digest,
+)
 from polylogue.schemas.audit.models import AuditReport
 from polylogue.schemas.generation.models import GenerationResult
 from polylogue.schemas.operator.models import (
@@ -20,11 +28,19 @@
     SchemaPromoteRequest,
     SchemaPromoteResult,
 )
+from polylogue.storage.index_generation import RebuildLease
+from tests.infra.schema_inference import seed_schema_inference_archive as _seed_archive
 
 
 @dataclass(frozen=True)
 class _ConfigStub:
     db_path: Path
+    archive_root: Path | None = None
+
+
+@contextmanager
+def _allow_schema_generation(*_args: object, **_kwargs: object) -> Iterator[dict[str, object]]:
+    yield {}
 
 
 def test_schema_audit_returns_success_for_passing_report(
@@ -181,8 +197,21 @@ def fake_infer(request: SchemaInferRequest) -> SchemaInferResult:
 
     monkeypatch.setattr(schema_generate, "get_config", fake_get_config)
     monkeypatch.setattr(schema_generate, "infer_schema", fake_infer)
+    monkeypatch.setattr(schema_generate, "authorize_schema_generation", _allow_schema_generation)
 
-    assert schema_generate.main(["--provider", "chatgpt", "--max-samples", "2"]) == 0
+    assert (
+        schema_generate.main(
+            [
+                "--provider",
+                "chatgpt",
+                "--max-samples",
+                "2",
+                "--schema-inference-receipt",
+                str(tmp_path / "receipt.json"),
+            ]
+        )
+        == 0
+    )
 
     assert captured == [
         SchemaInferRequest(
@@ -228,8 +257,22 @@ def fake_infer(request: SchemaInferRequest) -> SchemaInferResult:
     receipt_path = tmp_path / "receipt.json"
     monkeypatch.setattr(schema_generate, "get_config", fake_get_config)
     monkeypatch.setattr(schema_generate, "infer_schema", fake_infer)
+    monkeypatch.setattr(schema_generate, "authorize_schema_generation", _allow_schema_generation)
 
-    assert schema_generate.main(["--provider", "chatgpt", "--progress", "--receipt", str(receipt_path)]) == 0
+    assert (
+        schema_generate.main(
+            [
+                "--provider",
+                "chatgpt",
+                "--progress",
+                "--receipt",
+                str(receipt_path),
+                "--schema-inference-receipt",
+                str(tmp_path / "schema-inference-gate-receipt.json"),
+            ]
+        )
+        == 0
+    )
 
     receipt = json.loads(receipt_path.read_text())
     assert receipt["generation"]["status"] == "succeeded"
@@ -270,11 +313,125 @@ def fake_infer(request: SchemaInferRequest) -> SchemaInferResult:
 
     monkeypatch.setattr(schema_generate, "get_config", fake_get_config)
     monkeypatch.setattr(schema_generate, "infer_schema", fake_infer)
+    monkeypatch.setattr(schema_generate, "authorize_schema_generation", _allow_schema_generation)
 
-    assert schema_generate.main(["--provider", "chatgpt", "--cluster"]) == 1
+    assert (
+        schema_generate.main(
+            ["--provider", "chatgpt", "--cluster", "--schema-inference-receipt", str(tmp_path / "receipt.json")]
+        )
+        == 1
+    )
     assert "No samples found for clustering" in capsys.readouterr().err
 
 
+def _authoritative_schema_gate(tmp_path: Path, name: str) -> tuple[Path, Path]:
+    root = tmp_path / name
+    ground_truth = _seed_archive(root)
+    receipt = tmp_path / f"{name}-receipt" / "schema-inference-gate-receipt.json"
+    result = run_schema_inference_gate(
+        root,
+        receipt_path=receipt,
+        ground_truth_roots={"codex-session": (ground_truth,)},
+    )
+    assert result.passed, result.payload["pass_fail_reasons"]
+    return root, receipt
+
+
+def _successful_inference(request: SchemaInferRequest) -> SchemaInferResult:
+    return SchemaInferResult(
+        generation=GenerationResult(
+            provider=request.provider,
+            schema={"type": "object"},
+            sample_count=1,
+        )
+    )
+
+
+def test_schema_generate_direct_command_bypass_refuses_without_authoritative_receipt() -> None:
+    with pytest.raises(SystemExit):
+        schema_generate.main(["--provider", "chatgpt"])
+
+
+def test_schema_generate_valid_route_consumes_fresh_gate_receipt(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    root, receipt = _authoritative_schema_gate(tmp_path, "valid")
+    monkeypatch.setattr(schema_generate, "get_config", lambda: _ConfigStub(root / "index.db", root))
+    monkeypatch.setattr(schema_generate, "infer_schema", _successful_inference)
+
+    assert schema_generate.main(["--provider", "chatgpt", "--schema-inference-receipt", str(receipt)]) == 0
+    assert "Generated schema package set for chatgpt" in capsys.readouterr().out
+
+
+def test_schema_generate_rejects_receipt_for_a_different_archive(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    _archive_a, receipt_a = _authoritative_schema_gate(tmp_path, "archive-a")
+    archive_b, _receipt_b = _authoritative_schema_gate(tmp_path, "archive-b")
+    monkeypatch.setattr(schema_generate, "get_config", lambda: _ConfigStub(archive_b / "index.db", archive_b))
+    monkeypatch.setattr(schema_generate, "infer_schema", pytest.fail)
+
+    assert schema_generate.main(["--provider", "chatgpt", "--schema-inference-receipt", str(receipt_a)]) == 1
+    error = capsys.readouterr().err
+    assert "different archive" in error or "attestation" in error
+
+
+@pytest.mark.parametrize("mutation", ["replaced", "stale"])
+def test_schema_generate_rejects_replaced_or_stale_receipt(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    mutation: str,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    root, receipt = _authoritative_schema_gate(tmp_path, mutation)
+    payload = json.loads(receipt.read_text(encoding="utf-8"))
+    if mutation == "replaced":
+        payload["receipt_nonce"] = "attacker-replacement"
+    else:
+        payload["generated_at"] = datetime(2020, 1, 1, tzinfo=UTC).isoformat()
+        payload["receipt_sha256"] = schema_inference_gate_receipt_digest(payload)
+    receipt.write_text(json.dumps(payload, sort_keys=True), encoding="utf-8")
+    monkeypatch.setattr(schema_generate, "get_config", lambda: _ConfigStub(root / "index.db", root))
+    monkeypatch.setattr(schema_generate, "infer_schema", pytest.fail)
+
+    assert schema_generate.main(["--provider", "chatgpt", "--schema-inference-receipt", str(receipt)]) == 1
+    error = capsys.readouterr().err
+    assert "digest mismatch" in error or "stale" in error or "attestation" in error
+
+
+def test_schema_generate_rejects_stale_non_source_tier(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    root, receipt = _authoritative_schema_gate(tmp_path, "stale-user-tier")
+    with sqlite3.connect(root / "user.db") as connection:
+        connection.execute("PRAGMA user_version = 0")
+    monkeypatch.setattr(schema_generate, "get_config", lambda: _ConfigStub(root / "index.db", root))
+    monkeypatch.setattr(schema_generate, "infer_schema", pytest.fail)
+
+    assert schema_generate.main(["--provider", "chatgpt", "--schema-inference-receipt", str(receipt)]) == 1
+    assert "tier identity is stale" in capsys.readouterr().err
+
+
+def test_schema_generate_refuses_concurrent_writer(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    root, receipt = _authoritative_schema_gate(tmp_path, "concurrent-writer")
+    monkeypatch.setattr(schema_generate, "get_config", lambda: _ConfigStub(root / "index.db", root))
+    monkeypatch.setattr(schema_generate, "infer_schema", pytest.fail)
+
+    with RebuildLease(root):
+        assert schema_generate.main(["--provider", "chatgpt", "--schema-inference-receipt", str(receipt)]) == 1
+    assert "exclusive offline archive ownership" in capsys.readouterr().err
+
+
 def test_schema_promote_forwards_cluster_request(
     monkeypatch: pytest.MonkeyPatch,
     tmp_path: Path,
diff --git a/tests/unit/maintenance/test_schema_inference_gate.py b/tests/unit/maintenance/test_schema_inference_gate.py
index 14cf969667..cb81bb8567 100644
--- a/tests/unit/maintenance/test_schema_inference_gate.py
+++ b/tests/unit/maintenance/test_schema_inference_gate.py
@@ -4,6 +4,7 @@
 
 import hashlib
 import sqlite3
+from datetime import UTC, datetime, timedelta
 from pathlib import Path
 from typing import TypedDict, cast
 
@@ -14,9 +15,12 @@
     RECEIPT_FILENAME,
     SchemaInferenceGateError,
     run_schema_inference_gate,
+    validate_schema_inference_gate_receipt,
 )
 from polylogue.storage.blob_store import BlobStore
+from polylogue.storage.index_generation import RebuildLease
 from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root
+from tests.infra.schema_inference import seed_schema_inference_archive
 
 
 class _RawGroundTruthProvenance(TypedDict):
@@ -72,61 +76,7 @@ class _GateReceipt(TypedDict):
     pass_fail_reasons: list[str]
 
 
-def _seed_archive(root: Path) -> Path:
-    """Create one source raw whose actual external file and blob agree."""
-
-    initialize_active_archive_root(root)
-    ground_truth = root.parent / f"{root.name}-codex-ground-truth"
-    ground_truth.mkdir()
-    payload = b"actual external codex raw"
-    source_file = ground_truth / "session.jsonl"
-    source_file.write_bytes(payload)
-    blob_hash, blob_size = BlobStore(root / "blob").write_from_bytes(payload)
-    with sqlite3.connect(root / "source.db") as conn:
-        conn.execute(
-            """
-            INSERT INTO raw_sessions(
-                raw_id, origin, native_id, source_path, blob_hash, blob_size,
-                acquired_at_ms, logical_source_key, revision_authority
-            ) VALUES ('raw-1', 'codex-session', 'session', ?, ?, ?, 100,
-                      'codex:session', 'byte_proven')
-            """,
-            (str(source_file), bytes.fromhex(blob_hash), blob_size),
-        )
-        conn.execute(
-            """
-            INSERT INTO raw_session_memberships(
-                raw_id, logical_source_key, provider_session_id, source_revision,
-                normalized_content_hash, message_count, decision, decided_at_ms
-            ) VALUES ('raw-1', 'codex:session', 'session', 'rev-1', ?, 1, 'applied', 100)
-            """,
-            (b"m" * 32,),
-        )
-    with sqlite3.connect(root / "index.db") as conn:
-        conn.execute(
-            """
-            INSERT INTO sessions(native_id, origin, raw_id, content_hash, message_count)
-            VALUES ('session', 'codex-session', 'raw-1', ?, 1)
-            """,
-            (b"s" * 32,),
-        )
-        conn.execute(
-            """
-            INSERT INTO messages(session_id, position, role, material_origin, content_hash)
-            VALUES ('codex-session:session', 0, 'user', 'human_authored', ?)
-            """,
-            (b"n" * 32,),
-        )
-        conn.execute(
-            """
-            INSERT INTO blocks(message_id, session_id, position, block_type, text)
-            VALUES ('codex-session:session:0.0', 'codex-session:session', 0, 'text', 'hello')
-            """
-        )
-        conn.execute("ANALYZE blocks")
-        conn.execute("ANALYZE messages")
-        conn.execute("ANALYZE action_pairs")
-    return ground_truth
+_seed_archive = seed_schema_inference_archive
 
 
 def _run(
@@ -137,10 +87,11 @@ def _run(
     ground_truth_roots: dict[str, tuple[Path, ...]] | None = None,
 ) -> _GateReceipt:
     external_root = ground_truth or root.parent / f"{root.name}-codex-ground-truth"
+    receipt_count = sum(1 for child in tmp_path.iterdir() if child.name.startswith("schema-gate-receipt-"))
     result = run_schema_inference_gate(
         root,
-        receipt_path=tmp_path / RECEIPT_FILENAME,
-        ground_truth_roots=ground_truth_roots or {"codex-session": (external_root,)},
+        receipt_path=tmp_path / f"schema-gate-receipt-{receipt_count}" / RECEIPT_FILENAME,
+        ground_truth_roots=({"codex-session": (external_root,)} if ground_truth_roots is None else ground_truth_roots),
     )
     return cast(_GateReceipt, result.payload)
 
@@ -181,6 +132,69 @@ def test_clean_archive_runs_actual_blobstore_verifier_and_external_reconciliatio
     } == before
 
 
+def test_empty_archive_cannot_authorize_schema_inference(tmp_path: Path) -> None:
+    root = tmp_path / "archive"
+    initialize_active_archive_root(root)
+
+    payload = _run(root, tmp_path, ground_truth_roots={})
+
+    assert payload["verdict"] == "FAIL"
+    assert "schema inference requires at least one reconciled source raw" in payload["pass_fail_reasons"]
+
+
+def test_gate_rejects_stale_non_source_tier_schema_identity(tmp_path: Path) -> None:
+    root = tmp_path / "archive"
+    ground_truth = _seed_archive(root)
+    with sqlite3.connect(root / "user.db") as conn:
+        conn.execute("PRAGMA user_version = 0")
+
+    payload = _run(root, tmp_path, ground_truth=ground_truth)
+
+    assert payload["verdict"] == "FAIL"
+    assert any("user.db schema identity is stale" in reason for reason in payload["pass_fail_reasons"])
+
+
+def test_gate_receipt_path_is_immutable(tmp_path: Path) -> None:
+    root = tmp_path / "archive"
+    ground_truth = _seed_archive(root)
+    receipt = tmp_path / RECEIPT_FILENAME
+    run_schema_inference_gate(root, receipt_path=receipt, ground_truth_roots={"codex-session": (ground_truth,)})
+
+    with pytest.raises(SchemaInferenceGateError, match="immutable"):
+        run_schema_inference_gate(root, receipt_path=receipt, ground_truth_roots={"codex-session": (ground_truth,)})
+
+
+def test_authoritative_receipt_expires_from_its_signed_generation_time(tmp_path: Path) -> None:
+    root = tmp_path / "archive"
+    ground_truth = _seed_archive(root)
+    receipt = tmp_path / RECEIPT_FILENAME
+    payload = run_schema_inference_gate(
+        root,
+        receipt_path=receipt,
+        ground_truth_roots={"codex-session": (ground_truth,)},
+    ).payload
+    generated_at = datetime.fromisoformat(str(payload["generated_at"])).astimezone(UTC)
+
+    with pytest.raises(SchemaInferenceGateError, match="stale or from the future"):
+        validate_schema_inference_gate_receipt(
+            receipt,
+            archive_root=root,
+            now=generated_at + timedelta(hours=24, seconds=1),
+        )
+
+
+def test_gate_refuses_concurrent_writer_lease(tmp_path: Path) -> None:
+    root = tmp_path / "archive"
+    ground_truth = _seed_archive(root)
+    with RebuildLease(root):
+        with pytest.raises(SchemaInferenceGateError, match="exclusive offline archive ownership"):
+            run_schema_inference_gate(
+                root,
+                receipt_path=tmp_path / RECEIPT_FILENAME,
+                ground_truth_roots={"codex-session": (ground_truth,)},
+            )
+
+
 def test_receipt_target_can_never_replace_a_live_tier(tmp_path: Path) -> None:
     root = tmp_path / "archive"
     ground_truth = _seed_archive(root)
@@ -241,6 +255,19 @@ def test_extra_external_file_is_rejected_by_bidirectional_reconciliation(tmp_pat
     assert any("external file(s) have no source raw match" in reason for reason in payload["pass_fail_reasons"])
 
 
+def test_archive_owned_blob_cannot_pose_as_external_ground_truth(tmp_path: Path) -> None:
+    root = tmp_path / "archive"
+    _seed_archive(root)
+    blob_path = next(path for path in (root / "blob").rglob("*") if path.is_file())
+
+    payload = _run(root, tmp_path, ground_truth=blob_path)
+
+    assert payload["verdict"] == "FAIL"
+    assert any(
+        "must be external to the archive and blob namespace" in reason for reason in payload["pass_fail_reasons"]
+    )
+
+
 def test_cross_origin_external_source_is_rejected_and_recorded(tmp_path: Path) -> None:
     root = tmp_path / "archive"
     ground_truth = _seed_archive(root)