teleo-infrastructure/tests/test_gcp_leoclean_nosend_package.py

"""Offline contracts for the isolated GCP leoclean no-send service image."""

from __future__ import annotations

import copy
import importlib.util
import json
import os
import shutil
import subprocess
import sys
from pathlib import Path

import pytest

from ops import gcp_leoclean_nosend_package as package

ROOT = Path(__file__).resolve().parents[1]
REVISION = package._git_commit(ROOT)
IMAGE_DIGEST = "2" * 64
CLAIM_CEILING = "Offline package proof only; live GCP identity, database access, restart, and parity remain unproven."


def write_json(path: Path, value: object, mode: int = 0o600) -> None:
    path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    path.chmod(mode)


def git_source_fixture(tmp_path: Path, files: dict[str, str]) -> tuple[Path, str, dict[str, str]]:
    root = tmp_path / "source-repository"
    root.mkdir()
    subprocess.run(["git", "init", "--quiet", str(root)], check=True)
    for relative, content in files.items():
        path = root / relative
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(content, encoding="utf-8")
    subprocess.run(["git", "-C", str(root), "add", "--", *sorted(files)], check=True)
    subprocess.run(
        [
            "git",
            "-C",
            str(root),
            "-c",
            "user.name=fwazb",
            "-c",
            "user.email=fawaz.butt95@gmail.com",
            "commit",
            "--quiet",
            "-m",
            "Create source fixture",
        ],
        check=True,
    )
    revision = package._git_commit(root)
    expected = {relative: package.sha256_file(root / relative) for relative in files}
    return root, revision, expected


def load_entrypoint_module(tmp_path: Path):
    image_root = tmp_path / "entrypoint-image"
    image_root.mkdir()
    entrypoint_path = image_root / "entrypoint.py"
    shutil.copy2(ROOT / "docker" / "leoclean-nosend" / "entrypoint.py", entrypoint_path)
    shutil.copy2(ROOT / "ops" / "gcp_leoclean_nosend_package.py", image_root / "package_contract.py")
    scripts = image_root / "scripts"
    scripts.mkdir()
    for relative in package.IDENTITY_CONTRACT_SOURCES:
        shutil.copy2(ROOT / relative, image_root / relative)
    spec = importlib.util.spec_from_file_location(f"leoclean_entrypoint_{tmp_path.name}", entrypoint_path)
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def artifact_fixture(tmp_path: Path) -> tuple[Path, Path]:
    artifact = tmp_path / "artifact"
    artifact.mkdir()
    (artifact / "bootstrap.py").write_text("#!/usr/bin/env python3\n", encoding="utf-8")
    (artifact / "bootstrap.py").chmod(0o755)
    hermes = artifact / "hermes-agent"
    hermes.mkdir()
    (hermes / "uv.lock").write_text(
        'version = 1\n\n[[package]]\nname = "pyyaml"\nversion = "6.0.3"\n',
        encoding="utf-8",
    )
    uv_lock_sha256 = package.sha256_file(hermes / "uv.lock")
    write_json(
        artifact / "runtime-lock.json",
        {
            "base_images": {
                "python": package.PYTHON_IMAGE.removeprefix("docker.io/library/"),
                "uv": package.UV_IMAGE,
            },
            "hermes": {
                "commit": package.HERMES_COMMIT,
                "uv_lock_sha256": uv_lock_sha256,
                "patched_source_tree_sha256": "4" * 64,
            },
        },
        mode=0o644,
    )
    profile = artifact / "profile-template"
    profile.mkdir()
    (profile / "config.yaml").write_text(
        """model:
  provider: openrouter
  default: anthropic/claude-sonnet-4-6
  max_tokens: 8192
smart_model_routing: false
agent:
  max_turns: 90
memory:
  enabled: false
  memory_enabled: false
  user_profile_enabled: false
  nudge_interval: 0
skills:
  creation_nudge_interval: 0
platform_toolsets:
  api_server: [livingip-leoclean-nosend]
known_plugin_toolsets:
  api_server: []
mcp_servers: {}
display:
  tool_progress: false
""",
        encoding="utf-8",
    )
    (profile / "config.yaml").chmod(0o600)
    skills = profile / "skills" / "teleo-kb-bridge"
    skills.mkdir(parents=True)
    (skills / "SKILL.md").write_text("# fixture skill\n", encoding="utf-8")
    write_json(
        artifact / "runtime-contract.json",
        {
            "schema": "livingip.leocleanNoSendRuntimeContract.v1",
            "transport_authority": "absent",
            "toolset": {"allowed_tools": ["skills_list", "skill_view", "terminal"]},
            "database": {
                "direct_canonical_writes": "denied",
                "direct_stage_table_writes": "denied",
                "proposal_staging": "function_only",
                "proposal_function": package.PROPOSAL_FUNCTION,
            },
        },
        mode=0o644,
    )
    content = package.tree_manifest(artifact)
    manifest = {
        "schema": package.ARTIFACT_MANIFEST_SCHEMA,
        "teleo_git_head": REVISION,
        "hermes_commit": package.HERMES_COMMIT,
        "claim_ceiling": CLAIM_CEILING,
        "content": content,
    }
    write_json(artifact / "artifact-manifest.json", manifest, mode=0o644)
    receipt = tmp_path / "receipt.json"
    write_json(
        receipt,
        {
            "schema": package.ARTIFACT_RECEIPT_SCHEMA,
            "status": "pass",
            "verification_mode": "release",
            "teleo_git_head": REVISION,
            "artifact_sha256": content["sha256"],
            "hermes_commit": package.HERMES_COMMIT,
            "hermes_source_sha256": "4" * 64,
            "config_sha256": package.sha256_file(profile / "config.yaml"),
            "runtime_probe": runtime_probe_fixture(package.sha256_file(profile / "config.yaml")),
            "claim_ceiling": CLAIM_CEILING,
        },
        mode=0o644,
    )
    return artifact, receipt


def runtime_surface_fixture() -> dict[str, object]:
    return copy.deepcopy(package.NO_SEND_TOOL_SURFACE)


def runtime_probe_fixture(config_sha256: str) -> dict[str, object]:
    surface = runtime_surface_fixture()
    return {
        "schema": "livingip.leocleanNoSendRuntimeProbe.v1",
        "status": "pass",
        "config_sha256": config_sha256,
        "gateway_adapter_count": 0,
        "connected_platforms": [],
        "python_version": package.PYTHON_VERSION,
        "model": package.DEFAULT_MODEL,
        "provider_credential_names_present": ["OPENROUTER_API_KEY"],
        "provider_credential_sentinel_redaction": "pass",
        "effective_policy": {
            "cheap_model_route": None,
            "memory_enabled": False,
            "memory_nudge_interval": 0,
            "skill_creation_nudge_interval": 0,
            "smart_model_routing": False,
            "user_profile_enabled": False,
        },
        "tool_surface": surface,
        "lifecycle": {
            "started": True,
            "stopped": True,
            "post_start_registry": surface,
            "post_discovery_registry": surface,
        },
    }


def test_git_source_closure_accepts_exact_revision_and_ignores_unrelated_dirty_files(tmp_path: Path) -> None:
    root, revision, expected = git_source_fixture(tmp_path, {"package/source.py": "reviewed = True\n"})
    (root / "unrelated.txt").write_text("not in the image closure\n", encoding="utf-8")

    package._verify_git_source_closure(root, revision, expected)


@pytest.mark.parametrize("mutation", ["unstaged", "staged", "deleted", "mode"])
def test_git_source_closure_rejects_dirty_required_source(tmp_path: Path, mutation: str) -> None:
    root, revision, expected = git_source_fixture(tmp_path, {"package/source.py": "reviewed = True\n"})
    source = root / "package" / "source.py"
    if mutation == "deleted":
        source.unlink()
    elif mutation == "mode":
        source.chmod(0o755)
    else:
        source.write_text("reviewed = False\n", encoding="utf-8")
        if mutation == "staged":
            subprocess.run(["git", "-C", str(root), "add", "--", "package/source.py"], check=True)

    with pytest.raises(package.PackageError, match="closure is dirty"):
        package._verify_git_source_closure(root, revision, expected)


def test_git_source_closure_rejects_untracked_required_source(tmp_path: Path) -> None:
    root, revision, expected = git_source_fixture(tmp_path, {"package/source.py": "reviewed = True\n"})
    untracked = root / "package" / "validator.py"
    untracked.write_text("unreviewed = True\n", encoding="utf-8")
    expected["package/validator.py"] = package.sha256_file(untracked)

    with pytest.raises(package.PackageError):
        package._verify_git_source_closure(root, revision, expected)


def test_git_source_closure_rejects_head_and_hash_mismatch(tmp_path: Path) -> None:
    root, revision, expected = git_source_fixture(tmp_path, {"package/source.py": "reviewed = True\n"})
    (root / "unrelated.txt").write_text("second commit\n", encoding="utf-8")
    subprocess.run(["git", "-C", str(root), "add", "unrelated.txt"], check=True)
    subprocess.run(
        [
            "git",
            "-C",
            str(root),
            "-c",
            "user.name=fwazb",
            "-c",
            "user.email=fawaz.butt95@gmail.com",
            "commit",
            "--quiet",
            "-m",
            "Advance source fixture",
        ],
        check=True,
    )

    with pytest.raises(package.PackageError, match="reviewed revision"):
        package._verify_git_source_closure(root, revision, expected)

    current = package._git_commit(root)
    wrong_hash = {"package/source.py": "9" * 64}
    with pytest.raises(package.PackageError, match="package binding"):
        package._verify_git_source_closure(root, current, wrong_hash)


def identity_fixture(tmp_path: Path, artifact: Path, receipt: Path) -> tuple[Path, dict[str, object]]:
    runtime = package.validate_artifact(artifact, receipt, REVISION)
    contract = package.build_expected_identity_contract(
        artifact=artifact,
        receipt_path=receipt,
        runtime=runtime,
        source_revision=REVISION,
        source_root=ROOT,
    )
    fixture = ROOT / "fixtures" / "working-leo" / "leo-identity-v1"
    manifest = package.strict_identity_manifest.build_identity_manifest(
        identity_runtime_contract=contract,
        database_fingerprint_path=fixture / "leo-database-fingerprint-v1.json",
        constitution_path=fixture / "leo-constitution-v1.json",
        database_identity_path=fixture / "leo-database-identity-v1.json",
        source_root=ROOT,
    )
    identity = tmp_path / "identity"
    package.compile_identity_bundle(
        manifest=manifest,
        expected_identity_contract=contract,
        source_root=ROOT,
        output=identity,
    )
    return identity, contract


def fully_rehash_identity_bundle(identity: Path) -> None:
    manifest_path = identity / "identity-manifest.json"
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    inputs_hash = package.canonical_sha256(manifest["identity_inputs"])
    manifest["identity_inputs_sha256"] = inputs_hash
    fixture = ROOT / "fixtures" / "working-leo" / "leo-identity-v1"
    fingerprint = package.strict_identity_manifest.load_json(
        fixture / "leo-database-fingerprint-v1.json", "database fingerprint"
    )
    rules = package.strict_identity_manifest.canonical_rules(
        package.strict_identity_manifest.load_json(fixture / "leo-constitution-v1.json", "constitution")
    )
    database_identity = package.strict_identity_manifest.load_json(
        fixture / "leo-database-identity-v1.json", "database identity"
    )
    records = package.strict_identity_manifest.canonical_database_records(
        database_identity,
        fingerprint_sha256=fingerprint["fingerprint_sha256"],
    )
    provenance = package.strict_identity_manifest.database_identity_provenance(
        database_identity,
        fingerprint=fingerprint,
    )
    rendered = package.strict_identity_manifest.render_identity_views(
        identity_inputs_sha256=inputs_hash,
        database_fingerprint_sha256=fingerprint["fingerprint_sha256"],
        database_provenance=provenance,
        rules=rules,
        records=records,
    )
    for name, content in rendered.items():
        (identity / name).write_text(content, encoding="utf-8")
        (identity / name).chmod(0o600)
        manifest["compiled_views"][name]["sha256"] = package.sha256_file(identity / name)
        manifest["compiled_views"][name]["generated_from_identity_inputs_sha256"] = inputs_hash
    stable = {key: value for key, value in manifest.items() if key != "manifest_sha256"}
    manifest["manifest_sha256"] = package.canonical_sha256(stable)
    write_json(manifest_path, manifest)
    lock_path = identity / "identity-lock.json"
    lock = json.loads(lock_path.read_text(encoding="utf-8"))
    lock["manifest_sha256"] = manifest["manifest_sha256"]
    lock["identity_inputs_sha256"] = inputs_hash
    lock["compiled_views"] = {name: {"sha256": package.sha256_file(identity / name)} for name in sorted(rendered)}
    write_json(lock_path, lock)


def fully_rehash_identity_contract(contract: dict[str, object]) -> None:
    runtime = contract["runtime"]
    assert isinstance(runtime, dict)
    runtime["identity_runtime_sha256"] = package.canonical_sha256(package._identity_runtime_payload(contract))
    stable = {key: value for key, value in contract.items() if key != "contract_sha256"}
    contract["contract_sha256"] = package.canonical_sha256(stable)


