#!/usr/bin/env python3
"""Materialize and verify deletion-safe Chronos transfer evidence."""

from __future__ import annotations

import argparse
import csv
import hashlib
import io
import json
import os
import re
import subprocess
import sys
import tempfile
from collections import Counter
from pathlib import Path, PurePosixPath
from typing import Iterable


SOURCE_REVISION = "3155be0e275437a8eedc1aa93497955a7d30465b"
STATE_LOGICAL_IDS_SHA256 = "9bce499e286514baeb7ad9bc7c59119a3b5725f69763706dfeba06569d7404b9"

MANIFEST_HEADER = [
    "path",
    "kind",
    "disposition",
    "bundle_path",
    "iop_action",
    "neutral_successor",
    "evidence_map",
]
DELTA_HEADER = [
    "path",
    "original_disposition",
    "effective_disposition",
    "iop_action",
    "neutral_successor",
    "rationale",
]
MATRIX_HEADER = [
    "path",
    "kind",
    "source_manifest",
    "source_disposition",
    "source_bundle_path",
    "source_iop_action",
    "source_neutral_successor",
    "source_evidence_map",
    "effective_disposition",
    "effective_bundle_path",
    "effective_iop_action",
    "effective_neutral_successor",
    "effective_evidence_map",
    "decision_source",
    "decision_rationale",
]

SOURCE_FILES = {
    "manifest": "scripts/fixtures/iop-agent-chronos-ownership-manifest.tsv",
    "delta": "scripts/fixtures/iop-agent-chronos-boundary-disposition-delta-v1.tsv",
    "addendum": "scripts/fixtures/iop-agent-chronos-boundary-surface-addendum-v1.tsv",
    "state_schema": "scripts/fixtures/iop-agent-chronos-transfer-state-v1.schema.json",
    "state_fixture": "scripts/fixtures/iop-agent-chronos-transfer-state-v1.json",
    "stable_receipt": "scripts/fixtures/iop-agent-chronos-transfer-bundle-receipt-v1.json",
}
LOCAL_FILES = {
    "manifest": "iop-agent-chronos-ownership-manifest.tsv",
    "delta": "boundary-disposition-delta-v1.tsv",
    "addendum": "boundary-surface-addendum-v1.tsv",
    "state_schema": "iop-agent-chronos-transfer-state-v1.schema.json",
    "state_fixture": "iop-agent-chronos-transfer-state-v1.json",
    "stable_receipt": "iop-agent-chronos-transfer-bundle-receipt-v1.json",
    "matrix": "effective-disposition-matrix-v1.tsv",
}
LOCAL_SIDECARS = {
    "manifest": "iop-agent-chronos-ownership-manifest.sha256",
    "delta": "boundary-disposition-delta-v1.sha256",
    "addendum": "boundary-surface-addendum-v1.sha256",
    "state_schema": "iop-agent-chronos-transfer-state-v1.schema.json.sha256",
    "state_fixture": "iop-agent-chronos-transfer-state-v1.json.sha256",
    "stable_receipt": "iop-agent-chronos-transfer-bundle-receipt-v1.json.sha256",
    "matrix": "effective-disposition-matrix-v1.sha256",
}
EXPECTED_DIGESTS = {
    "manifest": "d8598134dee99d96fc05a045091dd3d8932d4708d2c980c2a78c8b11825d2ccf",
    "delta": "c807d2aa87c0ffb54c3c43bb244be91926206b6a7a40ae8abf9d4d39b9fa667f",
    "addendum": "c678d415daf01a3f68260a3912ffc0b4596f6b03d4f20a38cf796deb9623b670",
    "state_schema": "e509b541b26b54401a33a3a1ae0ea8b8581933d64d9f1465bb714969c7eb63d4",
    "state_fixture": "8a71dcc22cee5ed9990cb3a204ca8bc3bc8ffb949618a8a8bb6664a500016148",
    "stable_receipt": "dab36aac1e3876b7b6a80d0b26a8e89ed59acd06d578db825e0cd79c290625de",
}
SOURCE_SIDECARS = {
    "delta": "scripts/fixtures/iop-agent-chronos-boundary-disposition-delta-v1.sha256",
    "addendum": "scripts/fixtures/iop-agent-chronos-boundary-surface-addendum-v1.sha256",
}
DELTA_EXTRA_PATHS = {
    "agent-contract/inner/iop-agent-cli-runtime.md",
    "agent-spec/runtime/iop-agent-cli-runtime.md",
}

PREDECESSORS = {
    "task03": {
        "complete": "agent-task/archive/2026/08/m-iop-agent-chronos-extraction-decoupling/03+02_inventory_closure/complete.log",
        "review": "code_review_cloud_G04_1.log",
        "complete_sha256": "7657b23f2ea2c0efb045677aac611c35d05ac890b0f4dc346857db6ec30229b8",
        "review_sha256": "646bc7331ba94711974b11ae59a95198924c343707161b820d9bf0cc252051be",
    },
    "task06": {
        "complete": "agent-task/archive/2026/08/m-iop-agent-chronos-extraction-decoupling/06+04,05_bundle_verification/complete.log",
        "review": "code_review_cloud_G07_2.log",
        "complete_sha256": "df0956c99765e4e9a6cdcf96d52568adc2cb08c6b7a0d34ec2c07527094a2836",
        "review_sha256": "c64851c73ec93897e091d4442cb421eebc2d9b9aee32f7f51916166124aeca91",
    },
    "task09": {
        "complete": "agent-task/archive/2026/08/m-iop-agent-chronos-extraction-decoupling/09+08_remove_agent_ui_close/complete.log",
        "review": "code_review_cloud_G05_3.log",
        "complete_sha256": "580fb464b6c0a805a3a9c836ad29328fdc76a9b79cfa1e38660d8f63b26612a6",
        "review_sha256": "923af26de17695fbf331199d5f9640638ed5a218a51646b56efcab6057a6e8dc",
    },
    "task10": {
        "complete": "agent-task/archive/2026/08/m-iop-agent-chronos-extraction-decoupling/10+09_neutral_execution/complete.log",
        "review": "code_review_cloud_G08_7.log",
        "complete_sha256": "2f9a2583181ec2fd8d972bfa6a2d1cc33e25e8c8e92f32862f553a07b04e5fbe",
        "review_sha256": "a36348ff24b97571cf7d33d464b3f782b54fd53a2427c84dc566d9a785b214fd",
    },
    "task16": {
        "complete": "agent-task/archive/2026/08/m-iop-agent-chronos-extraction-decoupling/16_manifest_baseline_repair/complete.log",
        "review": "code_review_cloud_G04_0.log",
        "complete_sha256": "29efd7ee432f7dd2eb57eacc7f5e842fa988ef042be9df9f4fba932fb63ae274",
        "review_sha256": "f500c314a7ad7afcd47f587155c692845c7e523203938e13d47effc07f421221",
    },
}
IMPORT_AUDIT_PATH = (
    "agent-task/archive/2026/08/m-iop-agent-chronos-extraction-decoupling/"
    "10+09_neutral_execution/node-boundary-audit.log"
)
IMPORT_AUDIT_SHA256 = "462bc1566b37327470d1faa23b1adf7311a5ede12d43364cd0f379bf229db23e"

