#!/usr/bin/env python3
"""Build and verify the three anonymous journal Additional Files.

The payloads are selected only from frozen file allowlists. The script never
recursively copies a directory and never redistributes raw or derived TIMSS
student data.
"""

from __future__ import annotations

import argparse
import csv
import getpass
import hashlib
import io
import os
import posixpath
import re
import socket
import stat
import subprocess
import sys
import tempfile
import unicodedata
import zipfile
from contextlib import contextmanager
from pathlib import Path, PurePosixPath
from urllib.parse import unquote, urlsplit
from xml.etree import ElementTree as ET

ROOT = Path(__file__).resolve().parents[1]
SUBMISSION = ROOT / "submission"
MAX_ADDITIONAL_FILE_BYTES = 20_000_000
MAX_ARCHIVE_MEMBERS = 512
MAX_ARCHIVE_UNCOMPRESSED_BYTES = 100_000_000
MAX_ARCHIVE_MEMBER_BYTES = 20_000_000
MAX_ARCHIVE_NAME_BYTES = 240
MAX_ARCHIVE_COMPRESSION_RATIO = 250.0
ALLOWED_ZIP_COMPRESSION = {zipfile.ZIP_STORED, zipfile.ZIP_DEFLATED}
FIXED_ZIP_TIME = (1980, 1, 1, 0, 0, 0)
PUBLISH_LOCK = SUBMISSION / ".additional-files.lock"

DESTINATIONS = {
    "Additional_file_1_supplement.docx": SUBMISSION / "Additional_file_1_supplement.docx",
    "Additional_file_2_machine_readable_tables.zip": SUBMISSION / "Additional_file_2_machine_readable_tables.zip",
    "Additional_file_3_anonymous_reproducibility.zip": SUBMISSION / "Additional_file_3_anonymous_reproducibility.zip",
}
OVERALL_MANIFEST = SUBMISSION / "Additional_files_SHA256.csv"
SOURCE_SNAPSHOT_MANIFEST = "provenance/source_snapshot_manifest.csv"

ANALYSIS_FILES = (
    "analysis/00_config.R",
    "analysis/01_acquire_data.R",
    "analysis/02_audit_data.R",
    "analysis/03_primary_transitions.R",
    "analysis/04_standardization.R",
    "analysis/05_sensitivity.R",
    "analysis/06_outputs.R",
    "analysis/lib_timss.R",
    "analysis/run_all.R",
    "analysis/standardization_specifications.md",
    "analysis/xlsx_extract.py",
)

MANUSCRIPT_GENERATED_FILES = (
    "manuscript/generated/main_table_1.md",
    "manuscript/generated/main_table_2.md",
    "manuscript/generated/main_table_3.md",
    "manuscript/generated/supplement_sample_summary.md",
    "manuscript/generated/supplement_validation_summary.md",
    "manuscript/generated/supplement_overlap_summary.md",
    "manuscript/generated/supplement_grade89_summary.md",
    "manuscript/generated/supplement_table_inventory.md",
)

MANUSCRIPT_SOURCE_FILES = (
    "manuscript/manuscript.qmd",
    "manuscript/supplement.qmd",
    "manuscript/references.bib",
    "manuscript/apa.csl",
) + MANUSCRIPT_GENERATED_FILES

TOOL_FILES = (
    "tools/build_additional_files.py",
    "tools/finalize_docx.py",
)

REPRODUCIBILITY_SOURCE_FILES = ANALYSIS_FILES + MANUSCRIPT_SOURCE_FILES + TOOL_FILES + (
    "BUILD_RECIPE.txt",
)

RENDERED_DOCX_FILES = (
    "manuscript/manuscript.docx",
    "manuscript/supplement.docx",
)

TABLE_FILES = (
    "table_manifest.csv",
    "table_1_sample_participation.csv",
    "table_2_primary_standardized_recovery.csv",
    "table_3_robustness_summary.csv",
    "table_s1_full_transition_matrices.csv",
    "table_s2_state_distributions.csv",
    "table_s3_continuous_changes.csv",
    "table_s4_standardization_specification_decomposition.csv",
    "table_s5_standardization_alternative_contrasts.csv",
    "table_s6_standardization_overlap_by_draw.csv",
    "table_s7_standardization_overlap_summary.csv",
    "table_s8_threshold_sensitivity.csv",
    "table_s9_hysteresis_sensitivity.csv",
    "table_s10_observed_ipw_sensitivity.csv",
    "table_s11_mnar_pattern_mixture.csv",
    "table_s12_mnar_tipping_diagnostics.csv",
    "table_s13_grade89_joint_transitions.csv",
    "table_s14_grade89_domain_extension.csv",
    "table_s15_measurement_crossing_diagnostic.csv",
    "table_s16_validation_summary.csv",
)

REPRODUCIBILITY_ONLY_TABLE_FILES = (
    "standardized_pairwise_contrasts_exploratory.csv",
)

FIGURE_STEMS = (
    "figure_1_primary_recovery",
    "figure_2_raw_vs_standardized",
    "figure_3_threshold_sensitivity",
    "figure_4_missingness_robustness",
    "figure_s1_transition_matrices",
    "figure_s2_overlap_diagnostics",
    "figure_s3_standardization_sensitivity",
    "figure_s4_mnar_sensitivity",
    "figure_s5_grade89_extension",
    "figure_s6_measurement_crossing",
)
FIGURE_FILES = ("figure_manifest.csv",) + tuple(
    f"{stem}.{extension}"
    for stem in FIGURE_STEMS
    for extension in ("pdf", "png")
)

DIAGNOSTIC_FILES = (
    "audit_summary.txt",
    "baseline_state_draws.csv",
    "data_inventory.csv",
    "figure_artifact_validation.csv",
    "figure_interval_validation.csv",
    "official_result_validation.csv",
    "official_result_validation_long.csv",
    "publication_output_validation.csv",
    "response_ipw_diagnostics.csv",
    "sensitivity_covariance_validation.csv",
    "sensitivity_validation_checks.csv",
    "standardization_coefficients_by_draw.csv",
    "standardization_covariance_validation.csv",
    "standardization_model_diagnostics_by_draw.csv",
    "standardization_replicate_diagnostics.csv",
    "standardization_specification_diagnostics_by_draw.csv",
    "standardization_support_by_draw.csv",
    "table_artifact_validation.csv",
    "variable_map.csv",
    "xlsx_cache_lineage.csv",
)
SANITIZED_ENVIRONMENT_SOURCE = "outputs/diagnostics/software_versions_anonymous.txt"

SOURCE_SNAPSHOT_FILES = tuple(dict.fromkeys(
    REPRODUCIBILITY_SOURCE_FILES
    + tuple(f"outputs/diagnostics/{name}" for name in DIAGNOSTIC_FILES)
    + tuple(f"outputs/tables/{name}" for name in TABLE_FILES)
    + tuple(
        f"outputs/tables/{name}" for name in REPRODUCIBILITY_ONLY_TABLE_FILES
    )
    + tuple(f"outputs/figures/{name}" for name in FIGURE_FILES)
    + RENDERED_DOCX_FILES
    + (SANITIZED_ENVIRONMENT_SOURCE, "data/raw/download_manifest.csv")
))

RAW_SOURCE_FILENAMES = {
    "data_r_g45": "T23L_Data_R_G4-5.zip",
    "data_r_g89": "T23L_Data_R_G8-9.zip",
    "codebook_g45": "T23L_Codebook_G4-5.xlsx",
    "codebook_g89": "T23L_Codebook_G8-9.xlsx",
    "user_guide": "T23L_UG-International-Database.pdf",
    "r_programs": "T23L_Programs_R.zip",
    "technical_report": "T23L_TechnicalReport.pdf",
    "chapter13": "T23_TR_Chap13.pdf",
    "report_math_g45": "report_1-1-1a-b-ach-g5m.xlsx",
    "report_math_benchmark_g45": "report_1-1-3-ach-bmk-g5m.xlsx",
    "report_science_g45": "report_2-1-1a-b-ach-g5s.xlsx",
    "report_science_benchmark_g45": "report_2-1-3-ach-bmk-g5s.xlsx",
    "report_math_g89": "report_1-2-1a-b-ach-g9m.xlsx",
    "report_science_g89": "report_2-2-1a-b-ach-g9s.xlsx",
    "report_sample_a1": "report_A-1_students-assessed.xlsx",
    "report_participation_a5": "report_A-5_samp-participation.xlsx",
}

EXPECTED_TABLE_NUMBERS = tuple(
    [f"Table {number}" for number in range(1, 4)]
    + [f"Table S{number}" for number in range(1, 17)]
)
EXPECTED_FIGURE_NUMBERS = tuple(
    [f"Figure {number}" for number in range(1, 5)]
    + [f"Figure S{number}" for number in range(1, 7)]
)