def refresh_image_identity_binding(image_input: dict[str, object], identity: Path) -> None:
    binding = image_input["identity"]
    assert isinstance(binding, dict)
    manifest = json.loads((identity / "identity-manifest.json").read_text(encoding="utf-8"))
    binding.update(
        {
            "bundle_sha256": package.identity_bundle_manifest(identity)["sha256"],
            "manifest_sha256": manifest["manifest_sha256"],
            "identity_inputs_sha256": manifest["identity_inputs_sha256"],
            "identity_lock_sha256": package.sha256_file(identity / "identity-lock.json"),
            "soul_sha256": package.sha256_file(identity / "SOUL.md"),
            "identity_view_sha256": package.sha256_file(identity / "identity.json"),
        }
    )
    stable = {key: value for key, value in image_input.items() if key != "input_sha256"}
    image_input["input_sha256"] = package.canonical_sha256(stable)


def image_input_fixture(tmp_path: Path) -> dict[str, object]:
    artifact, receipt = artifact_fixture(tmp_path)
    identity, _contract = identity_fixture(tmp_path, artifact, receipt)
    return package.build_image_input(
        artifact=artifact,
        receipt=receipt,
        identity=identity,
        ca=ROOT / "ops" / "gcp-teleo-pgvector-standby-server-ca.pem",
        source_revision=REVISION,
        dockerfile=ROOT / "docker" / "leoclean-nosend" / "Dockerfile",
        entrypoint=ROOT / "docker" / "leoclean-nosend" / "entrypoint.py",
        package_contract=ROOT / "ops" / "gcp_leoclean_nosend_package.py",
        contract_source_root=ROOT,
        identity_source_root=ROOT,
    )


def docker_inspection_fixture(
    image_input: dict[str, object],
    reference: str,
    *,
    config_digest: str = "sha256:" + "3" * 64,
) -> dict[str, object]:
    runtime = copy.deepcopy(package.IMAGE_RUNTIME_CONFIG)
    healthcheck = runtime["healthcheck"]
    return {
        "Id": config_digest,
        "RepoDigests": [reference],
        "Architecture": "amd64",
        "Os": "linux",
        "Variant": None,
        "Config": {
            "Labels": package.expected_image_labels(image_input),
            "User": runtime["user"],
            "Env": runtime["environment"],
            "WorkingDir": runtime["working_dir"],
            "ExposedPorts": runtime["exposed_ports"],
            "Volumes": runtime["volumes"],
            "Shell": runtime["shell"],
            "OnBuild": runtime["on_build"],
            "ArgsEscaped": runtime["args_escaped"],
            "StopTimeout": runtime["stop_timeout"],
            "Tty": runtime["tty"],
            "OpenStdin": runtime["open_stdin"],
            "StdinOnce": runtime["stdin_once"],
            "AttachStdin": runtime["attach_stdin"],
            "AttachStdout": runtime["attach_stdout"],
            "AttachStderr": runtime["attach_stderr"],
            "NetworkDisabled": runtime["network_disabled"],
            "Hostname": runtime["hostname"],
            "Domainname": runtime["domainname"],
            "MacAddress": runtime["mac_address"],
            "Entrypoint": runtime["entrypoint"],
            "Cmd": runtime["cmd"],
            "Healthcheck": {
                "Test": healthcheck["test"],
                "Interval": healthcheck["interval_ns"],
                "Timeout": healthcheck["timeout_ns"],
                "StartPeriod": healthcheck["start_period_ns"],
                "StartInterval": healthcheck["start_interval_ns"],
                "Retries": healthcheck["retries"],
            },
            "StopSignal": runtime["stop_signal"],
        },
    }


def docker_authority(tmp_path: Path) -> tuple[Path, str, Path]:
    docker_binary = tmp_path / "bin" / "docker"
    docker_config = tmp_path / "docker-config"
    docker_config.mkdir()
    return docker_binary, "unix:///tmp/livingip-docker.sock", docker_config


def docker_inventory_fixture(
    reference: str,
    *,
    present: bool,
    config_digest: str = "sha256:" + "3" * 64,
) -> bytes:
    if not present:
        return b""
    return (
        json.dumps(
            {
                "Repository": package.IMAGE_REPOSITORY,
                "Digest": reference.rsplit("@", 1)[1],
                "ID": config_digest,
            }
        ).encode()
        + b"\n"
    )


def build_push_receipt_fixture(
    image_input: dict[str, object],
    reference: str | None = None,
    *,
    config_digest: str = "sha256:" + "3" * 64,
) -> dict[str, object]:
    exact_reference = reference or f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    return package.build_build_push_receipt(
        image_input,
        post_push_image=docker_inspection_fixture(
            image_input,
            exact_reference,
            config_digest=config_digest,
        ),
    )


def prepared_context_fixture(tmp_path: Path) -> tuple[Path, dict[str, object]]:
    artifact, receipt = artifact_fixture(tmp_path)
    identity, _contract = identity_fixture(tmp_path, artifact, receipt)
    context = tmp_path / "prepared-context"
    image_input = package.prepare_context(
        repo_root=ROOT,
        artifact=artifact,
        receipt=receipt,
        identity=identity,
        identity_source_root=ROOT,
        output=context,
        source_revision=REVISION,
    )
    return context, image_input


def mocked_build_push_docker(
    image_input: dict[str, object],
    *,
    post_push_references: list[str] | None = None,
    fail_after_side_effect: str | None = None,
) -> tuple[object, dict[str, object], list[tuple[list[str], dict[str, object]]]]:
    config_digest = "sha256:" + "3" * 64
    exact_reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    references = list(post_push_references if post_push_references is not None else [exact_reference])
    build_tag = f"input-{image_input['input_sha256']}"
    build_reference = f"{package.LOCAL_BUILD_REPOSITORY}:{build_tag}"
    candidate_tag = f"candidate-{config_digest.removeprefix('sha256:')}"
    candidate_reference = f"{package.IMAGE_REPOSITORY}:{candidate_tag}"
    state: dict[str, object] = {"build": False, "candidate": False, "digests": False, "pushes": 0}
    calls: list[tuple[list[str], dict[str, object]]] = []

    def inspection(reference: str) -> bytes:
        value = docker_inspection_fixture(image_input, reference, config_digest=config_digest)
        value["RepoDigests"] = references if state["digests"] else []
        return json.dumps([value]).encode()

    def fake_run(arguments: list[str], **kwargs: object) -> subprocess.CompletedProcess[bytes]:
        operation = arguments[5:]
        calls.append((operation, kwargs))
        if operation[:2] == ["image", "ls"]:
            repository = operation[-1]
            if "--digests" in operation:
                rows = []
                if state["digests"] and repository == package.IMAGE_REPOSITORY:
                    for reference in references:
                        if reference.startswith(f"{package.IMAGE_REPOSITORY}@"):
                            rows.append(
                                json.dumps(
                                    {
                                        "Repository": package.IMAGE_REPOSITORY,
                                        "Digest": reference.rsplit("@", 1)[1],
                                        "ID": config_digest,
                                    }
                                )
                            )
                payload = (("\n".join(rows) + "\n") if rows else "").encode()
                return subprocess.CompletedProcess(arguments, 0, stdout=payload, stderr=b"")
            rows = []
            if repository == package.LOCAL_BUILD_REPOSITORY and state["build"]:
                rows.append(
                    json.dumps({"Repository": package.LOCAL_BUILD_REPOSITORY, "Tag": build_tag, "ID": config_digest})
                )
            if repository == package.IMAGE_REPOSITORY and state["candidate"]:
                rows.append(
                    json.dumps({"Repository": package.IMAGE_REPOSITORY, "Tag": candidate_tag, "ID": config_digest})
                )
            payload = (("\n".join(rows) + "\n") if rows else "").encode()
            return subprocess.CompletedProcess(arguments, 0, stdout=payload, stderr=b"")
        if operation[0] == "build":
            state["build"] = True
            if fail_after_side_effect == "build":
                raise subprocess.TimeoutExpired(arguments, 1800, stderr=b"credential=secret")
            return subprocess.CompletedProcess(arguments, 0, stdout=b"sha256:ignored\n", stderr=b"")
        if operation[:2] == ["image", "tag"]:
            assert operation[2:] == [build_reference, candidate_reference]
            state["candidate"] = True
            if fail_after_side_effect == "tag":
                raise subprocess.TimeoutExpired(arguments, 60, stderr=b"credential=secret")
            return subprocess.CompletedProcess(arguments, 0, stdout=b"", stderr=b"")
        if operation[:2] == ["image", "push"]:
            assert operation[2:] == ["--quiet", candidate_reference]
            state["digests"] = True
            state["pushes"] = int(state["pushes"]) + 1
            if fail_after_side_effect == "push":
                raise subprocess.TimeoutExpired(arguments, 1800, stderr=b"credential=secret")
            return subprocess.CompletedProcess(
                arguments,
                0,
                stdout=("sha256:" + "9" * 64 + "\n").encode(),
                stderr=b"",
            )
        if operation[:2] == ["image", "inspect"]:
            reference = operation[2]
            present = (
                (reference == build_reference and state["build"])
                or (reference == candidate_reference and state["candidate"])
                or (reference in references and state["digests"])
            )
            assert present, reference
            return subprocess.CompletedProcess(arguments, 0, stdout=inspection(reference), stderr=b"")
        if operation[:2] == ["image", "rm"]:
            reference = operation[2]
            assert "--force" not in operation
            if "@" in reference:
                state["digests"] = False
            elif reference == candidate_reference:
                state["candidate"] = False
            elif reference == build_reference:
                state["build"] = False
            else:
                raise AssertionError(reference)
            return subprocess.CompletedProcess(arguments, 0, stdout=b"removed\n", stderr=b"")
        raise AssertionError(operation)

    return fake_run, state, calls


def test_image_input_binds_exact_runtime_identity_and_staging_target(tmp_path: Path) -> None:
    value = image_input_fixture(tmp_path)
    package.validate_image_input(value)

    assert value["target"] == {
        "project": "teleo-501523",
        "zone": "europe-west6-a",
        "instance": "teleo-staging-1",
        "service": "leoclean-gcp-nosend.service",
        "service_account": "sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com",
        "platform": "linux/amd64",
    }
    assert value["runtime"]["python_version"] == "3.11.9"
    assert value["runtime"]["postgres_version"] == "16.14"
    assert value["identity"]["canonical_authority_granted"] is False
    database_policy = value["identity"]["runtime_contract"]["permissions"]["runtime_policy"]["database"]
    assert database_policy == {
        "direct_canonical_writes": "denied",
        "direct_stage_table_writes": "denied",
        "proposal_staging": "function_only",
        "proposal_function": "kb_stage.stage_leoclean_proposal(text,text,text,text,jsonb)",
        "live_effective_permissions_proven": False,
    }
    assert value["base_images"]["postgres"].endswith(
        "@sha256:92620daddcd947f8d5ab5ba66e848702fe443d87fed30c4cea8e389fd78dfc55"
    )


def test_artifact_requires_release_receipt_exact_revision_and_unchanged_content(tmp_path: Path) -> None:
    artifact, receipt = artifact_fixture(tmp_path)
    package.validate_artifact(artifact, receipt, REVISION)

    development = json.loads(receipt.read_text(encoding="utf-8"))
    development["verification_mode"] = "development"
    write_json(receipt, development, mode=0o644)
    with pytest.raises(package.PackageError, match="not release"):
        package.validate_artifact(artifact, receipt, REVISION)

    development["verification_mode"] = "release"
    write_json(receipt, development, mode=0o644)
    (artifact / "bootstrap.py").write_text("mutated\n", encoding="utf-8")
    (artifact / "bootstrap.py").chmod(0o755)
    with pytest.raises(package.PackageError, match="content manifest drifted"):
        package.validate_artifact(artifact, receipt, REVISION)


def test_artifact_binds_runtime_probe_to_the_exact_packaged_config(tmp_path: Path) -> None:
    artifact, receipt = artifact_fixture(tmp_path)
    value = json.loads(receipt.read_text(encoding="utf-8"))
    value["runtime_probe"]["config_sha256"] = "9" * 64
    write_json(receipt, value, mode=0o644)

    with pytest.raises(package.PackageError, match="runtime probe config differs"):
        package.validate_artifact(artifact, receipt, REVISION)


def test_identity_bundle_rejects_drift_extra_files_and_false_authority(tmp_path: Path) -> None:
    artifact, receipt = artifact_fixture(tmp_path)
    identity, contract = identity_fixture(tmp_path, artifact, receipt)
    validate = lambda root: package.validate_identity_bundle(  # noqa: E731
        root,
        expected_revision=REVISION,
        expected_identity_contract=contract,
        source_root=ROOT,
    )
    validate(identity)

    (identity / "SOUL.md").write_text("drift\n", encoding="utf-8")
    (identity / "SOUL.md").chmod(0o600)
    with pytest.raises(package.PackageError, match="view derivation drifted"):
        validate(identity)

    identity, _contract = identity_fixture(tmp_path / "second", artifact, receipt)
    (identity / "extra.txt").write_text("extra\n", encoding="utf-8")
    with pytest.raises(package.PackageError, match="entry set"):
        validate(identity)

    identity, _contract = identity_fixture(tmp_path / "third", artifact, receipt)
    manifest_path = identity / "identity-manifest.json"
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    manifest["identity_inputs"]["canonical_database"] = {
        "authority": "synthetic_noncanonical_fixture",
        "canonical_authority_granted": True,
    }
    stable = {key: value for key, value in manifest.items() if key != "manifest_sha256"}
    manifest["manifest_sha256"] = package.canonical_sha256(stable)
    write_json(manifest_path, manifest)
    lock_path = identity / "identity-lock.json"
    lock = json.loads(lock_path.read_text(encoding="utf-8"))
    lock["manifest_sha256"] = manifest["manifest_sha256"]
    write_json(lock_path, lock)
    with pytest.raises(package.PackageError, match="semantic/source validation"):
        validate(identity)