RECEIPT_KEYS = (
    "receipt_version",
    "receipt_status",
    "artifact_language",
    "source_revision",
    "rollback_revision",
    "artifact_root",
    "manifest_source_path",
    "manifest_path",
    "manifest_sha256_path",
    "manifest_sha256",
    "manifest_row_count",
    "manifest_file_count",
    "manifest_state_count",
    "manifest_retain_generic_count",
    "delta_source_path",
    "delta_path",
    "delta_sha256_path",
    "delta_sha256",
    "delta_row_count",
    "delta_original_retain_generic_count",
    "delta_original_transfer_count",
    "delta_effective_retain_generic_count",
    "delta_effective_transfer_count",
    "addendum_source_path",
    "addendum_path",
    "addendum_sha256_path",
    "addendum_sha256",
    "addendum_row_count",
    "addendum_file_count",
    "addendum_state_count",
    "effective_matrix_path",
    "effective_matrix_sha256_path",
    "effective_matrix_sha256",
    "effective_matrix_row_count",
    "effective_matrix_delta_decision_count",
    "effective_matrix_addendum_decision_count",
    "effective_matrix_unclassified_count",
    "manifest_delta_overlap_count",
    "manifest_addendum_overlap_count",
    "delta_addendum_overlap_count",
    "state_schema_source_path",
    "state_schema_path",
    "state_schema_sha256_path",
    "state_schema_sha256",
    "state_fixture_source_path",
    "state_fixture_path",
    "state_fixture_sha256_path",
    "state_fixture_sha256",
    "state_original_manifest_logical_id_count",
    "state_addendum_logical_id_count",
    "state_logical_id_count",
    "state_logical_ids_sha256",
    "state_record_count",
    "state_category_count",
    "state_quarantine_count",
    "stable_receipt_source_path",
    "stable_receipt_path",
    "stable_receipt_sha256_path",
    "stable_receipt_sha256",
    "owner_transfer_evidence",
    "task06_owner_state_record_count",
    "task06_owner_state_category_count",
    "task06_owner_quarantine_count",
    "task06_owner_logical_id_count",
    "task06_owner_logical_ids_sha256",
    "task03_complete_log_path",
    "task03_complete_log_sha256",
    "task03_review_path",
    "task03_review_sha256",
    "task03_historical_row_count",
    "task03_historical_file_count",
    "task03_historical_state_count",
    "task03_universe_residuals",
    "task03_duplicate_rows",
    "task03_pass",
    "task06_complete_log_path",
    "task06_complete_log_sha256",
    "task06_review_path",
    "task06_review_sha256",
    "task06_pass",
    "task09_complete_log_path",
    "task09_complete_log_sha256",
    "task09_review_path",
    "task09_review_sha256",
    "task09_pass",
    "task10_complete_log_path",
    "task10_complete_log_sha256",
    "task10_review_path",
    "task10_review_sha256",
    "task10_pass",
    "task10_import_audit_path",
    "task10_import_audit_sha256",
    "import_graph_residuals",
    "task16_complete_log_path",
    "task16_complete_log_sha256",
    "task16_review_path",
    "task16_review_sha256",
    "task16_pass",
    "task16_current_row_count",
    "task16_current_file_count",
    "task16_current_state_count",
    "task16_current_retain_generic_count",
    "raw_owner_bundle_required",
    "external_inputs",
    "external_mutations",
    "receipt_path_leaks",
    "iop_node_owner",
    "chronos_connection_surfaces",
    "canonical_promotion",
    "downstream_lock",
    "staging_lock_identity",
    "staging_lock_result",
    "canonical_lock_identity",
    "canonical_lock_result",
)

RECEIPT_BODY = """# Pre-deletion Transfer Receipt v1

This receipt pins the current disposition inputs, the effective matrix, the redacted owner-transfer proof, and exact predecessor verdict evidence before live migration fixtures are deleted.

The receipt is evidence-only. Canonical promotion and downstream lock synchronization remain pending.
"""

AUDIT_KEYS = (
    "schema",
    "audit_status",
    "task03_pass",
    "task06_pass",
    "task09_pass",
    "task10_pass",
    "task16_pass",
    "manifest_digest_match",
    "delta_digest_match",
    "addendum_digest_match",
    "state_schema_digest_match",
    "state_fixture_digest_match",
    "stable_receipt_digest_match",
    "effective_matrix_digest_match",
    "manifest_rows",
    "manifest_retain_generic",
    "delta_rows",
    "addendum_rows",
    "effective_matrix_rows",
    "effective_matrix_unclassified",
    "manifest_delta_overlap",
    "manifest_addendum_overlap",
    "delta_addendum_overlap",
    "state_records",
    "state_categories",
    "state_quarantine",
    "import_graph_residuals",
    "raw_owner_bundle_required",
    "external_inputs",
    "external_mutations",
    "receipt_path_leaks",
    "canonical_promotion",
    "check_live",
    "check_archived",
)


class VerificationError(RuntimeError):
    """A deterministic receipt assertion failed."""


def require(condition: bool, message: str) -> None:
    if not condition:
        raise VerificationError(message)