TEXT_SUFFIXES = {".r", ".py", ".qmd", ".md", ".bib", ".csl", ".csv", ".txt"}
PII_SCAN_EXEMPT = {"manuscript/references.bib"}
PII_PLACEHOLDER_PATTERNS = (
    re.compile(r"\[(?:EMAIL|FULL NAME|AUTHOR \d+ FULL|CORRESPONDING AUTHOR|COMPLETE CORRESPONDENCE ADDRESS|DEPARTMENT, INSTITUTION|ORCID OR NONE)", re.I),
    re.compile(r"\[(?:AFFILIATION|POSTAL ADDRESS|PHONE|TELEPHONE)[^]]*\]", re.I),
)
EMAIL_PATTERN = re.compile(r"\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b", re.I)
ORCID_PATTERN = re.compile(r"\b\d{4}-\d{4}-\d{4}-\d{3}[\dX]\b", re.I)
LOCAL_PATH_PATTERNS = (
    re.compile(re.escape("/" + "Users" + "/"), re.I),
    re.compile(re.escape("/" + "home" + "/"), re.I),
    re.compile(r"\b[A-Z]:[\\/]", re.I),
    re.compile(re.escape("file" + "://"), re.I),
)
FORBIDDEN_ARCHIVE_PREFIXES = (
    "data/raw/",
    "data/derived/",
    "submission/",
    ".git/",
    ".quarto/",
)


def fail(message: str) -> None:
    raise RuntimeError(message)


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


def _secure_open_regular(path: Path) -> tuple[int, os.stat_result]:
    try:
        before = path.lstat()
    except FileNotFoundError:
        fail(f"Required file is missing: {path}")
    if stat.S_ISLNK(before.st_mode) or not stat.S_ISREG(before.st_mode):
        fail(f"File is not a regular non-symlink: {path}")
    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
    descriptor = os.open(path, flags)
    opened = os.fstat(descriptor)
    if (
        not stat.S_ISREG(opened.st_mode)
        or opened.st_dev != before.st_dev
        or opened.st_ino != before.st_ino
    ):
        os.close(descriptor)
        fail(f"File changed identity while being opened: {path}")
    return descriptor, opened


def _verify_open_file_stable(path: Path, descriptor: int, opened: os.stat_result) -> None:
    finished = os.fstat(descriptor)
    try:
        current = path.lstat()
    except FileNotFoundError:
        fail(f"File disappeared while being read: {path}")
    stable_fields = ("st_dev", "st_ino", "st_size", "st_mtime_ns", "st_ctime_ns")
    if any(getattr(finished, field) != getattr(opened, field) for field in stable_fields) or any(
        getattr(current, field) != getattr(opened, field) for field in stable_fields
    ):
        fail(f"File changed while being read: {path}")


def secure_file_bytes(path: Path, maximum_bytes: int | None = None) -> bytes:
    descriptor, opened = _secure_open_regular(path)
    try:
        if maximum_bytes is not None and opened.st_size > maximum_bytes:
            fail(f"File exceeds its read budget: {path}")
        chunks: list[bytes] = []
        total = 0
        while True:
            block = os.read(descriptor, 1024 * 1024)
            if not block:
                break
            total += len(block)
            if maximum_bytes is not None and total > maximum_bytes:
                fail(f"File exceeds its read budget: {path}")
            chunks.append(block)
        _verify_open_file_stable(path, descriptor, opened)
        data = b"".join(chunks)
        if len(data) != opened.st_size:
            fail(f"File size changed while being read: {path}")
        return data
    finally:
        os.close(descriptor)


def hash_file(path: Path, algorithm: str = "sha256") -> str:
    digest = hashlib.new(algorithm)
    descriptor, opened = _secure_open_regular(path)
    try:
        while True:
            block = os.read(descriptor, 1024 * 1024)
            if not block:
                break
            digest.update(block)
        _verify_open_file_stable(path, descriptor, opened)
    finally:
        os.close(descriptor)
    return digest.hexdigest()


def hash_file_md5_sha256(path: Path) -> tuple[int, str, str]:
    md5 = hashlib.md5(usedforsecurity=False)
    sha256 = hashlib.sha256()
    descriptor, opened = _secure_open_regular(path)
    try:
        while True:
            block = os.read(descriptor, 1024 * 1024)
            if not block:
                break
            md5.update(block)
            sha256.update(block)
        _verify_open_file_stable(path, descriptor, opened)
    finally:
        os.close(descriptor)
    return opened.st_size, md5.hexdigest(), sha256.hexdigest()


def csv_bytes(fieldnames: tuple[str, ...], rows: list[dict[str, object]]) -> bytes:
    output = io.StringIO(newline="")
    writer = csv.DictWriter(output, fieldnames=fieldnames, lineterminator="\n")
    writer.writeheader()
    writer.writerows(rows)
    return output.getvalue().encode("utf-8")


def read_csv(
    relative_path: str, snapshot: dict[str, bytes] | None = None
) -> list[dict[str, str]]:
    data = source_bytes(relative_path) if snapshot is None else snapshot_bytes(snapshot, relative_path)
    try:
        text = data.decode("utf-8")
    except UnicodeDecodeError:
        fail(f"CSV source is not UTF-8: {relative_path}")
    return list(csv.DictReader(io.StringIO(text, newline="")))


def normalized_relative_name(name: str) -> str:
    if not name or "\x00" in name or "\\" in name or unicodedata.normalize("NFC", name) != name:
        fail(f"Unsafe relative path: {name!r}")
    path = PurePosixPath(name)
    if path.is_absolute() or any(part in {"", ".", ".."} for part in path.parts):
        fail(f"Unsafe relative path: {name!r}")
    if any(part != part.strip() or ":" in part for part in path.parts):
        fail(f"Unsafe relative path component: {name!r}")
    return path.as_posix()


def normalized_archive_name(name: str) -> str:
    normalized = normalized_relative_name(name)
    lowered = normalized.casefold()
    if any(lowered.startswith(prefix) for prefix in FORBIDDEN_ARCHIVE_PREFIXES):
        fail(f"Forbidden archive member: {name}")
    return normalized


def validate_archive_names(names: list[str]) -> None:
    normalized = [normalized_archive_name(name) for name in names]
    if any(original != canonical for original, canonical in zip(names, normalized)):
        fail("Archive contains a noncanonical member path")
    if len(normalized) != len(set(normalized)):
        fail("Duplicate archive path")
    folded = [unicodedata.normalize("NFC", name).casefold() for name in normalized]
    if len(folded) != len(set(folded)):
        fail("Case-insensitive or Unicode archive-path collision")


def validate_zip_infos(infos: list[zipfile.ZipInfo], label: str) -> None:
    if not infos or len(infos) > MAX_ARCHIVE_MEMBERS:
        fail(f"{label} member count is outside the safety budget")
    names = [info.filename for info in infos]
    validate_archive_names(names)
    total_uncompressed = 0
    for info in infos:
        if len(info.filename.encode("utf-8")) > MAX_ARCHIVE_NAME_BYTES:
            fail(f"{label} member name exceeds the safety budget: {info.filename}")
        if info.filename.endswith("/") or info.is_dir():
            fail(f"{label} contains an unexpected directory member: {info.filename}")
        if info.flag_bits & 0x1:
            fail(f"{label} contains an encrypted member: {info.filename}")
        if info.compress_type not in ALLOWED_ZIP_COMPRESSION:
            fail(f"{label} uses an unsupported compression method: {info.filename}")
        mode = info.external_attr >> 16
        if stat.S_ISLNK(mode):
            fail(f"{label} member is a symlink: {info.filename}")
        if info.file_size < 0 or info.file_size > MAX_ARCHIVE_MEMBER_BYTES:
            fail(f"{label} member exceeds the uncompressed-size budget: {info.filename}")
        total_uncompressed += info.file_size
        if total_uncompressed > MAX_ARCHIVE_UNCOMPRESSED_BYTES:
            fail(f"{label} exceeds the total uncompressed-size budget")
        if info.file_size > 0 and info.compress_size <= 0:
            fail(f"{label} has an invalid compressed size: {info.filename}")
        if info.file_size >= 100_000:
            ratio = info.file_size / max(1, info.compress_size)
            if ratio > MAX_ARCHIVE_COMPRESSION_RATIO:
                fail(f"{label} member exceeds the compression-ratio budget: {info.filename}")


def require_regular_source(relative_path: str) -> Path:
    archive_style = normalized_relative_name(relative_path)
    candidate = ROOT
    for part in PurePosixPath(archive_style).parts:
        candidate = candidate / part
        try:
            info = candidate.lstat()
        except FileNotFoundError as error:
            fail(f"Required source is missing: {relative_path}")
            raise error
        if stat.S_ISLNK(info.st_mode):
            fail(f"Symlink is forbidden in package source: {relative_path}")
    if not stat.S_ISREG(candidate.stat().st_mode):
        fail(f"Package source is not a regular file: {relative_path}")
    try:
        candidate.resolve().relative_to(ROOT.resolve())
    except ValueError:
        fail(f"Package source escapes project root: {relative_path}")
    return candidate


def source_bytes(relative_path: str) -> bytes:
    path = require_regular_source(relative_path)
    data = secure_file_bytes(path)
    scan_payload(relative_path, data)
    return data


def snapshot_sources(relative_paths: tuple[str, ...]) -> dict[str, bytes]:
    if len(relative_paths) != len(set(relative_paths)):
        fail("Source snapshot contains duplicate paths")
    return {relative: source_bytes(relative) for relative in relative_paths}


def snapshot_bytes(snapshot: dict[str, bytes], relative_path: str) -> bytes:
    try:
        return snapshot[relative_path]
    except KeyError:
        fail(f"Source snapshot lacks required input: {relative_path}")
    raise AssertionError("unreachable")