@pytest.mark.parametrize(
    ("path", "value"),
    [
        (("source_commit",), "2" * 40),
        (("runtime", "hermes_source_sha256"), "7" * 64),
        (("model", "default_model"), "attacker/model"),
        (("skills", "content_sha256"), "8" * 64),
        (("permissions", "runtime_policy", "general_network_egress", "denial_tested"), True),
    ],
)
def test_fully_rehashed_identity_runtime_mutations_fail_closed(
    tmp_path: Path, path: tuple[str, ...], value: object
) -> None:
    artifact, receipt = artifact_fixture(tmp_path)
    identity, contract = identity_fixture(tmp_path, artifact, receipt)
    manifest_path = identity / "identity-manifest.json"
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    target = manifest["identity_inputs"]
    for key in path[:-1]:
        target = target[key]
    target[path[-1]] = value
    if path == ("source_commit",):
        manifest["identity_inputs"]["runtime"]["teleo_git_head"] = value
    write_json(manifest_path, manifest)
    fully_rehash_identity_bundle(identity)

    with pytest.raises(package.PackageError):
        package.validate_identity_bundle(
            identity,
            expected_revision=REVISION,
            expected_identity_contract=contract,
            source_root=ROOT,
        )


def test_custom_no_send_policy_requires_explicit_validator_authority(tmp_path: Path) -> None:
    artifact, receipt = artifact_fixture(tmp_path)
    identity, contract = identity_fixture(tmp_path, artifact, receipt)
    manifest = json.loads((identity / "identity-manifest.json").read_text(encoding="utf-8"))

    with pytest.raises(package.strict_identity_manifest.IdentityManifestError, match="exact local readback policy"):
        package.strict_identity_manifest.validate_identity_manifest(manifest)
    package.strict_identity_manifest.validate_identity_manifest(
        manifest,
        expected_runtime_policy_value=contract["permissions"]["runtime_policy"],
    )
    with pytest.raises(
        package.strict_identity_manifest.IdentityManifestError, match="expected identity runtime policy"
    ):
        package.strict_identity_manifest.validate_identity_manifest(manifest, expected_runtime_policy_value={})


@pytest.mark.parametrize("mutation", ["legacy_policy", "malformed_behavior_inputs"])
def test_runtime_contract_adapter_rejects_weaker_or_malformed_contracts(tmp_path: Path, mutation: str) -> None:
    artifact, receipt = artifact_fixture(tmp_path)
    _identity, contract = identity_fixture(tmp_path, artifact, receipt)
    mutated = copy.deepcopy(contract)
    if mutation == "legacy_policy":
        mutated["permissions"]["runtime_policy"] = package.strict_identity_manifest.expected_runtime_policy()
    else:
        del mutated["behavior_inputs"]["tool_surface"]
    fully_rehash_identity_contract(mutated)

    with pytest.raises(package.strict_identity_manifest.IdentityManifestError):
        package.strict_identity_manifest.identity_runtime_inputs_from_contract(mutated)


def test_identity_source_content_drift_fails_closed(tmp_path: Path) -> None:
    artifact, receipt = artifact_fixture(tmp_path)
    identity, contract = identity_fixture(tmp_path, artifact, receipt)
    source_root = tmp_path / "source-root"
    fixture_relative = Path("fixtures/working-leo/leo-identity-v1")
    shutil.copytree(ROOT / fixture_relative, source_root / fixture_relative)
    package.validate_identity_bundle(
        identity,
        expected_revision=REVISION,
        expected_identity_contract=contract,
        source_root=source_root,
    )
    constitution = source_root / fixture_relative / "leo-constitution-v1.json"
    constitution.write_text(constitution.read_text(encoding="utf-8") + "\n", encoding="utf-8")

    with pytest.raises(package.PackageError, match="semantic/source validation"):
        package.validate_identity_bundle(
            identity,
            expected_revision=REVISION,
            expected_identity_contract=contract,
            source_root=source_root,
        )


def test_identity_source_intermediate_symlink_fails_closed(tmp_path: Path) -> None:
    artifact, receipt = artifact_fixture(tmp_path)
    identity, contract = identity_fixture(tmp_path, artifact, receipt)
    outside = tmp_path / "outside"
    shutil.copytree(ROOT / "fixtures", outside / "fixtures")
    source_root = tmp_path / "source-root"
    source_root.mkdir()
    (source_root / "fixtures").symlink_to(outside / "fixtures", target_is_directory=True)

    with pytest.raises(package.PackageError, match="semantic/source validation"):
        package.validate_identity_bundle(
            identity,
            expected_revision=REVISION,
            expected_identity_contract=contract,
            source_root=source_root,
        )


@pytest.mark.parametrize("mutation", ["skills", "behavior_artifact_and_uv"])
def test_installed_validation_rederives_fully_rehashed_identity_contract(tmp_path: Path, mutation: str) -> None:
    artifact, receipt = artifact_fixture(tmp_path)
    identity, _contract = identity_fixture(tmp_path, artifact, receipt)
    image_input = package.build_image_input(
        artifact=artifact,
        receipt=receipt,
        identity=identity,
        ca=ROOT / "ops" / "gcp-teleo-pgvector-standby-server-ca.pem",
        source_revision=REVISION,
        dockerfile=ROOT / "docker" / "leoclean-nosend" / "Dockerfile",
        entrypoint=ROOT / "docker" / "leoclean-nosend" / "entrypoint.py",
        package_contract=ROOT / "ops" / "gcp_leoclean_nosend_package.py",
        contract_source_root=ROOT,
        identity_source_root=ROOT,
    )
    embedded = image_input["identity"]["runtime_contract"]
    manifest_path = identity / "identity-manifest.json"
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    if mutation == "skills":
        embedded["skills"] = {"content_sha256": "8" * 64, "file_count": 999}
        manifest["identity_inputs"]["skills"] = copy.deepcopy(embedded["skills"])
    else:
        embedded["behavior_inputs"]["artifact_sha256"] = "8" * 64
        embedded["behavior_inputs"]["uv_lock_sha256"] = "9" * 64
        image_input["runtime"]["artifact_sha256"] = "8" * 64
        image_input["runtime"]["uv_lock_sha256"] = "9" * 64
    fully_rehash_identity_contract(embedded)
    manifest["identity_inputs"]["runtime"]["identity_runtime_sha256"] = embedded["runtime"]["identity_runtime_sha256"]
    write_json(manifest_path, manifest)
    fully_rehash_identity_bundle(identity)
    refresh_image_identity_binding(image_input, identity)

    package.validate_image_input(image_input)
    with pytest.raises(package.PackageError):
        package.validate_installed_runtime_identity(
            image_input,
            artifact=artifact,
            receipt=receipt,
            identity=identity,
            contract_source_root=ROOT,
            identity_source_root=ROOT,
        )


def test_release_requires_exact_digest_repository_and_strict_schema(tmp_path: Path) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    inspection = package.build_image_inspection(
        image_input, reference, docker_inspection_fixture(image_input, reference)
    )
    release = package.build_release_descriptor(
        image_input, inspection, build_push_receipt_fixture(image_input, inspection["reference"])
    )
    package.validate_release_descriptor(release)

    assert release["image"]["reference"] == reference
    assert release["image"]["digest"] == f"sha256:{IMAGE_DIGEST}"
    assert release["image"]["config_digest"] == "sha256:" + "3" * 64
    assert release["image"]["platform"] == "linux/amd64"
    assert release["inspection"] == inspection

    with pytest.raises(package.PackageError, match="exact staging digest"):
        package.build_release_descriptor(
            image_input,
            {**inspection, "reference": f"{package.IMAGE_REPOSITORY}:latest"},
            build_push_receipt_fixture(image_input, reference),
        )
    with pytest.raises(package.PackageError, match="exact staging digest"):
        package.build_release_descriptor(
            image_input,
            {
                **inspection,
                "reference": f"europe-west6-docker.pkg.dev/teleo-501523/teleo/other@sha256:{IMAGE_DIGEST}",
            },
            build_push_receipt_fixture(image_input, reference),
        )

    extra = copy.deepcopy(release)
    extra["unexpected"] = True
    with pytest.raises(package.PackageError, match="fields are not exact"):
        package.validate_release_descriptor(extra)

    wrong_label = copy.deepcopy(release)
    wrong_label["image"]["labels"]["livingip.artifact.sha256"] = "9" * 64
    stable = {key: value for key, value in wrong_label.items() if key != "release_sha256"}
    wrong_label["release_sha256"] = package.canonical_sha256(stable)
    with pytest.raises(package.PackageError, match="labels are not exact"):
        package.validate_release_descriptor(wrong_label)

    wrong_inspection = copy.deepcopy(release)
    wrong_inspection["inspection"]["source_revision"] = "9" * 40
    stable = {key: value for key, value in wrong_inspection.items() if key != "release_sha256"}
    wrong_inspection["release_sha256"] = package.canonical_sha256(stable)
    with pytest.raises(package.PackageError, match="source revision"):
        package.validate_release_descriptor(wrong_inspection)


def test_build_push_receipt_is_strict_self_hashed_and_repeatable(tmp_path: Path) -> None:
    image_input = image_input_fixture(tmp_path)
    first = build_push_receipt_fixture(image_input)
    second = build_push_receipt_fixture(image_input)

    assert first == second
    assert first["schema"] == package.BUILD_PUSH_RECEIPT_SCHEMA
    assert first["registry"] == {
        "repository": package.IMAGE_REPOSITORY,
        "manifest_digest": f"sha256:{IMAGE_DIGEST}",
    }
    stable = {key: value for key, value in first.items() if key != "receipt_sha256"}
    assert first["receipt_sha256"] == package.canonical_sha256(stable)
    package.validate_build_push_receipt(first, image_input=image_input)


@pytest.mark.parametrize(
    ("mutation", "error"),
    [
        ("extra", "fields are not exact"),
        ("schema", "schema"),
        ("revision", "source revision"),
        ("input", "input binding"),
        ("config", "config digest"),
        ("platform", "platform"),
        ("labels", "labels"),
        ("runtime", "runtime config"),
        ("repository", "repository"),
        ("manifest", "manifest digest"),
        ("self_hash", "self-hash"),
    ],
)
def test_build_push_receipt_rejects_every_binding_drift(
    tmp_path: Path,
    mutation: str,
    error: str,
) -> None:
    image_input = image_input_fixture(tmp_path)
    receipt = build_push_receipt_fixture(image_input)
    if mutation == "extra":
        receipt["unexpected"] = True
    elif mutation == "schema":
        receipt["schema"] = "livingip.leocleanNoSendBuildPushReceipt.v0"
    elif mutation == "revision":
        receipt["source_revision"] = "9" * 40
    elif mutation == "input":
        receipt["input_sha256"] = "9" * 64
    elif mutation == "config":
        receipt["local_image"]["config_digest"] = "invalid"
    elif mutation == "platform":
        receipt["local_image"]["platform"] = "linux/arm64"
    elif mutation == "labels":
        receipt["local_image"]["labels"]["livingip.runtime"] = "other"
    elif mutation == "runtime":
        receipt["local_image"]["runtime_config"]["cmd"] = ["other"]
    elif mutation == "repository":
        receipt["registry"]["repository"] = "example.invalid/other"
    elif mutation == "manifest":
        receipt["registry"]["manifest_digest"] = "invalid"
    else:
        receipt["receipt_sha256"] = "9" * 64
    if mutation != "self_hash":
        stable = {key: value for key, value in receipt.items() if key != "receipt_sha256"}
        receipt["receipt_sha256"] = package.canonical_sha256(stable)

    with pytest.raises(package.PackageError, match=error):
        package.validate_build_push_receipt(receipt, image_input=image_input)


def test_prepared_image_context_revalidates_exact_reviewed_closure(tmp_path: Path) -> None:
    context, expected = prepared_context_fixture(tmp_path)

    observed, build_args = package.validate_prepared_image_context(context)
    package.validate_reviewed_context_source(ROOT, context, observed)

    assert observed == expected
    assert build_args == json.loads((context / "build-args.json").read_text(encoding="utf-8"))


@pytest.mark.parametrize(
    "mutation",
    ["extra", "build_args", "dockerfile", "identity_source", "script", "empty_directory", "directory_mode"],
)
def test_prepared_image_context_rejects_mutated_build_inputs(tmp_path: Path, mutation: str) -> None:
    context, _image_input = prepared_context_fixture(tmp_path)
    if mutation == "extra":
        (context / "unreviewed.txt").write_text("unreviewed\n", encoding="utf-8")
        error = "entry set"
    elif mutation == "build_args":
        build_args = json.loads((context / "build-args.json").read_text(encoding="utf-8"))
        build_args["INPUT_SHA256"] = "9" * 64
        write_json(context / "build-args.json", build_args, mode=0o644)
        error = "build arguments"
    elif mutation == "dockerfile":
        (context / "Dockerfile").write_text("FROM scratch\n", encoding="utf-8")
        error = "source drifted"
    elif mutation == "identity_source":
        source = next(path for path in (context / "identity-sources").rglob("*") if path.is_file())
        source.write_text("{}\n", encoding="utf-8")
        error = "identity"
    elif mutation == "script":
        (context / "scripts" / "unreviewed.py").write_text("raise RuntimeError\n", encoding="utf-8")
        error = "validator entry set"
    elif mutation == "empty_directory":
        (context / "artifact" / "unreviewed-runtime-directory").mkdir()
        error = "directory entry set"
    else:
        (context / "artifact").chmod(0o700)
        error = "directory mode"

    with pytest.raises(package.PackageError, match=error):
        package.validate_prepared_image_context(context)


