ai-contract-platform/yuqei-infra/tests/test_rc06_release_trust_bootstrap.py

from __future__ import annotations

import base64
import copy
import hashlib
import importlib.util
import json
import os
import re
import stat
import subprocess
import sys
from datetime import UTC, datetime, timedelta
from pathlib import Path

import pytest
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding, rsa, utils
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey


REPO_ROOT = Path(__file__).resolve().parents[2]
SCRIPT = REPO_ROOT / "yuqei-infra" / "scripts" / "rc06-release-trust-bootstrap.py"
EXISTING_BATCH_TEST = REPO_ROOT / "yuqei-infra" / "tests" / "test_rc06_publication_batch.py"
AUTHORITY_LAUNCHER = REPO_ROOT / "yuqei-infra" / "scripts" / "Launch-ReleaseAuthority.ps1"


def _acl_snapshot(path: Path) -> bytes | int:
    if os.name != "nt":
        return stat.S_IMODE(path.stat().st_mode)
    completed = subprocess.run(
        ["icacls", os.fspath(path)],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        check=False,
    )
    if completed.returncode != 0:
        pytest.fail(f"cannot snapshot fixture ACL: {completed.stderr!r}")
    return completed.stdout


def _load_script():
    spec = importlib.util.spec_from_file_location("rc06_release_trust_bootstrap", SCRIPT)
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


authority = _load_script()

TEST_REVIEWER_IDENTITY = {
    **authority.batch.SOLE_RELEASE_REVIEWER,
    "role": "发布负责人测试角色",
    "organization": "测试组织（非真实主体）",
    "authorization_scope": "仅批准测试夹具中的精确 RC-06 批次，不授权公众切流",
}


def _load_existing_batch_test_module():
    spec = importlib.util.spec_from_file_location(
        "rc06_publication_batch_fixture_helpers",
        EXISTING_BATCH_TEST,
    )
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


def _protect(raw: bytes, role: str) -> bytes:
    return b"fixture-protected\0" + role.encode("ascii") + b"\0" + raw[::-1]


def _unprotect(raw: bytes, role: str) -> bytes:
    prefix = b"fixture-protected\0" + role.encode("ascii") + b"\0"
    if not raw.startswith(prefix):
        raise ValueError("wrong fixture protection domain")
    return raw[len(prefix) :][::-1]


def _reviewer_registration(path: Path, *, private_key: Ed25519PrivateKey | None = None) -> dict:
    if private_key is None and path.exists():
        return json.loads(path.read_bytes())
    private_key = private_key or Ed25519PrivateKey.generate()
    public_key = private_key.public_key()
    public_raw = public_key.public_bytes(
        serialization.Encoding.Raw,
        serialization.PublicFormat.Raw,
    )
    public_pem = public_key.public_bytes(
        serialization.Encoding.PEM,
        serialization.PublicFormat.SubjectPublicKeyInfo,
    )
    fingerprint = authority.batch._sha256_bytes(public_raw)
    registration = {
        "schema_version": authority.REVIEWER_REGISTRATION_SCHEMA,
        "key_id": f"chen-yuqi-rc06-{fingerprint[:16]}",
        "algorithm": "Ed25519",
        "reviewer": dict(authority.batch.SOLE_RELEASE_REVIEWER),
        "created_at": "2026-09-27T00:00:00Z",
        "public_key_pem": public_pem.decode("ascii"),
        "public_key_raw_sha256": fingerprint,
        "public_key_pem_sha256": authority.batch._sha256_bytes(public_pem),
    }
    path.write_bytes(authority._canonical(registration))
    return registration


def _online_service_registration(path: Path) -> dict:
    private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    public_pem = private_key.public_key().public_bytes(
        serialization.Encoding.PEM,
        serialization.PublicFormat.SubjectPublicKeyInfo,
    )
    public_spki = private_key.public_key().public_bytes(
        serialization.Encoding.DER,
        serialization.PublicFormat.SubjectPublicKeyInfo,
    )
    registration = {
        "schema_version": authority.ONLINE_SERVICE_REGISTRATION_SCHEMA,
        "provider": "aliyun",
        "key_reference": "kms://fixture/rc06-online",
        "key_version": "version-1",
        "algorithm": "RSA_PSS_SHA_256",
        "component": authority.batch.TRUSTED_ONLINE_APPLY_SERVICE_COMPONENT,
        "created_at": "2026-09-27T02:30:00Z",
        "public_key_pem": public_pem.decode("ascii"),
        "public_key_spki_sha256": authority.batch._sha256_bytes(public_spki),
    }
    path.write_bytes(authority._canonical(registration))
    return registration


def _legal_keyring(path: Path) -> tuple[dict, Ed25519PrivateKey]:
    private_key = Ed25519PrivateKey.generate()
    public_raw = private_key.public_key().public_bytes(
        serialization.Encoding.Raw,
        serialization.PublicFormat.Raw,
    )
    keyring = {
        "schema_version": "yuqei.scene-source-legal-review-keyring.v1",
        "keys": [
            {
                "key_id": "fixture-legal-reviewer",
                "status": "trusted",
                "revoked_at": None,
                "valid_from": "2026-09-26T00:00:00Z",
                "valid_until": "2026-09-28T00:00:00Z",
                "reviewer_name": "测试法源复核人（非真实身份）",
                "reviewer_organization": "测试组织（非真实法律服务主体）",
                "reviewer_qualification": "测试资格声明（不代表真实法律资质）",
                "credential_reference": "TEST-FIXTURE-NOT-A-CREDENTIAL",
                "public_key_raw_base64": base64.b64encode(public_raw).decode("ascii"),
            }
        ],
    }
    path.write_bytes(authority._canonical(keyring))
    return keyring, private_key


class _FakeCanonicalLegalReviewValidator:
    @staticmethod
    def verify_request(request: dict, _repo: Path) -> dict:
        projection = {key: value for key, value in request.items() if key != "request_sha256"}
        if request.get("request_sha256") != authority.batch._sha256_json(projection):
            raise ValueError("review request embedded hash drift")
        if len(request.get("occurrences", [])) != 43:
            raise ValueError("review request must contain 43 occurrences")
        return request

    @staticmethod
    def verify_signed_decision(
        request: dict, decision: dict, _keyring: dict, *, checked_at: str
    ) -> dict:
        if decision.get("request_sha256") != request["request_sha256"]:
            raise ValueError("decision request binding drift")
        if datetime.fromisoformat(checked_at) > datetime.fromisoformat(
            decision["expires_at"]
        ):
            raise ValueError("decision expired")
        return {
            "status": "verified_qualified_human_review",
            "candidate_commit": request["candidate_commit"],
            "request_sha256": request["request_sha256"],
            "reviewer_name": decision["reviewer"]["name"],
            "reviewed_occurrence_count": 43,
            "all_current_effectiveness_verified": True,
            "all_transaction_applicability_verified": True,
            "expires_at": decision["expires_at"],
            "authorizes_publication_by_itself": False,
        }


def _kms_envelope(
    private_key,
    *,
    schema: str,
    prefix: str,
    domain: bytes,
    payload: dict,
    request_id: str,
) -> dict:
    digest = hashlib.sha256(domain + authority._canonical(payload)).digest()
    signature_value = private_key.sign(
        digest,
        padding.PSS(
            mgf=padding.MGF1(hashes.SHA256()),
            salt_length=hashes.SHA256().digest_size,
        ),
        utils.Prehashed(hashes.SHA256()),
    )
    signature = {
        "provider": "fixture-kms",
        "key_reference": "kms://fixture/rc06-online",
        "key_version": "version-1",
        "algorithm": "RSA_PSS_SHA_256",
        "value_base64": base64.b64encode(signature_value).decode("ascii"),
        "request_id": request_id,
    }
    attestation_payload = {
        "schema_version": authority.batch.KMS_REQUEST_ID_ATTESTATION_PAYLOAD_SCHEMA_VERSION,
        "receipt_payload_sha256": authority.batch._sha256_json(payload),
        "primary_signature_sha256": authority.batch._sha256_bytes(signature_value),
        "primary_provider": signature["provider"],
        "primary_key_reference": signature["key_reference"],
        "primary_key_version": signature["key_version"],
        "primary_algorithm": signature["algorithm"],
        "primary_request_id": request_id,
    }
    attestation_domain = (
        authority.batch.ONLINE_REQUEST_ID_ATTESTATION_DOMAIN
        if domain == authority.batch.ONLINE_TRANSACTION_SIGNATURE_DOMAIN
        else authority.batch.RUNTIME_REQUEST_ID_ATTESTATION_DOMAIN
    )
    attestation_digest = hashlib.sha256(
        attestation_domain + authority._canonical(attestation_payload)
    ).digest()
    attestation_signature_value = private_key.sign(
        attestation_digest,
        padding.PSS(
            mgf=padding.MGF1(hashes.SHA256()),
            salt_length=hashes.SHA256().digest_size,
        ),
        utils.Prehashed(hashes.SHA256()),
    )
    request_id_attestation = {
        "schema_version": authority.batch.KMS_REQUEST_ID_ATTESTATION_SCHEMA_VERSION,
        "payload": attestation_payload,
        "signature": {
            "provider": signature["provider"],
            "key_reference": signature["key_reference"],
            "key_version": signature["key_version"],
            "algorithm": signature["algorithm"],
            "value_base64": base64.b64encode(attestation_signature_value).decode("ascii"),
        },
    }
    return {
        "schema_version": schema,
        "receipt_id": prefix
        + authority.batch._sha256_bytes(
            authority._canonical(payload)
            + authority._canonical(signature)
            + authority._canonical(request_id_attestation)
        )[:24],
        "payload": payload,
        "signature": signature,
        "request_id_attestation": request_id_attestation,
    }


def _initialize(tmp_path: Path):
    state = tmp_path / "authority-state"
    reviewer_file = tmp_path / "reviewer-registration.json"
    _reviewer_registration(reviewer_file)
    result = authority.initialize_authority(
        state,
        reviewer_file,
        protect=_protect,
        unprotect=_unprotect,
        created_at=datetime(2026, 9, 27, 1, 0, tzinfo=UTC),
    )
    return state, result, authority.load_authority(
        state,
        unprotect=_unprotect,
        unseal_roles=authority.AUTHORITY_ROLES,
    )


def _copy_pin_sources(tmp_path: Path) -> Path:
    repo = tmp_path / "candidate"
    paths = (
        "yuqei-infra/scripts/build-rc06-publication-batch.py",
        "yuqei-infra/deploy/plan-first-release-snapshot-restore.py",
        "yuqei-infra/deploy/plan-rollback-rehearsal.py",
    )
    for relative in paths:
        source = REPO_ROOT / relative
        target = repo / relative
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_bytes(source.read_bytes())
    return repo