def private_deny_tokens() -> tuple[str, ...]:
    supplied = os.environ.get("ANONYMOUS_PACKAGE_DENYLIST", "")
    tokens = [token.strip().casefold() for token in supplied.split(",") if len(token.strip()) >= 3]
    return tuple(sorted(set(tokens)))


def scan_payload(relative_path: str, data: bytes) -> None:
    suffix = Path(relative_path).suffix.casefold()
    local_tokens = {
        str(Path.home()).casefold(),
        getpass.getuser().casefold(),
        socket.gethostname().casefold(),
    }
    lower_data = data.lower()
    for token in local_tokens:
        if token and token.encode("utf-8", errors="ignore") in lower_data:
            fail(f"Machine-identifying token detected in {relative_path}")
    if suffix not in TEXT_SUFFIXES:
        binary_local_markers = (
            b"/" + b"users" + b"/",
            b"/" + b"home" + b"/",
            b"file" + b"://",
        )
        for marker in binary_local_markers:
            if marker in lower_data:
                fail(f"Local path detected in binary payload {relative_path}")
        return
    try:
        text = data.decode("utf-8")
    except UnicodeDecodeError:
        fail(f"Allowlisted text file is not UTF-8: {relative_path}")
    for pattern in LOCAL_PATH_PATTERNS:
        if pattern.search(text):
            fail(f"Absolute or local path detected in {relative_path}")
    if relative_path not in PII_SCAN_EXEMPT:
        if EMAIL_PATTERN.search(text) or ORCID_PATTERN.search(text):
            fail(f"Email-like or ORCID-like PII detected in {relative_path}")
        if any(pattern.search(text) for pattern in PII_PLACEHOLDER_PATTERNS):
            fail(f"PII placeholder detected in {relative_path}")
        folded = text.casefold()
        for token in private_deny_tokens():
            if token in folded:
                fail(f"Private denylist token detected in {relative_path}")
    if "session_info.txt" in relative_path:
        fail("Machine-specific session_info.txt is forbidden")


def parse_truth(value: str) -> bool:
    return value.strip().casefold() in {"true", "t", "1", "yes"}


def png_dimensions_bytes(data: bytes, label: str) -> tuple[int, int]:
    header = data[:24]
    if len(header) != 24 or header[:8] != b"\x89PNG\r\n\x1a\n" or header[12:16] != b"IHDR":
        fail(f"Invalid PNG signature: {label}")
    return int.from_bytes(header[16:20], "big"), int.from_bytes(header[20:24], "big")


def validate_publication_artifacts(snapshot: dict[str, bytes]) -> None:
    output_checks = read_csv("outputs/diagnostics/publication_output_validation.csv", snapshot)
    if len(output_checks) != 17 or not all(parse_truth(row["passed"]) for row in output_checks):
        fail("The 17 publication-output gates are not all present and passing")

    table_manifest = read_csv("outputs/tables/table_manifest.csv", snapshot)
    if len(table_manifest) != 19:
        fail("Table manifest must contain exactly 19 publication tables")
    manifest_files = tuple(row["file"] for row in table_manifest)
    if manifest_files != TABLE_FILES[1:]:
        fail("Table manifest filenames or order differ from the frozen allowlist")
    if tuple(row["number"] for row in table_manifest) != EXPECTED_TABLE_NUMBERS:
        fail("Table publication numbering differs from the frozen catalog")
    roles = [row["role"] for row in table_manifest]
    if roles.count("Main") != 3 or roles.count("Supplement") != 16:
        fail("Table roles are not exactly three main and 16 supplement")
    table_ids = {row["table_id"] for row in table_manifest}
    if len(table_ids) != len(table_manifest):
        fail("Table manifest contains duplicate table IDs")

    table_checks = read_csv("outputs/diagnostics/table_artifact_validation.csv", snapshot)
    if len(table_checks) != 19 or not all(parse_truth(row["passed"]) for row in table_checks):
        fail("The 19 table artifact validations are not all present and passing")
    checks_by_artifact = {row["artifact"]: row for row in table_checks}
    if set(checks_by_artifact) != table_ids or len(checks_by_artifact) != len(table_checks):
        fail("Table artifact validation IDs differ from the table manifest")
    for row in table_manifest:
        relative = f"outputs/tables/{row['file']}"
        data = snapshot_bytes(snapshot, relative)
        check = checks_by_artifact[row["table_id"]]
        reader = csv.reader(io.StringIO(data.decode("utf-8"), newline=""))
        header = next(reader, None)
        observed_rows = sum(1 for _ in reader)
        if header is None or observed_rows != int(row["declared_rows"]):
            fail(f"Table row-count mismatch: {relative}")
        if observed_rows != int(check["observed_rows"]) or len(header) != int(check["observed_columns"]):
            fail(f"Table schema differs from its validation record: {relative}")
        if len(data) != int(check["file_bytes"]):
            fail(f"Table bytes differ from its validation record: {relative}")

    figure_manifest = read_csv("outputs/figures/figure_manifest.csv", snapshot)
    if len(figure_manifest) != 10:
        fail("Figure manifest must contain exactly ten figures")
    if tuple(row["figure_id"] for row in figure_manifest) != FIGURE_STEMS:
        fail("Figure IDs or order differ from the frozen allowlist")
    if tuple(row["number"] for row in figure_manifest) != EXPECTED_FIGURE_NUMBERS:
        fail("Figure publication numbering differs from the frozen catalog")
    if [row["role"] for row in figure_manifest].count("Main") != 4 or [row["role"] for row in figure_manifest].count("Supplement") != 6:
        fail("Figure roles are not exactly four main and six supplement")

    figure_checks = read_csv("outputs/diagnostics/figure_artifact_validation.csv", snapshot)
    if len(figure_checks) != 10 or not all(parse_truth(row["passed"]) for row in figure_checks):
        fail("Figure artifact validations are not all present and passing")
    figure_checks_by_artifact = {row["artifact"]: row for row in figure_checks}
    if set(figure_checks_by_artifact) != set(FIGURE_STEMS) or len(figure_checks_by_artifact) != len(figure_checks):
        fail("Figure artifact validation IDs differ from the figure manifest")
    for row in figure_manifest:
        stem = row["figure_id"]
        expected_pdf = f"outputs/figures/{stem}.pdf"
        expected_png = f"outputs/figures/{stem}.png"
        if row["pdf_path"] != expected_pdf or row["png_path"] != expected_png:
            fail(f"Figure paths differ from frozen conventions: {stem}")
        pdf_data = snapshot_bytes(snapshot, expected_pdf)
        png_data = snapshot_bytes(snapshot, expected_png)
        if not pdf_data.startswith(b"%PDF-"):
            fail(f"Invalid PDF signature: {expected_pdf}")
        width, height = png_dimensions_bytes(png_data, expected_png)
        check = figure_checks_by_artifact[stem]
        if len(pdf_data) != int(check["pdf_bytes"]) or len(png_data) != int(check["png_bytes"]):
            fail(f"Figure bytes differ from validation record: {stem}")
        if width != int(check["png_width_pixels"]) or height != int(check["png_height_pixels"]):
            fail(f"Figure dimensions differ from validation record: {stem}")
        if width < 2000 or height < 1200:
            fail(f"Figure resolution is below the frozen minimum: {stem}")

    producer_sources = [require_regular_source(path) for path in ANALYSIS_FILES]
    validation_path = require_regular_source("outputs/diagnostics/publication_output_validation.csv")
    newest_producer = max(path.stat().st_mtime_ns for path in producer_sources)
    required_outputs = [
        require_regular_source(f"outputs/tables/{name}")
        for name in TABLE_FILES + REPRODUCIBILITY_ONLY_TABLE_FILES
    ] + [
        require_regular_source(f"outputs/figures/{name}") for name in FIGURE_FILES
    ] + [validation_path]
    if min(path.stat().st_mtime_ns for path in required_outputs) < newest_producer:
        fail("A publication artifact predates an analysis producer; rerun analysis/run_all.R")
    derived_rds = tuple((ROOT / "data/derived").glob("*.rds"))
    if derived_rds and max(path.stat().st_mtime_ns for path in derived_rds) > validation_path.stat().st_mtime_ns:
        fail("A derived analysis object is newer than publication validation; rerun analysis/06_outputs.R")


def run_version_command(command: list[str]) -> str:
    result = subprocess.run(command, cwd=ROOT, capture_output=True, text=True, check=True)
    return (result.stdout + "\n" + result.stderr).strip()