def test_prepared_image_context_rejects_every_file_mode_class(tmp_path: Path) -> None:
    context, _image_input = prepared_context_fixture(tmp_path)
    cases = [
        ("Dockerfile", 0o755, 0o644, "file mode drifted"),
        ("scripts/leo_identity_manifest.py", 0o755, 0o644, "file mode drifted"),
        (
            next(
                path.relative_to(context).as_posix()
                for path in (context / "identity-sources").rglob("*")
                if path.is_file()
            ),
            0o755,
            0o644,
            "file mode drifted",
        ),
        ("identity/SOUL.md", 0o644, 0o600, "file mode drifted"),
        ("artifact/artifact-manifest.json", 0o600, 0o644, "file mode drifted"),
        ("artifact/bootstrap.py", 0o644, 0o755, "artifact content manifest drifted"),
    ]

    for relative, changed_mode, expected_mode, error in cases:
        target = context / relative
        target.chmod(changed_mode)
        with pytest.raises(package.PackageError, match=error):
            package.validate_prepared_image_context(context)
        target.chmod(expected_mode)

    package.validate_prepared_image_context(context)


def test_build_push_derives_receipt_and_retains_all_daemon_references(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    image_input = image_input_fixture(tmp_path)
    build_args = {
        "ARTIFACT_SHA256": image_input["runtime"]["artifact_sha256"],
        "IDENTITY_SHA256": image_input["identity"]["bundle_sha256"],
        "INPUT_SHA256": image_input["input_sha256"],
        "TELEO_REVISION": image_input["runtime"]["teleo_git_head"],
    }
    context = tmp_path / "context"
    context.mkdir()
    repo_root = tmp_path / "repo"
    repo_root.mkdir()
    output = tmp_path / "build-push-receipt.json"
    docker_binary, docker_host, docker_config = docker_authority(tmp_path)
    fake_run, state, calls = mocked_build_push_docker(image_input)
    reviewed: list[tuple[Path, Path, dict[str, object]]] = []
    monkeypatch.setattr(package, "validate_prepared_image_context", lambda _context: (image_input, build_args))
    monkeypatch.setattr(
        package,
        "validate_reviewed_context_source",
        lambda root, observed_context, value: reviewed.append((root, observed_context, value)),
    )
    monkeypatch.setattr(package.subprocess, "run", fake_run)
    monkeypatch.setenv("DOCKER_TLS_VERIFY", "1")
    monkeypatch.setenv("BUILDKIT_PROGRESS", "plain")
    monkeypatch.setenv("BUILDX_BUILDER", "unreviewed")

    receipt = package.build_and_push_staging_image(
        repo_root,
        context,
        docker_binary=docker_binary,
        docker_host=docker_host,
        docker_config=docker_config,
        output_receipt=output,
        execute_staging_push=True,
    )

    assert json.loads(output.read_text(encoding="utf-8")) == receipt
    assert receipt["registry"]["manifest_digest"] == f"sha256:{IMAGE_DIGEST}"
    assert receipt["registry"]["manifest_digest"] != "sha256:" + "9" * 64
    assert reviewed == [(repo_root, context, image_input), (repo_root, context, image_input)]
    assert state == {"build": True, "candidate": True, "digests": True, "pushes": 1}
    build = next(operation for operation, _kwargs in calls if operation and operation[0] == "build")
    expected_prefix = [
        "build",
        "--quiet",
        "--platform",
        package.PLATFORM,
        "--pull=false",
        "--file",
        str(context / "Dockerfile"),
        "--tag",
        f"{package.LOCAL_BUILD_REPOSITORY}:input-{image_input['input_sha256']}",
    ]
    assert build[: len(expected_prefix)] == expected_prefix
    assert build[-1] == str(context)
    encoded_build_args = [build[index + 1] for index, item in enumerate(build) if item == "--build-arg"]
    assert encoded_build_args == [f"{name}={build_args[name]}" for name in sorted(build_args)]
    mutating = [
        operation
        for operation, _kwargs in calls
        if operation
        and (operation[0] == "build" or operation[:2] in (["image", "tag"], ["image", "push"], ["image", "rm"]))
    ]
    assert [operation[:2] for operation in mutating] == [
        ["build", "--quiet"],
        ["image", "tag"],
        ["image", "push"],
    ]
    assert not any(operation[:2] == ["image", "rm"] for operation, _kwargs in calls)
    for _operation, kwargs in calls:
        assert kwargs["stdin"] == subprocess.DEVNULL
        assert not any(name.startswith(("DOCKER_", "BUILDKIT_", "BUILDX_")) for name in kwargs["env"])


def test_build_push_requires_explicit_execution_flag_before_validation(tmp_path: Path) -> None:
    with pytest.raises(package.PackageError, match="explicit execution authorization"):
        package.build_and_push_staging_image(
            tmp_path,
            tmp_path,
            docker_binary=tmp_path / "docker",
            docker_host="unix:///tmp/docker.sock",
            docker_config=tmp_path,
            output_receipt=tmp_path / "receipt.json",
            execute_staging_push=False,
        )


@pytest.mark.parametrize(
    "references",
    [
        [],
        [
            f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}",
            f"{package.IMAGE_REPOSITORY}@sha256:{'4' * 64}",
        ],
    ],
)
def test_build_push_fails_closed_without_one_exact_post_push_digest(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    references: list[str],
) -> None:
    image_input = image_input_fixture(tmp_path)
    build_args = {
        "ARTIFACT_SHA256": image_input["runtime"]["artifact_sha256"],
        "IDENTITY_SHA256": image_input["identity"]["bundle_sha256"],
        "INPUT_SHA256": image_input["input_sha256"],
        "TELEO_REVISION": image_input["runtime"]["teleo_git_head"],
    }
    context = tmp_path / "context"
    context.mkdir()
    repo_root = tmp_path / "repo"
    repo_root.mkdir()
    docker_binary, docker_host, docker_config = docker_authority(tmp_path)
    fake_run, state, _calls = mocked_build_push_docker(
        image_input,
        post_push_references=references,
    )
    monkeypatch.setattr(package, "validate_prepared_image_context", lambda _context: (image_input, build_args))
    monkeypatch.setattr(package, "validate_reviewed_context_source", lambda *_args: None)
    monkeypatch.setattr(package.subprocess, "run", fake_run)

    output = tmp_path / "failed-build-push-receipt.json"
    with pytest.raises(package.PackageError, match="build/push failed"):
        package.build_and_push_staging_image(
            repo_root,
            context,
            docker_binary=docker_binary,
            docker_host=docker_host,
            docker_config=docker_config,
            output_receipt=output,
            execute_staging_push=True,
        )

    assert not output.exists()
    assert state["build"] is True
    assert state["candidate"] is True


@pytest.mark.parametrize("failure", ["build", "tag", "push"])
def test_build_push_retains_daemon_references_when_cli_fails_after_side_effect(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    failure: str,
) -> None:
    image_input = image_input_fixture(tmp_path)
    build_args = {
        "ARTIFACT_SHA256": image_input["runtime"]["artifact_sha256"],
        "IDENTITY_SHA256": image_input["identity"]["bundle_sha256"],
        "INPUT_SHA256": image_input["input_sha256"],
        "TELEO_REVISION": image_input["runtime"]["teleo_git_head"],
    }
    context = tmp_path / "context"
    context.mkdir()
    repo_root = tmp_path / "repo"
    repo_root.mkdir()
    docker_binary, docker_host, docker_config = docker_authority(tmp_path)
    fake_run, state, calls = mocked_build_push_docker(
        image_input,
        fail_after_side_effect=failure,
    )
    monkeypatch.setattr(package, "validate_prepared_image_context", lambda _context: (image_input, build_args))
    monkeypatch.setattr(package, "validate_reviewed_context_source", lambda *_args: None)
    monkeypatch.setattr(package.subprocess, "run", fake_run)
    output = tmp_path / "failed-side-effect-receipt.json"

    with pytest.raises(package.PackageError, match="build/push failed") as raised:
        package.build_and_push_staging_image(
            repo_root,
            context,
            docker_binary=docker_binary,
            docker_host=docker_host,
            docker_config=docker_config,
            output_receipt=output,
            execute_staging_push=True,
        )

    assert "credential=secret" not in str(raised.value)
    assert not output.exists()
    assert state == {
        "build": True,
        "candidate": failure in {"tag", "push"},
        "digests": failure == "push",
        "pushes": 1 if failure == "push" else 0,
    }
    removals = [operation for operation, _kwargs in calls if operation[:2] == ["image", "rm"]]
    assert removals == []


def test_build_push_receipt_publication_is_atomic_repeatable_and_no_clobber(tmp_path: Path) -> None:
    image_input = image_input_fixture(tmp_path)
    receipt = build_push_receipt_fixture(image_input)
    first = tmp_path / "first.json"
    second = tmp_path / "second.json"

    package.publish_build_push_receipt(first, receipt, image_input=image_input)
    package.publish_build_push_receipt(second, receipt, image_input=image_input)

    assert first.read_bytes() == second.read_bytes()
    assert first.stat().st_mode & 0o777 == second.stat().st_mode & 0o777 == 0o600
    before = first.stat(follow_symlinks=False)
    with pytest.raises(package.PackageError, match="already exists"):
        package.publish_build_push_receipt(first, receipt, image_input=image_input)
    after = first.stat(follow_symlinks=False)
    assert (after.st_dev, after.st_ino) == (before.st_dev, before.st_ino)
    retained = list(tmp_path.glob(".*.tmp"))
    assert len(retained) == 2
    assert {entry.stat().st_ino for entry in retained} == {first.stat().st_ino, second.stat().st_ino}
    assert first.stat().st_nlink == second.stat().st_nlink == 2


@pytest.mark.parametrize("mode", [0o770, 0o777])
def test_build_push_receipt_rejects_group_or_world_writable_parent(
    tmp_path: Path,
    mode: int,
) -> None:
    image_input = image_input_fixture(tmp_path)
    receipt = build_push_receipt_fixture(image_input)
    shared = tmp_path / "shared"
    shared.mkdir(mode=mode)
    shared.chmod(mode)
    output = shared / "receipt.json"

    with pytest.raises(package.PackageError, match="operator-owned directory"):
        package.publish_build_push_receipt(output, receipt, image_input=image_input)

    assert not output.exists()