def _formal_apply_materials(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict:
    reviewer_private = Ed25519PrivateKey.generate()
    reviewer_file = tmp_path / "reviewer-registration.json"
    registration = _reviewer_registration(reviewer_file, private_key=reviewer_private)
    state, _result, keys = _initialize(tmp_path)
    helpers = _load_existing_batch_test_module()
    inventory = helpers.batch.load_runtime_inventory(REPO_ROOT)
    manifest = helpers._manifest(inventory, resolved=True)
    manifest_raw = authority._canonical(manifest)
    manifest_file = tmp_path / "rc06-publication-manifest.json"
    manifest_file.write_bytes(manifest_raw)
    manifest_sha = authority.batch._sha256_bytes(manifest_raw)
    monkeypatch.setattr(
        authority.batch,
        "TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256",
        keys[authority.ROLE_RC06_ROOT].public_sha256,
    )
    monkeypatch.setattr(
        authority.batch,
        "TRUSTED_APPLY_SERVICE_PUBLIC_KEY_SHA256",
        keys[authority.ROLE_APPLY_SERVICE].public_sha256,
    )
    online_private = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    online_public_pem = online_private.public_key().public_bytes(
        serialization.Encoding.PEM,
        serialization.PublicFormat.SubjectPublicKeyInfo,
    )
    online_public_spki = online_private.public_key().public_bytes(
        serialization.Encoding.DER,
        serialization.PublicFormat.SubjectPublicKeyInfo,
    )
    online_public_file = tmp_path / "online-service-public.pem"
    online_public_file.write_bytes(online_public_pem)
    monkeypatch.setattr(
        authority.batch, "TRUSTED_ONLINE_APPLY_SERVICE_PROVIDER", "fixture-kms"
    )
    monkeypatch.setattr(
        authority.batch,
        "TRUSTED_ONLINE_APPLY_SERVICE_KEY_REFERENCE",
        "kms://fixture/rc06-online",
    )
    monkeypatch.setattr(
        authority.batch, "TRUSTED_ONLINE_APPLY_SERVICE_KEY_VERSION", "version-1"
    )
    monkeypatch.setattr(
        authority.batch, "TRUSTED_ONLINE_APPLY_SERVICE_ALGORITHM", "RSA_PSS_SHA_256"
    )
    monkeypatch.setattr(
        authority.batch,
        "TRUSTED_ONLINE_APPLY_SERVICE_PUBLIC_KEY_SPKI_SHA256",
        authority.batch._sha256_bytes(online_public_spki),
    )
    monkeypatch.setattr(
        authority.batch,
        "TRUSTED_ONLINE_APPLY_ENVIRONMENT_IDS",
        ("isolated-uat-fixture",),
    )
    monkeypatch.setattr(authority.batch, "FROZEN_APPROVAL_KEYRING_EPOCH", 1)
    keyring_result = authority.build_rc06_candidate_keyring(
        state,
        reviewer_file,
        manifest_file,
        manifest_sha,
        unprotect=_unprotect,
        now=datetime(2026, 9, 27, 3, 0, tzinfo=UTC),
    )
    keyring_directory = Path(keyring_result["directory"])
    keyring_file = keyring_directory / "rc06-approval-keyring.json"
    keyring_raw = keyring_file.read_bytes()
    keyring = json.loads(keyring_raw)
    approved_at = datetime(2026, 9, 27, 3, 10, tzinfo=UTC)
    expires_at = datetime(2026, 9, 27, 5, 0, tzinfo=UTC)
    approval_payload = authority.batch.build_approval_payload(
        manifest,
        manifest_raw_sha256=manifest_sha,
        key_id=registration["key_id"],
        approval_keyring_epoch=1,
        approval_keyring_payload_sha256=keyring["keyring_payload_sha256"],
        approved_at=approved_at,
        expires_at=expires_at,
        approver={**TEST_REVIEWER_IDENTITY, "human_confirmed": True},
    )
    approval_signature = reviewer_private.sign(
        authority.batch.APPROVAL_SIGNATURE_DOMAIN + authority._canonical(approval_payload)
    )
    approval = {
        "schema_version": authority.batch.APPROVAL_SCHEMA_VERSION,
        "approval_id": (
            "rc06-batch-approval-"
            + authority.batch._sha256_bytes(
                authority._canonical(approval_payload) + approval_signature
            )[:24]
        ),
        "payload": approval_payload,
        "signature": base64.b64encode(approval_signature).decode("ascii"),
    }
    approval_file = tmp_path / "rc06-batch-approval.json"
    approval_raw = authority._canonical(approval)
    approval_file.write_bytes(approval_raw)
    approval_sha = authority.batch._sha256_bytes(approval_raw)

    legal_keyring_file = tmp_path / "scene-legal-keyring.json"
    legal_keyring, _legal_private = _legal_keyring(legal_keyring_file)
    legal_keyring_raw = legal_keyring_file.read_bytes()
    legal_keyring_sha = authority.batch._sha256_bytes(legal_keyring_raw)
    monkeypatch.setattr(
        authority.batch,
        "TRUSTED_SCENE_LEGAL_REVIEW_KEYRING_SHA256",
        legal_keyring_sha,
    )
    monkeypatch.setattr(
        authority.batch,
        "_load_scene_legal_review_module",
        lambda: _FakeCanonicalLegalReviewValidator,
    )
    occurrences = [
        {
            "occurrence_id": f"fixture-occurrence-{index:02d}",
            "source_record_sha256": f"{index + 1:064x}",
        }
        for index in range(43)
    ]
    legal_request = {
        "schema_version": "yuqei.scene-source-legal-review-request.v1",
        "status": "pending_qualified_human_review",
        "candidate_commit": manifest["candidate"]["commit"],
        "created_at": "2026-09-27T03:00:00Z",
        "scope": {
            "scene_count": 30,
            "reference_occurrence_count": 43,
            "distinct_source_record_count": 43,
            "distinct_url_count": 43,
        },
        "source_files": ["fixture-only/not-production.json"],
        "occurrences": occurrences,
        "guardrails": {
            "not_legal_approval": True,
            "authorizes_publication": False,
            "ai_must_not_supply_human_decisions": True,
            "all_occurrences_require_qualified_human_review": True,
        },
    }
    legal_request["request_sha256"] = authority.batch._sha256_json(legal_request)
    legal_decision = {
        "schema_version": "yuqei.scene-source-legal-review-decision.v1",
        "request_sha256": legal_request["request_sha256"],
        "candidate_commit": manifest["candidate"]["commit"],
        "reviewed_at": "2026-09-27T03:30:00Z",
        "expires_at": "2026-09-27T05:00:00Z",
        "reviewer": {"name": "测试法源复核人（非真实身份）"},
        "items": [],
        "signature": {
            "algorithm": "Ed25519",
            "key_id": "fixture-legal-reviewer",
            "value_base64": base64.b64encode(b"fixture-signature").decode("ascii"),
        },
    }
    legal_request_file = tmp_path / "scene-legal-request.json"
    legal_decision_file = tmp_path / "scene-legal-decision.json"
    legal_request_file.write_bytes(authority._canonical(legal_request))
    legal_decision_file.write_bytes(authority._canonical(legal_decision))
    legal_request_sha = authority.batch._sha256_bytes(legal_request_file.read_bytes())
    legal_decision_sha = authority.batch._sha256_bytes(legal_decision_file.read_bytes())
    challenge_result = authority.issue_apply_challenge(
        state,
        manifest_file=manifest_file,
        manifest_sha256=manifest_sha,
        approval_file=approval_file,
        approval_sha256=approval_sha,
        approval_keyring_file=keyring_file,
        approval_keyring_sha256=authority.batch._sha256_bytes(keyring_raw),
        approval_root_public_key_file=(
            keyring_directory / "rc06-approval-root-public-key.raw"
        ),
        legal_review_request_file=legal_request_file,
        legal_review_request_sha256=legal_request_sha,
        legal_review_decision_file=legal_decision_file,
        legal_review_decision_sha256=legal_decision_sha,
        legal_review_keyring_file=legal_keyring_file,
        legal_review_keyring_sha256=legal_keyring_sha,
        environment_id="isolated-uat-fixture",
        tenant_id="tenant-fixture",
        unprotect=_unprotect,
        now=datetime(2026, 9, 27, 3, 55, tzinfo=UTC),
    )
    challenge_file = Path(challenge_result["challenge_file"])
    challenge = json.loads(challenge_file.read_bytes())
    item_results = authority.batch._expected_apply_item_results(manifest)
    runtime_results = authority.batch.expected_runtime_snapshot_results(manifest)
    challenge_payload = challenge["payload"]
    images = {
        item["role"]: item["manifest_digest"]
        for item in manifest["candidate"]["images"]
    }
    participant_evidence = []
    for participant, committed_at, write_count, runtime_count, image, results in (
        (
            "ai-platform",
            "2026-09-27T03:58:00Z",
            14,
            14,
            images["ai-platform-api"],
            [item for item in item_results if item["kind"] == "prompt_module"],
        ),
        (
            "contract-platform",
            "2026-09-27T03:59:00Z",
            60,
            0,
            images["contract-api"],
            [item for item in item_results if item["kind"] == "asset"],
        ),
    ):
        projection = {
            "participant": participant,
            "state": "committed",
            "transaction_id": challenge_payload["transaction_id"],
            "candidate_commit": manifest["candidate"]["commit"],
            "release_image_digest": image,
            "committed_at": committed_at,
            "write_count": write_count,
            "runtime_snapshot_count": runtime_count,
            "committed_item_results_sha256": authority.batch._sha256_json(results),
        }
        participant_evidence.append(
            {
                **projection,
                "durable_commit_evidence_sha256": authority.batch._sha256_json(
                    projection
                ),
            }
        )
    online_payload = {
        "schema_version": authority.batch.ONLINE_TRANSACTION_PAYLOAD_SCHEMA_VERSION,
        "candidate_commit": manifest["candidate"]["commit"],
        "candidate_tag": manifest["candidate"]["tag"],
        "batch_manifest_raw_sha256": manifest_sha,
        "batch_manifest_payload_sha256": manifest["manifest_payload_sha256"],
        "approval_receipt_sha256": approval_sha,
        "legal_source_review_binding_sha256": challenge_payload[
            "legal_source_review_binding_sha256"
        ],
        "environment_id": "isolated-uat-fixture",
        "tenant_id": "tenant-fixture",
        "idempotency_key": challenge_payload["idempotency_key"],
        "transaction_id": challenge_payload["transaction_id"],
        "transaction_mode": authority.batch.TRANSACTION_MODE,
        "cross_database_atomicity": False,
        "coordinator_deployment": {
            "component": authority.batch.TRUSTED_ONLINE_APPLY_SERVICE_COMPONENT,
            "candidate_commit": manifest["candidate"]["commit"],
            "release_image_digest": images["contract-api"],
        },
        "participant_evidence": participant_evidence,
        "committed_at": "2026-09-27T04:00:00Z",
        "authority_challenge": challenge,
        "generated_by": {"type": "system", "component": "rc06-publication-service"},
        "item_results": item_results,
        "write_count": 74,
        "status": "committed_and_verified",
    }
    online_receipt = _kms_envelope(
        online_private,
        schema=authority.batch.ONLINE_TRANSACTION_RECEIPT_SCHEMA_VERSION,
        prefix="rc06-online-transaction-",
        domain=authority.batch.ONLINE_TRANSACTION_SIGNATURE_DOMAIN,
        payload=online_payload,
        request_id="fixture-kms-online",
    )
    online_receipts_file = tmp_path / "online-transaction-receipts.json"
    online_receipts_raw = authority._canonical(online_receipt)
    online_receipts_file.write_bytes(online_receipts_raw)
    runtime_payload = {
        "schema_version": authority.batch.RUNTIME_PAYLOAD_SCHEMA_VERSION,
        "batch_manifest_raw_sha256": manifest_sha,
        "batch_manifest_payload_sha256": manifest["manifest_payload_sha256"],
        "approval_receipt_sha256": approval_sha,
        "legal_source_review_binding_sha256": challenge_payload[
            "legal_source_review_binding_sha256"
        ],
        "environment_id": "isolated-uat-fixture",
        "tenant_id": "tenant-fixture",
        "idempotency_key": challenge_payload["idempotency_key"],
        "transaction_id": challenge_payload["transaction_id"],
        "online_transaction_receipt_sha256": authority.batch._sha256_json(
            online_receipt
        ),
        "captured_at": "2026-09-27T04:01:00Z",
        "generated_by": {
            "type": "system",
            "component": authority.batch.TRUSTED_ONLINE_APPLY_SERVICE_COMPONENT,
        },
        "transaction_mode": authority.batch.TRANSACTION_MODE,
        "cross_database_atomicity": False,
        "publication_item_count": 74,
        "runtime_snapshot_dependencies": runtime_results,
        "runtime_snapshot_dependency_count": 14,
        "status": "published_and_resolved",
    }
    runtime_receipt = _kms_envelope(
        online_private,
        schema=authority.batch.RUNTIME_RECEIPT_SCHEMA_VERSION,
        prefix="rc06-runtime-snapshot-",
        domain=authority.batch.RUNTIME_SIGNATURE_DOMAIN,
        payload=runtime_payload,
        request_id="fixture-kms-runtime",
    )
    runtime_receipt_file = tmp_path / "runtime-snapshot-receipt.json"
    runtime_receipt_raw = authority._canonical(runtime_receipt)
    runtime_receipt_file.write_bytes(runtime_receipt_raw)
    return {
        "state": state,
        "keys": keys,
        "reviewer_private": reviewer_private,
        "registration": registration,
        "manifest": manifest,
        "manifest_file": manifest_file,
        "manifest_sha": manifest_sha,
        "keyring": keyring,
        "keyring_file": keyring_file,
        "keyring_sha": authority.batch._sha256_bytes(keyring_raw),
        "root_file": keyring_directory / "rc06-approval-root-public-key.raw",
        "approval": approval,
        "approval_file": approval_file,
        "approval_sha": approval_sha,
        "challenge": challenge,
        "challenge_file": challenge_file,
        "legal_request_file": legal_request_file,
        "legal_request_sha": legal_request_sha,
        "legal_decision_file": legal_decision_file,
        "legal_decision_sha": legal_decision_sha,
        "legal_keyring_file": legal_keyring_file,
        "legal_keyring_sha": legal_keyring_sha,
        "online_private": online_private,
        "online_public_file": online_public_file,
        "online_receipt": online_receipt,
        "online_receipts_file": online_receipts_file,
        "online_receipts_sha": authority.batch._sha256_bytes(online_receipts_raw),
        "runtime_receipt": runtime_receipt,
        "runtime_receipt_file": runtime_receipt_file,
        "runtime_receipt_sha": authority.batch._sha256_bytes(runtime_receipt_raw),
    }


def test_authority_initialization_creates_three_distinct_protected_keys_and_is_idempotent(
    tmp_path: Path,
) -> None:
    state, result, keys = _initialize(tmp_path)

    assert result["status"] == "initialized_not_pinned_not_signed"
    assert result["private_key_export_supported"] is False
    assert set(keys) == set(authority.AUTHORITY_ROLES)
    assert len({item.public_raw for item in keys.values()}) == 3
    key_dir = state / authority.KEYS_DIRECTORY_NAME
    assert set(child.name for child in key_dir.iterdir()) == (
        authority._expected_authority_names() | {authority.signer.BUNDLE_COMPLETE_FILENAME}
    )
    for role, item in keys.items():
        protected_raw = (key_dir / authority._private_filename(role)).read_bytes()
        private_der = authority._private_der(item.private_key)
        assert private_der not in protected_raw
        assert b"BEGIN PRIVATE KEY" not in protected_raw
        envelope = json.loads(protected_raw)
        assert envelope["protection"] == authority.PROTECTION_NAME
        assert base64.b64decode(envelope["ciphertext_base64"], validate=True) != private_der

    unsealed_roles: list[str] = []

    def observed_unprotect(raw: bytes, role: str) -> bytes:
        unsealed_roles.append(role)
        return _unprotect(raw, role)

    repeated = authority.initialize_authority(
        state,
        tmp_path / "reviewer-registration.json",
        protect=lambda _raw, _role: pytest.fail("idempotent initialization generated a new key"),
        unprotect=observed_unprotect,
    )
    assert repeated["status"] == "already_initialized_and_verified"
    assert repeated["roles"] == result["roles"]
    assert unsealed_roles == []


def test_authority_rejects_partial_or_tampered_bundle(tmp_path: Path) -> None:
    state, _result, _keys = _initialize(tmp_path)
    key_dir = state / authority.KEYS_DIRECTORY_NAME
    (key_dir / authority._public_filename(authority.ROLE_APPLY_SERVICE)).unlink()

    with pytest.raises(authority.AuthorityError, match="incomplete"):
        authority.initialize_authority(
            state,
            tmp_path / "reviewer-registration.json",
            protect=_protect,
            unprotect=_unprotect,
        )


def test_authority_manifest_requires_exact_role_array_objects(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    state, _result, _keys = _initialize(tmp_path)
    manifest_path = state / authority.KEYS_DIRECTORY_NAME / authority.AUTHORITY_MANIFEST_FILENAME
    original = json.loads(manifest_path.read_bytes())
    monkeypatch.setattr(authority.signer, "_verify_complete_bundle", lambda *args, **kwargs: None)

    extra = copy.deepcopy(original)
    extra["roles"].append(copy.deepcopy(extra["roles"][0]))
    manifest_path.write_bytes(authority._canonical(extra))
    with pytest.raises(authority.AuthorityError, match="role inventory drift"):
        authority.load_authority(state, unprotect=_unprotect, unseal_roles=())

    missing_field = copy.deepcopy(original)
    del missing_field["roles"][0]["algorithm"]
    manifest_path.write_bytes(authority._canonical(missing_field))
    with pytest.raises(authority.AuthorityError, match="role inventory drift"):
        authority.load_authority(state, unprotect=_unprotect, unseal_roles=())


def test_recovery_keyrings_share_only_the_recovery_root_and_verify_both_schemas(
    tmp_path: Path,
) -> None:
    state, _result, keys = _initialize(tmp_path)
    reviewer_file = tmp_path / "reviewer-registration.json"
    reviewer = _reviewer_registration(reviewer_file)
    now = datetime(2026, 9, 27, 2, 0, tzinfo=UTC)

    result = authority.build_recovery_keyrings(
        state,
        reviewer_file,
        epoch=1,
        unprotect=_unprotect,
        now=now,
        validity=timedelta(days=30),
    )

    assert result["status"] == "recovery_keyrings_created_not_pinned"
    assert result["recovery_root_public_key_sha256"] == keys[
        authority.ROLE_RECOVERY_ROOT
    ].public_sha256
    assert result["first_release_keyring_payload_sha256"] != result[
        "rollback_keyring_payload_sha256"
    ]
    bundle = Path(result["directory"])
    first_release = json.loads((bundle / "first-release-recovery-keyring.json").read_bytes())
    rollback = json.loads((bundle / "rollback-rehearsal-keyring.json").read_bytes())
    assert first_release["root_key_id"] == authority.FIRST_RELEASE_ROOT_KEY_ID
    assert rollback["root_key_id"] == authority.ROLLBACK_ROOT_KEY_ID
    assert "algorithm" not in first_release["keys"][0]
    assert rollback["keys"][0]["algorithm"] == "ed25519"
    assert first_release["keys"][0]["public_key_base64"] == rollback["keys"][0][
        "public_key_base64"
    ]
    assert reviewer["public_key_raw_sha256"] not in {
        item.public_sha256 for item in keys.values()
    }

    repeated = authority.build_recovery_keyrings(
        state,
        reviewer_file,
        epoch=1,
        unprotect=_unprotect,
        now=now + timedelta(hours=1),
    )
    assert repeated["status"] == "recovery_keyrings_already_created_and_verified_not_pinned"
    assert repeated["first_release_keyring_payload_sha256"] == result[
        "first_release_keyring_payload_sha256"
    ]


def test_recovery_keyrings_reject_reviewer_reuse_of_any_technical_key(tmp_path: Path) -> None:
    state, _result, keys = _initialize(tmp_path)
    technical_private = keys[authority.ROLE_APPLY_SERVICE].private_key
    reviewer_file = tmp_path / "reviewer-registration.json"
    _reviewer_registration(reviewer_file, private_key=technical_private)

    with pytest.raises(authority.AuthorityError, match="initialization binding"):
        authority.build_recovery_keyrings(
            state,
            reviewer_file,
            epoch=1,
            unprotect=_unprotect,
            now=datetime(2026, 9, 27, 2, 0, tzinfo=UTC),
        )


def test_pin_preview_and_apply_update_only_public_hashes_and_epoch_in_temp_sources(
    tmp_path: Path,
) -> None:
    state, _result, keys = _initialize(tmp_path)
    reviewer_file = tmp_path / "reviewer-registration.json"
    _reviewer_registration(reviewer_file)
    recovery = authority.build_recovery_keyrings(
        state,
        reviewer_file,
        epoch=1,
        unprotect=_unprotect,
        now=datetime(2026, 9, 27, 2, 0, tzinfo=UTC),
    )
    repo = _copy_pin_sources(tmp_path)
    online_registration_file = tmp_path / "online-service-registration.json"
    online_registration = _online_service_registration(online_registration_file)
    legal_keyring_file = tmp_path / "scene-legal-keyring.json"
    _legal_keyring(legal_keyring_file)
    online_environments = ("isolated-uat-fixture",)
    before = {
        path.relative_to(repo): path.read_bytes()
        for path in repo.rglob("*.py")
    }

    plan = authority.plan_pin_updates(
        repo,
        state,
        Path(recovery["directory"]),
        epoch=1,
        unprotect=_unprotect,
        online_service_registration_file=online_registration_file,
        online_environment_ids=online_environments,
        legal_review_keyring_file=legal_keyring_file,
    )
    preview = authority.pin_plan_summary(plan)
    assert preview["status"] == "preview_only_not_applied"
    assert all(path.read_bytes() == before[path.relative_to(repo)] for path in repo.rglob("*.py"))
    assert preview["values"]["online_apply_adapter_pinned"] is True
    assert preview["values"]["online_apply_environment_ids"] == online_environments
    assert preview["values"]["scene_legal_review_keyring_pinned"] is True

    applied = authority.apply_pin_plan(plan)
    assert applied["status"] == "pins_applied_recoverable_transaction_not_frozen_not_signed"
    assert applied["transaction_semantics"] == "recoverable_multi_file_not_atomic"
    assert Path(applied["audit_journal"]).is_file()
    assert applied["changed_file_count"] == 3
    rc06 = (repo / "yuqei-infra/scripts/build-rc06-publication-batch.py").read_text(
        encoding="utf-8"
    )
    first = (repo / "yuqei-infra/deploy/plan-first-release-snapshot-restore.py").read_text(
        encoding="utf-8"
    )
    rollback = (repo / "yuqei-infra/deploy/plan-rollback-rehearsal.py").read_text(
        encoding="utf-8"
    )
    assert f'TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = "{keys[authority.ROLE_RC06_ROOT].public_sha256}"' in rc06
    assert f'TRUSTED_APPLY_SERVICE_PUBLIC_KEY_SHA256 = "{keys[authority.ROLE_APPLY_SERVICE].public_sha256}"' in rc06
    assert f'TRUSTED_ONLINE_APPLY_SERVICE_PROVIDER = "{online_registration["provider"]}"' in rc06
    assert f'TRUSTED_ONLINE_APPLY_SERVICE_KEY_REFERENCE = "{online_registration["key_reference"]}"' in rc06
    assert f'TRUSTED_ONLINE_APPLY_SERVICE_KEY_VERSION = "{online_registration["key_version"]}"' in rc06
    assert f'TRUSTED_ONLINE_APPLY_SERVICE_ALGORITHM = "{online_registration["algorithm"]}"' in rc06
    assert f'TRUSTED_ONLINE_APPLY_SERVICE_PUBLIC_KEY_SPKI_SHA256 = "{online_registration["public_key_spki_sha256"]}"' in rc06
    assert f'TRUSTED_SCENE_LEGAL_REVIEW_KEYRING_SHA256 = "{authority.batch._sha256_bytes(legal_keyring_file.read_bytes())}"' in rc06
    assert "FROZEN_APPROVAL_KEYRING_EPOCH = 1" in rc06
    for text in (first, rollback):
        assert f'TRUSTED_ROOT_PUBLIC_KEY_SHA256 = "{keys[authority.ROLE_RECOVERY_ROOT].public_sha256}"' in text
        assert re.search(r"(?m)^MIN_KEYRING_EPOCH: int \| None = 1$", text)
        assert "PRIVATE KEY" not in text

    repeated = authority.plan_pin_updates(
        repo,
        state,
        Path(recovery["directory"]),
        epoch=1,
        unprotect=_unprotect,
        online_service_registration_file=online_registration_file,
        online_environment_ids=online_environments,
        legal_review_keyring_file=legal_keyring_file,
    )
    assert all(change.original == change.proposed for change in repeated.changes)
    assert authority.apply_pin_plan(repeated)["changed_file_count"] == 0


def test_pin_planner_refuses_unknown_existing_pin_without_overwriting(tmp_path: Path) -> None:
    state, _result, _keys = _initialize(tmp_path)
    reviewer_file = tmp_path / "reviewer-registration.json"
    _reviewer_registration(reviewer_file)
    recovery = authority.build_recovery_keyrings(
        state,
        reviewer_file,
        epoch=1,
        unprotect=_unprotect,
        now=datetime(2026, 9, 27, 2, 0, tzinfo=UTC),
    )
    repo = _copy_pin_sources(tmp_path)
    rc06 = repo / "yuqei-infra/scripts/build-rc06-publication-batch.py"
    original = rc06.read_text(encoding="utf-8")
    rc06.write_text(
        original.replace(
            'TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = ""',
            'TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = "' + "f" * 64 + '"',
        ),
        encoding="utf-8",
    )

    # A different already-provisioned pin cannot be replaced by a first-time
    # provisioning run.
    with pytest.raises(authority.AuthorityError, match="different epoch|unexpected existing value"):
        authority.plan_pin_updates(
            repo,
            state,
            Path(recovery["directory"]),
            epoch=1,
            unprotect=_unprotect,
        )


@pytest.mark.parametrize(
    ("injection", "message"),
    (
        (
            '\nTRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = ""\n',
            "one unique module-level literal",
        ),
        (
            '\nglobals()["TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256"] = "' + "f" * 64 + '"\n',
            "dynamic namespace mutation",
        ),
        (
            '\nexec("TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = \\\"' + "f" * 64 + '\\\"")\n',
            "untrusted dynamic exec",
        ),
        (
            "\nexec(compile("
            "\"TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = '"
            + "f" * 64
            + "'\", '<pin-write>', 'exec'))\n",
            "untrusted dynamic exec",
        ),
        (
            "\nfrom attacker_pin_side_effect import *\n",
            "star import",
        ),
        (
            "\nfrom builtins import exec as _danger\n"
            "_danger(\"TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = '"
            + "f" * 64
            + "'\")\n",
            "aliases dangerous built-in exec",
        ),
        (
            "\nimport sys as _shadow_sys\n"
            "from builtins import setattr as _danger\n"
            "_danger(_shadow_sys.modules[__name__], "
            "'TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256', '"
            + "f" * 64
            + "')\n",
            "aliases dangerous built-in setattr",
        ),
        (
            "\n_danger = exec\n"
            "_danger(\"TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = '"
            + "f" * 64
            + "'\")\n",
            "aliases dangerous built-in exec",
        ),
        (
            "\nimport builtins as _builtins\nimport sys as _shadow_sys\n"
            "_danger = getattr(_builtins, 'exec')\n"
            "if sys.argv[0] != '-c':\n"
            "    _danger(\"TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = '"
            + "f" * 64
            + "'\", _shadow_sys.modules[__name__].__dict__)\n",
            "looks up dangerous attribute exec",
        ),
        (
            "\nimport sys as _shadow_sys\n"
            "if __name__ != '_yuqei_pin_runtime_validation':\n"
            "    _shadow_sys.modules[__name__].TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = '"
            + "f" * 64
            + "'\n",
            "mutates frozen pin.*attribute",
        ),
        (
            "\n_shadow = {}\n"
            "_shadow['TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256'] = '"
            + "f" * 64
            + "'\n",
            "mutates frozen pin.*subscript",
        ),
        (
            "\nsetattr(object(), 'TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256', '"
            + "f" * 64
            + "')\n",
            "dynamically mutates a frozen pin",
        ),
        (
            "\nimport sys as _shadow_sys\n"
            "_shadow_sys.modules[__name__].__dict__.update({"
            "'TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256': '"
            + "f" * 64
            + "'})\n",
            "dynamic module namespace mutation",
        ),
        (
            "\nimport sys as _shadow_sys\n"
            "getattr(_shadow_sys.modules[__name__], '__dict__').update({"
            "'TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256': '"
            + "f" * 64
            + "'})\n",
            "dynamic module namespace mutation",
        ),
        (
            "\nimport sys as _shadow_sys\n"
            "_namespace = _shadow_sys.modules[__name__].__dict__\n"
            "if _shadow_sys.argv[0] != '-c':\n"
            "    _namespace.update({"
            "'TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256': '"
            + "f" * 64
            + "'})\n",
            "dynamic module namespace mutation",
        ),
        (
            "\nimport sys as _shadow_sys\n"
            "_namespace = _shadow_sys.modules[__name__].__dict__\n"
            "_namespace_alias = _namespace\n"
            "_namespace_alias['TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256'] = '"
            + "f" * 64
            + "'\n",
            "dynamic namespace mutation",
        ),
        (
            "\nimport sys as _shadow_sys\n"
            "_namespace: dict = _shadow_sys.modules[__name__].__dict__\n"
            "_writer = _namespace.update\n"
            "_writer_alias = _writer\n"
            "_writer_alias({'TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256': '"
            + "f" * 64
            + "'})\n",
            "aliased module namespace mutator update",
        ),
        (
            "\nimport sys as _shadow_sys\n"
            "if (_namespace := _shadow_sys.modules[__name__].__dict__):\n"
            "    _namespace.setdefault("
            "'TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256', '"
            + "f" * 64
            + "')\n",
            "dynamic module namespace mutation",
        ),
        (
            "\nimport sys as _shadow_sys\n"
            "object.__setattr__(_shadow_sys.modules[__name__], "
            "'TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256', '"
            + "f" * 64
            + "')\n",
            "dynamically mutates a frozen pin",
        ),
        (
            "\nimport sys as _shadow_sys\n"
            "_shadow_sys.modules[__name__].__setattr__("
            "'TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256', '"
            + "f" * 64
            + "')\n",
            "dynamically mutates a frozen pin",
        ),
        (
            "\nimport sys as _shadow_sys\n"
            "_shadow_sys.modules[__name__].__delattr__("
            "'TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256')\n",
            "dynamically mutates a frozen pin",
        ),
        (
            "\nimport operator as _operator\nimport sys as _shadow_sys\n"
            "_operator.setitem(_shadow_sys.modules[__name__].__dict__, "
            "'TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256', '"
            + "f" * 64
            + "')\n",
            "dynamically mutates a frozen pin",
        ),
    ),
)
def test_pin_planner_rejects_duplicate_dynamic_and_import_side_effect_assignments(
    tmp_path: Path,
    injection: str,
    message: str,
) -> None:
    state, repo, _plan = _pin_plan_fixture(tmp_path, include_online=False)
    rc06 = repo / "yuqei-infra/scripts/build-rc06-publication-batch.py"
    rc06.write_text(rc06.read_text(encoding="utf-8") + injection, encoding="utf-8")
    with pytest.raises(authority.AuthorityError, match=message):
        authority.plan_pin_updates(
            repo,
            state,
            state / "keyrings" / "recovery-epoch-1",
            epoch=1,
            unprotect=_unprotect,
        )


def test_pin_planner_isolated_runtime_detects_import_side_effect_override(
    tmp_path: Path,
) -> None:
    state, repo, _plan = _pin_plan_fixture(tmp_path, include_online=False)
    rc06 = repo / "yuqei-infra/scripts/build-rc06-publication-batch.py"
    side_effect = rc06.parent / "attacker_pin_side_effect.py"
    side_effect.write_text(
        "import pathlib\n"
        "import sys\n"
        "target = pathlib.Path(__file__).with_name('build-rc06-publication-batch.py').resolve()\n"
        "for module in tuple(sys.modules.values()):\n"
        "    module_file = getattr(module, '__file__', None)\n"
        "    if module_file and pathlib.Path(module_file).resolve() == target:\n"
        "        module.TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = '" + "f" * 64 + "'\n"
        "        break\n",
        encoding="utf-8",
    )
    rc06.write_text(
        rc06.read_text(encoding="utf-8") + "\nimport attacker_pin_side_effect\n",
        encoding="utf-8",
    )
    with pytest.raises(authority.AuthorityError, match="runtime values drift"):
        authority.plan_pin_updates(
            repo,
            state,
            state / "keyrings" / "recovery-epoch-1",
            epoch=1,
            unprotect=_unprotect,
        )


@pytest.mark.parametrize(
    ("filename", "module_name"),
    (
        ("plan-rollback-rehearsal.py", "name"),
        ("plan-first-release-snapshot-restore.py", "module_name"),
    ),
)
def test_pin_planner_rejects_modified_verified_module_exec_loader(
    tmp_path: Path,
    filename: str,
    module_name: str,
) -> None:
    state, repo, _plan = _pin_plan_fixture(tmp_path, include_online=False)
    planner = repo / "yuqei-infra/deploy" / filename
    original = planner.read_text(encoding="utf-8")
    needle = f'    module = types.ModuleType({module_name})\n'
    replacement = (
        "    if sys.argv[0] != '-c':\n"
        "        module = sys.modules[__name__]\n"
        "        raw = b\"TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = '"
        + "f" * 64
        + "'\"\n"
        "    else:\n"
        f"        module = types.ModuleType({module_name})\n"
    )
    assert needle in original
    planner.write_text(original.replace(needle, replacement, 1), encoding="utf-8")

    with pytest.raises(authority.AuthorityError, match="untrusted dynamic exec"):
        authority.plan_pin_updates(
            repo,
            state,
            state / "keyrings" / "recovery-epoch-1",
            epoch=1,
            unprotect=_unprotect,
        )


def test_pin_runtime_validation_uses_a_fresh_unpredictable_module_name(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    _state, repo, _plan = _pin_plan_fixture(tmp_path, include_online=False)
    rc06 = repo / "yuqei-infra/scripts/build-rc06-publication-batch.py"
    raw = rc06.read_bytes()
    expected = authority._validate_source_constant_contract(rc06, repo, raw)
    real_run = authority.subprocess.run
    module_names: list[str] = []

    def observe_run(command, *args, **kwargs):
        module_names.append(command[-1])
        return real_run(command, *args, **kwargs)

    monkeypatch.setattr(authority.subprocess, "run", observe_run)
    authority._validate_isolated_runtime_constants(rc06, repo, raw, expected)
    authority._validate_isolated_runtime_constants(rc06, repo, raw, expected)
    assert len(module_names) == 2
    assert module_names[0] != module_names[1]
    assert all(re.fullmatch(r"_[0-9a-f]{64}", name) for name in module_names)


def test_pin_apply_rolls_back_completed_files_when_later_replace_fails(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    _, _, plan = _pin_plan_fixture(tmp_path)
    real_replace = authority._write_bound_source_replacement
    calls = 0

    def fail_second(path: Path, repo: Path, expected: bytes, raw: bytes) -> None:
        nonlocal calls
        calls += 1
        if calls == 2:
            raise OSError("fixture replacement failure")
        real_replace(path, repo, expected, raw)

    monkeypatch.setattr(authority, "_write_bound_source_replacement", fail_second)
    with pytest.raises(authority.AuthorityError, match="transaction failed"):
        authority.apply_pin_plan(plan)
    assert calls >= 2
    assert all(change.path.read_bytes() == change.original for change in plan.changes)


def test_rc06_candidate_keyring_requires_frozen_pin_and_binds_exact_manifest(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    state, _result, keys = _initialize(tmp_path)
    reviewer_file = tmp_path / "reviewer-registration.json"
    registration = _reviewer_registration(reviewer_file)
    manifest_file = tmp_path / "manifest.json"
    manifest_raw = authority._canonical({"fixture": "validated elsewhere in this unit test"})
    manifest_file.write_bytes(manifest_raw)
    manifest_sha = authority.batch._sha256_bytes(manifest_raw)
    monkeypatch.setattr(authority.batch, "validate_manifest", lambda value: value)

    with pytest.raises(authority.AuthorityError, match="frozen RC-06 root pin"):
        authority.build_rc06_candidate_keyring(
            state,
            reviewer_file,
            manifest_file,
            manifest_sha,
            unprotect=_unprotect,
            now=datetime(2026, 9, 27, 3, 0, tzinfo=UTC),
        )

    monkeypatch.setattr(
        authority.batch,
        "TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256",
        keys[authority.ROLE_RC06_ROOT].public_sha256,
    )
    monkeypatch.setattr(authority.batch, "FROZEN_APPROVAL_KEYRING_EPOCH", 1)
    result = authority.build_rc06_candidate_keyring(
        state,
        reviewer_file,
        manifest_file,
        manifest_sha,
        unprotect=_unprotect,
        now=datetime(2026, 9, 27, 3, 0, tzinfo=UTC),
    )
    assert result["status"] == "rc06_candidate_keyring_created_not_approved"
    bundle = Path(result["directory"])
    keyring = json.loads((bundle / "rc06-approval-keyring.json").read_bytes())
    assert keyring["authorized_batch_manifest_raw_sha256"] == manifest_sha
    assert keyring["epoch"] == 1
    assert keyring["keys"][0]["key_id"] == registration["key_id"]
    assert keyring["root_key_id"] == authority.batch.TRUSTED_APPROVAL_ROOT_KEY_ID

    repeated = authority.build_rc06_candidate_keyring(
        state,
        reviewer_file,
        manifest_file,
        manifest_sha,
        unprotect=_unprotect,
        now=datetime(2026, 9, 27, 4, 0, tzinfo=UTC),
    )
    assert repeated["status"] == "rc06_candidate_keyring_already_created_and_verified_not_approved"
    with pytest.raises(authority.AuthorityError, match="SHA-256 mismatch"):
        authority.build_rc06_candidate_keyring(
            state,
            reviewer_file,
            manifest_file,
            "f" * 64,
            unprotect=_unprotect,
            now=datetime(2026, 9, 27, 4, 0, tzinfo=UTC),
        )


def test_apply_private_key_provision_command_fails_closed_without_export(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    capsys: pytest.CaptureFixture[str],
) -> None:
    monkeypatch.setattr(authority, "_assert_runtime", lambda: None)
    result = authority.main(
        [
            "provision-apply-private-key",
            "--state-directory",
            os.fspath(tmp_path / "authority-state"),
        ]
    )
    captured = capsys.readouterr()
    assert result == 2
    assert "not implemented" in captured.err
    assert '"private_material_exported":false' in captured.err
    assert list(tmp_path.iterdir()) == []


def test_apply_payload_signing_is_bound_verified_atomic_and_idempotent(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    materials = _formal_apply_materials(tmp_path, monkeypatch)
    output = tmp_path / "signed-apply-receipts"
    result = authority.sign_apply_payloads(
        materials["state"],
        manifest_file=materials["manifest_file"],
        manifest_sha256=materials["manifest_sha"],
        approval_file=materials["approval_file"],
        approval_sha256=materials["approval_sha"],
        approval_keyring_file=materials["keyring_file"],
        approval_keyring_sha256=materials["keyring_sha"],
        approval_root_public_key_file=materials["root_file"],
        online_service_public_key_file=materials["online_public_file"],
        online_transaction_receipts_file=materials["online_receipts_file"],
        online_transaction_receipts_sha256=materials["online_receipts_sha"],
        runtime_snapshot_receipt_file=materials["runtime_receipt_file"],
        runtime_snapshot_receipt_sha256=materials["runtime_receipt_sha"],
        output_directory=output,
        unprotect=_unprotect,
        now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
    )
    assert result["status"] == "apply_receipts_signed_and_verified_not_published"
    assert result["receipt_count"] == 1
    assert result["business_api_calls_by_this_tool"] == 0
    assert result["trusted_online_transaction_receipt_count"] == 1
    assert result["private_keys_in_bundle"] is False
    receipt = json.loads((output / "apply-receipt-001.json").read_bytes())
    assert authority.batch.validate_apply_receipt(
        materials["manifest"],
        materials["manifest_sha"],
        materials["approval"],
        materials["keyring"],
        materials["keys"][authority.ROLE_RC06_ROOT].public_key,
        materials["keys"][authority.ROLE_APPLY_SERVICE].public_key,
            materials["online_private"].public_key(),
            materials["approval_sha"],
            receipt,
            legal_review_binding_sha256=materials["challenge"]["payload"][
                "legal_source_review_binding_sha256"
            ],
            legal_review_expires_at=materials["challenge"]["payload"][
                "legal_source_review_expires_at"
            ],
            environment_id="isolated-uat-fixture",
            tenant_id="tenant-fixture",
        ) == receipt
    repeated = authority.sign_apply_payloads(
        materials["state"],
        manifest_file=materials["manifest_file"],
        manifest_sha256=materials["manifest_sha"],
        approval_file=materials["approval_file"],
        approval_sha256=materials["approval_sha"],
        approval_keyring_file=materials["keyring_file"],
        approval_keyring_sha256=materials["keyring_sha"],
        approval_root_public_key_file=materials["root_file"],
        online_service_public_key_file=materials["online_public_file"],
        online_transaction_receipts_file=materials["online_receipts_file"],
        online_transaction_receipts_sha256=materials["online_receipts_sha"],
        runtime_snapshot_receipt_file=materials["runtime_receipt_file"],
        runtime_snapshot_receipt_sha256=materials["runtime_receipt_sha"],
        output_directory=output,
        unprotect=_unprotect,
        now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
    )
    assert repeated["status"] == "apply_receipts_already_signed_and_verified_not_published"
    assert repeated["receipts"] == result["receipts"]

    with pytest.raises(
        authority.AuthorityError,
        match="already consumed or reserved for another output",
    ):
        authority.sign_apply_payloads(
            materials["state"],
            manifest_file=materials["manifest_file"],
            manifest_sha256=materials["manifest_sha"],
            approval_file=materials["approval_file"],
            approval_sha256=materials["approval_sha"],
            approval_keyring_file=materials["keyring_file"],
            approval_keyring_sha256=materials["keyring_sha"],
            approval_root_public_key_file=materials["root_file"],
            online_service_public_key_file=materials["online_public_file"],
            online_transaction_receipts_file=materials["online_receipts_file"],
            online_transaction_receipts_sha256=materials["online_receipts_sha"],
            runtime_snapshot_receipt_file=materials["runtime_receipt_file"],
            runtime_snapshot_receipt_sha256=materials["runtime_receipt_sha"],
            output_directory=tmp_path / "replayed-to-another-output",
            unprotect=_unprotect,
            now=datetime(2026, 9, 27, 4, 2, tzinfo=UTC),
        )


def test_apply_challenge_is_unique_per_authorization_context_and_recovers_publish_interrupt(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    materials = _formal_apply_materials(tmp_path, monkeypatch)
    common = {
        "manifest_file": materials["manifest_file"],
        "manifest_sha256": materials["manifest_sha"],
        "approval_file": materials["approval_file"],
        "approval_sha256": materials["approval_sha"],
        "approval_keyring_file": materials["keyring_file"],
        "approval_keyring_sha256": materials["keyring_sha"],
        "approval_root_public_key_file": materials["root_file"],
        "legal_review_request_file": materials["legal_request_file"],
        "legal_review_request_sha256": materials["legal_request_sha"],
        "legal_review_decision_file": materials["legal_decision_file"],
        "legal_review_decision_sha256": materials["legal_decision_sha"],
        "legal_review_keyring_file": materials["legal_keyring_file"],
        "legal_review_keyring_sha256": materials["legal_keyring_sha"],
        "environment_id": "isolated-uat-fixture",
        "tenant_id": "tenant-fixture",
        "unprotect": _unprotect,
    }
    repeated = authority.issue_apply_challenge(
        materials["state"],
        now=datetime(2026, 9, 27, 3, 56, tzinfo=UTC),
        **common,
    )
    assert repeated["status"] == "apply_challenge_already_issued_not_replaced"
    assert repeated["challenge_id"] == materials["challenge"]["challenge_id"]
    assert len(list((materials["state"] / "apply-challenge-contexts").glob("*.json"))) == 1
    assert len(list((materials["state"] / "apply-challenges").glob("*.json"))) == 1

    second_payload = dict(materials["approval"]["payload"])
    second_payload["approved_at"] = "2026-09-27T03:11:00Z"
    second_signature = materials["reviewer_private"].sign(
        authority.batch.APPROVAL_SIGNATURE_DOMAIN + authority._canonical(second_payload)
    )
    second_approval = {
        "schema_version": authority.batch.APPROVAL_SCHEMA_VERSION,
        "approval_id": (
            "rc06-batch-approval-"
            + authority.batch._sha256_bytes(
                authority._canonical(second_payload) + second_signature
            )[:24]
        ),
        "payload": second_payload,
        "signature": base64.b64encode(second_signature).decode("ascii"),
    }
    second_file = tmp_path / "second-context-approval.json"
    second_raw = authority._canonical(second_approval)
    second_file.write_bytes(second_raw)
    second_common = {
        **common,
        "approval_file": second_file,
        "approval_sha256": authority.batch._sha256_bytes(second_raw),
    }

    real_publish = authority._publish_persisted_apply_challenge
    interrupted = False

    def interrupt_after_context(*args, **kwargs):
        nonlocal interrupted
        if not interrupted:
            interrupted = True
            raise KeyboardInterrupt()
        return real_publish(*args, **kwargs)

    monkeypatch.setattr(
        authority,
        "_publish_persisted_apply_challenge",
        interrupt_after_context,
    )
    with pytest.raises(KeyboardInterrupt):
        authority.issue_apply_challenge(
            materials["state"],
            now=datetime(2026, 9, 27, 3, 57, tzinfo=UTC),
            **second_common,
        )
    assert len(list((materials["state"] / "apply-challenge-contexts").glob("*.json"))) == 2
    assert len(list((materials["state"] / "apply-challenges").glob("*.json"))) == 1

    recovered = authority.issue_apply_challenge(
        materials["state"],
        now=datetime(2026, 9, 27, 3, 58, tzinfo=UTC),
        **second_common,
    )
    recovered_again = authority.issue_apply_challenge(
        materials["state"],
        now=datetime(2026, 9, 27, 3, 59, tzinfo=UTC),
        **second_common,
    )
    assert recovered["challenge_id"] == recovered_again["challenge_id"]
    assert recovered["authorization_context_id"] == recovered_again[
        "authorization_context_id"
    ]
    assert len(list((materials["state"] / "apply-challenges").glob("*.json"))) == 2


def test_apply_payload_signing_rejects_tamper_cross_approval_and_partial_bundle(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    materials = _formal_apply_materials(tmp_path, monkeypatch)

    tampered = copy.deepcopy(materials["online_receipt"])
    tampered["payload"]["batch_manifest_payload_sha256"] = "f" * 64
    tampered_file = tmp_path / "tampered-payload.json"
    tampered_raw = authority._canonical(tampered)
    tampered_file.write_bytes(tampered_raw)
    tampered_output = tmp_path / "tampered-output"
    with pytest.raises(authority.AuthorityError, match="not bound to this batch"):
        authority.sign_apply_payloads(
            materials["state"],
            manifest_file=materials["manifest_file"],
            manifest_sha256=materials["manifest_sha"],
            approval_file=materials["approval_file"],
            approval_sha256=materials["approval_sha"],
            approval_keyring_file=materials["keyring_file"],
            approval_keyring_sha256=materials["keyring_sha"],
            approval_root_public_key_file=materials["root_file"],
            online_service_public_key_file=materials["online_public_file"],
            online_transaction_receipts_file=tampered_file,
            online_transaction_receipts_sha256=authority.batch._sha256_bytes(tampered_raw),
            runtime_snapshot_receipt_file=materials["runtime_receipt_file"],
            runtime_snapshot_receipt_sha256=materials["runtime_receipt_sha"],
            output_directory=tampered_output,
            unprotect=_unprotect,
            now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
        )
    assert not tampered_output.exists()

    challenge_tampered = copy.deepcopy(materials["online_receipt"])
    challenge_tampered["payload"]["authority_challenge"]["payload"][
        "operation"
    ] = "attacker-operation"
    challenge_tampered = _kms_envelope(
        materials["online_private"],
        schema=authority.batch.ONLINE_TRANSACTION_RECEIPT_SCHEMA_VERSION,
        prefix="rc06-online-transaction-",
        domain=authority.batch.ONLINE_TRANSACTION_SIGNATURE_DOMAIN,
        payload=challenge_tampered["payload"],
        request_id="fixture-kms-challenge-tamper",
    )
    challenge_tampered_file = tmp_path / "tampered-challenge-receipt.json"
    challenge_tampered_raw = authority._canonical(challenge_tampered)
    challenge_tampered_file.write_bytes(challenge_tampered_raw)
    with pytest.raises(
        authority.AuthorityError,
        match="challenge.*(binding drift|unique authorization-context challenge)",
    ):
        authority.sign_apply_payloads(
            materials["state"],
            manifest_file=materials["manifest_file"],
            manifest_sha256=materials["manifest_sha"],
            approval_file=materials["approval_file"],
            approval_sha256=materials["approval_sha"],
            approval_keyring_file=materials["keyring_file"],
            approval_keyring_sha256=materials["keyring_sha"],
            approval_root_public_key_file=materials["root_file"],
            online_service_public_key_file=materials["online_public_file"],
            online_transaction_receipts_file=challenge_tampered_file,
            online_transaction_receipts_sha256=authority.batch._sha256_bytes(
                challenge_tampered_raw
            ),
            runtime_snapshot_receipt_file=materials["runtime_receipt_file"],
            runtime_snapshot_receipt_sha256=materials["runtime_receipt_sha"],
            output_directory=tmp_path / "tampered-challenge-output",
            unprotect=_unprotect,
            now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
        )

    second_payload = dict(materials["approval"]["payload"])
    second_payload["approved_at"] = "2026-09-27T03:15:00Z"
    second_signature = materials["reviewer_private"].sign(
        authority.batch.APPROVAL_SIGNATURE_DOMAIN + authority._canonical(second_payload)
    )
    second_approval = {
        "schema_version": authority.batch.APPROVAL_SCHEMA_VERSION,
        "approval_id": (
            "rc06-batch-approval-"
            + authority.batch._sha256_bytes(
                authority._canonical(second_payload) + second_signature
            )[:24]
        ),
        "payload": second_payload,
        "signature": base64.b64encode(second_signature).decode("ascii"),
    }
    second_approval_file = tmp_path / "second-approval.json"
    second_approval_raw = authority._canonical(second_approval)
    second_approval_file.write_bytes(second_approval_raw)
    with pytest.raises(
        authority.AuthorityError,
        match="not bound to the approval receipt|cannot open apply challenge context",
    ):
        authority.sign_apply_payloads(
            materials["state"],
            manifest_file=materials["manifest_file"],
            manifest_sha256=materials["manifest_sha"],
            approval_file=second_approval_file,
            approval_sha256=authority.batch._sha256_bytes(second_approval_raw),
            approval_keyring_file=materials["keyring_file"],
            approval_keyring_sha256=materials["keyring_sha"],
            approval_root_public_key_file=materials["root_file"],
            online_service_public_key_file=materials["online_public_file"],
            online_transaction_receipts_file=materials["online_receipts_file"],
            online_transaction_receipts_sha256=materials["online_receipts_sha"],
            runtime_snapshot_receipt_file=materials["runtime_receipt_file"],
            runtime_snapshot_receipt_sha256=materials["runtime_receipt_sha"],
            output_directory=tmp_path / "cross-approval-output",
            unprotect=_unprotect,
            now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
        )

    partial = copy.deepcopy(materials["online_receipt"])
    partial["payload"]["item_results"] = partial["payload"]["item_results"][:-1]
    payload_list = [materials["online_receipt"], partial]
    payload_list_file = tmp_path / "payload-list.json"
    payload_list_raw = authority._canonical(payload_list)
    payload_list_file.write_bytes(payload_list_raw)
    partial_output = tmp_path / "partial-output"
    with pytest.raises(authority.AuthorityError, match="exactly one signed receipt"):
        authority.sign_apply_payloads(
            materials["state"],
            manifest_file=materials["manifest_file"],
            manifest_sha256=materials["manifest_sha"],
            approval_file=materials["approval_file"],
            approval_sha256=materials["approval_sha"],
            approval_keyring_file=materials["keyring_file"],
            approval_keyring_sha256=materials["keyring_sha"],
            approval_root_public_key_file=materials["root_file"],
            online_service_public_key_file=materials["online_public_file"],
            online_transaction_receipts_file=payload_list_file,
            online_transaction_receipts_sha256=authority.batch._sha256_bytes(payload_list_raw),
            runtime_snapshot_receipt_file=materials["runtime_receipt_file"],
            runtime_snapshot_receipt_sha256=materials["runtime_receipt_sha"],
            output_directory=partial_output,
            unprotect=_unprotect,
            now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
        )
    assert not partial_output.exists()


def test_apply_challenge_consumption_recovers_after_finalize_interrupt(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    materials = _formal_apply_materials(tmp_path, monkeypatch)
    output = tmp_path / "interrupted-consumption-output"
    real_finalize = authority._finalize_apply_consumption
    interrupted = False

    def interrupt_finalize(*args, **kwargs):
        nonlocal interrupted
        if not interrupted:
            interrupted = True
            raise KeyboardInterrupt()
        return real_finalize(*args, **kwargs)

    monkeypatch.setattr(authority, "_finalize_apply_consumption", interrupt_finalize)
    with pytest.raises(KeyboardInterrupt):
        authority.sign_apply_payloads(
            materials["state"],
            manifest_file=materials["manifest_file"],
            manifest_sha256=materials["manifest_sha"],
            approval_file=materials["approval_file"],
            approval_sha256=materials["approval_sha"],
            approval_keyring_file=materials["keyring_file"],
            approval_keyring_sha256=materials["keyring_sha"],
            approval_root_public_key_file=materials["root_file"],
            online_service_public_key_file=materials["online_public_file"],
            online_transaction_receipts_file=materials["online_receipts_file"],
            online_transaction_receipts_sha256=materials["online_receipts_sha"],
            runtime_snapshot_receipt_file=materials["runtime_receipt_file"],
            runtime_snapshot_receipt_sha256=materials["runtime_receipt_sha"],
            output_directory=output,
            unprotect=_unprotect,
            now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
        )
    assert output.is_dir()
    consumption_path = (
        materials["state"]
        / "apply-challenge-consumption"
        / f"{materials['challenge']['challenge_id']}.json"
    )
    assert json.loads(consumption_path.read_bytes())["status"] == "reserved"

    monkeypatch.setattr(authority, "_finalize_apply_consumption", real_finalize)
    calls: list[str] = []

    def observed_unprotect(raw: bytes, role: str) -> bytes:
        calls.append(role)
        return _unprotect(raw, role)

    recovered = authority.sign_apply_payloads(
        materials["state"],
        manifest_file=materials["manifest_file"],
        manifest_sha256=materials["manifest_sha"],
        approval_file=materials["approval_file"],
        approval_sha256=materials["approval_sha"],
        approval_keyring_file=materials["keyring_file"],
        approval_keyring_sha256=materials["keyring_sha"],
        approval_root_public_key_file=materials["root_file"],
        online_service_public_key_file=materials["online_public_file"],
        online_transaction_receipts_file=materials["online_receipts_file"],
        online_transaction_receipts_sha256=materials["online_receipts_sha"],
        runtime_snapshot_receipt_file=materials["runtime_receipt_file"],
        runtime_snapshot_receipt_sha256=materials["runtime_receipt_sha"],
        output_directory=output,
        unprotect=observed_unprotect,
        now=datetime(2026, 9, 27, 4, 2, tzinfo=UTC),
    )
    assert recovered["status"] == "apply_receipts_already_signed_and_verified_not_published"
    assert json.loads(consumption_path.read_bytes())["status"] == "consumed"
    assert calls == []


def test_authority_binds_exact_reviewer_registration_and_derived_key_id(
    tmp_path: Path,
) -> None:
    state, _result, _keys = _initialize(tmp_path)
    bound_file = tmp_path / "reviewer-registration.json"
    changed = json.loads(bound_file.read_bytes())
    changed["created_at"] = "2026-09-27T00:00:01Z"
    changed_file = tmp_path / "changed-reviewer-registration.json"
    changed_file.write_bytes(authority._canonical(changed))

    with pytest.raises(authority.AuthorityError, match="initialization binding"):
        authority.initialize_authority(
            state,
            changed_file,
            protect=_protect,
            unprotect=_unprotect,
        )
    with pytest.raises(authority.AuthorityError, match="initialization binding"):
        authority.build_recovery_keyrings(
            state,
            changed_file,
            epoch=1,
            unprotect=_unprotect,
            now=datetime(2026, 9, 27, 2, 0, tzinfo=UTC),
        )
    assert not (state / "keyrings").exists()

    invalid_id = json.loads(bound_file.read_bytes())
    invalid_id["key_id"] = "chen-yuqi-rc06-attacker"
    invalid_file = tmp_path / "invalid-key-id.json"
    invalid_file.write_bytes(authority._canonical(invalid_id))
    with pytest.raises(authority.AuthorityError, match="public-key derived"):
        authority.load_reviewer_registration(invalid_file)

    noncanonical = tmp_path / "noncanonical-registration.json"
    noncanonical.write_text(json.dumps(json.loads(bound_file.read_bytes()), indent=2), encoding="utf-8")
    with pytest.raises(authority.AuthorityError, match="canonical JSON"):
        authority.load_reviewer_registration(noncanonical)


def test_public_status_and_apply_validation_use_least_private_key_unsealing(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    materials = _formal_apply_materials(tmp_path, monkeypatch)
    calls: list[str] = []

    def observed_unprotect(raw: bytes, role: str) -> bytes:
        calls.append(role)
        return _unprotect(raw, role)

    public_keys = authority.load_authority(
        materials["state"],
        unprotect=observed_unprotect,
        unseal_roles=(),
    )
    assert calls == []
    assert all(item.private_key is None for item in public_keys.values())

    output = tmp_path / "least-privilege-apply"
    authority.sign_apply_payloads(
        materials["state"],
        manifest_file=materials["manifest_file"],
        manifest_sha256=materials["manifest_sha"],
        approval_file=materials["approval_file"],
        approval_sha256=materials["approval_sha"],
        approval_keyring_file=materials["keyring_file"],
        approval_keyring_sha256=materials["keyring_sha"],
        approval_root_public_key_file=materials["root_file"],
        online_service_public_key_file=materials["online_public_file"],
        online_transaction_receipts_file=materials["online_receipts_file"],
        online_transaction_receipts_sha256=materials["online_receipts_sha"],
        runtime_snapshot_receipt_file=materials["runtime_receipt_file"],
        runtime_snapshot_receipt_sha256=materials["runtime_receipt_sha"],
        output_directory=output,
        unprotect=observed_unprotect,
        now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
    )
    assert calls == [authority.ROLE_APPLY_SERVICE]

    calls.clear()
    repeated = authority.sign_apply_payloads(
        materials["state"],
        manifest_file=materials["manifest_file"],
        manifest_sha256=materials["manifest_sha"],
        approval_file=materials["approval_file"],
        approval_sha256=materials["approval_sha"],
        approval_keyring_file=materials["keyring_file"],
        approval_keyring_sha256=materials["keyring_sha"],
        approval_root_public_key_file=materials["root_file"],
        online_service_public_key_file=materials["online_public_file"],
        online_transaction_receipts_file=materials["online_receipts_file"],
        online_transaction_receipts_sha256=materials["online_receipts_sha"],
        runtime_snapshot_receipt_file=materials["runtime_receipt_file"],
        runtime_snapshot_receipt_sha256=materials["runtime_receipt_sha"],
        output_directory=output,
        unprotect=observed_unprotect,
        now=datetime(2026, 9, 27, 6, 0, tzinfo=UTC),
    )
    assert repeated["status"] == "apply_receipts_already_signed_and_verified_not_published"
    assert calls == []


def test_apply_signing_rejects_unprovisioned_custom_environment_and_expired_window_before_unseal(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    materials = _formal_apply_materials(tmp_path, monkeypatch)
    calls: list[str] = []

    def observed_unprotect(raw: bytes, role: str) -> bytes:
        calls.append(role)
        return _unprotect(raw, role)

    common = {
        "manifest_file": materials["manifest_file"],
        "manifest_sha256": materials["manifest_sha"],
        "approval_file": materials["approval_file"],
        "approval_sha256": materials["approval_sha"],
        "approval_keyring_file": materials["keyring_file"],
        "approval_keyring_sha256": materials["keyring_sha"],
        "approval_root_public_key_file": materials["root_file"],
        "online_service_public_key_file": materials["online_public_file"],
        "online_transaction_receipts_file": materials["online_receipts_file"],
        "online_transaction_receipts_sha256": materials["online_receipts_sha"],
        "runtime_snapshot_receipt_file": materials["runtime_receipt_file"],
        "runtime_snapshot_receipt_sha256": materials["runtime_receipt_sha"],
        "unprotect": observed_unprotect,
    }

    monkeypatch.setattr(
        authority.batch, "TRUSTED_ONLINE_APPLY_SERVICE_PUBLIC_KEY_SPKI_SHA256", ""
    )
    with pytest.raises(authority.AuthorityError, match="trust is not provisioned"):
        authority.sign_apply_payloads(
            materials["state"],
            output_directory=tmp_path / "unprovisioned-output",
            now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
            **common,
        )
    assert calls == []
    monkeypatch.setattr(
        authority.batch,
        "TRUSTED_ONLINE_APPLY_SERVICE_PUBLIC_KEY_SPKI_SHA256",
        authority.batch._sha256_bytes(
            materials["online_private"].public_key().public_bytes(
                serialization.Encoding.DER,
                serialization.PublicFormat.SubjectPublicKeyInfo,
            )
        ),
    )

    custom = copy.deepcopy(materials["online_receipt"])
    custom["payload"]["environment_id"] = "attacker-environment"
    custom = _kms_envelope(
        materials["online_private"],
        schema=authority.batch.ONLINE_TRANSACTION_RECEIPT_SCHEMA_VERSION,
        prefix="rc06-online-transaction-",
        domain=authority.batch.ONLINE_TRANSACTION_SIGNATURE_DOMAIN,
        payload=custom["payload"],
        request_id="fixture-kms-custom-environment",
    )
    custom_file = tmp_path / "custom-environment-receipt.json"
    custom_raw = authority._canonical(custom)
    custom_file.write_bytes(custom_raw)
    with pytest.raises(
        authority.AuthorityError,
        match="frozen allowlist|environment or tenant binding drift",
    ):
        authority.sign_apply_payloads(
            materials["state"],
            output_directory=tmp_path / "custom-output",
            now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
            **{
                **common,
                "online_transaction_receipts_file": custom_file,
                "online_transaction_receipts_sha256": authority.batch._sha256_bytes(
                    custom_raw
                ),
            },
        )
    assert calls == []

    future = copy.deepcopy(materials["online_receipt"])
    future["payload"]["committed_at"] = "2026-09-27T04:45:00Z"
    future = _kms_envelope(
        materials["online_private"],
        schema=authority.batch.ONLINE_TRANSACTION_RECEIPT_SCHEMA_VERSION,
        prefix="rc06-online-transaction-",
        domain=authority.batch.ONLINE_TRANSACTION_SIGNATURE_DOMAIN,
        payload=future["payload"],
        request_id="fixture-kms-future-commit",
    )
    future_file = tmp_path / "future-online-receipt.json"
    future_raw = authority._canonical(future)
    future_file.write_bytes(future_raw)
    with pytest.raises(
        authority.AuthorityError,
        match="does not satisfy its apply challenge|after the current signing time",
    ):
        authority.sign_apply_payloads(
            materials["state"],
            output_directory=tmp_path / "future-output",
            now=datetime(2026, 9, 27, 4, 1, tzinfo=UTC),
            **{
                **common,
                "online_transaction_receipts_file": future_file,
                "online_transaction_receipts_sha256": authority.batch._sha256_bytes(
                    future_raw
                ),
            },
        )
    assert calls == []

    delayed = authority.sign_apply_payloads(
        materials["state"],
        output_directory=tmp_path / "post-challenge-countersign-output",
        now=datetime(2026, 9, 27, 4, 6, tzinfo=UTC),
        **common,
    )
    assert delayed["status"] == "apply_receipts_signed_and_verified_not_published"
    assert calls == [authority.ROLE_APPLY_SERVICE]


def _pin_plan_fixture(tmp_path: Path, *, include_online: bool = True):
    state, _result, _keys = _initialize(tmp_path)
    reviewer_file = tmp_path / "reviewer-registration.json"
    recovery = authority.build_recovery_keyrings(
        state,
        reviewer_file,
        epoch=1,
        unprotect=_unprotect,
        now=datetime(2026, 9, 27, 2, 0, tzinfo=UTC),
    )
    repo = _copy_pin_sources(tmp_path)
    legal_keyring_file = tmp_path / "scene-legal-keyring.json"
    _legal_keyring(legal_keyring_file)
    online_kwargs = {}
    if include_online:
        online_registration_file = tmp_path / "online-service-registration.json"
        _online_service_registration(online_registration_file)
        online_kwargs = {
            "online_service_registration_file": online_registration_file,
            "online_environment_ids": ("isolated-uat-fixture",),
        }
    plan = authority.plan_pin_updates(
        repo,
        state,
        Path(recovery["directory"]),
        epoch=1,
        unprotect=_unprotect,
        legal_review_keyring_file=legal_keyring_file,
        **online_kwargs,
    )
    return state, repo, plan


def test_pin_transaction_handles_keyboard_interrupt_and_concurrent_writer(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    state, repo, plan = _pin_plan_fixture(tmp_path)
    real_replace = authority._write_bound_source_replacement
    interrupted = False

    def interrupt_first_source(path: Path, repo: Path, expected: bytes, raw: bytes) -> None:
        nonlocal interrupted
        if not interrupted and any(path == change.path for change in plan.changes):
            interrupted = True
            raise KeyboardInterrupt()
        real_replace(path, repo, expected, raw)

    monkeypatch.setattr(authority, "_write_bound_source_replacement", interrupt_first_source)
    with pytest.raises(KeyboardInterrupt):
        authority.apply_pin_plan(plan)
    assert all(change.path.read_bytes() == change.original for change in plan.changes)
    monkeypatch.setattr(authority, "_write_bound_source_replacement", real_replace)

    with authority._repository_pin_lock(repo, state):
        with pytest.raises(authority.AuthorityError, match="repository lock"):
            authority.apply_pin_plan(plan)


def test_pin_apply_rechecks_ancestor_reparse_state_immediately_before_write(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    state, _repo, plan = _pin_plan_fixture(tmp_path)
    first = next(change for change in plan.changes if change.original != change.proposed)
    real_assert = authority.batch._assert_no_symlink_components
    source_checks = 0

    def detect_reparse(path: Path, label: str) -> None:
        nonlocal source_checks
        if path == first.path:
            source_checks += 1
            if source_checks == 2:
                raise authority.batch.PublicationBatchError(
                    "fixture ancestor became a reparse point"
                )
        real_assert(path, label)

    monkeypatch.setattr(
        authority.batch,
        "_assert_no_symlink_components",
        detect_reparse,
    )
    with pytest.raises(authority.AuthorityError, match="reparse point"):
        authority.apply_pin_plan(plan)
    assert source_checks == 2
    assert all(change.path.read_bytes() == change.original for change in plan.changes)


def test_bound_source_publish_rejects_parent_path_swap_without_touching_outside_file(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    state, _repo, plan = _pin_plan_fixture(tmp_path)
    target = next(change.path for change in plan.changes if change.original != change.proposed)
    outside = tmp_path / "outside-protected.py"
    outside.write_bytes(b"outside-content-must-not-change\n")
    os.chmod(outside, 0o444)
    outside_before = outside.read_bytes()
    outside_mode = stat.S_IMODE(outside.stat().st_mode)
    outside_attributes = getattr(outside.stat(), "st_file_attributes", None)
    outside_acl = _acl_snapshot(outside)
    real_assert_bound = authority._assert_descriptor_still_bound
    swapped = False
    swap_blocked_by_open_handle = False

    def swap_parent_after_open(descriptor, path, repo, label, *, original_resolved):
        nonlocal swapped, swap_blocked_by_open_handle
        if path == target and not swapped:
            swapped = True
            original_parent = path.parent.with_name(path.parent.name + "-opened-original")
            try:
                path.parent.rename(original_parent)
                path.parent.mkdir()
                os.link(outside, path)
            except PermissionError as exc:
                swap_blocked_by_open_handle = True
                raise authority.AuthorityError(
                    "source path or parent identity changed; operating system blocked the swap"
                ) from exc
        return real_assert_bound(
            descriptor,
            path,
            repo,
            label,
            original_resolved=original_resolved,
        )

    monkeypatch.setattr(
        authority,
        "_assert_descriptor_still_bound",
        swap_parent_after_open,
    )
    with pytest.raises(authority.AuthorityError, match="path or parent identity changed"):
        authority.apply_pin_plan(plan)
    assert swapped is True
    assert swap_blocked_by_open_handle is True or not target.parent.samefile(
        target.parent.with_name(target.parent.name + "-opened-original")
    )
    assert outside.read_bytes() == outside_before
    assert stat.S_IMODE(outside.stat().st_mode) == outside_mode
    assert getattr(outside.stat(), "st_file_attributes", None) == outside_attributes
    assert _acl_snapshot(outside) == outside_acl
    control = authority._pin_control_directory(plan.repo, state)
    assert not (control / "transaction.json").exists()


def test_pin_transaction_persistent_journal_blocks_and_recovers_after_crash(
    tmp_path: Path,
) -> None:
    state, repo, plan = _pin_plan_fixture(tmp_path)
    first = next(change for change in plan.changes if change.original != change.proposed)
    with authority._repository_pin_lock(repo, state) as control:
        journal = authority._journal_payload(plan)
        journal_path = control / "transaction.json"
        authority._create_pin_journal(journal_path, journal)
        authority._write_bound_source_replacement(
            first.path,
            repo,
            first.original,
            first.proposed,
        )
        journal["status"] = "applying"
        journal["completed_paths"].append(first.path.relative_to(repo).as_posix())
        authority._write_pin_journal(journal_path, journal)

    with pytest.raises(authority.AuthorityError, match="unfinished.*journal"):
        authority.apply_pin_plan(plan)
    recovered = authority.recover_pin_transaction(repo, state)
    assert recovered["status"] == "source_pin_transaction_recovered_to_originals"
    assert first.path.read_bytes() == first.original
    assert Path(recovered["audit_journal"]).is_file()
    assert authority.recover_pin_transaction(repo, state)["status"] == (
        "no_source_pin_recovery_required"
    )


def test_pin_initial_journal_publish_interrupt_is_inside_recovery_boundary(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    state, repo, plan = _pin_plan_fixture(tmp_path)
    real_create = authority._create_pin_journal
    interrupted = False

    def publish_then_interrupt(path: Path, payload: dict) -> None:
        nonlocal interrupted
        real_create(path, payload)
        if not interrupted:
            interrupted = True
            raise KeyboardInterrupt()

    monkeypatch.setattr(authority, "_create_pin_journal", publish_then_interrupt)
    with pytest.raises(KeyboardInterrupt):
        authority.apply_pin_plan(plan)
    assert all(change.path.read_bytes() == change.original for change in plan.changes)
    control = authority._pin_control_directory(repo, state)
    assert not (control / "transaction.json").exists()
    assert list((control / "history").glob("*.rolled_back.json"))


def test_partial_pin_journal_is_quarantined_and_blocks_future_apply(
    tmp_path: Path,
) -> None:
    state, repo, plan = _pin_plan_fixture(tmp_path)
    with authority._repository_pin_lock(repo, state) as control:
        journal_path = control / "transaction.json"
        authority._publish_private_no_replace(
            journal_path,
            b'{"partial":',
            "partial test journal",
        )
    before = {change.path: change.path.read_bytes() for change in plan.changes}
    result = authority.recover_pin_transaction(repo, state)
    assert result["status"] == (
        "source_pin_transaction_quarantined_manual_recovery_required"
    )
    assert Path(result["quarantine_marker"]).is_file()
    assert Path(result["quarantine_file"]).read_bytes() == b'{"partial":'
    assert all(path.read_bytes() == raw for path, raw in before.items())
    with pytest.raises(authority.AuthorityError, match="quarantine requires manual recovery"):
        authority.apply_pin_plan(plan)


def test_pin_commit_point_survives_archive_failure_without_rollback(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    state, repo, plan = _pin_plan_fixture(tmp_path)
    real_archive = authority._archive_pin_journal

    def fail_committed_archive(path: Path, journal: dict, terminal_status: str) -> Path:
        if terminal_status == "committed":
            raise OSError("fixture archive failure")
        return real_archive(path, journal, terminal_status)

    monkeypatch.setattr(authority, "_archive_pin_journal", fail_committed_archive)
    with pytest.raises(authority.AuthorityError, match="durably committed"):
        authority.apply_pin_plan(plan)

    assert all(change.path.read_bytes() == change.proposed for change in plan.changes)
    control = authority._pin_control_directory(repo, state)
    journal_path = control / "transaction.json"
    journal = json.loads(journal_path.read_bytes())
    assert journal["status"] == "committed"

    monkeypatch.setattr(authority, "_archive_pin_journal", real_archive)
    recovered = authority.recover_pin_transaction(repo, state)
    assert recovered["status"] == "source_pin_committed_transaction_archived"
    assert recovered["restored_files"] == []
    assert all(change.path.read_bytes() == change.proposed for change in plan.changes)
    history = control / "history"
    assert list(history.glob("*.committed.json"))
    assert not list(history.glob("*.rolled_back.json"))


def test_pin_recovery_validates_entire_journal_before_restoring_any_file(
    tmp_path: Path,
) -> None:
    state, repo, plan = _pin_plan_fixture(tmp_path)
    changed = [change for change in plan.changes if change.original != change.proposed]
    assert len(changed) >= 2

    with authority._repository_pin_lock(repo, state) as control:
        journal = authority._journal_payload(plan)
        for change in changed[:2]:
            authority._write_bound_source_replacement(
                change.path,
                repo,
                change.original,
                change.proposed,
            )
            journal["completed_paths"].append(change.path.relative_to(repo).as_posix())
        journal["status"] = "applying"
        journal["changes"][-1]["original_base64"] = base64.b64encode(
            b"invalid-journal-original"
        ).decode("ascii")
        authority._create_pin_journal(control / "transaction.json", journal)

    result = authority.recover_pin_transaction(repo, state)
    assert result["status"] == (
        "source_pin_transaction_quarantined_manual_recovery_required"
    )
    assert all(change.path.read_bytes() == change.proposed for change in changed[:2])


def test_pin_plan_can_freeze_online_service_key_and_exact_environment_allowlist(
    tmp_path: Path,
) -> None:
    state, repo, base_plan = _pin_plan_fixture(tmp_path, include_online=False)
    recovery_directory = state / "keyrings" / "recovery-epoch-1"
    registration_file = tmp_path / "online-service-registration.json"
    registration = _online_service_registration(registration_file)
    plan = authority.plan_pin_updates(
        repo,
        state,
        recovery_directory,
        epoch=1,
        unprotect=_unprotect,
        online_service_registration_file=registration_file,
        online_environment_ids=("uat-cn-north-1", "uat-cn-south-1"),
        legal_review_keyring_file=tmp_path / "scene-legal-keyring.json",
    )
    assert plan.values["online_apply_adapter_pinned"] is True
    assert plan.values["online_apply_environment_ids"] == (
        "uat-cn-north-1",
        "uat-cn-south-1",
    )
    rc06_change = next(
        change
        for change in plan.changes
        if change.path.name == "build-rc06-publication-batch.py"
    )
    proposed = rc06_change.proposed.decode("utf-8")
    assert (
        f'TRUSTED_ONLINE_APPLY_SERVICE_PUBLIC_KEY_SPKI_SHA256 = "{registration["public_key_spki_sha256"]}"'
        in proposed
    )
    assert (
        'TRUSTED_ONLINE_APPLY_ENVIRONMENT_IDS: tuple[str, ...] = '
        '("uat-cn-north-1", "uat-cn-south-1")'
    ) in proposed
    assert base_plan.values["online_apply_adapter_pinned"] is False

    with pytest.raises(authority.AuthorityError, match="supplied together"):
        authority.plan_pin_updates(
            repo,
            state,
            recovery_directory,
            epoch=1,
            unprotect=_unprotect,
            online_service_registration_file=registration_file,
            online_environment_ids=(),
        )


def test_pin_apply_fails_closed_without_online_adapter_and_rejects_preseeded_trust(
    tmp_path: Path,
) -> None:
    state, repo, plan = _pin_plan_fixture(tmp_path, include_online=False)
    assert authority.pin_plan_summary(plan)["status"] == (
        "preview_blocked_online_apply_trust_not_provisioned"
    )
    with pytest.raises(authority.AuthorityError, match="trust is not provisioned"):
        authority.apply_pin_plan(plan)

    rc06 = repo / "yuqei-infra/scripts/build-rc06-publication-batch.py"
    source = rc06.read_text(encoding="utf-8")
    rc06.write_text(
        source.replace(
            'TRUSTED_ONLINE_APPLY_SERVICE_PUBLIC_KEY_SPKI_SHA256 = ""',
            'TRUSTED_ONLINE_APPLY_SERVICE_PUBLIC_KEY_SPKI_SHA256 = "' + "f" * 64 + '"',
        ),
        encoding="utf-8",
    )
    recovery_directory = state / "keyrings" / "recovery-epoch-1"
    with pytest.raises(authority.AuthorityError, match="unexpected existing value"):
        authority.plan_pin_updates(
            repo,
            state,
            recovery_directory,
            epoch=1,
            unprotect=_unprotect,
        )


def test_pin_planner_rejects_hardlinked_source_and_recovery_refuses_unknown_bytes(
    tmp_path: Path,
) -> None:
    state, repo, plan = _pin_plan_fixture(tmp_path)
    first = next(change for change in plan.changes if change.original != change.proposed)
    alias = first.path.with_name(first.path.name + ".hardlink")
    try:
        os.link(first.path, alias)
    except OSError:
        pytest.skip("hard links are unavailable on this platform")
    with pytest.raises(
        authority.AuthorityError,
        match="single-link regular|path or parent identity changed",
    ):
        authority._read_source_file(first.path, repo, first.path.name)
    alias.unlink()

    with authority._repository_pin_lock(repo, state) as control:
        journal = authority._journal_payload(plan)
        journal_path = control / "transaction.json"
        authority._create_pin_journal(journal_path, journal)
        first.path.write_bytes(b"third-party-drift")
    with pytest.raises(authority.AuthorityError, match="unrecognized content"):
        authority.recover_pin_transaction(repo, state)
    control = authority._pin_control_directory(repo, state)
    assert (control / "transaction.json").is_file()


def test_authority_launcher_uses_authority_script_and_forwards_non_secret_arguments() -> None:
    text = AUTHORITY_LAUNCHER.read_text(encoding="utf-8-sig")
    assert '"rc06-release-trust-bootstrap.py"' in text
    assert '"issue-apply-challenge"' in text
    assert '"recover-pin-source"' in text
    assert "ValueFromRemainingArguments = $true" in text
    assert "$authorityScript $Command @CommandArguments" in text
    assert "rc06-single-reviewer-signing.py" not in text
    assert "password" not in text.casefold()
    assert "passphrase" not in text.casefold()


@pytest.mark.skipif(os.name != "nt", reason="Windows CurrentUser DPAPI is platform-specific")
def test_windows_current_user_dpapi_round_trip_has_role_entropy() -> None:
    raw = b"not-a-real-private-key-fixture"
    protected = authority._protect_current_user(raw, authority.ROLE_RC06_ROOT)
    assert protected != raw
    assert authority._unprotect_current_user(protected, authority.ROLE_RC06_ROOT) == raw
    with pytest.raises(authority.AuthorityError):
        authority._unprotect_current_user(protected, authority.ROLE_APPLY_SERVICE)