def sanitized_environment_bytes(recorded_data: bytes | None = None) -> bytes:
    r_output = run_version_command(["Rscript", "--version"])
    quarto_output = run_version_command(["quarto", "--version"])
    pandoc_output = run_version_command(["quarto", "pandoc", "--version"])
    tectonic_output = run_version_command(["tectonic", "--version"])
    r_match = re.search(r"version\s+(\d+\.\d+\.\d+)", r_output, re.I)
    quarto_match = re.search(r"\b(\d+\.\d+\.\d+)\b", quarto_output)
    pandoc_match = re.search(r"pandoc\s+(\d+\.\d+(?:\.\d+)?)", pandoc_output, re.I)
    tectonic_match = re.search(r"\bTectonic\s+(\d+\.\d+\.\d+)\b", tectonic_output, re.I)
    if not r_match or not quarto_match or not pandoc_match or not tectonic_match:
        fail("Could not parse generic runtime versions")
    text = (
        f"R: {r_match.group(1)}\n"
        f"Python: {sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}\n"
        f"Quarto: {quarto_match.group(1)}\n"
        f"Pandoc (bundled with Quarto): {pandoc_match.group(1)}\n"
        f"Tectonic: {tectonic_match.group(1)}\n"
        "R dependencies: base packages only\n"
        "Python dependencies: standard library only\n"
    ).encode("utf-8")
    recorded = recorded_data if recorded_data is not None else source_bytes(SANITIZED_ENVIRONMENT_SOURCE)
    if recorded != text:
        fail(f"{SANITIZED_ENVIRONMENT_SOURCE} does not match the current sanitized runtime capture")
    return text


def official_source_checksum_bytes(manifest_data: bytes | None = None) -> bytes:
    if manifest_data is None:
        manifest_data = source_bytes("data/raw/download_manifest.csv")
    manifest_rows = list(csv.DictReader(io.StringIO(manifest_data.decode("utf-8"), newline="")))
    rows_by_key = {row["key"]: row for row in manifest_rows}
    if set(rows_by_key) != set(RAW_SOURCE_FILENAMES) or len(manifest_rows) != len(RAW_SOURCE_FILENAMES):
        fail("Raw download manifest does not contain the exact 16 frozen official sources")
    output_rows: list[dict[str, object]] = []
    for key in sorted(RAW_SOURCE_FILENAMES):
        expected_filename = RAW_SOURCE_FILENAMES[key]
        row = rows_by_key[key]
        if row["filename"] != expected_filename:
            fail(f"Raw filename mismatch for {key}")
        if row["expected_md5"].casefold() != row["md5"].casefold():
            fail(f"Observed and frozen MD5 disagree for {key}")
        if not row["url"].startswith("https://timss2023.org/"):
            fail(f"Unexpected official-source host for {key}")
        source = require_regular_source(f"data/raw/{expected_filename}")
        observed_size, observed_md5, observed_sha256 = hash_file_md5_sha256(source)
        if observed_size != int(row["bytes"]):
            fail(f"Raw source byte count differs from manifest for {key}")
        if observed_md5 != row["expected_md5"].casefold():
            fail(f"Raw source MD5 differs from frozen fingerprint for {key}")
        output_rows.append({
            "key": key,
            "filename": expected_filename,
            "url": row["url"],
            "bytes": observed_size,
            "md5": observed_md5,
            "sha256": observed_sha256,
        })
    return csv_bytes(("key", "filename", "url", "bytes", "md5", "sha256"), output_rows)


def zip_info(name: str) -> zipfile.ZipInfo:
    info = zipfile.ZipInfo(normalized_archive_name(name), FIXED_ZIP_TIME)
    info.create_system = 3
    info.external_attr = (stat.S_IFREG | 0o644) << 16
    info.compress_type = zipfile.ZIP_DEFLATED
    info.extra = b""
    info.comment = b""
    return info


def source_snapshot_manifest(snapshot: dict[str, bytes]) -> bytes:
    rows = [
        {
            "source_path": name,
            "bytes": len(snapshot[name]),
            "sha256": sha256_bytes(snapshot[name]),
        }
        for name in sorted(snapshot)
    ]
    return csv_bytes(("source_path", "bytes", "sha256"), rows)


def checksum_manifest(entries: dict[str, bytes]) -> bytes:
    rows = [
        {"archive_path": name, "bytes": len(entries[name]), "sha256": sha256_bytes(entries[name])}
        for name in sorted(entries)
    ]
    return csv_bytes(("archive_path", "bytes", "sha256"), rows)