def sha256_bytes(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def sha256(path: Path) -> str:
    return sha256_bytes(path.read_bytes())


def bytewise_sorted(values: Iterable[str]) -> list[str]:
    return sorted(values, key=lambda value: value.encode("utf-8"))


def atomic_write(path: Path, data: bytes, mode: int = 0o644) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    descriptor, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
    temporary = Path(temporary_name)
    try:
        with os.fdopen(descriptor, "wb") as stream:
            stream.write(data)
            stream.flush()
            os.fsync(stream.fileno())
        os.chmod(temporary, mode)
        os.replace(temporary, path)
    finally:
        if temporary.exists():
            temporary.unlink()


def read_tsv(path: Path, header: list[str]) -> list[dict[str, str]]:
    data = path.read_bytes()
    require(data.endswith(b"\n"), f"{path.name}: missing final newline")
    require(b"\r" not in data, f"{path.name}: CR bytes are not canonical")
    text = data.decode("utf-8")
    lines = text.splitlines()
    require(bool(lines), f"{path.name}: empty TSV")
    require(lines[0].split("\t") == header, f"{path.name}: unexpected header")
    require(all(line and not line.endswith("\t") for line in lines), f"{path.name}: blank line or trailing tab")
    rows = list(csv.DictReader(lines, delimiter="\t"))
    require(
        all(None not in row and set(row) == set(header) and all(value != "" for value in row.values()) for row in rows),
        f"{path.name}: malformed row",
    )
    paths = [row["path"] for row in rows]
    require(paths == bytewise_sorted(paths), f"{path.name}: paths are not bytewise sorted")
    require(len(paths) == len(set(paths)), f"{path.name}: duplicate path")
    return rows


def render_tsv(header: list[str], rows: list[dict[str, str]]) -> bytes:
    output = io.StringIO(newline="")
    writer = csv.DictWriter(output, fieldnames=header, delimiter="\t", lineterminator="\n", extrasaction="raise")
    writer.writeheader()
    writer.writerows(rows)
    return output.getvalue().encode("utf-8")


def verify_digest(path: Path, expected: str) -> None:
    require(re.fullmatch(r"[0-9a-f]{64}", expected) is not None, "invalid expected digest")
    actual = sha256(path)
    require(actual == expected, f"{path.name}: sha256 {actual} != expected {expected}")


def sidecar_bytes(artifact_name: str, digest: str) -> bytes:
    return f"{digest}  {artifact_name}\n".encode("ascii")


def verify_sidecar(artifact: Path, sidecar: Path) -> str:
    data = sidecar.read_bytes()
    require(data.endswith(b"\n") and data.count(b"\n") == 1, f"{sidecar.name}: expected one line")
    match = re.fullmatch(rb"([0-9a-f]{64})  ([^/\s]+)\n", data)
    require(match is not None, f"{sidecar.name}: malformed canonical sidecar")
    declared_digest = match.group(1).decode("ascii")
    declared_name = match.group(2).decode("utf-8")
    require(declared_name == artifact.name, f"{sidecar.name}: sidecar must name the local basename")
    actual = sha256(artifact)
    require(declared_digest == actual, f"{sidecar.name}: digest mismatch")
    return actual


def verify_source_sidecar(root: Path, key: str) -> None:
    artifact_rel = SOURCE_FILES[key]
    artifact = root / artifact_rel
    sidecar = root / SOURCE_SIDECARS[key]
    expected = f"{EXPECTED_DIGESTS[key]}  {artifact_rel}\n".encode("ascii")
    require(sidecar.read_bytes() == expected, f"{sidecar.name}: source sidecar contract mismatch")
    verify_digest(artifact, EXPECTED_DIGESTS[key])


def validate_manifest_delta_addendum(
    manifest_path: Path, delta_path: Path, addendum_path: Path
) -> tuple[list[dict[str, str]], list[dict[str, str]], list[dict[str, str]], dict[str, int]]:
    manifest = read_tsv(manifest_path, MANIFEST_HEADER)
    delta = read_tsv(delta_path, DELTA_HEADER)
    addendum = read_tsv(addendum_path, MANIFEST_HEADER)

    manifest_paths = {row["path"] for row in manifest}
    delta_paths = {row["path"] for row in delta}
    addendum_paths = {row["path"] for row in addendum}
    file_count = sum(row["kind"] != "state" for row in manifest)
    state_count = sum(row["kind"] == "state" for row in manifest)
    retain_count = sum(row["disposition"] == "retain-generic" for row in manifest)
    require((len(manifest), file_count, state_count, retain_count) == (300, 290, 10, 135), "manifest counts mismatch")

    for row in manifest:
        is_state = row["path"].startswith("state:")
        require(is_state == (row["kind"] == "state"), f"manifest state-kind mismatch: {row['path']}")
        require(row["disposition"] in {"transfer", "retain-generic", "reference"}, f"manifest disposition mismatch: {row['path']}")
        if is_state:
            name = row["path"][6:]
            require(
                row["disposition"] == "transfer"
                and row["bundle_path"] == f"legacy-state/{name}"
                and row["iop_action"] == "remove"
                and row["neutral_successor"] == "-",
                f"manifest state contract mismatch: {row['path']}",
            )

    expected_delta_paths = {
        row["path"] for row in manifest if row["disposition"] == "retain-generic"
    } | DELTA_EXTRA_PATHS
    require(len(delta) == 137 and delta_paths == expected_delta_paths, "delta coverage mismatch")
    manifest_by_path = {row["path"]: row for row in manifest}
    for row in delta:
        original = manifest_by_path.get(row["path"])
        require(original is not None, f"delta path is absent from manifest: {row['path']}")
        require(row["original_disposition"] == original["disposition"], f"delta source mismatch: {row['path']}")
        require(row["effective_disposition"] in {"retain-generic", "transfer"}, f"delta disposition mismatch: {row['path']}")
        require(row["iop_action"] in {"rewrite", "remove"}, f"delta action mismatch: {row['path']}")
        if row["iop_action"] == "remove":
            require(row["neutral_successor"] == "-", f"delta removal has successor: {row['path']}")
        else:
            require(row["neutral_successor"] != "-", f"delta rewrite lacks successor: {row['path']}")

    require(len(addendum) == 137, "addendum row count mismatch")
    require(not (manifest_paths & addendum_paths), "manifest/addendum overlap")
    require(not (delta_paths & addendum_paths), "delta/addendum overlap")
    addendum_state_paths = {row["path"] for row in addendum if row["kind"] == "state"}
    require(addendum_state_paths == {"state:cache-root", "state:temp-root"}, "addendum state set mismatch")
    for row in addendum:
        if row["kind"] == "state":
            name = row["path"][6:]
            require(
                row["disposition"] == "transfer"
                and row["bundle_path"] == f"legacy-state/{name}"
                and row["iop_action"] == "remove"
                and row["neutral_successor"] == "-"
                and row["evidence_map"] == "S01,S02,S03,S05",
                f"addendum state contract mismatch: {row['path']}",
            )
        else:
            require(
                row["kind"] == "file"
                and row["disposition"] == "retain-generic"
                and row["bundle_path"] == row["path"]
                and row["iop_action"] == "rewrite"
                and row["neutral_successor"] == row["path"]
                and row["evidence_map"] == "S01,S03,S04",
                f"addendum file contract mismatch: {row['path']}",
            )

    original_dispositions = Counter(row["original_disposition"] for row in delta)
    effective_dispositions = Counter(row["effective_disposition"] for row in delta)
    require(original_dispositions == {"retain-generic": 135, "transfer": 2}, "delta original counts mismatch")
    require(effective_dispositions == {"retain-generic": 78, "transfer": 59}, "delta effective counts mismatch")
    stats = {
        "manifest_rows": len(manifest),
        "manifest_files": file_count,
        "manifest_states": state_count,
        "manifest_retain": retain_count,
        "delta_rows": len(delta),
        "delta_original_retain": original_dispositions["retain-generic"],
        "delta_original_transfer": original_dispositions["transfer"],
        "delta_effective_retain": effective_dispositions["retain-generic"],
        "delta_effective_transfer": effective_dispositions["transfer"],
        "addendum_rows": len(addendum),
        "addendum_files": sum(row["kind"] == "file" for row in addendum),
        "addendum_states": sum(row["kind"] == "state" for row in addendum),
        "manifest_delta_overlap": len(manifest_paths & delta_paths),
        "manifest_addendum_overlap": len(manifest_paths & addendum_paths),
        "delta_addendum_overlap": len(delta_paths & addendum_paths),
    }
    require(
        (stats["manifest_delta_overlap"], stats["manifest_addendum_overlap"], stats["delta_addendum_overlap"])
        == (137, 0, 0),
        "source overlap counts mismatch",
    )
    return manifest, delta, addendum, stats


def build_matrix(
    manifest: list[dict[str, str]], delta: list[dict[str, str]], addendum: list[dict[str, str]]
) -> tuple[list[dict[str, str]], dict[str, int]]:
    delta_by_path = {row["path"]: row for row in delta}
    matrix: list[dict[str, str]] = []
    for source in manifest:
        decision = delta_by_path.get(source["path"])
        if decision:
            effective_disposition = decision["effective_disposition"]
            effective_action = decision["iop_action"]
            effective_successor = decision["neutral_successor"]
            decision_source = "boundary-disposition-delta-v1"
            rationale = decision["rationale"]
        else:
            effective_disposition = source["disposition"]
            effective_action = source["iop_action"]
            effective_successor = source["neutral_successor"]
            decision_source = "current-manifest"
            rationale = "unchanged-current-baseline"
        matrix.append(
            {
                "path": source["path"],
                "kind": source["kind"],
                "source_manifest": "current-manifest",
                "source_disposition": source["disposition"],
                "source_bundle_path": source["bundle_path"],
                "source_iop_action": source["iop_action"],
                "source_neutral_successor": source["neutral_successor"],
                "source_evidence_map": source["evidence_map"],
                "effective_disposition": effective_disposition,
                "effective_bundle_path": source["bundle_path"],
                "effective_iop_action": effective_action,
                "effective_neutral_successor": effective_successor,
                "effective_evidence_map": source["evidence_map"],
                "decision_source": decision_source,
                "decision_rationale": rationale,
            }
        )
    for source in addendum:
        matrix.append(
            {
                "path": source["path"],
                "kind": source["kind"],
                "source_manifest": "d04-addendum",
                "source_disposition": source["disposition"],
                "source_bundle_path": source["bundle_path"],
                "source_iop_action": source["iop_action"],
                "source_neutral_successor": source["neutral_successor"],
                "source_evidence_map": source["evidence_map"],
                "effective_disposition": source["disposition"],
                "effective_bundle_path": source["bundle_path"],
                "effective_iop_action": source["iop_action"],
                "effective_neutral_successor": source["neutral_successor"],
                "effective_evidence_map": source["evidence_map"],
                "decision_source": "boundary-surface-addendum-v1",
                "decision_rationale": "d04-active-surface-coverage",
            }
        )
    matrix.sort(key=lambda row: row["path"].encode("utf-8"))
    paths = [row["path"] for row in matrix]
    require(len(matrix) == 437 and len(paths) == len(set(paths)), "effective matrix path universe mismatch")
    decision_counts = Counter(row["decision_source"] for row in matrix)
    require(
        decision_counts
        == {
            "boundary-disposition-delta-v1": 137,
            "current-manifest": 163,
            "boundary-surface-addendum-v1": 137,
        },
        "effective matrix decision counts mismatch",
    )
    stats = {
        "matrix_rows": len(matrix),
        "matrix_delta_decisions": decision_counts["boundary-disposition-delta-v1"],
        "matrix_addendum_decisions": decision_counts["boundary-surface-addendum-v1"],
        "matrix_unclassified": sum(not row["decision_source"] for row in matrix),
    }
    return matrix, stats


def validate_state_and_stable_receipt(
    schema_path: Path,
    state_path: Path,
    stable_receipt_path: Path,
    manifest: list[dict[str, str]],
    addendum: list[dict[str, str]],
) -> tuple[dict[str, int], dict[str, object]]:
    schema = json.loads(schema_path.read_text(encoding="utf-8"))
    require(schema.get("$schema") == "https://json-schema.org/draft/2020-12/schema", "state schema draft mismatch")
    require(schema.get("type") == "object" and schema.get("additionalProperties") is False, "state schema root mismatch")

    payload = json.loads(state_path.read_text(encoding="utf-8"))
    require(set(payload) == {"version", "records"} and payload["version"] == 1, "state fixture root mismatch")
    records = payload["records"]
    require(isinstance(records, list) and len(records) == 12, "state fixture record count mismatch")
    required = {
        "logical_id",
        "bundle_path",
        "source_manifest",
        "category",
        "source_revision",
        "digest",
        "size",
        "mode",
        "resumable",
    }
    ids: list[str] = []
    for record in records:
        require(isinstance(record, dict) and required <= set(record), "state fixture record shape mismatch")
        require(set(record) <= required | {"quarantine_reason"}, "state fixture record has unknown key")
        logical_id = record["logical_id"]
        require(re.fullmatch(r"state:[a-z0-9-]+", logical_id) is not None, "state logical id mismatch")
        require(record["bundle_path"] == f"legacy-state/{logical_id[6:]}", "state bundle path mismatch")
        require(record["source_revision"] == SOURCE_REVISION, "state source revision mismatch")
        require(re.fullmatch(r"[0-9a-f]{64}", record["digest"]) is not None, "state payload digest mismatch")
        require(isinstance(record["size"], int) and record["size"] >= 0, "state payload size mismatch")
        require(record["mode"] == "read-only", "state mode mismatch")
        if record["category"] == "quarantine":
            require(record["resumable"] is False and bool(record.get("quarantine_reason")), "quarantine contract mismatch")
        else:
            require(record["resumable"] is True and "quarantine_reason" not in record, "resumable state contract mismatch")
        ids.append(logical_id)
    require(len(ids) == len(set(ids)), "duplicate state logical id")
    logical_digest = sha256_bytes(("\n".join(bytewise_sorted(ids)) + "\n").encode("utf-8"))
    require(logical_digest == STATE_LOGICAL_IDS_SHA256, "state logical-id digest mismatch")
    original_states = {row["path"] for row in manifest if row["kind"] == "state"}
    addendum_states = {row["path"] for row in addendum if row["kind"] == "state"}
    require(len(original_states) == 10 and len(addendum_states) == 2, "state source counts mismatch")
    require(set(ids) == original_states | addendum_states, "state source linkage mismatch")
    source_counts = Counter(record["source_manifest"] for record in records)
    require(source_counts == {"original": 10, "d04-addendum": 2}, "state source labels mismatch")
    categories = {record["category"] for record in records}
    quarantine = sum(record["category"] == "quarantine" for record in records)
    require(len(categories) == 8 and quarantine == 1, "state category or quarantine count mismatch")

    stable = json.loads(stable_receipt_path.read_text(encoding="utf-8"))
    stable_keys = {
        "version",
        "inventory_source_revision",
        "manifest_sha",
        "baseline_fixture_bundle_sha",
        "owner_config_sha",
        "owner_snapshot_sha",
        "real_bundle_sha",
        "real_state_export_sha",
        "state_record_count",
        "state_category_count",
        "quarantine_record_count",
        "state_logical_id_count",
        "state_logical_ids_sha",
        "archive_mode",
        "source_mode",
    }
    require(set(stable) == stable_keys and stable["version"] == 1, "stable receipt key contract mismatch")
    require(stable["inventory_source_revision"] == SOURCE_REVISION, "stable receipt source revision mismatch")
    require(stable["manifest_sha"] == EXPECTED_DIGESTS["manifest"], "stable receipt manifest mismatch")
    require(
        stable["state_record_count"] == 12
        and stable["state_category_count"] == 8
        and stable["quarantine_record_count"] == 1
        and stable["state_logical_id_count"] == 12
        and stable["state_logical_ids_sha"] == STATE_LOGICAL_IDS_SHA256,
        "stable receipt state facts mismatch",
    )
    require(stable["archive_mode"] == "0600" and stable["source_mode"] == "pinned-git-tree", "stable receipt mode mismatch")
    for key in (
        "baseline_fixture_bundle_sha",
        "owner_config_sha",
        "owner_snapshot_sha",
        "real_bundle_sha",
        "real_state_export_sha",
    ):
        require(re.fullmatch(r"[0-9a-f]{64}", str(stable[key])) is not None, f"stable receipt digest mismatch: {key}")
    stats = {
        "state_records": len(records),
        "state_categories": len(categories),
        "state_quarantine": quarantine,
        "state_original": len(original_states),
        "state_addendum": len(addendum_states),
    }
    return stats, stable


def parse_final_loop_review(complete_text: str) -> tuple[str, str]:
    match = re.search(r"^## Loop History\s*$\n(?P<body>.*?)(?=^## |\Z)", complete_text, re.MULTILINE | re.DOTALL)
    require(match is not None, "complete log lacks Loop History")
    rows: list[list[str]] = []
    for line in match.group("body").splitlines():
        if not line.startswith("|"):
            continue
        cells = [cell.strip() for cell in line.strip("|").split("|")]
        if len(cells) < 3 or cells[0] == "Plan" or set(cells[0]) <= {"-", ":"}:
            continue
        rows.append(cells)
    require(bool(rows), "complete log Loop History has no data rows")
    final = rows[-1]
    review_match = re.fullmatch(r"`([^`]+)`", final[1])
    require(review_match is not None, "final Loop History review is not an exact filename")
    return review_match.group(1), final[2]


def parse_overall_verdict(review_text: str) -> str:
    verdicts = re.findall(r"^- Overall Verdict:\s*(PASS|WARN|FAIL)\s*$", review_text, re.MULTILINE)
    heading_matches = list(re.finditer(r"^### Overall Verdict\s*$", review_text, re.MULTILINE))
    for heading in heading_matches:
        remainder = review_text[heading.end() :]
        value_match = re.match(r"\s*(PASS|WARN|FAIL)\s*(?:\n|\Z)", remainder)
        require(value_match is not None, "Overall Verdict heading lacks a verdict")
        verdicts.append(value_match.group(1))
    require(len(verdicts) == 1, "review must contain exactly one Overall Verdict")
    return verdicts[0]


def validate_predecessors(root: Path) -> dict[str, dict[str, str]]:
    result: dict[str, dict[str, str]] = {}
    for task, expected in PREDECESSORS.items():
        complete_rel = str(expected["complete"])
        complete_path = root / complete_rel
        review_rel = str(PurePosixPath(complete_rel).parent / str(expected["review"]))
        review_path = root / review_rel
        require(complete_path.is_file(), f"{task}: exact complete log is missing")
        require(review_path.is_file(), f"{task}: exact final review is missing")
        verify_digest(complete_path, str(expected["complete_sha256"]))
        verify_digest(review_path, str(expected["review_sha256"]))
        complete_text = complete_path.read_text(encoding="utf-8")
        review_name, loop_verdict = parse_final_loop_review(complete_text)
        require(review_name == expected["review"], f"{task}: final Loop History review mismatch")
        require(loop_verdict == "PASS", f"{task}: final Loop History verdict is not PASS")
        require(parse_overall_verdict(review_path.read_text(encoding="utf-8")) == "PASS", f"{task}: review verdict is not PASS")
        result[task] = {
            "complete": complete_rel,
            "complete_sha256": str(expected["complete_sha256"]),
            "review": review_rel,
            "review_sha256": str(expected["review_sha256"]),
        }

    task03_text = (root / result["task03"]["complete"]).read_text(encoding="utf-8")
    require("rows=303 file_rows=293 state_rows=10" in task03_text, "Task 03 historical row evidence mismatch")
    require("missing=0 orphans=0" in task03_text and "duplicates=0" in task03_text, "Task 03 universe evidence mismatch")
    task06_text = (root / result["task06"]["complete"]).read_text(encoding="utf-8")
    require("12 canonical state records across 8 categories" in task06_text, "Task 06 owner state evidence mismatch")
    require("one non-resumable quarantine record" in task06_text, "Task 06 quarantine evidence mismatch")
    task10_text = (root / result["task10"]["complete"]).read_text(encoding="utf-8")
    require("all-zero audit evidence passed review" in task10_text, "Task 10 audit evidence mismatch")
    task16_text = (root / result["task16"]["complete"]).read_text(encoding="utf-8")
    require("300 rows (`file=290`, `state=10`, `retain-generic=135`)" in task16_text, "Task 16 current counts mismatch")
    require(EXPECTED_DIGESTS["manifest"] in task16_text, "Task 16 current manifest digest mismatch")
    return result


def parse_import_audit(path: Path) -> None:
    verify_digest(path, IMPORT_AUDIT_SHA256)
    lines = path.read_text(encoding="utf-8").splitlines()
    metadata: dict[str, str] = {}
    index = 0
    while index < len(lines) and not lines[index].startswith("[command "):
        require("=" in lines[index], "Task 10 audit metadata is malformed")
        key, value = lines[index].split("=", 1)
        require(key not in metadata, f"Task 10 audit duplicate key: {key}")
        metadata[key] = value
        index += 1
    expected_keys = {
        "schema",
        "manifest_sha256",
        "delta_sha256",
        "addendum_sha256",
        "state_logical_id_sha256",
        "state_count",
        "provider_commands",
        "session_id_semantics",
        "cancel_semantics",
        "clean_build",
        "standard_inference",
        "s03",
        "s04",
    }
    require(set(metadata) == expected_keys, "Task 10 audit metadata keys mismatch")
    require(metadata["schema"] == "iop-node-boundary-audit-v1", "Task 10 audit schema mismatch")
    require(metadata["manifest_sha256"] == EXPECTED_DIGESTS["manifest"], "Task 10 audit manifest mismatch")
    require(metadata["delta_sha256"] == EXPECTED_DIGESTS["delta"], "Task 10 audit delta mismatch")
    require(metadata["addendum_sha256"] == EXPECTED_DIGESTS["addendum"], "Task 10 audit addendum mismatch")
    require(metadata["state_logical_id_sha256"] == STATE_LOGICAL_IDS_SHA256, "Task 10 audit state mismatch")
    require(all(metadata[key] == "PASS" for key in ("clean_build", "standard_inference", "s03", "s04")), "Task 10 audit closure mismatch")

    commands: dict[str, tuple[str, int]] = {}
    while index < len(lines):
        start = re.fullmatch(r"\[command ([0-9]{2})\]", lines[index])
        require(start is not None, "Task 10 audit command start mismatch")
        command_id = start.group(1)
        require(command_id not in commands and index + 3 < len(lines), "Task 10 audit duplicate or truncated command")
        require(lines[index + 1].startswith("text="), "Task 10 audit command text mismatch")
        exit_match = re.fullmatch(r"exit_code=([0-9]+)", lines[index + 2])
        require(exit_match is not None and lines[index + 3] == "output_begin", "Task 10 audit command metadata mismatch")
        command_text = lines[index + 1][5:]
        index += 4
        while index < len(lines) and lines[index] != "output_end":
            index += 1
        require(index + 1 < len(lines), "Task 10 audit command output is truncated")
        require(lines[index + 1] == f"[/command {command_id}]", "Task 10 audit command close mismatch")
        commands[command_id] = (command_text, int(exit_match.group(1)))
        index += 2
    require(set(commands) == {f"{number:02d}" for number in range(1, 31)}, "Task 10 audit command set mismatch")
    require(all(exit_code == 0 for _, exit_code in commands.values()), "Task 10 audit has a nonzero command")
    require("go list -deps" in commands["21"][0], "Task 10 dependency audit command mismatch")
    require("chronos" in commands["27"][0].lower(), "Task 10 Chronos surface audit command mismatch")


def validate_live_sources(root: Path) -> tuple[
    list[dict[str, str]],
    list[dict[str, str]],
    list[dict[str, str]],
    dict[str, int],
    dict[str, int],
    dict[str, object],
]:
    for key, relative in SOURCE_FILES.items():
        path = root / relative
        require(path.is_file(), f"live source is missing: {relative}")
        verify_digest(path, EXPECTED_DIGESTS[key])
    for key in SOURCE_SIDECARS:
        verify_source_sidecar(root, key)
    manifest, delta, addendum, source_stats = validate_manifest_delta_addendum(
        root / SOURCE_FILES["manifest"], root / SOURCE_FILES["delta"], root / SOURCE_FILES["addendum"]
    )
    state_stats, stable = validate_state_and_stable_receipt(
        root / SOURCE_FILES["state_schema"],
        root / SOURCE_FILES["state_fixture"],
        root / SOURCE_FILES["stable_receipt"],
        manifest,
        addendum,
    )
    return manifest, delta, addendum, source_stats, state_stats, stable


def run_global_boundary_verifier(root: Path) -> None:
    command = [
        sys.executable,
        "scripts/verify-iop-agent-chronos-boundary.py",
        "--manifest",
        SOURCE_FILES["manifest"],
        "--expected-manifest-sha256",
        EXPECTED_DIGESTS["manifest"],
        "--delta",
        SOURCE_FILES["delta"],
        "--delta-sha256",
        SOURCE_SIDECARS["delta"],
        "--addendum",
        SOURCE_FILES["addendum"],
        "--addendum-sha256",
        SOURCE_SIDECARS["addendum"],
        "--state-fixture",
        SOURCE_FILES["state_fixture"],
        "--expected-state-logical-id-sha256",
        STATE_LOGICAL_IDS_SHA256,
        "--iop-root",
        ".",
        "--check-only",
    ]
    result = subprocess.run(command, cwd=root, text=True, capture_output=True, check=False)
    require(result.returncode == 0, f"global boundary verifier failed: {result.stderr.strip() or result.stdout.strip()}")
    require(
        "boundary_verification=PASS manifest_rows=300 delta_rows=137 addendum_rows=137 state_records=12"
        in result.stdout,
        "global boundary verifier output mismatch",
    )


def validate_local_artifacts(task_dir: Path) -> tuple[
    list[dict[str, str]],
    list[dict[str, str]],
    list[dict[str, str]],
    dict[str, int],
    dict[str, int],
    dict[str, int],
    dict[str, object],
    dict[str, str],
]:
    digests: dict[str, str] = {}
    for key, filename in LOCAL_FILES.items():
        artifact = task_dir / filename
        require(artifact.is_file(), f"task-local artifact is missing: {filename}")
        digests[key] = verify_sidecar(artifact, task_dir / LOCAL_SIDECARS[key])
        if key in EXPECTED_DIGESTS:
            require(digests[key] == EXPECTED_DIGESTS[key], f"task-local {key} digest mismatch")

    manifest, delta, addendum, source_stats = validate_manifest_delta_addendum(
        task_dir / LOCAL_FILES["manifest"], task_dir / LOCAL_FILES["delta"], task_dir / LOCAL_FILES["addendum"]
    )
    state_stats, stable = validate_state_and_stable_receipt(
        task_dir / LOCAL_FILES["state_schema"],
        task_dir / LOCAL_FILES["state_fixture"],
        task_dir / LOCAL_FILES["stable_receipt"],
        manifest,
        addendum,
    )
    expected_matrix, matrix_stats = build_matrix(manifest, delta, addendum)
    expected_matrix_bytes = render_tsv(MATRIX_HEADER, expected_matrix)
    require((task_dir / LOCAL_FILES["matrix"]).read_bytes() == expected_matrix_bytes, "effective matrix bytes mismatch")
    actual_matrix = read_tsv(task_dir / LOCAL_FILES["matrix"], MATRIX_HEADER)
    require(actual_matrix == expected_matrix, "effective matrix rows mismatch")
    return manifest, delta, addendum, source_stats, matrix_stats, state_stats, stable, digests


def verify_live_copy_identity(root: Path, task_dir: Path) -> None:
    for key, source_rel in SOURCE_FILES.items():
        require(
            (root / source_rel).read_bytes() == (task_dir / LOCAL_FILES[key]).read_bytes(),
            f"live/task-local byte identity mismatch: {key}",
        )


def verify_effective_worktree(root: Path, matrix: list[dict[str, str]]) -> None:
    for row in matrix:
        if row["kind"] == "state":
            continue
        action = row["effective_iop_action"]
        path = root / row["path"]
        successor = row["effective_neutral_successor"]
        if action == "remove":
            require(not path.exists(), f"effective removal has a residual path: {row['path']}")
        elif action in {"rewrite", "rename"}:
            require(successor != "-" and (root / successor).exists(), f"effective successor is missing: {row['path']}")
        else:
            require(action == "none", f"unsupported effective action: {row['path']}")


def predecessor_receipt_values(predecessors: dict[str, dict[str, str]]) -> dict[str, str]:
    values: dict[str, str] = {}
    for task, evidence in predecessors.items():
        values[f"{task}_complete_log_path"] = evidence["complete"]
        values[f"{task}_complete_log_sha256"] = evidence["complete_sha256"]
        values[f"{task}_review_path"] = evidence["review"]
        values[f"{task}_review_sha256"] = evidence["review_sha256"]
        values[f"{task}_pass"] = "true"
    return values


def expected_receipt_values(
    source_stats: dict[str, int],
    matrix_stats: dict[str, int],
    state_stats: dict[str, int],
    stable: dict[str, object],
    digests: dict[str, str],
    predecessors: dict[str, dict[str, str]],
) -> dict[str, str]:
    values = {
        "receipt_version": "1",
        "receipt_status": "PASS",
        "artifact_language": "English",
        "source_revision": SOURCE_REVISION,
        "rollback_revision": SOURCE_REVISION,
        "artifact_root": ".",
        "manifest_source_path": SOURCE_FILES["manifest"],
        "manifest_path": LOCAL_FILES["manifest"],
        "manifest_sha256_path": LOCAL_SIDECARS["manifest"],
        "manifest_sha256": digests["manifest"],
        "manifest_row_count": str(source_stats["manifest_rows"]),
        "manifest_file_count": str(source_stats["manifest_files"]),
        "manifest_state_count": str(source_stats["manifest_states"]),
        "manifest_retain_generic_count": str(source_stats["manifest_retain"]),
        "delta_source_path": SOURCE_FILES["delta"],
        "delta_path": LOCAL_FILES["delta"],
        "delta_sha256_path": LOCAL_SIDECARS["delta"],
        "delta_sha256": digests["delta"],
        "delta_row_count": str(source_stats["delta_rows"]),
        "delta_original_retain_generic_count": str(source_stats["delta_original_retain"]),
        "delta_original_transfer_count": str(source_stats["delta_original_transfer"]),
        "delta_effective_retain_generic_count": str(source_stats["delta_effective_retain"]),
        "delta_effective_transfer_count": str(source_stats["delta_effective_transfer"]),
        "addendum_source_path": SOURCE_FILES["addendum"],
        "addendum_path": LOCAL_FILES["addendum"],
        "addendum_sha256_path": LOCAL_SIDECARS["addendum"],
        "addendum_sha256": digests["addendum"],
        "addendum_row_count": str(source_stats["addendum_rows"]),
        "addendum_file_count": str(source_stats["addendum_files"]),
        "addendum_state_count": str(source_stats["addendum_states"]),
        "effective_matrix_path": LOCAL_FILES["matrix"],
        "effective_matrix_sha256_path": LOCAL_SIDECARS["matrix"],
        "effective_matrix_sha256": digests["matrix"],
        "effective_matrix_row_count": str(matrix_stats["matrix_rows"]),
        "effective_matrix_delta_decision_count": str(matrix_stats["matrix_delta_decisions"]),
        "effective_matrix_addendum_decision_count": str(matrix_stats["matrix_addendum_decisions"]),
        "effective_matrix_unclassified_count": str(matrix_stats["matrix_unclassified"]),
        "manifest_delta_overlap_count": str(source_stats["manifest_delta_overlap"]),
        "manifest_addendum_overlap_count": str(source_stats["manifest_addendum_overlap"]),
        "delta_addendum_overlap_count": str(source_stats["delta_addendum_overlap"]),
        "state_schema_source_path": SOURCE_FILES["state_schema"],
        "state_schema_path": LOCAL_FILES["state_schema"],
        "state_schema_sha256_path": LOCAL_SIDECARS["state_schema"],
        "state_schema_sha256": digests["state_schema"],
        "state_fixture_source_path": SOURCE_FILES["state_fixture"],
        "state_fixture_path": LOCAL_FILES["state_fixture"],
        "state_fixture_sha256_path": LOCAL_SIDECARS["state_fixture"],
        "state_fixture_sha256": digests["state_fixture"],
        "state_original_manifest_logical_id_count": str(state_stats["state_original"]),
        "state_addendum_logical_id_count": str(state_stats["state_addendum"]),
        "state_logical_id_count": str(state_stats["state_records"]),
        "state_logical_ids_sha256": STATE_LOGICAL_IDS_SHA256,
        "state_record_count": str(state_stats["state_records"]),
        "state_category_count": str(state_stats["state_categories"]),
        "state_quarantine_count": str(state_stats["state_quarantine"]),
        "stable_receipt_source_path": SOURCE_FILES["stable_receipt"],
        "stable_receipt_path": LOCAL_FILES["stable_receipt"],
        "stable_receipt_sha256_path": LOCAL_SIDECARS["stable_receipt"],
        "stable_receipt_sha256": digests["stable_receipt"],
        "owner_transfer_evidence": "task06-official-pass",
        "task06_owner_state_record_count": str(stable["state_record_count"]),
        "task06_owner_state_category_count": str(stable["state_category_count"]),
        "task06_owner_quarantine_count": str(stable["quarantine_record_count"]),
        "task06_owner_logical_id_count": str(stable["state_logical_id_count"]),
        "task06_owner_logical_ids_sha256": str(stable["state_logical_ids_sha"]),
        "task03_historical_row_count": "303",
        "task03_historical_file_count": "293",
        "task03_historical_state_count": "10",
        "task03_universe_residuals": "0",
        "task03_duplicate_rows": "0",
        "task10_import_audit_path": IMPORT_AUDIT_PATH,
        "task10_import_audit_sha256": IMPORT_AUDIT_SHA256,
        "import_graph_residuals": "0",
        "task16_current_row_count": "300",
        "task16_current_file_count": "290",
        "task16_current_state_count": "10",
        "task16_current_retain_generic_count": "135",
        "raw_owner_bundle_required": "false",
        "external_inputs": "0",
        "external_mutations": "0",
        "receipt_path_leaks": "0",
        "iop_node_owner": "model-provider-device",
        "chronos_connection_surfaces": "0",
        "canonical_promotion": "pending",
        "downstream_lock": "chronos:chronos-architecture-ownership-boundary",
        "staging_lock_identity": "iop-s1:agent-roadmap/phase/automation-runtime-bridge/milestones/iop-agent-chronos-extraction-decoupling.md",
        "staging_lock_result": "none",
        "canonical_lock_identity": "iop:agent-roadmap/phase/automation-runtime-bridge/milestones/iop-agent-chronos-extraction-decoupling.md",
        "canonical_lock_result": "rely-on:chronos:chronos-architecture-ownership-boundary",
    }
    values.update(predecessor_receipt_values(predecessors))
    require(set(values) == set(RECEIPT_KEYS), "internal receipt key construction mismatch")
    return values


def render_receipt(values: dict[str, str]) -> bytes:
    lines = ["---"] + [f"{key}: {values[key]}" for key in RECEIPT_KEYS] + ["---", "", RECEIPT_BODY.rstrip(), ""]
    return "\n".join(lines).encode("utf-8")


def validate_relative_path(value: str, *, basename_only: bool = False) -> None:
    require(value and "\\" not in value and "$" not in value and "~" not in value, f"unsafe path value: {value}")
    path = PurePosixPath(value)
    require(not path.is_absolute() and ".." not in path.parts and "." not in path.parts, f"unsafe relative path: {value}")
    if basename_only:
        require(len(path.parts) == 1, f"artifact path must be a basename: {value}")


def parse_receipt(path: Path) -> dict[str, str]:
    text = path.read_text(encoding="utf-8")
    lines = text.splitlines()
    require(lines and lines[0] == "---", "receipt lacks YAML front matter")
    try:
        close = lines.index("---", 1)
    except ValueError as exc:
        raise VerificationError("receipt front matter is not closed") from exc
    pairs: list[tuple[str, str]] = []
    for line in lines[1:close]:
        match = re.fullmatch(r"([a-z0-9_]+): ([^\r\n]+)", line)
        require(match is not None, f"receipt front matter line is malformed: {line}")
        pairs.append((match.group(1), match.group(2)))
    keys = [key for key, _ in pairs]
    require(keys == list(RECEIPT_KEYS), "receipt keys are missing, duplicated, unknown, or out of order")
    require(len(keys) == len(set(keys)), "receipt contains a duplicate key")
    require("\n".join(lines[close + 2 :]).rstrip() == RECEIPT_BODY.rstrip(), "receipt body mismatch")
    values = dict(pairs)

    basename_path_keys = {
        "manifest_path",
        "manifest_sha256_path",
        "delta_path",
        "delta_sha256_path",
        "addendum_path",
        "addendum_sha256_path",
        "effective_matrix_path",
        "effective_matrix_sha256_path",
        "state_schema_path",
        "state_schema_sha256_path",
        "state_fixture_path",
        "state_fixture_sha256_path",
        "stable_receipt_path",
        "stable_receipt_sha256_path",
    }
    for key, value in values.items():
        if key.endswith("_path"):
            validate_relative_path(value, basename_only=key in basename_path_keys)
        if key.endswith("_sha256"):
            require(re.fullmatch(r"[0-9a-f]{64}", value) is not None, f"receipt digest is not lowercase hex: {key}")
    require(values["artifact_root"] == ".", "receipt artifact root mismatch")
    leak_patterns = (
        r"/(?:config|home|Users|tmp)/",
        r"(?:^|\s)\.\.(?:/|\s|$)",
        r"\$\{?",
        r"~/",
    )
    require(not any(re.search(pattern, text) for pattern in leak_patterns), "receipt contains a host or dynamic path")
    secret_assignment = re.compile(
        r"(?i)(?:token|password|credential|private[_ -]?key|api[_ -]?key|bearer)\s*[:=]\s*\S+"
    )
    require(secret_assignment.search(text) is None, "receipt contains a secret-like assignment")
    return values


def expected_audit_values() -> dict[str, str]:
    return {
        "schema": "iop-chronos-pre-deletion-audit-v1",
        "audit_status": "PASS",
        "task03_pass": "1",
        "task06_pass": "1",
        "task09_pass": "1",
        "task10_pass": "1",
        "task16_pass": "1",
        "manifest_digest_match": "1",
        "delta_digest_match": "1",
        "addendum_digest_match": "1",
        "state_schema_digest_match": "1",
        "state_fixture_digest_match": "1",
        "stable_receipt_digest_match": "1",
        "effective_matrix_digest_match": "1",
        "manifest_rows": "300",
        "manifest_retain_generic": "135",
        "delta_rows": "137",
        "addendum_rows": "137",
        "effective_matrix_rows": "437",
        "effective_matrix_unclassified": "0",
        "manifest_delta_overlap": "137",
        "manifest_addendum_overlap": "0",
        "delta_addendum_overlap": "0",
        "state_records": "12",
        "state_categories": "8",
        "state_quarantine": "1",
        "import_graph_residuals": "0",
        "raw_owner_bundle_required": "0",
        "external_inputs": "0",
        "external_mutations": "0",
        "receipt_path_leaks": "0",
        "canonical_promotion": "pending",
        "check_live": "PASS",
        "check_archived": "PASS",
    }


def render_audit(values: dict[str, str]) -> bytes:
    require(set(values) == set(AUDIT_KEYS), "internal audit key construction mismatch")
    return ("\n".join(f"{key}={values[key]}" for key in AUDIT_KEYS) + "\n").encode("utf-8")


def parse_audit(path: Path) -> dict[str, str]:
    data = path.read_bytes()
    require(data.endswith(b"\n") and b"\r" not in data, "pre-deletion audit is not canonical text")
    pairs: list[tuple[str, str]] = []
    for line in data.decode("utf-8").splitlines():
        match = re.fullmatch(r"([a-z0-9_]+)=([^\r\n]+)", line)
        require(match is not None, f"pre-deletion audit line is malformed: {line}")
        pairs.append((match.group(1), match.group(2)))
    keys = [key for key, _ in pairs]
    require(keys == list(AUDIT_KEYS) and len(keys) == len(set(keys)), "pre-deletion audit keys mismatch")
    return dict(pairs)


def receipt_and_audit_check(
    root: Path,
    task_dir: Path,
    source_stats: dict[str, int],
    matrix_stats: dict[str, int],
    state_stats: dict[str, int],
    stable: dict[str, object],
    digests: dict[str, str],
    predecessors: dict[str, dict[str, str]],
) -> None:
    receipt_path = task_dir / "pre-deletion-transfer-receipt-v1.md"
    audit_path = task_dir / "pre-deletion-audit.log"
    actual_receipt = parse_receipt(receipt_path)
    expected_receipt = expected_receipt_values(
        source_stats, matrix_stats, state_stats, stable, digests, predecessors
    )
    require(actual_receipt == expected_receipt, "receipt values mismatch")
    actual_audit = parse_audit(audit_path)
    require(actual_audit == expected_audit_values(), "pre-deletion audit values mismatch")
    require(actual_receipt["receipt_path_leaks"] == "0", "receipt leak marker mismatch")
    require(actual_receipt["external_inputs"] == actual_receipt["external_mutations"] == "0", "external boundary mismatch")
    require(actual_receipt["raw_owner_bundle_required"] == "false", "raw owner bundle must not be required")
    parse_import_audit(root / IMPORT_AUDIT_PATH)


def materialize(root: Path, task_dir: Path) -> None:
    predecessors = validate_predecessors(root)
    parse_import_audit(root / IMPORT_AUDIT_PATH)
    run_global_boundary_verifier(root)
    manifest, delta, addendum, _, _, _ = validate_live_sources(root)
    matrix, _ = build_matrix(manifest, delta, addendum)
    matrix_bytes = render_tsv(MATRIX_HEADER, matrix)

    for key, source_rel in SOURCE_FILES.items():
        atomic_write(task_dir / LOCAL_FILES[key], (root / source_rel).read_bytes())
    atomic_write(task_dir / LOCAL_FILES["matrix"], matrix_bytes)
    for key, filename in LOCAL_FILES.items():
        digest = sha256(task_dir / filename)
        atomic_write(task_dir / LOCAL_SIDECARS[key], sidecar_bytes(filename, digest))

    local_manifest, _, _, source_stats, matrix_stats, state_stats, stable, digests = validate_local_artifacts(task_dir)
    verify_live_copy_identity(root, task_dir)
    verify_effective_worktree(root, read_tsv(task_dir / LOCAL_FILES["matrix"], MATRIX_HEADER))
    receipt_values = expected_receipt_values(
        source_stats, matrix_stats, state_stats, stable, digests, predecessors
    )
    atomic_write(task_dir / "pre-deletion-transfer-receipt-v1.md", render_receipt(receipt_values))
    atomic_write(task_dir / "pre-deletion-audit.log", render_audit(expected_audit_values()))
    require(len(local_manifest) == 300, "local manifest validation did not complete")
    receipt_and_audit_check(
        root, task_dir, source_stats, matrix_stats, state_stats, stable, digests, predecessors
    )


def check_archived(root: Path, task_dir: Path) -> None:
    predecessors = validate_predecessors(root)
    parse_import_audit(root / IMPORT_AUDIT_PATH)
    _, _, _, source_stats, matrix_stats, state_stats, stable, digests = validate_local_artifacts(task_dir)
    receipt_and_audit_check(
        root, task_dir, source_stats, matrix_stats, state_stats, stable, digests, predecessors
    )


def check_live(root: Path, task_dir: Path) -> None:
    check_archived(root, task_dir)
    run_global_boundary_verifier(root)
    validate_live_sources(root)
    verify_live_copy_identity(root, task_dir)
    matrix = read_tsv(task_dir / LOCAL_FILES["matrix"], MATRIX_HEADER)
    verify_effective_worktree(root, matrix)


def resolve_context(workspace: str) -> tuple[Path, Path]:
    root = Path(workspace).resolve()
    require((root / "go.mod").is_file(), "workspace does not contain go.mod")
    task_dir = Path(__file__).resolve().parent
    try:
        task_dir.relative_to(root)
    except ValueError as exc:
        raise VerificationError("verifier is outside the workspace") from exc
    return root, task_dir


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("mode", choices=("materialize", "check-live", "check-archived"))
    parser.add_argument("--workspace", required=True)
    return parser


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    try:
        root, task_dir = resolve_context(args.workspace)
        if args.mode == "materialize":
            materialize(root, task_dir)
        elif args.mode == "check-live":
            check_live(root, task_dir)
        else:
            check_archived(root, task_dir)
    except (OSError, UnicodeError, ValueError, json.JSONDecodeError, VerificationError) as exc:
        print(f"pre_deletion_receipt=FAIL mode={args.mode} reason={exc}", file=sys.stderr)
        return 1
    print(
        f"pre_deletion_receipt=PASS mode={args.mode} "
        "manifest_rows=300 delta_rows=137 addendum_rows=137 effective_matrix_rows=437"
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())