def test_build_push_receipt_refuses_replaced_staged_inode(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    image_input = image_input_fixture(tmp_path)
    receipt = build_push_receipt_fixture(image_input)
    output = tmp_path / "published-build-push-receipt.json"
    original_link = package.os.link

    def replace_then_link(source: str, destination: str, **kwargs: object) -> None:
        directory_descriptor = int(kwargs["src_dir_fd"])
        os.unlink(source, dir_fd=directory_descriptor)
        replacement = os.open(
            source,
            os.O_WRONLY | os.O_CREAT | os.O_EXCL,
            0o600,
            dir_fd=directory_descriptor,
        )
        try:
            os.write(replacement, b"attacker replacement")
            os.fsync(replacement)
        finally:
            os.close(replacement)
        original_link(source, destination, **kwargs)

    monkeypatch.setattr(package.os, "link", replace_then_link)

    with pytest.raises(package.PackageError, match="identity or posture drifted"):
        package.publish_build_push_receipt(output, receipt, image_input=image_input)

    assert output.read_bytes() == b"attacker replacement"
    staged = list(tmp_path.glob(".*.tmp"))
    assert len(staged) == 1
    assert staged[0].read_bytes() == b"attacker replacement"
    assert staged[0].stat().st_ino == output.stat().st_ino


def test_build_push_receipt_rollback_preserves_replaced_destination(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    image_input = image_input_fixture(tmp_path)
    receipt = build_push_receipt_fixture(image_input)
    output = tmp_path / "published-build-push-receipt.json"
    replacement_payload = b"same-user replacement"
    original_link = package.os.link

    def link_then_replace(source: str, destination: str, **kwargs: object) -> None:
        original_link(source, destination, **kwargs)
        directory_descriptor = int(kwargs["dst_dir_fd"])
        os.unlink(destination, dir_fd=directory_descriptor)
        replacement = os.open(
            destination,
            os.O_WRONLY | os.O_CREAT | os.O_EXCL,
            0o600,
            dir_fd=directory_descriptor,
        )
        try:
            os.write(replacement, replacement_payload)
            os.fsync(replacement)
        finally:
            os.close(replacement)

    monkeypatch.setattr(package.os, "link", link_then_replace)

    with pytest.raises(package.PackageError, match="identity or posture drifted"):
        package.publish_build_push_receipt(output, receipt, image_input=image_input)

    assert output.read_bytes() == replacement_payload
    staged = list(tmp_path.glob(".*.tmp"))
    assert len(staged) == 1
    assert staged[0].stat().st_ino != output.stat().st_ino


def test_build_push_receipt_never_automatically_unlinks_paths(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    image_input = image_input_fixture(tmp_path)
    receipt = build_push_receipt_fixture(image_input)
    output = tmp_path / "published-build-push-receipt.json"

    def forbidden_unlink(*_args: object, **_kwargs: object) -> None:
        raise AssertionError("receipt publication must not unlink pathnames")

    monkeypatch.setattr(package.os, "unlink", forbidden_unlink)

    package.publish_build_push_receipt(output, receipt, image_input=image_input)

    assert output.stat().st_nlink == 2


def test_private_receipt_reader_detects_path_replacement_after_open(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    image_input = image_input_fixture(tmp_path)
    receipt = build_push_receipt_fixture(image_input)
    receipt_path = tmp_path / "published-build-push-receipt.json"
    package.publish_build_push_receipt(receipt_path, receipt, image_input=image_input)
    original_read = package.os.read
    replaced = False

    def replace_then_read(descriptor: int, size: int) -> bytes:
        nonlocal replaced
        if not replaced:
            replaced = True
            receipt_path.unlink()
            write_json(receipt_path, receipt, mode=0o600)
        return original_read(descriptor, size)

    monkeypatch.setattr(package.os, "read", replace_then_read)

    with pytest.raises(package.PackageError, match="changed while it was read"):
        package.load_private_json(receipt_path, "build/push receipt")


def test_private_receipt_reader_rejects_writable_file(
    tmp_path: Path,
) -> None:
    image_input = image_input_fixture(tmp_path)
    receipt_path = tmp_path / "receipt.json"
    write_json(receipt_path, build_push_receipt_fixture(image_input), mode=0o660)

    with pytest.raises(package.PackageError, match="file posture is unsafe"):
        package.load_private_json(receipt_path, "build/push receipt")


def test_build_push_receipt_publication_rolls_back_failed_parent_fsync(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    image_input = image_input_fixture(tmp_path)
    receipt = build_push_receipt_fixture(image_input)
    output = tmp_path / "published-build-push-receipt.json"
    original = package._fsync_directory_descriptor
    parent_calls = 0

    def fail_first_parent(descriptor: int) -> None:
        nonlocal parent_calls
        parent_calls += 1
        if parent_calls == 1:
            raise OSError("injected")
        original(descriptor)

    monkeypatch.setattr(package, "_fsync_directory_descriptor", fail_first_parent)

    with pytest.raises(package.PackageError, match="cannot publish"):
        package.publish_build_push_receipt(output, receipt, image_input=image_input)

    assert parent_calls == 1
    assert output.exists()
    retained = list(tmp_path.glob(".*.tmp"))
    assert len(retained) == 1
    assert retained[0].stat().st_ino == output.stat().st_ino


def test_release_cross_binds_receipt_registry_and_independent_inspection(tmp_path: Path) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    inspection = package.build_image_inspection(
        image_input,
        reference,
        docker_inspection_fixture(image_input, reference),
    )
    receipt = build_push_receipt_fixture(image_input, reference)
    release = package.build_release_descriptor(image_input, inspection, receipt)
    release["build_push_receipt"]["registry"]["manifest_digest"] = "sha256:" + "4" * 64
    stable_receipt = {key: value for key, value in release["build_push_receipt"].items() if key != "receipt_sha256"}
    release["build_push_receipt"]["receipt_sha256"] = package.canonical_sha256(stable_receipt)
    stable_release = {key: value for key, value in release.items() if key != "release_sha256"}
    release["release_sha256"] = package.canonical_sha256(stable_release)

    with pytest.raises(package.PackageError, match="differs from its build/push receipt"):
        package.validate_release_descriptor(release)


@pytest.mark.parametrize(
    ("field", "value", "error"),
    [
        ("Architecture", "arm64", "platform"),
        ("Os", "windows", "platform"),
        ("Variant", "v8", "platform"),
        ("RepoDigests", [], "repository digest"),
        ("RepoDigests", [f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"] * 2, "repository digest"),
        ("Id", "sha256:invalid", "config digest"),
        ("Config", {"Labels": {}}, "labels"),
    ],
)
def test_image_inspection_rejects_platform_digest_and_label_drift(
    tmp_path: Path,
    field: str,
    value: object,
    error: str,
) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    inspected = docker_inspection_fixture(image_input, reference)
    inspected[field] = value

    with pytest.raises(package.PackageError, match=error):
        package.build_image_inspection(image_input, reference, inspected)


@pytest.mark.parametrize(
    ("field", "value"),
    [
        ("User", "65532"),
        ("Env", ["PATH=/unreviewed"]),
        ("WorkingDir", "/tmp"),
        ("ExposedPorts", {"8080/tcp": {}}),
        ("Volumes", {"/host": {}}),
        ("Shell", ["/bin/sh", "-c"]),
        ("OnBuild", ["RUN id"]),
        ("ArgsEscaped", 0),
        ("StopTimeout", 30),
        ("Tty", True),
        ("OpenStdin", True),
        ("StdinOnce", True),
        ("AttachStdin", True),
        ("AttachStdout", True),
        ("AttachStderr", True),
        ("NetworkDisabled", True),
        ("Hostname", "other"),
        ("Domainname", "other"),
        ("MacAddress", "00:11:22:33:44:55"),
        ("WorkingDir", False),
        ("WorkingDir", 0),
        ("WorkingDir", []),
        ("WorkingDir", {}),
        ("Entrypoint", ["/bin/sh"]),
        ("Cmd", ["probe"]),
        ("StopSignal", "SIGKILL"),
        ("Unexpected", "unreviewed authority"),
    ],
)
def test_image_inspection_rejects_runtime_config_drift(tmp_path: Path, field: str, value: object) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    inspected = docker_inspection_fixture(image_input, reference)
    inspected["Config"][field] = value

    with pytest.raises(package.PackageError, match="runtime config"):
        package.build_image_inspection(image_input, reference, inspected)


@pytest.mark.parametrize(
    ("field", "value"),
    [
        ("Interval", 15_000_000_000.0),
        ("Timeout", 10_000_000_000.0),
        ("StartPeriod", 30_000_000_000.0),
        ("Retries", 4.0),
        ("StartInterval", False),
        ("Unexpected", 0),
    ],
)
def test_image_inspection_rejects_healthcheck_type_and_field_drift(
    tmp_path: Path,
    field: str,
    value: object,
) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    inspected = docker_inspection_fixture(image_input, reference)
    inspected["Config"]["Healthcheck"][field] = value

    with pytest.raises(package.PackageError, match="runtime config"):
        package.build_image_inspection(image_input, reference, inspected)


@pytest.mark.parametrize("user", [None, ""])
@pytest.mark.parametrize("start_interval", [None, 0])
def test_image_inspection_accepts_portable_root_and_start_interval(
    tmp_path: Path,
    user: str | None,
    start_interval: int | None,
) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    inspected = docker_inspection_fixture(image_input, reference)
    inspected["Config"]["User"] = user
    if start_interval is None:
        inspected["Config"]["Healthcheck"].pop("StartInterval")
    else:
        inspected["Config"]["Healthcheck"]["StartInterval"] = start_interval

    receipt = package.build_image_inspection(image_input, reference, inspected)

    assert receipt["runtime_config"] == package.IMAGE_RUNTIME_CONFIG


@pytest.mark.parametrize("args_escaped", [None, False, True])
def test_image_inspection_normalizes_linux_args_escaped(
    tmp_path: Path,
    args_escaped: bool | None,
) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    inspected = docker_inspection_fixture(image_input, reference)
    if args_escaped is None:
        inspected["Config"].pop("ArgsEscaped")
    else:
        inspected["Config"]["ArgsEscaped"] = args_escaped

    receipt = package.build_image_inspection(image_input, reference, inspected)

    assert receipt["runtime_config"]["args_escaped"] is False


def test_image_inspection_allows_unrelated_labels_but_rejects_extra_authority(tmp_path: Path) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    inspected = docker_inspection_fixture(image_input, reference)
    inspected["Config"]["Labels"]["org.example.inherited"] = "permitted"

    receipt = package.build_image_inspection(image_input, reference, inspected)

    assert receipt["labels"] == package.expected_image_labels(image_input)
    inspected["Config"]["Labels"]["livingip.unreviewed"] = "authority"
    with pytest.raises(package.PackageError, match="labels"):
        package.build_image_inspection(image_input, reference, inspected)


@pytest.mark.parametrize(
    "raw",
    [
        b'[{"Id":"first","Id":"second"}]',
        b'[{"value":NaN}]',
        b"\xff",
    ],
)
def test_docker_inspection_json_parser_rejects_ambiguous_payloads(raw: bytes) -> None:
    with pytest.raises(package.PackageError, match="output is invalid"):
        package._load_docker_inspection(raw)


def test_docker_inspection_json_parser_rejects_oversized_payload() -> None:
    raw = b" " * (package.MAX_DOCKER_INSPECTION_BYTES + 1)
    with pytest.raises(package.PackageError, match="too large"):
        package._load_docker_inspection(raw)


@pytest.mark.parametrize(
    "raw",
    [
        '{"outer":{"value":"first","value":"second"}}',
        '{"value":NaN}',
        '{"value":Infinity}',
        '{"value":-Infinity}',
    ],
)
def test_trust_bound_json_loader_rejects_duplicate_keys_and_nonfinite_values(
    tmp_path: Path,
    raw: str,
) -> None:
    path = tmp_path / "untrusted.json"
    path.write_text(raw, encoding="utf-8")

    with pytest.raises(package.PackageError, match="invalid JSON"):
        package.load_json(path, "untrusted input")


def test_canonical_json_rejects_nonfinite_values() -> None:
    with pytest.raises(package.PackageError, match="non-finite"):
        package.canonical_sha256({"value": float("nan")})


@pytest.mark.parametrize("command", ["finalize", "validate"])
@pytest.mark.parametrize("raw", ['{"schema":"first","schema":"second"}', '{"value":NaN}'])
def test_cli_rejects_ambiguous_trust_bound_json_before_docker(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    command: str,
    raw: str,
) -> None:
    path = tmp_path / "ambiguous.json"
    path.write_text(raw, encoding="utf-8")
    monkeypatch.setattr(package, "inspect_registry_image", lambda *_args, **_kwargs: pytest.fail("Docker used"))
    if command == "validate":
        arguments = ["validate", "--release", str(path)]
    else:
        docker_binary, docker_host, docker_config = docker_authority(tmp_path)
        arguments = [
            "finalize",
            "--image-input",
            str(path),
            "--build-push-receipt",
            str(path),
            "--docker-binary",
            str(docker_binary),
            "--docker-host",
            docker_host,
            "--docker-config",
            str(docker_config),
            "--output-bundle",
            str(tmp_path / "bundle"),
        ]

    assert package.main(arguments) == 65


@pytest.mark.parametrize(
    ("command", "forbidden_flag"),
    [
        ("build-push", "--manifest-digest"),
        ("build-push", "--image-reference"),
        ("finalize", "--expected-config-digest"),
        ("finalize", "--image-reference"),
    ],
)
def test_build_push_and_finalize_expose_no_caller_digest_authority(
    tmp_path: Path,
    command: str,
    forbidden_flag: str,
) -> None:
    docker_binary, docker_host, docker_config = docker_authority(tmp_path)
    if command == "build-push":
        arguments = [
            command,
            "--image-context",
            str(tmp_path),
            "--docker-binary",
            str(docker_binary),
            "--docker-host",
            docker_host,
            "--docker-config",
            str(docker_config),
            "--output-receipt",
            str(tmp_path / "receipt.json"),
            forbidden_flag,
            "sha256:" + "9" * 64,
        ]
    else:
        arguments = [
            command,
            "--image-input",
            str(tmp_path / "image-input.json"),
            "--build-push-receipt",
            str(tmp_path / "receipt.json"),
            "--docker-binary",
            str(docker_binary),
            "--docker-host",
            docker_host,
            "--docker-config",
            str(docker_config),
            "--output-bundle",
            str(tmp_path / "bundle"),
            forbidden_flag,
            "sha256:" + "9" * 64,
        ]

    with pytest.raises(SystemExit) as raised:
        package._parse_args(arguments)

    assert raised.value.code == 2


@pytest.mark.parametrize(
    "inventory",
    [
        b'{"Repository":"first","Repository":"second"}\n',
        b'{"Repository":NaN}\n',
        b"\xff\n",
    ],
)
def test_local_image_inventory_rejects_ambiguous_json(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    inventory: bytes,
) -> None:
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    docker_binary, docker_host, docker_config = docker_authority(tmp_path)

    def fake_run(*_args, **_kwargs):
        return subprocess.CompletedProcess([], 0, stdout=inventory, stderr=b"")

    monkeypatch.setattr(package, "_run_docker", fake_run)

    with pytest.raises(package.PackageError, match="local image inventory output is invalid"):
        package._local_image_reference_config_id(
            docker_binary,
            docker_host,
            docker_config,
            reference,
        )


def test_local_image_inventory_rejects_inventory_and_inspection_disagreement(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    docker_binary, docker_host, docker_config = docker_authority(tmp_path)
    operations: list[list[str]] = []

    def fake_run(arguments: list[str], **_kwargs):
        operation = arguments[5:]
        operations.append(operation)
        if operation[:2] == ["image", "ls"]:
            return subprocess.CompletedProcess(
                arguments,
                0,
                stdout=docker_inventory_fixture(reference, present=True),
                stderr=b"",
            )
        if operation[:2] == ["image", "inspect"]:
            return subprocess.CompletedProcess(
                arguments,
                0,
                stdout=json.dumps([{"Id": "sha256:" + "4" * 64}]).encode(),
                stderr=b"",
            )
        raise AssertionError(operation)

    monkeypatch.setattr(package.subprocess, "run", fake_run)

    with pytest.raises(package.PackageError, match="inventory and inspection disagree"):
        package._local_image_reference_config_id(
            docker_binary,
            docker_host,
            docker_config,
            reference,
        )

    assert not any(operation and operation[0] == "pull" for operation in operations)
    assert not any(operation[:2] == ["image", "rm"] for operation in operations)


@pytest.mark.parametrize("config_digest", ["sha256:" + "3" * 64, "sha256:" + "a" * 64])
def test_registry_inspection_absent_pull_dependent_retains_new_reference(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    config_digest: str,
) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    docker_binary, docker_host, docker_config = docker_authority(tmp_path)
    prefix = [str(docker_binary), "--host", docker_host, "--config", str(docker_config)]
    calls: list[list[str]] = []
    environments: list[dict[str, str]] = []
    state = {"present": False, "dependent": False, "inspection_calls": 0}
    payload = json.dumps([docker_inspection_fixture(image_input, reference, config_digest=config_digest)]).encode()
    monkeypatch.setenv("DOCKER_HOST", "tcp://unreviewed.example:2375")
    monkeypatch.setenv("DOCKER_CONTEXT", "unreviewed")

    def fake_run(arguments: list[str], **kwargs):
        calls.append(arguments)
        environments.append(kwargs["env"])
        assert arguments[:5] == prefix
        operation = arguments[5:]
        if operation[:2] == ["image", "ls"]:
            return subprocess.CompletedProcess(
                arguments,
                0,
                stdout=docker_inventory_fixture(
                    reference,
                    present=state["present"],
                    config_digest=config_digest,
                ),
                stderr=b"",
            )
        if operation[:2] == ["image", "inspect"]:
            state["inspection_calls"] += 1
            if state["inspection_calls"] == 2:
                state["dependent"] = True
            return subprocess.CompletedProcess(
                arguments,
                0,
                stdout=payload,
                stderr=b"",
            )
        if operation[0] == "pull":
            state["present"] = True
            return subprocess.CompletedProcess(arguments, 0, stdout=b"pulled\n", stderr=b"")
        if operation[:2] == ["image", "rm"]:
            raise AssertionError("digest reference must never be removed")
        raise AssertionError(operation)

    monkeypatch.setattr(package.subprocess, "run", fake_run)

    inspection = package.inspect_registry_image(
        image_input,
        build_push_receipt_fixture(image_input, reference, config_digest=config_digest),
        docker_binary=docker_binary,
        docker_host=docker_host,
        docker_config=docker_config,
    )

    assert inspection["status"] == "pass"
    assert state == {"present": True, "dependent": True, "inspection_calls": 2}
    assert [call[5:] for call in calls if call[5] == "pull"] == [
        ["pull", "--quiet", "--platform", package.PLATFORM, reference]
    ]
    assert all(not any(name.startswith("DOCKER_") for name in environment) for environment in environments)
    assert not any(call[5:7] == ["image", "rm"] for call in calls)


def test_registry_inspection_retains_preexisting_reference(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    docker_binary, docker_host, docker_config = docker_authority(tmp_path)
    payload = json.dumps([docker_inspection_fixture(image_input, reference)]).encode()
    state = {"present": True, "pulls": 0}

    def fake_run(arguments: list[str], **_kwargs):
        operation = arguments[5:]
        if operation[:2] == ["image", "ls"]:
            return subprocess.CompletedProcess(
                arguments,
                0,
                stdout=docker_inventory_fixture(reference, present=state["present"]),
                stderr=b"",
            )
        if operation[:2] == ["image", "inspect"]:
            return subprocess.CompletedProcess(arguments, 0, stdout=payload, stderr=b"")
        if operation[0] == "pull":
            state["pulls"] += 1
            return subprocess.CompletedProcess(arguments, 0, stdout=b"pulled\n", stderr=b"")
        if operation[:2] == ["image", "rm"]:
            raise AssertionError("digest reference must never be removed")
        raise AssertionError(operation)

    monkeypatch.setattr(package.subprocess, "run", fake_run)

    inspection = package.inspect_registry_image(
        image_input,
        build_push_receipt_fixture(image_input, reference),
        docker_binary=docker_binary,
        docker_host=docker_host,
        docker_config=docker_config,
    )

    assert inspection["status"] == "pass"
    assert state == {"present": True, "pulls": 1}


def test_registry_inspection_rejects_config_digest_mismatch_and_retains_reference(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    docker_binary, docker_host, docker_config = docker_authority(tmp_path)
    state = {"present": False, "removed": False}
    payload = json.dumps([docker_inspection_fixture(image_input, reference)]).encode()

    def fake_run(arguments: list[str], **_kwargs):
        operation = arguments[5:]
        if operation[:2] == ["image", "ls"]:
            return subprocess.CompletedProcess(
                arguments,
                0,
                stdout=docker_inventory_fixture(reference, present=state["present"]),
                stderr=b"",
            )
        if operation[:2] == ["image", "inspect"]:
            return subprocess.CompletedProcess(arguments, 0, stdout=payload, stderr=b"")
        if operation[0] == "pull":
            state["present"] = True
            return subprocess.CompletedProcess(arguments, 0, stdout=b"pulled\n", stderr=b"")
        if operation[:2] == ["image", "rm"]:
            state.update(present=False, removed=True)
            return subprocess.CompletedProcess(arguments, 0, stdout=b"removed\n", stderr=b"")
        raise AssertionError(operation)

    monkeypatch.setattr(package.subprocess, "run", fake_run)

    with pytest.raises(package.PackageError, match="inspection failed") as raised:
        package.inspect_registry_image(
            image_input,
            build_push_receipt_fixture(image_input, reference, config_digest="sha256:" + "4" * 64),
            docker_binary=docker_binary,
            docker_host=docker_host,
            docker_config=docker_config,
        )

    assert "receipted build config digest" in str(raised.value.__cause__)
    assert state == {"present": True, "removed": False}


def test_registry_inspection_pull_failure_is_redacted_and_does_not_remove_reference(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    docker_binary, docker_host, docker_config = docker_authority(tmp_path)
    state = {"present": False}
    payload = json.dumps([docker_inspection_fixture(image_input, reference)]).encode()

    def fake_run(arguments: list[str], **_kwargs):
        operation = arguments[5:]
        if operation[:2] == ["image", "ls"]:
            return subprocess.CompletedProcess(
                arguments,
                0,
                stdout=docker_inventory_fixture(reference, present=state["present"]),
                stderr=b"",
            )
        if operation[:2] == ["image", "inspect"]:
            return subprocess.CompletedProcess(arguments, 0, stdout=payload, stderr=b"")
        if operation[0] == "pull":
            state["present"] = True
            raise subprocess.CalledProcessError(1, arguments, stderr=b"credential=value")
        if operation[:2] == ["image", "rm"]:
            state["present"] = False
            return subprocess.CompletedProcess(arguments, 0, stdout=b"removed\n", stderr=b"")
        raise AssertionError(operation)

    monkeypatch.setattr(package.subprocess, "run", fake_run)

    with pytest.raises(package.PackageError, match="inspection failed") as raised:
        package.inspect_registry_image(
            image_input,
            build_push_receipt_fixture(image_input, reference),
            docker_binary=docker_binary,
            docker_host=docker_host,
            docker_config=docker_config,
        )

    assert state["present"] is True
    assert "credential=value" not in str(raised.value)


def test_registry_inspection_validation_failure_retains_reference(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    docker_binary, docker_host, docker_config = docker_authority(tmp_path)
    inspected = docker_inspection_fixture(image_input, reference)
    inspected["Config"]["Env"] = ["PATH=/unreviewed"]
    payload = json.dumps([inspected]).encode()
    state = {"present": False, "removed": False}

    def fake_run(arguments: list[str], **_kwargs):
        operation = arguments[5:]
        if operation[:2] == ["image", "ls"]:
            return subprocess.CompletedProcess(
                arguments,
                0,
                stdout=docker_inventory_fixture(reference, present=state["present"]),
                stderr=b"",
            )
        if operation[:2] == ["image", "inspect"]:
            return subprocess.CompletedProcess(
                arguments,
                0,
                stdout=payload,
                stderr=b"",
            )
        if operation[0] == "pull":
            state["present"] = True
            return subprocess.CompletedProcess(arguments, 0, stdout=b"pulled\n", stderr=b"")
        if operation[:2] == ["image", "rm"]:
            state.update(present=False, removed=True)
            return subprocess.CompletedProcess(arguments, 0, stdout=b"removed\n", stderr=b"")
        raise AssertionError(operation)

    monkeypatch.setattr(package.subprocess, "run", fake_run)

    with pytest.raises(package.PackageError, match="inspection failed"):
        package.inspect_registry_image(
            image_input,
            build_push_receipt_fixture(image_input, reference),
            docker_binary=docker_binary,
            docker_host=docker_host,
            docker_config=docker_config,
        )

    assert state == {"present": True, "removed": False}


def test_registry_inspection_never_calls_image_rm_after_success(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    docker_binary, docker_host, docker_config = docker_authority(tmp_path)
    payload = json.dumps([docker_inspection_fixture(image_input, reference)]).encode()
    state = {"present": False}

    def fake_run(arguments: list[str], **_kwargs):
        operation = arguments[5:]
        if operation[:2] == ["image", "ls"]:
            return subprocess.CompletedProcess(
                arguments,
                0,
                stdout=docker_inventory_fixture(reference, present=state["present"]),
                stderr=b"",
            )
        if operation[:2] == ["image", "inspect"]:
            return subprocess.CompletedProcess(
                arguments,
                0,
                stdout=payload,
                stderr=b"",
            )
        if operation[0] == "pull":
            state["present"] = True
            return subprocess.CompletedProcess(arguments, 0, stdout=b"pulled\n", stderr=b"")
        if operation[:2] == ["image", "rm"]:
            raise subprocess.CalledProcessError(1, arguments, stderr=b"busy")
        raise AssertionError(operation)

    monkeypatch.setattr(package.subprocess, "run", fake_run)

    inspection = package.inspect_registry_image(
        image_input,
        build_push_receipt_fixture(image_input, reference),
        docker_binary=docker_binary,
        docker_host=docker_host,
        docker_config=docker_config,
    )

    assert inspection["status"] == "pass"
    assert state["present"] is True


def test_registry_inspection_has_no_post_inspection_cleanup_window(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    docker_binary, docker_host, docker_config = docker_authority(tmp_path)
    state = {"present": False, "inventory_calls": 0, "removed": False}
    current_config = "sha256:" + "3" * 64

    def fake_run(arguments: list[str], **_kwargs):
        nonlocal current_config
        operation = arguments[5:]
        if operation[:2] == ["image", "ls"]:
            state["inventory_calls"] += 1
            if state["inventory_calls"] == 3:
                current_config = "sha256:" + "4" * 64
            return subprocess.CompletedProcess(
                arguments,
                0,
                stdout=docker_inventory_fixture(
                    reference,
                    present=state["present"],
                    config_digest=current_config,
                ),
                stderr=b"",
            )
        if operation[:2] == ["image", "inspect"]:
            payload = json.dumps(
                [docker_inspection_fixture(image_input, reference, config_digest=current_config)]
            ).encode()
            return subprocess.CompletedProcess(arguments, 0, stdout=payload, stderr=b"")
        if operation[0] == "pull":
            state["present"] = True
            return subprocess.CompletedProcess(arguments, 0, stdout=b"pulled\n", stderr=b"")
        if operation[:2] == ["image", "rm"]:
            state["removed"] = True
            return subprocess.CompletedProcess(arguments, 0, stdout=b"removed\n", stderr=b"")
        raise AssertionError(operation)

    monkeypatch.setattr(package.subprocess, "run", fake_run)

    inspection = package.inspect_registry_image(
        image_input,
        build_push_receipt_fixture(image_input, reference),
        docker_binary=docker_binary,
        docker_host=docker_host,
        docker_config=docker_config,
    )

    assert inspection["status"] == "pass"
    assert state["present"] is True
    assert state["removed"] is False
    assert state["inventory_calls"] == 1


def test_registry_inspection_accepts_preexisting_reference_without_removal(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    docker_binary, docker_host, docker_config = docker_authority(tmp_path)
    operations: list[list[str]] = []
    payload = json.dumps([docker_inspection_fixture(image_input, reference)]).encode()

    def fake_run(arguments: list[str], **_kwargs):
        operation = arguments[5:]
        operations.append(operation)
        if operation[:2] == ["image", "ls"]:
            return subprocess.CompletedProcess(
                arguments,
                0,
                stdout=docker_inventory_fixture(reference, present=True),
                stderr=b"",
            )
        if operation[:2] == ["image", "inspect"]:
            return subprocess.CompletedProcess(arguments, 0, stdout=payload, stderr=b"")
        if operation[0] == "pull":
            return subprocess.CompletedProcess(arguments, 0, stdout=b"pulled\n", stderr=b"")
        if operation[:2] == ["image", "rm"]:
            raise AssertionError("digest reference must never be removed")
        raise AssertionError(operation)

    monkeypatch.setattr(package.subprocess, "run", fake_run)

    inspection = package.inspect_registry_image(
        image_input,
        build_push_receipt_fixture(image_input, reference),
        docker_binary=docker_binary,
        docker_host=docker_host,
        docker_config=docker_config,
    )

    assert inspection["status"] == "pass"
    assert any(operation and operation[0] == "pull" for operation in operations)
    assert not any(operation[:2] == ["image", "rm"] for operation in operations)


@pytest.mark.parametrize("mutation", ["revision", "input", "config"])
def test_registry_inspection_rejects_receipt_binding_drift_before_docker(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    mutation: str,
) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    docker_binary, docker_host, docker_config = docker_authority(tmp_path)
    monkeypatch.setattr(package.subprocess, "run", lambda *_args, **_kwargs: pytest.fail("Docker was contacted"))
    receipt = build_push_receipt_fixture(image_input, reference)
    if mutation == "revision":
        receipt["source_revision"] = "9" * 40
    elif mutation == "input":
        receipt["input_sha256"] = "9" * 64
    else:
        receipt["local_image"]["config_digest"] = "invalid"
    stable = {key: value for key, value in receipt.items() if key != "receipt_sha256"}
    receipt["receipt_sha256"] = package.canonical_sha256(stable)

    with pytest.raises(package.PackageError):
        package.inspect_registry_image(
            image_input,
            receipt,
            docker_binary=docker_binary,
            docker_host=docker_host,
            docker_config=docker_config,
        )


@pytest.mark.parametrize(
    ("binary", "host", "config_kind"),
    [
        (Path("docker"), "unix:///tmp/docker.sock", "directory"),
        (Path("/usr/bin/docker"), "tcp://127.0.0.1:2375", "directory"),
        (Path("/usr/bin/docker"), "unix:///tmp/docker.sock", "missing"),
    ],
)
def test_registry_inspection_rejects_implicit_or_remote_docker_authority(
    tmp_path: Path,
    binary: Path,
    host: str,
    config_kind: str,
) -> None:
    config = tmp_path / "docker-config"
    if config_kind == "directory":
        config.mkdir()

    with pytest.raises(package.PackageError, match="Docker"):
        package._validate_docker_authority(binary, host, config)


def test_registry_inspection_rejects_cache_lock_contention_before_docker(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    docker_binary, docker_host, docker_config = docker_authority(tmp_path)
    descriptor = os.open(docker_config, os.O_RDONLY)
    package.fcntl.flock(descriptor, package.fcntl.LOCK_EX | package.fcntl.LOCK_NB)
    monkeypatch.setattr(package.subprocess, "run", lambda *_args, **_kwargs: pytest.fail("Docker was contacted"))

    try:
        with pytest.raises(package.PackageError, match="inspection failed") as raised:
            package.inspect_registry_image(
                image_input,
                build_push_receipt_fixture(image_input, reference),
                docker_binary=docker_binary,
                docker_host=docker_host,
                docker_config=docker_config,
            )
        assert "cache authority" in str(raised.value.__cause__)
    finally:
        package.fcntl.flock(descriptor, package.fcntl.LOCK_UN)
        os.close(descriptor)


def test_finalize_failure_creates_no_output(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    capsys: pytest.CaptureFixture[str],
) -> None:
    image_input = image_input_fixture(tmp_path)
    image_input_path = tmp_path / "image-input.json"
    write_json(image_input_path, image_input, mode=0o644)
    receipt_path = tmp_path / "build-push-receipt.json"
    package.publish_build_push_receipt(
        receipt_path,
        build_push_receipt_fixture(image_input),
        image_input=image_input,
    )
    docker_binary, docker_host, docker_config = docker_authority(tmp_path)
    output_bundle = tmp_path / "nested" / "release-bundle"

    def fake_run(arguments: list[str], **_kwargs):
        operation = arguments[5:]
        if operation[:2] == ["image", "ls"]:
            return subprocess.CompletedProcess(arguments, 0, stdout=b"", stderr=b"")
        if operation[0] == "pull":
            raise subprocess.CalledProcessError(1, arguments, stderr=b"credential=value")
        raise AssertionError(operation)

    monkeypatch.setattr(package.subprocess, "run", fake_run)

    result = package.main(
        [
            "finalize",
            "--image-input",
            str(image_input_path),
            "--build-push-receipt",
            str(receipt_path),
            "--docker-binary",
            str(docker_binary),
            "--docker-host",
            docker_host,
            "--docker-config",
            str(docker_config),
            "--output-bundle",
            str(output_bundle),
        ]
    )

    assert result == 65
    assert not output_bundle.exists()
    error = capsys.readouterr().err
    assert "immutable registry image inspection failed" in error
    assert "credential=value" not in error


def test_finalize_embeds_inspection_before_rendering_unit(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    capsys: pytest.CaptureFixture[str],
) -> None:
    image_input = image_input_fixture(tmp_path)
    image_input_path = tmp_path / "image-input.json"
    write_json(image_input_path, image_input, mode=0o644)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    build_push_receipt = build_push_receipt_fixture(image_input, reference)
    receipt_path = tmp_path / "build-push-receipt.json"
    package.publish_build_push_receipt(
        receipt_path,
        build_push_receipt,
        image_input=image_input,
    )
    inspection = package.build_image_inspection(
        image_input, reference, docker_inspection_fixture(image_input, reference)
    )
    docker_binary, docker_host, docker_config = docker_authority(tmp_path)
    output_bundle = tmp_path / "release" / "release-bundle"
    output_bundle.parent.mkdir()
    observed: dict[str, object] = {}

    def capture_inspection(
        observed_input: dict[str, object],
        observed_receipt: dict[str, object],
        **kwargs: object,
    ) -> dict[str, object]:
        observed.update(
            {
                "image_input": observed_input,
                "build_push_receipt": observed_receipt,
                **kwargs,
            }
        )
        return inspection

    monkeypatch.setattr(package, "inspect_registry_image", capture_inspection)

    result = package.main(
        [
            "finalize",
            "--image-input",
            str(image_input_path),
            "--build-push-receipt",
            str(receipt_path),
            "--docker-binary",
            str(docker_binary),
            "--docker-host",
            docker_host,
            "--docker-config",
            str(docker_config),
            "--output-bundle",
            str(output_bundle),
        ]
    )

    assert result == 0
    release = json.loads((output_bundle / "release.json").read_text(encoding="utf-8"))
    assert release["schema"] == package.RELEASE_SCHEMA
    assert release["inspection"] == inspection
    assert reference in (output_bundle / package.SERVICE).read_text(encoding="utf-8")
    assert observed == {
        "image_input": image_input,
        "build_push_receipt": build_push_receipt,
        "docker_binary": docker_binary,
        "docker_host": docker_host,
        "docker_config": docker_config,
    }
    assert json.loads(capsys.readouterr().out) == {"schema": package.RELEASE_SCHEMA, "status": "pass"}


@pytest.mark.parametrize("failure", ["unit_write", "rename"])
def test_finalize_bundle_failure_leaves_no_partial_output_or_temporary_residue(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    failure: str,
) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    inspection = package.build_image_inspection(
        image_input, reference, docker_inspection_fixture(image_input, reference)
    )
    release = package.build_release_descriptor(
        image_input, inspection, build_push_receipt_fixture(image_input, inspection["reference"])
    )
    output_bundle = tmp_path / "final" / "bundle"
    output_bundle.parent.mkdir()
    if failure == "unit_write":
        original_write_text = Path.write_text

        def fail_unit_write(path: Path, *args, **kwargs):
            if path.name == package.SERVICE:
                raise OSError("injected unit write failure")
            return original_write_text(path, *args, **kwargs)

        monkeypatch.setattr(Path, "write_text", fail_unit_write)
    else:
        monkeypatch.setattr(
            package,
            "_rename_noreplace",
            lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("injected")),
        )

    with pytest.raises(package.PackageError, match="finalized bundle"):
        package.publish_finalize_bundle(output_bundle, release)

    assert not output_bundle.exists()
    assert not list(output_bundle.parent.glob(".*.tmp"))


def test_finalize_bundle_is_byte_repeatable(tmp_path: Path) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    inspection = package.build_image_inspection(
        image_input, reference, docker_inspection_fixture(image_input, reference)
    )
    release = package.build_release_descriptor(
        image_input, inspection, build_push_receipt_fixture(image_input, inspection["reference"])
    )
    first = tmp_path / "first" / "bundle"
    second = tmp_path / "second" / "bundle"
    first.parent.mkdir()
    second.parent.mkdir()

    package.publish_finalize_bundle(first, release)
    package.publish_finalize_bundle(second, release)

    for name in ("release.json", package.SERVICE):
        assert (first / name).read_bytes() == (second / name).read_bytes()
        assert (first / name).stat().st_mode & 0o777 == 0o644
    assert first.stat().st_mode & 0o777 == second.stat().st_mode & 0o777 == 0o755


@pytest.mark.parametrize("destination_kind", ["file", "directory", "symlink"])
def test_finalize_bundle_preserves_preexisting_destination(tmp_path: Path, destination_kind: str) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    inspection = package.build_image_inspection(
        image_input, reference, docker_inspection_fixture(image_input, reference)
    )
    release = package.build_release_descriptor(
        image_input, inspection, build_push_receipt_fixture(image_input, inspection["reference"])
    )
    destination = tmp_path / "bundle"
    marker = tmp_path / "marker"
    marker.write_text("preserve\n", encoding="utf-8")
    if destination_kind == "file":
        destination.write_text("preserve\n", encoding="utf-8")
    elif destination_kind == "directory":
        destination.mkdir()
        (destination / "marker").write_text("preserve\n", encoding="utf-8")
    else:
        destination.symlink_to(marker)
    before = destination.lstat()

    with pytest.raises(package.PackageError, match="already exists"):
        package.publish_finalize_bundle(destination, release)

    after = destination.lstat()
    assert (after.st_dev, after.st_ino) == (before.st_dev, before.st_ino)
    if destination_kind == "file":
        assert destination.read_text(encoding="utf-8") == "preserve\n"
    elif destination_kind == "directory":
        assert (destination / "marker").read_text(encoding="utf-8") == "preserve\n"
    else:
        assert destination.is_symlink() and destination.readlink() == marker


def test_finalize_bundle_requires_preexisting_output_parent(tmp_path: Path) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    inspection = package.build_image_inspection(
        image_input, reference, docker_inspection_fixture(image_input, reference)
    )
    release = package.build_release_descriptor(
        image_input, inspection, build_push_receipt_fixture(image_input, inspection["reference"])
    )
    destination = tmp_path / "missing" / "bundle"

    with pytest.raises(package.PackageError, match="existing safe absolute directory"):
        package.publish_finalize_bundle(destination, release)

    assert not destination.parent.exists()


def test_finalize_bundle_race_cannot_replace_destination(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    inspection = package.build_image_inspection(
        image_input, reference, docker_inspection_fixture(image_input, reference)
    )
    release = package.build_release_descriptor(
        image_input, inspection, build_push_receipt_fixture(image_input, inspection["reference"])
    )
    destination = tmp_path / "bundle"
    original = package._rename_noreplace
    raced_identity: tuple[int, int] | None = None

    def race(source: Path, output: Path) -> None:
        nonlocal raced_identity
        output.mkdir()
        observed = output.stat(follow_symlinks=False)
        raced_identity = (observed.st_dev, observed.st_ino)
        original(source, output)

    monkeypatch.setattr(package, "_rename_noreplace", race)

    with pytest.raises(package.PackageError, match="cannot publish"):
        package.publish_finalize_bundle(destination, release)

    observed = destination.stat(follow_symlinks=False)
    assert (observed.st_dev, observed.st_ino) == raced_identity
    assert not list(destination.iterdir())
    assert not list(tmp_path.glob(".*.tmp"))


@pytest.mark.parametrize("failure_call", [1, 2, 3, 4])
def test_finalize_bundle_rolls_back_and_fsyncs_parent_after_each_durability_failure(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    failure_call: int,
) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    inspection = package.build_image_inspection(
        image_input, reference, docker_inspection_fixture(image_input, reference)
    )
    release = package.build_release_descriptor(
        image_input, inspection, build_push_receipt_fixture(image_input, inspection["reference"])
    )
    destination = tmp_path / "final" / "bundle"
    destination.parent.mkdir()
    original = package._fsync_path
    calls: list[tuple[Path, bool]] = []

    def fail_once(path: Path, *, directory: bool = False) -> None:
        calls.append((path, directory))
        if len(calls) == failure_call:
            raise OSError("injected durability failure")
        original(path, directory=directory)

    monkeypatch.setattr(package, "_fsync_path", fail_once)

    with pytest.raises(package.PackageError, match="cannot publish"):
        package.publish_finalize_bundle(destination, release)

    assert not destination.exists()
    assert not list(destination.parent.glob(".*.tmp"))
    if failure_call == 4:
        assert len(calls) == 5
        assert calls[-1] == (destination.parent, True)


def test_finalize_bundle_reports_rollback_fsync_failure(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    inspection = package.build_image_inspection(
        image_input, reference, docker_inspection_fixture(image_input, reference)
    )
    release = package.build_release_descriptor(
        image_input, inspection, build_push_receipt_fixture(image_input, inspection["reference"])
    )
    destination = tmp_path / "final" / "bundle"
    destination.parent.mkdir()
    original = package._fsync_path
    calls = 0

    def fail_parent(path: Path, *, directory: bool = False) -> None:
        nonlocal calls
        calls += 1
        if path == destination.parent and directory:
            raise OSError("injected parent fsync failure")
        original(path, directory=directory)

    monkeypatch.setattr(package, "_fsync_path", fail_parent)

    with pytest.raises(package.PackageError, match="rollback failed"):
        package.publish_finalize_bundle(destination, release)

    assert calls == 5
    assert not destination.exists()


def test_finalize_bundle_surfaces_temporary_cleanup_failure(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    inspection = package.build_image_inspection(
        image_input, reference, docker_inspection_fixture(image_input, reference)
    )
    release = package.build_release_descriptor(
        image_input, inspection, build_push_receipt_fixture(image_input, inspection["reference"])
    )
    output_bundle = tmp_path / "final" / "bundle"
    output_bundle.parent.mkdir()
    original_write_text = Path.write_text

    def fail_unit_write(path: Path, *args, **kwargs):
        if path.name == package.SERVICE:
            raise OSError("injected unit write failure")
        return original_write_text(path, *args, **kwargs)

    monkeypatch.setattr(Path, "write_text", fail_unit_write)
    monkeypatch.setattr(package.shutil, "rmtree", lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("injected")))

    with pytest.raises(package.PackageError, match="temporary cleanup failed"):
        package.publish_finalize_bundle(output_bundle, release)

    assert not output_bundle.exists()
    assert list(output_bundle.parent.glob(".*.tmp"))


@pytest.mark.parametrize(
    "schema",
    ["livingip.leocleanNoSendRelease.v1", "livingip.leocleanNoSendRelease.v2"],
)
def test_older_release_schemas_are_rejected_without_compatibility_shim(tmp_path: Path, schema: str) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    inspection = package.build_image_inspection(
        image_input, reference, docker_inspection_fixture(image_input, reference)
    )
    release = package.build_release_descriptor(
        image_input, inspection, build_push_receipt_fixture(image_input, inspection["reference"])
    )
    release["schema"] = schema

    with pytest.raises(package.PackageError, match="schema"):
        package.validate_release_descriptor(release)


def test_rendered_unit_is_isolated_digest_pinned_and_has_no_credential_or_transport_surface(tmp_path: Path) -> None:
    image_input = image_input_fixture(tmp_path)
    reference = f"{package.IMAGE_REPOSITORY}@sha256:{IMAGE_DIGEST}"
    inspection = package.build_image_inspection(
        image_input, reference, docker_inspection_fixture(image_input, reference)
    )
    release = package.build_release_descriptor(
        image_input, inspection, build_push_receipt_fixture(image_input, inspection["reference"])
    )
    unit = package.render_systemd_unit(release)

    assert "leoclean-gcp-nosend" in unit
    assert f"@sha256:{IMAGE_DIGEST}" in unit
    assert "--pull=never" in unit
    assert "--read-only" in unit
    assert "--cap-drop=ALL" in unit
    assert "--cap-add=CHOWN" in unit
    assert "--cap-add=SETGID" in unit
    assert "--cap-add=SETUID" in unit
    assert "--cap-add=SETPCAP" in unit
    assert "--network=bridge" in unit
    assert "--tmpfs=/run/leoclean-profile:" in unit
    assert "--tmpfs=/var/lib/postgresql/data:ro,noexec,nosuid,nodev,size=64k,uid=0,gid=0,mode=0000" in unit
    assert "NoNewPrivileges=yes" in unit
    assert "EnvironmentFile" not in unit
    assert "LoadCredential" not in unit
    assert f"--host={package.DOCKER_HOST}" in unit
    assert f"--config={package.SERVICE_DOCKER_CONFIG}" in unit
    assert f"--cidfile={package.SERVICE_CIDFILE}" in unit
    assert package.SERVICE_CONTROL_HELPER in unit
    assert "docker rm" not in unit
    assert "--force" not in unit
    assert "--volume=/var/run/docker.sock" not in unit
    assert "--publish" not in unit
    assert "teleo-prod-1" not in unit
    assert "77.42.65.182" not in unit
    assert "telegram" not in unit.casefold()


def test_prepare_context_is_secret_free_normalized_and_idempotently_refuses_overwrite(tmp_path: Path) -> None:
    artifact, receipt = artifact_fixture(tmp_path)
    identity, _contract = identity_fixture(tmp_path, artifact, receipt)
    output = tmp_path / "context"
    result = package.prepare_context(
        repo_root=ROOT,
        artifact=artifact,
        receipt=receipt,
        identity=identity,
        identity_source_root=ROOT,
        output=output,
        source_revision=REVISION,
    )

    package.validate_image_input(result)
    assert sorted(path.name for path in output.iterdir()) == [
        "Dockerfile",
        "artifact",
        "artifact-receipt.json",
        "build-args.json",
        "cloudsql-server-ca.pem",
        "entrypoint.py",
        "identity",
        "identity-sources",
        "image-input.json",
        "package_contract.py",
        "scripts",
    ]
    assert all((output / "identity" / name).stat().st_mode & 0o777 == 0o600 for name in package.IDENTITY_FILES)
    encoded = (output / "image-input.json").read_text(encoding="utf-8").casefold()
    assert "telegram_bot_token" not in encoded
    assert "postgres-password" not in encoded

    with pytest.raises(package.PackageError, match="already exists"):
        package.prepare_context(
            repo_root=ROOT,
            artifact=artifact,
            receipt=receipt,
            identity=identity,
            identity_source_root=ROOT,
            output=output,
            source_revision=REVISION,
        )


def test_dockerfile_has_only_digest_pinned_bases_and_no_floating_package_install() -> None:
    dockerfile = (ROOT / "docker" / "leoclean-nosend" / "Dockerfile").read_text(encoding="utf-8")
    from_lines = [line for line in dockerfile.splitlines() if line.startswith("FROM ")]

    assert len(from_lines) == 3
    assert all("@sha256:" in line for line in from_lines)
    assert "# syntax=" not in dockerfile
    assert "postgres:16.14-bookworm@sha256:92620d" in dockerfile
    assert "python:3.11.9-slim-bookworm@sha256:8fb099" in dockerfile
    assert "uv:0.9.30@sha256:538e0b" in dockerfile
    assert "apt-get" not in dockerfile
    assert "pip install" not in dockerfile
    assert "uv sync" in dockerfile
    assert "--frozen" in dockerfile
    assert "--no-editable" in dockerfile
    assert "HEALTHCHECK" in dockerfile
    assert "verify-build" in dockerfile
    assert "identity -type f -exec chmod 0444" in dockerfile


def test_pid_one_command_line_accepts_only_native_and_exact_x86_64_binfmt_wrappers() -> None:
    native = [b"/runtime/python", b"/runtime/bootstrap.py", b"service", b""]
    binfmt = [native[0], *native]

    assert package.pid_one_command_line_is_exact(native, native)
    assert package.pid_one_command_line_is_exact(binfmt, native)
    assert package.pid_one_command_line_is_exact([b"/usr/bin/qemu-x86_64", *binfmt], native)
    assert package.pid_one_command_line_is_exact([b"/usr/bin/qemu-x86_64-static", *binfmt], native)

    assert not package.pid_one_command_line_is_exact([b"/usr/bin/qemu-aarch64", *binfmt], native)
    assert not package.pid_one_command_line_is_exact([b"/usr/bin/qemu-x86_64", *native], native)
    assert not package.pid_one_command_line_is_exact([b"/usr/bin/qemu-x86_64", *binfmt, b"--extra"], native)
    assert not package.pid_one_command_line_is_exact(native, native[:-1])


def test_entrypoint_prepares_exact_profile_drops_privileges_and_health_checks_pid_one() -> None:
    source = (ROOT / "docker" / "leoclean-nosend" / "entrypoint.py").read_text(encoding="utf-8")

    assert "os.setgroups([])" in source
    assert "os.setgid(contract.RUNTIME_GID)" in source
    assert "os.setuid(contract.RUNTIME_UID)" in source
    assert "pr_capbset_drop = 24" in source
    assert "reverse=True" in source
    assert source.index("os.chown(runtime_identity, 0, 0)") < source.index(
        "os.chown(PROFILE, contract.RUNTIME_UID, contract.RUNTIME_GID)"
    )
    assert 'value.get("process_id") == 1' in source
    assert "contract.pid_one_command_line_is_exact(cmdline, expected_native)" in source
    assert "contract.NO_SEND_TOOL_SURFACE" in source
    assert 'image_input["runtime"]["config_sha256"]' in source
    assert "gateway_adapter_count" in source
    assert "os.execve" in source
    assert "OPENROUTER_API_KEY" not in source
    assert "TELEGRAM_BOT_TOKEN" not in source


def test_entrypoint_readiness_binds_packaged_and_live_config(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    entrypoint = load_entrypoint_module(tmp_path)
    profile = tmp_path / "profile"
    profile.mkdir()
    config = profile / "config.yaml"
    config.write_text("model: reviewed\n", encoding="utf-8")
    expected = entrypoint.contract.sha256_file(config)
    image_input = {"runtime": {"config_sha256": expected}}
    readiness = {
        "config_sha256": expected,
        "tool_surface": copy.deepcopy(entrypoint.contract.NO_SEND_TOOL_SURFACE),
    }
    monkeypatch.setattr(entrypoint, "PROFILE", profile)

    entrypoint._validate_readiness_behavior(image_input, readiness)

    wrong_receipt = copy.deepcopy(readiness)
    wrong_receipt["config_sha256"] = "9" * 64
    with pytest.raises(entrypoint.EntrypointError, match="readiness config"):
        entrypoint._validate_readiness_behavior(image_input, wrong_receipt)

    config.write_text("model: drifted\n", encoding="utf-8")
    with pytest.raises(entrypoint.EntrypointError, match="profile config drifted"):
        entrypoint._validate_readiness_behavior(image_input, readiness)

    drift_hash = entrypoint.contract.sha256_file(config)
    both_drifted = {**readiness, "config_sha256": drift_hash}
    with pytest.raises(entrypoint.EntrypointError, match="readiness config"):
        entrypoint._validate_readiness_behavior(image_input, both_drifted)

    config.unlink()
    with pytest.raises(entrypoint.EntrypointError, match="missing or unsafe"):
        entrypoint._validate_readiness_behavior(image_input, readiness)


@pytest.mark.parametrize(
    ("mutation", "value"),
    [
        ("extra", True),
        ("registry_sealed", False),
        ("send_message_present", True),
        ("terminal_handler", "unrestricted"),
        ("terminal_restricted_to", "shell"),
        ("toolset", "default"),
        ("tools", ["skill_view", "skills_list", "terminal", "send_message"]),
        ("plugins", []),
    ],
)
def test_entrypoint_readiness_rejects_exact_tool_surface_drift(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    mutation: str,
    value: object,
) -> None:
    entrypoint = load_entrypoint_module(tmp_path)
    profile = tmp_path / "profile"
    profile.mkdir()
    config = profile / "config.yaml"
    config.write_text("model: reviewed\n", encoding="utf-8")
    expected = entrypoint.contract.sha256_file(config)
    image_input = {"runtime": {"config_sha256": expected}}
    surface = copy.deepcopy(entrypoint.contract.NO_SEND_TOOL_SURFACE)
    surface[mutation] = value
    readiness = {"config_sha256": expected, "tool_surface": surface}
    monkeypatch.setattr(entrypoint, "PROFILE", profile)

    with pytest.raises(entrypoint.EntrypointError, match="tool surface drifted"):
        entrypoint._validate_readiness_behavior(image_input, readiness)


def test_entrypoint_readiness_rejects_plugin_field_drift(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    entrypoint = load_entrypoint_module(tmp_path)
    profile = tmp_path / "profile"
    profile.mkdir()
    config = profile / "config.yaml"
    config.write_text("model: reviewed\n", encoding="utf-8")
    expected = entrypoint.contract.sha256_file(config)
    monkeypatch.setattr(entrypoint, "PROFILE", profile)

    for field, value in (
        ("source", "project"),
        ("tools", ["send_message"]),
        ("hooks", ["pre_llm_call"]),
        ("enabled", False),
        ("error", "load failed"),
    ):
        surface = copy.deepcopy(entrypoint.contract.NO_SEND_TOOL_SURFACE)
        surface["plugins"][0][field] = value
        readiness = {"config_sha256": expected, "tool_surface": surface}
        with pytest.raises(entrypoint.EntrypointError, match="tool surface drifted"):
            entrypoint._validate_readiness_behavior(
                {"runtime": {"config_sha256": expected}},
                readiness,
            )


def test_entrypoint_loads_adjacent_contract_with_python_safe_path(tmp_path: Path) -> None:
    image_root = tmp_path / "image"
    image_root.mkdir()
    entrypoint = image_root / "entrypoint.py"
    shutil.copy2(ROOT / "docker" / "leoclean-nosend" / "entrypoint.py", entrypoint)
    shutil.copy2(ROOT / "ops" / "gcp_leoclean_nosend_package.py", image_root / "package_contract.py")
    scripts = image_root / "scripts"
    scripts.mkdir()
    for relative in package.IDENTITY_CONTRACT_SOURCES:
        shutil.copy2(ROOT / relative, image_root / relative)
    environment = {**os.environ, "PYTHONSAFEPATH": "1"}

    result = subprocess.run(
        [sys.executable, "-P", str(entrypoint), "verify"],
        cwd=tmp_path,
        env=environment,
        check=False,
        capture_output=True,
        text=True,
        timeout=10,
    )

    assert result.returncode == 65
    assert json.loads(result.stderr) == {"error": "container_contract_failed", "status": "fail"}
    assert "Traceback" not in result.stderr