def make_zip(entries_without_manifest: dict[str, bytes]) -> bytes:
    if "SHA256SUMS.csv" in entries_without_manifest:
        fail("SHA256SUMS.csv is reserved for the generated archive manifest")
    validate_archive_names(list(entries_without_manifest))
    entries = dict(entries_without_manifest)
    entries["SHA256SUMS.csv"] = checksum_manifest(entries_without_manifest)
    validate_archive_names(list(entries))
    output = io.BytesIO()
    with zipfile.ZipFile(output, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as archive:
        for name in sorted(entries):
            archive.writestr(zip_info(name), entries[name], compress_type=zipfile.ZIP_DEFLATED, compresslevel=9)
    data = output.getvalue()
    validate_zip_bytes(data, set(entries_without_manifest))
    return data


def validate_zip_bytes(
    data: bytes,
    expected_payload_names: set[str],
    expected_entries: dict[str, bytes] | None = None,
) -> None:
    if len(data) >= MAX_ADDITIONAL_FILE_BYTES:
        fail("Archive is not below the Additional File size limit")
    validate_archive_names(sorted(expected_payload_names))
    if expected_entries is not None and set(expected_entries) != expected_payload_names:
        fail("Expected archive byte mapping does not match the expected payload names")
    with zipfile.ZipFile(io.BytesIO(data), "r") as archive:
        if archive.comment:
            fail("Archive-level comments are forbidden")
        infos = archive.infolist()
        validate_zip_infos(infos, "Additional File archive")
        names = [info.filename for info in infos]
        if len(names) != len(set(names)):
            fail("Archive contains duplicate names")
        expected_names = expected_payload_names | {"SHA256SUMS.csv"}
        if set(names) != expected_names:
            missing = sorted(expected_names - set(names))
            extra = sorted(set(names) - expected_names)
            fail(f"Archive allowlist mismatch; missing={missing}, extra={extra}")
        for info in infos:
            if (
                info.date_time != FIXED_ZIP_TIME
                or info.extra
                or info.comment
                or info.compress_type != zipfile.ZIP_DEFLATED
            ):
                fail(f"Archive member lacks deterministic builder metadata: {info.filename}")
        if archive.testzip() is not None:
            fail("Archive CRC validation failed")
        manifest_data = archive.read("SHA256SUMS.csv")
        try:
            manifest_rows = list(csv.DictReader(io.StringIO(manifest_data.decode("utf-8"))))
        except (UnicodeDecodeError, csv.Error):
            fail("Internal SHA-256 manifest is not valid UTF-8 CSV")
        if not manifest_rows or tuple(manifest_rows[0]) != ("archive_path", "bytes", "sha256"):
            fail("Internal SHA-256 manifest schema is invalid")
        if {row["archive_path"] for row in manifest_rows} != expected_payload_names:
            fail("Internal SHA-256 manifest paths differ from payload paths")
        if len(manifest_rows) != len(expected_payload_names):
            fail("Internal SHA-256 manifest has duplicate or missing rows")
        for row in manifest_rows:
            if not re.fullmatch(r"[0-9a-f]{64}", row["sha256"]):
                fail(f"Invalid internal SHA-256 value: {row['archive_path']}")
            member = archive.read(row["archive_path"])
            if len(member) != int(row["bytes"]) or sha256_bytes(member) != row["sha256"]:
                fail(f"Internal SHA-256 verification failed: {row['archive_path']}")
            if expected_entries is not None and member != expected_entries[row["archive_path"]]:
                fail(f"Archive member differs from the current source bytes: {row['archive_path']}")
        if expected_entries is not None and manifest_data != checksum_manifest(expected_entries):
            fail("Internal SHA-256 manifest differs from the regenerated current-source manifest")


def validate_self_manifest_zip(data: bytes) -> set[str]:
    with zipfile.ZipFile(io.BytesIO(data), "r") as archive:
        infos = archive.infolist()
        validate_zip_infos(infos, "Existing Additional File archive")
        names = {info.filename for info in infos}
    if "SHA256SUMS.csv" not in names:
        fail("Existing archive lacks its internal SHA-256 manifest")
    payload_names = names - {"SHA256SUMS.csv"}
    validate_zip_bytes(data, payload_names)
    return payload_names


WORDPROCESSINGML_NAMESPACE = "http://schemas.openxmlformats.org/wordprocessingml/2006/main"
OFFICE_RELATIONSHIPS_NAMESPACE = "http://schemas.openxmlformats.org/officeDocument/2006/relationships"
PACKAGE_RELATIONSHIPS_NAMESPACE = "http://schemas.openxmlformats.org/package/2006/relationships"
CONTENT_TYPES_NAMESPACE = "http://schemas.openxmlformats.org/package/2006/content-types"
COMMENTS_RELATIONSHIP_TYPE = f"{OFFICE_RELATIONSHIPS_NAMESPACE}/comments"
HYPERLINK_RELATIONSHIP_SUFFIX = "/hyperlink"
COMMENTS_CONTENT_TYPE = "application/vnd.openxmlformats-officedocument.wordprocessingml.comments+xml"
COMMENT_ANCHOR_TAGS = {
    f"{{{WORDPROCESSINGML_NAMESPACE}}}commentRangeStart",
    f"{{{WORDPROCESSINGML_NAMESPACE}}}commentRangeEnd",
    f"{{{WORDPROCESSINGML_NAMESPACE}}}commentReference",
}
REVISION_TAGS = {
    f"{{{WORDPROCESSINGML_NAMESPACE}}}{local_name}"
    for local_name in ("ins", "del", "moveFrom", "moveTo", "trackRevisions")
}
FORBIDDEN_OPC_TYPE_MARKERS = (
    "macroenabled",
    "vbaproject",
    "oleobject",
    "activex",
    "attachedtemplate",
    "externallink",
    "digital-signature",
    "digitalsignature",
    "customxml",
    "custom-properties",
)


def parse_xml_bytes(data: bytes, name: str) -> ET.Element:
    upper = data.upper()
    if b"<!DOCTYPE" in upper or b"<!ENTITY" in upper:
        fail(f"DOCX XML declarations or entities are forbidden: {name}")
    try:
        return ET.fromstring(data)
    except ET.ParseError:
        fail(f"DOCX contains malformed XML: {name}")
    raise AssertionError("unreachable")


def parse_docx_xml(members: dict[str, bytes], name: str) -> ET.Element:
    if name not in members:
        fail(f"DOCX lacks required XML part: {name}")
    return parse_xml_bytes(members[name], name)


def relationship_source_part(relationship_part: str) -> str | None:
    if relationship_part == "_rels/.rels":
        return None
    path = PurePosixPath(relationship_part)
    if path.suffix.casefold() != ".rels" or path.parent.name != "_rels":
        fail(f"Relationship part is not in an OPC _rels directory: {relationship_part}")
    source_name = path.name[:-5]
    if not source_name:
        fail(f"Relationship part has no source filename: {relationship_part}")
    parent = path.parent.parent
    return (parent / source_name).as_posix() if parent.as_posix() != "." else source_name


def resolve_internal_opc_target(source: str | None, target: str) -> str:
    if not target or "\\" in target or "?" in target or "#" in target or "\x00" in target:
        fail(f"Unsafe internal OPC relationship target: {target!r}")
    decoded = unquote(target)
    if "\\" in decoded or "\x00" in decoded:
        fail(f"Unsafe encoded OPC relationship target: {target!r}")
    base = "" if source is None else posixpath.dirname(source)
    resolved = (
        posixpath.normpath(decoded).lstrip("/")
        if decoded.startswith("/")
        else posixpath.normpath(posixpath.join(base, decoded))
    )
    if not resolved or resolved in {".", ".."} or resolved.startswith("../"):
        fail(f"OPC relationship target escapes the package: {target!r}")
    return normalized_relative_name(resolved)


def validate_external_opc_target(relationship_type: str, target: str) -> None:
    if not relationship_type.casefold().endswith(HYPERLINK_RELATIONSHIP_SUFFIX):
        fail(f"Only hyperlink relationships may target external resources: {relationship_type}")
    if any(character.isspace() for character in target):
        fail("External hyperlink target contains whitespace")
    parsed = urlsplit(target)
    if (
        parsed.scheme.casefold() != "https"
        or not parsed.hostname
        or parsed.username is not None
        or parsed.password is not None
    ):
        fail(f"External hyperlink target is not an anonymous HTTPS URL: {target!r}")
    lowered = target.casefold()
    if any(pattern.search(target) for pattern in LOCAL_PATH_PATTERNS) or lowered.startswith(("file:", "data:", "javascript:")):
        fail(f"Unsafe external hyperlink target: {target!r}")


def validate_opc_package(members: dict[str, bytes]) -> None:
    """Validate OPC content-type coverage, relationship closure, and external-target policy."""
    required = {"[Content_Types].xml", "_rels/.rels", "word/document.xml"}
    missing = sorted(required - set(members))
    if missing:
        fail(f"DOCX lacks required OPC parts: {missing}")
    validate_archive_names(list(members))

    content_types = parse_docx_xml(members, "[Content_Types].xml")
    if content_types.tag != f"{{{CONTENT_TYPES_NAMESPACE}}}Types":
        fail("DOCX content-types part has an unexpected root")
    defaults: dict[str, str] = {}
    overrides: dict[str, str] = {}
    for declaration in content_types:
        content_type = declaration.attrib.get("ContentType", "")
        if not content_type or any(marker in content_type.casefold() for marker in FORBIDDEN_OPC_TYPE_MARKERS):
            fail("DOCX contains an invalid or active content type")
        if declaration.tag == f"{{{CONTENT_TYPES_NAMESPACE}}}Default":
            extension = declaration.attrib.get("Extension", "").casefold()
            if not extension or extension.startswith(".") or extension in defaults:
                fail("DOCX contains an invalid or duplicate default content type")
            defaults[extension] = content_type
        elif declaration.tag == f"{{{CONTENT_TYPES_NAMESPACE}}}Override":
            part_name = declaration.attrib.get("PartName", "")
            if not part_name.startswith("/"):
                fail("DOCX override content type lacks an absolute package part name")
            normalized = normalized_relative_name(part_name.lstrip("/"))
            if normalized != part_name.lstrip("/") or normalized in overrides:
                fail("DOCX contains an invalid or duplicate override content type")
            if normalized not in members:
                fail(f"DOCX content-type override targets a missing part: {normalized}")
            overrides[normalized] = content_type
        else:
            fail("DOCX content-types part contains an unknown declaration")

    for name in members:
        if name == "[Content_Types].xml":
            continue
        filename = PurePosixPath(name).name
        extension = filename.rsplit(".", 1)[-1].casefold() if "." in filename else ""
        if name not in overrides and extension not in defaults:
            fail(f"DOCX member lacks content-type coverage: {name}")

    relationship_parts = sorted(name for name in members if name.casefold().endswith(".rels"))
    relationships_by_source: dict[str | None, list[str]] = {}
    relationship_sources: set[str | None] = set()
    for relationship_part in relationship_parts:
        source = relationship_source_part(relationship_part)
        if source is not None and source not in members:
            fail(f"DOCX relationship part has a missing source: {relationship_part}")
        if source in relationship_sources:
            fail(f"DOCX has duplicate relationship parts for source: {source}")
        relationship_sources.add(source)
        root = parse_docx_xml(members, relationship_part)
        if root.tag != f"{{{PACKAGE_RELATIONSHIPS_NAMESPACE}}}Relationships":
            fail(f"DOCX relationship part has an unexpected root: {relationship_part}")
        ids: set[str] = set()
        internal_targets: list[str] = []
        for relationship in root:
            if relationship.tag != f"{{{PACKAGE_RELATIONSHIPS_NAMESPACE}}}Relationship":
                fail(f"DOCX relationship part contains an unknown element: {relationship_part}")
            relationship_id = relationship.attrib.get("Id", "")
            relationship_type = relationship.attrib.get("Type", "")
            target = relationship.attrib.get("Target", "")
            target_mode = relationship.attrib.get("TargetMode", "")
            if not relationship_id or relationship_id in ids or not relationship_type or not target:
                fail(f"DOCX relationship is incomplete or duplicated: {relationship_part}")
            ids.add(relationship_id)
            if any(marker in relationship_type.casefold() for marker in FORBIDDEN_OPC_TYPE_MARKERS):
                fail(f"DOCX contains an active relationship type: {relationship_type}")
            if target_mode.casefold() == "external":
                validate_external_opc_target(relationship_type, target)
            elif target_mode.casefold() in {"", "internal"}:
                resolved = resolve_internal_opc_target(source, target)
                if resolved not in members:
                    fail(f"DOCX relationship targets a missing part: {resolved}")
                internal_targets.append(resolved)
            else:
                fail(f"DOCX relationship uses an unsupported TargetMode: {target_mode}")
        relationships_by_source[source] = internal_targets

    if None not in relationships_by_source:
        fail("DOCX lacks package-root relationships")
    reachable: set[str] = set()
    queue = list(relationships_by_source[None])
    while queue:
        part = queue.pop()
        if part in reachable:
            continue
        reachable.add(part)
        queue.extend(relationships_by_source.get(part, []))
    payload_parts = {
        name
        for name in members
        if name != "[Content_Types].xml" and not name.casefold().endswith(".rels")
    }
    unreachable = sorted(payload_parts - reachable)
    if unreachable:
        fail(f"DOCX contains OPC parts unreachable from package relationships: {unreachable}")
    orphan_relationship_sources = sorted(
        source for source in relationship_sources if source is not None and source not in reachable
    )
    if orphan_relationship_sources:
        fail(f"DOCX contains relationship graphs for unreachable parts: {orphan_relationship_sources}")


def validate_sanitized_docx_features(members: dict[str, bytes]) -> None:
    """Reject review, identity, signature, custom-data, macro, and embedded-object features."""
    forbidden_member_markers = (
        "comments",
        "people",
        "person.xml",
        "vbaproject",
        "embeddings/",
        "_xmlsignatures/",
        "customxml/",
    )
    for name, content in members.items():
        lowered = name.casefold()
        if lowered == "docprops/custom.xml" or any(
            marker in lowered for marker in forbidden_member_markers
        ):
            fail(
                "DOCX contains forbidden comments, identities, signatures, custom data, "
                f"macros, or embedded objects: {name}"
            )
        if lowered.endswith((".xml", ".rels")):
            parse_xml_bytes(content, name)

    for name, content in members.items():
        lowered = name.casefold()
        if lowered.startswith("word/") and lowered.endswith(".xml"):
            root = parse_xml_bytes(content, name)
            if any(element.tag in COMMENT_ANCHOR_TAGS for element in root.iter()):
                fail(f"DOCX contains a comment anchor: {name}")
            if any(element.tag in REVISION_TAGS for element in root.iter()):
                fail(f"DOCX contains tracked revisions or revision settings: {name}")
        if lowered.endswith(".rels"):
            root = parse_xml_bytes(content, name)
            for relationship in root.iter(f"{{{PACKAGE_RELATIONSHIPS_NAMESPACE}}}Relationship"):
                if "comment" in relationship.attrib.get("Type", "").casefold():
                    fail(f"DOCX contains a comment relationship: {name}")

    content_types = parse_docx_xml(members, "[Content_Types].xml")
    for element in content_types:
        declaration = " ".join(
            (element.attrib.get("PartName", ""), element.attrib.get("ContentType", ""))
        ).casefold()
        if "comment" in declaration:
            fail("DOCX contains a comment content-type declaration")

    validate_opc_package(members)


def remove_empty_docx_comment_scaffold(members: dict[str, bytes]) -> None:
    """Remove only Pandoc's empty classic-comments package scaffold."""
    comments_part = "word/comments.xml"
    comment_named_parts = sorted(
        name for name in members if "comment" in PurePosixPath(name).name.casefold()
    )
    if not comment_named_parts:
        return
    if comment_named_parts != [comments_part]:
        fail("DOCX contains an unsupported or extended comments feature")

    comments_root = parse_docx_xml(members, comments_part)
    if comments_root.tag != f"{{{WORDPROCESSINGML_NAMESPACE}}}comments":
        fail("DOCX comments part has an unexpected root element")
    if next(comments_root.iter(f"{{{WORDPROCESSINGML_NAMESPACE}}}comment"), None) is not None:
        fail("DOCX contains real comments; remove them before packaging")

    for name, content in members.items():
        lowered = name.casefold()
        if name == comments_part or not (lowered.startswith("word/") and lowered.endswith(".xml")):
            continue
        root = parse_xml_bytes(content, name)
        if any(element.tag in COMMENT_ANCHOR_TAGS for element in root.iter()):
            fail("DOCX contains comment anchors; remove the comments before packaging")

    document_relationships_name = "word/_rels/document.xml.rels"
    comment_relationships: list[tuple[str, ET.Element, ET.Element]] = []
    relationship_roots: dict[str, ET.Element] = {}
    for name, content in members.items():
        if not name.casefold().endswith(".rels"):
            continue
        root = parse_xml_bytes(content, name)
        relationship_roots[name] = root
        for relationship in root.iter(f"{{{PACKAGE_RELATIONSHIPS_NAMESPACE}}}Relationship"):
            if "comment" in relationship.attrib.get("Type", "").casefold():
                comment_relationships.append((name, root, relationship))

    if len(comment_relationships) != 1:
        fail("Empty DOCX comments scaffold has an unexpected relationship graph")
    rels_name, rels_root, relationship = comment_relationships[0]
    target = relationship.attrib.get("Target", "")
    target_mode = relationship.attrib.get("TargetMode", "")
    if (
        rels_name != document_relationships_name
        or relationship.attrib.get("Type") != COMMENTS_RELATIONSHIP_TYPE
        or target_mode.casefold() == "external"
        or not target
        or "\\" in target
        or any(marker in target for marker in ("?", "#"))
    ):
        fail("Empty DOCX comments scaffold has an unsupported relationship")
    resolved_target = (
        posixpath.normpath(target).lstrip("/")
        if target.startswith("/")
        else posixpath.normpath(posixpath.join("word", target))
    )
    if resolved_target != comments_part:
        fail("Empty DOCX comments relationship targets an unexpected part")

    content_types_name = "[Content_Types].xml"
    content_types_root = parse_docx_xml(members, content_types_name)
    comment_declarations = [
        element
        for element in content_types_root
        if "comment" in " ".join(
            (element.attrib.get("PartName", ""), element.attrib.get("ContentType", ""))
        ).casefold()
    ]
    if len(comment_declarations) != 1:
        fail("Empty DOCX comments scaffold has an unexpected content-type graph")
    declaration = comment_declarations[0]
    if (
        declaration.tag != f"{{{CONTENT_TYPES_NAMESPACE}}}Override"
        or declaration.attrib.get("PartName") != "/word/comments.xml"
        or declaration.attrib.get("ContentType") != COMMENTS_CONTENT_TYPE
    ):
        fail("Empty DOCX comments scaffold has an unsupported content type")

    rels_root.remove(relationship)
    content_types_root.remove(declaration)
    del members[comments_part]
    members[document_relationships_name] = ET.tostring(
        rels_root, encoding="utf-8", xml_declaration=True
    )
    members[content_types_name] = ET.tostring(
        content_types_root, encoding="utf-8", xml_declaration=True
    )
    validate_sanitized_docx_features(members)


def validate_finalized_supplement_members(members: dict[str, bytes]) -> None:
    try:
        from finalize_docx import (
            DOCUMENTS,
            source_alt_texts,
            source_display_equations,
            validate_formatting,
        )
    except ImportError as error:
        fail(f"Could not import shared DOCX finalization validator: {error}")
    definition = DOCUMENTS["supplement"]
    source_display_equations(definition["source"], definition["expected_equations"])
    alternatives = source_alt_texts(definition["source"], definition["expected_figures"])
    validate_formatting(
        members,
        alternatives,
        definition["figure_prefix"],
        definition["expected_equations"],
        definition["equation_markers"],
        True,
    )


def validate_prior_supplement_for_replacement(members: dict[str, bytes]) -> None:
    """Validate a prior builder DOCX while tolerating only canonical legacy custom properties."""
    for name, content in members.items():
        if name.casefold().endswith((".xml", ".rels")):
            parse_xml_bytes(content, name)
    try:
        from finalize_docx import remove_custom_properties
    except ImportError as error:
        fail(f"Could not import legacy DOCX compatibility sanitizer: {error}")
    compatible_members = remove_custom_properties(dict(members))
    validate_sanitized_docx_features(compatible_members)


def sanitize_docx(source: Path, source_data: bytes | None = None) -> bytes:
    supplement_inputs = [
        require_regular_source("manuscript/supplement.qmd"),
        require_regular_source("manuscript/references.bib"),
        require_regular_source("manuscript/apa.csl"),
        require_regular_source("tools/finalize_docx.py"),
    ] + [
        require_regular_source(path)
        for path in MANUSCRIPT_GENERATED_FILES
        if PurePosixPath(path).name.startswith("supplement_")
    ] + [
        require_regular_source(f"outputs/figures/figure_s{number}_{suffix}.png")
        for number, suffix in (
            (1, "transition_matrices"),
            (2, "overlap_diagnostics"),
            (3, "standardization_sensitivity"),
            (4, "mnar_sensitivity"),
            (5, "grade89_extension"),
            (6, "measurement_crossing"),
        )
    ]
    if source.stat().st_mtime_ns < max(path.stat().st_mtime_ns for path in supplement_inputs):
        fail("manuscript/supplement.docx is stale; render it after all supplement inputs")

    if source_data is None:
        source_data = secure_file_bytes(source, MAX_ADDITIONAL_FILE_BYTES)
    if len(source_data) >= MAX_ADDITIONAL_FILE_BYTES:
        fail("Rendered supplement DOCX is not below 20 MB")
    with zipfile.ZipFile(io.BytesIO(source_data), "r") as archive:
        if archive.comment:
            fail("Source DOCX contains an archive-level comment")
        infos = archive.infolist()
        validate_zip_infos(infos, "Source DOCX")
        names = [info.filename for info in infos]
        if len(names) != len(set(names)):
            fail("Source DOCX contains duplicate ZIP members")
        if archive.testzip() is not None:
            fail("Source DOCX failed ZIP CRC validation")
        members = {name: archive.read(name) for name in names}

    remove_empty_docx_comment_scaffold(members)
    validate_sanitized_docx_features(members)

    for name, content in members.items():
        if name.casefold().endswith((".xml", ".rels")):
            text = content.decode("utf-8", errors="strict")
            for pattern in LOCAL_PATH_PATTERNS:
                if pattern.search(text):
                    fail(f"DOCX contains an absolute or local path: {name}")
            if EMAIL_PATTERN.search(text) or ORCID_PATTERN.search(text):
                fail(f"DOCX contains email-like or ORCID-like PII: {name}")
            folded = text.casefold()
            for token in private_deny_tokens():
                if token in folded:
                    fail(f"DOCX contains a private denylist token: {name}")

    core_name = "docProps/core.xml"
    if core_name in members:
        root = parse_xml_bytes(members[core_name], core_name)
        remove_locals = {"creator", "lastModifiedBy", "created", "modified", "lastPrinted"}
        for child in list(root):
            if child.tag.rsplit("}", 1)[-1] in remove_locals:
                root.remove(child)
        members[core_name] = ET.tostring(root, encoding="utf-8", xml_declaration=True)

    app_name = "docProps/app.xml"
    if app_name in members:
        root = parse_xml_bytes(members[app_name], app_name)
        for element in root.iter():
            if element.tag.rsplit("}", 1)[-1] in {"Company", "Manager"}:
                element.text = None
        members[app_name] = ET.tostring(root, encoding="utf-8", xml_declaration=True)

    document_name = "word/document.xml"
    if document_name not in members:
        fail("DOCX lacks word/document.xml")
    document_root = parse_xml_bytes(members[document_name], document_name)
    word_namespace = "http://schemas.openxmlformats.org/wordprocessingml/2006/main"
    visible_text = " ".join(
        element.text or "" for element in document_root.iter(f"{{{word_namespace}}}t")
    )
    required_title = "Grade 4-to-5 Joint Mathematics–Science Benchmark Transitions in Nine Systems: A Descriptive Common-Severity Analysis Using TIMSS 2023 Longitudinal Data"
    if required_title not in visible_text:
        fail("Rendered supplement DOCX does not contain the exact manuscript title")
    for number in range(1, 7):
        if visible_text.count(f"Figure S{number}.") != 1:
            fail(f"Rendered supplement DOCX does not contain exactly one Figure S{number} label")
    if re.search(r"\brecovery\b", visible_text, re.I):
        fail("Rendered supplement DOCX contains obsolete reader-facing recovery wording")
    if "AUTHOR CONFIRMATION REQUIRED" in visible_text:
        fail("Rendered supplement DOCX contains an unresolved author prompt")

    validate_sanitized_docx_features(members)
    validate_finalized_supplement_members(members)

    output = io.BytesIO()
    with zipfile.ZipFile(output, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as archive:
        for name in sorted(members):
            archive.writestr(zip_info(name), members[name], compress_type=zipfile.ZIP_DEFLATED, compresslevel=9)
    sanitized = output.getvalue()
    if len(sanitized) >= MAX_ADDITIONAL_FILE_BYTES:
        fail("Additional file 1 is not below 20 MB")
    with zipfile.ZipFile(io.BytesIO(sanitized), "r") as archive:
        validate_zip_infos(archive.infolist(), "Sanitized DOCX")
        if archive.testzip() is not None:
            fail("Sanitized DOCX failed ZIP CRC validation")
    scan_payload("Additional_file_1_supplement.docx", sanitized)
    return sanitized


def table_archive_entries(snapshot: dict[str, bytes]) -> dict[str, bytes]:
    return {
        f"tables/{name}": snapshot_bytes(snapshot, f"outputs/tables/{name}")
        for name in TABLE_FILES
    }


def reproducibility_archive_entries(
    runtime_data: bytes,
    source_checksums: bytes,
    snapshot: dict[str, bytes],
) -> dict[str, bytes]:
    entries: dict[str, bytes] = {}
    for relative in REPRODUCIBILITY_SOURCE_FILES:
        entries[relative] = snapshot_bytes(snapshot, relative)
    for name in DIAGNOSTIC_FILES:
        relative = f"outputs/diagnostics/{name}"
        entries[relative] = snapshot_bytes(snapshot, relative)
    for name in TABLE_FILES + REPRODUCIBILITY_ONLY_TABLE_FILES:
        relative = f"outputs/tables/{name}"
        entries[relative] = snapshot_bytes(snapshot, relative)
    for name in FIGURE_FILES:
        relative = f"outputs/figures/{name}"
        entries[relative] = snapshot_bytes(snapshot, relative)
    entries["provenance/official_source_checksums.csv"] = source_checksums
    entries["provenance/runtime_versions.txt"] = runtime_data
    entries[SOURCE_SNAPSHOT_MANIFEST] = source_snapshot_manifest(snapshot)
    validate_archive_names(list(entries))
    expected_names = expected_reproducibility_payload_names()
    if set(entries) != expected_names:
        missing = sorted(expected_names - set(entries))
        extra = sorted(set(entries) - expected_names)
        fail(f"Reproducibility payload allowlist mismatch; missing={missing}, extra={extra}")
    if any(name.startswith(("data/raw/", "data/derived/", "submission/")) for name in entries):
        fail("Reproducibility payload contains a forbidden data or submission path")
    return entries


def expected_reproducibility_payload_names() -> set[str]:
    names = set(REPRODUCIBILITY_SOURCE_FILES)
    names.update(f"outputs/diagnostics/{name}" for name in DIAGNOSTIC_FILES)
    names.update(
        f"outputs/tables/{name}"
        for name in TABLE_FILES + REPRODUCIBILITY_ONLY_TABLE_FILES
    )
    names.update(f"outputs/figures/{name}" for name in FIGURE_FILES)
    names.update({
        "provenance/official_source_checksums.csv",
        "provenance/runtime_versions.txt",
        SOURCE_SNAPSHOT_MANIFEST,
    })
    return names


def overall_manifest_bytes(artifacts: dict[str, bytes]) -> bytes:
    rows = [
        {"file": name, "bytes": len(artifacts[name]), "sha256": sha256_bytes(artifacts[name])}
        for name in sorted(artifacts)
    ]
    return csv_bytes(("file", "bytes", "sha256"), rows)


def docx_members_from_bytes(data: bytes, label: str) -> dict[str, bytes]:
    if len(data) >= MAX_ADDITIONAL_FILE_BYTES:
        fail(f"{label} is not below 20 MB")
    with zipfile.ZipFile(io.BytesIO(data), "r") as archive:
        if archive.comment:
            fail(f"{label} contains an archive-level comment")
        infos = archive.infolist()
        validate_zip_infos(infos, label)
        names = [info.filename for info in infos]
        if len(names) != len(set(names)):
            fail(f"{label} contains duplicate ZIP members")
        if archive.testzip() is not None:
            fail(f"{label} failed ZIP CRC validation")
        return {name: archive.read(name) for name in names}


def read_existing_artifacts() -> tuple[dict[str, bytes], bytes]:
    if not OVERALL_MANIFEST.is_file() or OVERALL_MANIFEST.is_symlink():
        fail("The external SHA-256 manifest is missing or unsafe")
    manifest_data = secure_file_bytes(OVERALL_MANIFEST, 100_000)
    artifacts: dict[str, bytes] = {}
    for name, path in DESTINATIONS.items():
        if not path.is_file() or path.is_symlink():
            fail(f"Generated Additional file is missing or unsafe: {name}")
        artifacts[name] = secure_file_bytes(path, MAX_ADDITIONAL_FILE_BYTES)
    try:
        rows = list(csv.DictReader(io.StringIO(manifest_data.decode("utf-8"), newline="")))
    except (UnicodeDecodeError, csv.Error):
        fail("The external SHA-256 manifest is not valid UTF-8 CSV")
    if (
        not rows
        or tuple(rows[0]) != ("file", "bytes", "sha256")
        or len(rows) != len(DESTINATIONS)
        or {row["file"] for row in rows} != set(DESTINATIONS)
        or len({row["file"] for row in rows}) != len(rows)
    ):
        fail("The external SHA-256 manifest has an unexpected schema or file set")
    by_name = {row["file"]: row for row in rows}
    for name, data in artifacts.items():
        row = by_name[name]
        if not re.fullmatch(r"[0-9a-f]{64}", row["sha256"]):
            fail(f"External SHA-256 value is invalid: {name}")
        if len(data) != int(row["bytes"]) or sha256_bytes(data) != row["sha256"]:
            fail(f"External SHA-256 verification failed: {name}")
    if manifest_data != overall_manifest_bytes(artifacts):
        fail("External SHA-256 manifest is not the deterministic manifest for the artifact set")
    return artifacts, manifest_data


def verify_existing_generated(compare_current_sources: bool = True) -> None:
    artifacts, _ = read_existing_artifacts()
    tables_data = artifacts["Additional_file_2_machine_readable_tables.zip"]
    reproducibility_data = artifacts["Additional_file_3_anonymous_reproducibility.zip"]
    supplement_data = artifacts["Additional_file_1_supplement.docx"]

    if compare_current_sources:
        snapshot = snapshot_sources(SOURCE_SNAPSHOT_FILES)
        validate_publication_artifacts(snapshot)
        runtime_data = sanitized_environment_bytes(
            snapshot_bytes(snapshot, SANITIZED_ENVIRONMENT_SOURCE)
        )
        source_checksums = official_source_checksum_bytes(
            snapshot_bytes(snapshot, "data/raw/download_manifest.csv")
        )
        expected_tables = table_archive_entries(snapshot)
        expected_reproducibility = reproducibility_archive_entries(
            runtime_data, source_checksums, snapshot
        )
        validate_zip_bytes(tables_data, set(expected_tables), expected_tables)
        validate_zip_bytes(
            reproducibility_data,
            set(expected_reproducibility),
            expected_reproducibility,
        )
        current_sanitized = sanitize_docx(
            require_regular_source("manuscript/supplement.docx"),
            snapshot_bytes(snapshot, "manuscript/supplement.docx"),
        )
        if supplement_data != current_sanitized:
            fail("Additional file 1 differs from freshly sanitized current supplement DOCX bytes")
        supplement_members = docx_members_from_bytes(supplement_data, "Additional file 1")
        validate_sanitized_docx_features(supplement_members)
        validate_finalized_supplement_members(supplement_members)
        for relative, before in snapshot.items():
            if source_bytes(relative) != before:
                fail(f"A package source changed during verification: {relative}")
    else:
        table_payload_names = validate_self_manifest_zip(tables_data)
        if (
            len(table_payload_names) != len(TABLE_FILES)
            or not all(name.startswith("tables/") and name.endswith(".csv") for name in table_payload_names)
            or "tables/table_manifest.csv" not in table_payload_names
        ):
            fail("Existing Additional file 2 is not a recognizable builder table payload")
        reproducibility_payload_names = validate_self_manifest_zip(reproducibility_data)
        required_reproducibility_core = {
            "analysis/run_all.R",
            "manuscript/manuscript.qmd",
            "manuscript/supplement.qmd",
            "provenance/official_source_checksums.csv",
            "provenance/runtime_versions.txt",
        }
        if not required_reproducibility_core.issubset(reproducibility_payload_names):
            fail("Existing Additional file 3 lacks the required reproducibility core")
        supplement_members = docx_members_from_bytes(supplement_data, "Existing Additional file 1")
        validate_prior_supplement_for_replacement(supplement_members)

    for name, data in artifacts.items():
        if len(data) >= MAX_ADDITIONAL_FILE_BYTES:
            fail(f"Additional file is not below 20 MB: {name}")


def fsync_directory(path: Path) -> None:
    flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
    descriptor = os.open(path, flags)
    try:
        os.fsync(descriptor)
    finally:
        os.close(descriptor)


def write_fsynced_new(path: Path, data: bytes, mode: int = 0o644) -> None:
    descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, mode)
    try:
        with os.fdopen(descriptor, "wb", closefd=False) as stream:
            stream.write(data)
            stream.flush()
            os.fsync(stream.fileno())
    finally:
        os.close(descriptor)


@contextmanager
def package_publish_lock():
    SUBMISSION.mkdir(parents=True, exist_ok=True)
    try:
        descriptor = os.open(PUBLISH_LOCK, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    except FileExistsError:
        fail(f"Additional Files publication lock already exists: {PUBLISH_LOCK.name}")
    lock_identity: tuple[int, int] | None = None
    try:
        lock_stat = os.fstat(descriptor)
        lock_identity = (lock_stat.st_dev, lock_stat.st_ino)
        os.write(descriptor, f"pid={os.getpid()}\n".encode("ascii"))
        os.fsync(descriptor)
        fsync_directory(SUBMISSION)
        yield
    finally:
        os.close(descriptor)
        try:
            current = PUBLISH_LOCK.lstat()
            if lock_identity == (current.st_dev, current.st_ino):
                PUBLISH_LOCK.unlink()
                fsync_directory(SUBMISSION)
        except FileNotFoundError:
            pass


def validate_artifact_set_bytes(artifacts: dict[str, bytes], manifest_data: bytes) -> None:
    if set(artifacts) != set(DESTINATIONS):
        fail("Generated artifact set differs from the three frozen destinations")
    if manifest_data != overall_manifest_bytes(artifacts):
        fail("External manifest does not match the generated artifact set")
    for name, data in artifacts.items():
        if len(data) >= MAX_ADDITIONAL_FILE_BYTES:
            fail(f"{name} is not below 20 MB")
    expected_table_names = {f"tables/{name}" for name in TABLE_FILES}
    validate_zip_bytes(
        artifacts["Additional_file_2_machine_readable_tables.zip"],
        expected_table_names,
    )
    validate_zip_bytes(
        artifacts["Additional_file_3_anonymous_reproducibility.zip"],
        expected_reproducibility_payload_names(),
    )
    members = docx_members_from_bytes(
        artifacts["Additional_file_1_supplement.docx"], "Staged Additional file 1"
    )
    validate_sanitized_docx_features(members)
    validate_finalized_supplement_members(members)


def publish(artifacts: dict[str, bytes], manifest_data: bytes, replace_generated: bool) -> None:
    validate_artifact_set_bytes(artifacts, manifest_data)
    with package_publish_lock():
        destinations_with_manifest = (*DESTINATIONS.values(), OVERALL_MANIFEST)
        existing = [path for path in destinations_with_manifest if path.exists()]
        if existing and not replace_generated:
            fail("Generated destinations already exist; use --replace-generated only for a prior builder set")
        if replace_generated:
            verify_existing_generated(compare_current_sources=False)
        elif existing:
            fail("Refusing to overwrite an existing file")

        with tempfile.TemporaryDirectory(prefix=".additional-files-publish-", dir=SUBMISSION) as temporary:
            temporary_path = Path(temporary)
            staged: dict[Path, Path] = {}
            for name in sorted(artifacts):
                stage = temporary_path / name
                write_fsynced_new(stage, artifacts[name])
                if secure_file_bytes(stage, MAX_ADDITIONAL_FILE_BYTES) != artifacts[name]:
                    fail(f"Staged artifact bytes changed before publication: {name}")
                staged[DESTINATIONS[name]] = stage
            manifest_stage = temporary_path / OVERALL_MANIFEST.name
            write_fsynced_new(manifest_stage, manifest_data)
            if secure_file_bytes(manifest_stage, 100_000) != manifest_data:
                fail("Staged external manifest bytes changed before publication")
            fsync_directory(temporary_path)

            backups: dict[Path, Path] = {}
            installed: list[Path] = []
            try:
                if replace_generated:
                    backup_directory = temporary_path / "backups"
                    backup_directory.mkdir(mode=0o700)
                    for destination in destinations_with_manifest:
                        backup = backup_directory / destination.name
                        os.replace(destination, backup)
                        backups[destination] = backup
                    fsync_directory(backup_directory)
                    fsync_directory(SUBMISSION)

                for destination in sorted(staged, key=lambda item: item.name):
                    os.replace(staged[destination], destination)
                    installed.append(destination)
                fsync_directory(SUBMISSION)
                # The external manifest is the set-level commit marker and is installed last.
                os.replace(manifest_stage, OVERALL_MANIFEST)
                installed.append(OVERALL_MANIFEST)
                fsync_directory(SUBMISSION)

                installed_artifacts, installed_manifest = read_existing_artifacts()
                if installed_artifacts != artifacts or installed_manifest != manifest_data:
                    fail("Installed Additional Files differ from the fully verified staged set")
            except Exception as error:
                rollback_errors: list[str] = []
                for destination in reversed(installed):
                    try:
                        destination.unlink(missing_ok=True)
                    except OSError as rollback_error:
                        rollback_errors.append(f"remove {destination.name}: {rollback_error}")
                for destination, backup in backups.items():
                    try:
                        if backup.exists():
                            os.replace(backup, destination)
                    except OSError as rollback_error:
                        rollback_errors.append(f"restore {destination.name}: {rollback_error}")
                try:
                    fsync_directory(SUBMISSION)
                except OSError as rollback_error:
                    rollback_errors.append(f"fsync submission directory: {rollback_error}")
                if rollback_errors:
                    fail(
                        f"Additional Files publication failed ({error}); rollback was incomplete: "
                        + "; ".join(rollback_errors)
                    )
                raise


def build(replace_generated: bool) -> None:
    snapshot = snapshot_sources(SOURCE_SNAPSHOT_FILES)
    validate_publication_artifacts(snapshot)
    runtime_data = sanitized_environment_bytes(
        snapshot_bytes(snapshot, SANITIZED_ENVIRONMENT_SOURCE)
    )
    source_checksums = official_source_checksum_bytes(
        snapshot_bytes(snapshot, "data/raw/download_manifest.csv")
    )
    table_entries = table_archive_entries(snapshot)
    reproducibility_entries = reproducibility_archive_entries(
        runtime_data, source_checksums, snapshot
    )

    docx_source = require_regular_source("manuscript/supplement.docx")
    artifacts = {
        "Additional_file_1_supplement.docx": sanitize_docx(
            docx_source, snapshot_bytes(snapshot, "manuscript/supplement.docx")
        ),
        "Additional_file_2_machine_readable_tables.zip": make_zip(table_entries),
        "Additional_file_3_anonymous_reproducibility.zip": make_zip(
            reproducibility_entries
        ),
    }
    for name, data in artifacts.items():
        if len(data) >= MAX_ADDITIONAL_FILE_BYTES:
            fail(f"{name} is not below 20 MB")

    for relative, before in snapshot.items():
        if source_bytes(relative) != before:
            fail(f"A package source changed during the build: {relative}")

    manifest_data = overall_manifest_bytes(artifacts)
    publish(artifacts, manifest_data, replace_generated)
    verify_existing_generated()

    print("Additional Files built and verified")
    for name in sorted(DESTINATIONS):
        path = DESTINATIONS[name]
        print(f"{name}: {path.stat().st_size} bytes; sha256={hash_file(path)}")
    print(f"{OVERALL_MANIFEST.name}: external SHA-256 manifest")


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    mode = parser.add_mutually_exclusive_group()
    mode.add_argument(
        "--replace-generated",
        action="store_true",
        help="replace only a complete prior package set that passes its external and internal SHA-256 manifests",
    )
    mode.add_argument(
        "--verify-only",
        action="store_true",
        help="verify the existing three Additional Files and their SHA-256 manifests without rebuilding",
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    try:
        if args.verify_only:
            with package_publish_lock():
                verify_existing_generated()
            print("Existing Additional Files and current source bytes verified")
        else:
            build(args.replace_generated)
    except (RuntimeError, OSError, ValueError, KeyError, csv.Error, zipfile.BadZipFile, subprocess.CalledProcessError) as error:
        print(f"ERROR: {error}", file=sys.stderr)
        return 1
    return 0


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