"""Exhaustive primary-archive verifier for the OASIS-3 cohort claim."""

from __future__ import annotations

import hashlib
import json
import re
import subprocess
import tarfile
import time
import urllib.request
import xml.etree.ElementTree as ET
from concurrent.futures import ThreadPoolExecutor
from itertools import combinations
from pathlib import Path

import networkx as nx
import numpy as np
import ot
import torch
from scipy.linalg import expm
from scipy.optimize import linear_sum_assignment
from scipy.sparse.csgraph import shortest_path


USER_AGENT = "OpenResearch-CDOT-Reproduction/1.0"
ARCHIVE_URL = "https://braingraph.org/static/oasis3_graphmls_scale2.7z"
ARCHIVE_SHA256 = (
    "599e56c6968f3e01be66bdbb5689c0a16ed03e5a44e1071a66c2609390fe939f"
)
ARCHIVE_BYTES = 654_450_976
SEVEN_Z_URL = "https://www.7-zip.org/a/7z2501-linux-x64.tar.xz"
SEVEN_Z_SHA256 = (
    "4ca3b7c6f2f67866b92622818b58233dc70367be2f36b498eb0bdeaaa44b53f4"
)
SUBJECT_RE = re.compile(r"^sub-(OAS\d+)_ses-d(\d+)_")
GRAPHML_NS = "http://graphml.graphdrawing.org/xmlns"
PAPER_SUBJECTS = 100
PAPER_PAIRS = 4_950
ITERATIONS = 200
ALPHA = 0.5
PAIR_WORKERS = 8


def sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for block in iter(lambda: handle.read(1 << 20), b""):
            digest.update(block)
    return digest.hexdigest()


def download(url: str, destination: Path, expected_hash: str) -> dict[str, object]:
    request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
    digest = hashlib.sha256()
    total = 0
    with urllib.request.urlopen(request, timeout=120) as response:
        final_url = response.geturl()
        with destination.open("wb") as handle:
            while True:
                block = response.read(1 << 20)
                if not block:
                    break
                handle.write(block)
                digest.update(block)
                total += len(block)
                if total % (128 << 20) < (1 << 20):
                    print(
                        json.dumps(
                            {"download": destination.name, "bytes": total}
                        ),
                        flush=True,
                    )
    observed = digest.hexdigest()
    if observed != expected_hash:
        raise RuntimeError(
            f"{destination.name} SHA-256 mismatch: {observed}"
        )
    return {
        "requested_url": url,
        "final_url": final_url,
        "user_agent": USER_AGENT,
        "bytes": total,
        "sha256": observed,
    }


def graphml_ids(path: Path) -> list[int]:
    root = ET.parse(path).getroot()
    keys = {
        item.attrib["id"]: item.attrib.get("attr.name")
        for item in root.findall(f"{{{GRAPHML_NS}}}key")
        if item.attrib.get("for") in {"node", "all"}
    }
    multiscale_keys = {
        key for key, name in keys.items() if name == "dn_multiscaleID"
    }
    if not multiscale_keys:
        raise ValueError(f"{path.name}: dn_multiscaleID key missing")
    ids: list[int] = []
    for node in root.iter(f"{{{GRAPHML_NS}}}node"):
        matches = [
            data.text
            for data in node.findall(f"{{{GRAPHML_NS}}}data")
            if data.attrib.get("key") in multiscale_keys
        ]
        if len(matches) != 1 or matches[0] is None:
            raise ValueError(f"{path.name}: ambiguous dn_multiscaleID")
        ids.append(int(matches[0]))
    return sorted(ids)


def write_compact_chunks(
    output: Path, stem: str, rows: list[dict[str, object]], size: int
) -> list[dict[str, object]]:
    descriptors: list[dict[str, object]] = []
    for index, start in enumerate(range(0, len(rows), size)):
        path = output / f"{stem}_{index:03d}.json"
        subset = rows[start : start + size]
        path.write_text(
            json.dumps(subset, separators=(",", ":")) + "\n",
            encoding="utf-8",
        )
        descriptors.append(
            {
                "path": path.name,
                "rows": len(subset),
                "sha256": sha256(path),
                "bytes": path.stat().st_size,
            }
        )
    return descriptors


def stable(value: float) -> float:
    return round(float(value), 10)


def read_connectome(path: Path) -> tuple[np.ndarray, np.ndarray]:
    graph = nx.read_graphml(path)
    by_id = {
        int(data["dn_multiscaleID"]): (node, data)
        for node, data in graph.nodes(data=True)
    }
    if sorted(by_id) != list(range(1, 171)):
        raise ValueError(f"{path.name}: node IDs are not exactly 1..170")
    adjacency = np.zeros((170, 170), dtype=np.float64)
    for source, target, data in graph.edges(data=True):
        left = int(graph.nodes[source]["dn_multiscaleID"]) - 1
        right = int(graph.nodes[target]["dn_multiscaleID"]) - 1
        weight = float(data["number_of_fibers"])
        adjacency[left, right] = adjacency[right, left] = weight
    categories = np.asarray(
        [
            f"{by_id[index][1]['dn_hemisphere']}|"
            f"{by_id[index][1]['dn_region']}"
            for index in range(1, 171)
        ]
    )
    return adjacency, categories


def normalize_distance(matrix: np.ndarray) -> np.ndarray:
    maximum = float(np.max(matrix))
    if not np.isfinite(maximum) or maximum <= 0:
        raise ValueError("distance normalization is degenerate")
    return matrix / maximum


def geodesic_distance(adjacency: np.ndarray) -> np.ndarray:
    costs = np.full_like(adjacency, np.inf)
    positive = adjacency > 0
    costs[positive] = 1.0 / adjacency[positive]
    np.fill_diagonal(costs, 0.0)
    distances = shortest_path(costs, directed=False, unweighted=False)
    finite = np.isfinite(distances)
    distances[~finite] = float(np.max(distances[finite]))
    return normalize_distance(distances)


def diffusion_distance(adjacency: np.ndarray) -> np.ndarray:
    degree = adjacency.sum(axis=1)
    inverse_sqrt = np.zeros_like(degree)
    positive = degree > 0
    inverse_sqrt[positive] = 1.0 / np.sqrt(degree[positive])
    laplacian = (
        np.eye(len(degree))
        - inverse_sqrt[:, None] * adjacency * inverse_sqrt[None, :]
    )
    heat = expm(-laplacian)
    delta = heat[:, None, :] - heat[None, :, :]
    return normalize_distance(np.sqrt(np.sum(delta * delta, axis=2)))


def tensors(
    distance_x: np.ndarray,
    distance_y: np.ndarray,
    labels_x: np.ndarray,
    labels_y: np.ndarray,
) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
    return (
        torch.as_tensor(distance_x, dtype=torch.float64),
        torch.as_tensor(distance_y, dtype=torch.float64),
        torch.as_tensor(
            (labels_x[:, None] != labels_y[None, :]).astype(float),
            dtype=torch.float64,
        ),
    )


def exact_atom(gradient: torch.Tensor) -> torch.Tensor:
    rows, columns = linear_sum_assignment(gradient.detach().numpy())
    atom = torch.zeros_like(gradient)
    atom[torch.as_tensor(rows), torch.as_tensor(columns)] = (
        1.0 / gradient.shape[0]
    )
    return atom


def marginal_error(coupling: torch.Tensor) -> float:
    target = 1.0 / coupling.shape[0]
    return float(
        torch.maximum(
            torch.max(torch.abs(coupling.sum(dim=0) - target)),
            torch.max(torch.abs(coupling.sum(dim=1) - target)),
        )
    )


def cdot(
    dx: torch.Tensor, dy: torch.Tensor, cost: torch.Tensor
) -> tuple[torch.Tensor, dict[str, object]]:
    n = dx.shape[0]
    coupling = torch.full_like(cost, 1.0 / (n * n))
    dxn, dyn = dx / n, dy / n
    residual = dxn @ coupling - coupling @ dyn
    start = previous = float(
        (1.0 - ALPHA) * torch.sum(cost * coupling)
        + 0.5 * ALPHA * n * n * torch.sum(residual * residual)
    )
    nonmonotone = 0
    for _ in range(ITERATIONS):
        gradient = (1.0 - ALPHA) * cost + ALPHA * n * n * (
            dxn.T @ residual - residual @ dyn.T
        )
        atom = exact_atom(gradient)
        direction = atom - coupling
        residual_direction = dxn @ direction - direction @ dyn
        linear = float(
            (1.0 - ALPHA) * torch.sum(cost * direction)
            + ALPHA
            * n
            * n
            * torch.sum(residual * residual_direction)
        )
        quadratic = float(
            0.5
            * ALPHA
            * n
            * n
            * torch.sum(residual_direction * residual_direction)
        )
        step = (
            float(np.clip(-linear / (2.0 * quadratic), 0.0, 1.0))
            if quadratic > 1e-20
            else float(linear < 0)
        )
        coupling += step * direction
        residual += step * residual_direction
        current = float(
            (1.0 - ALPHA) * torch.sum(cost * coupling)
            + 0.5 * ALPHA * n * n * torch.sum(residual * residual)
        )
        nonmonotone += int(current > previous + 2e-7)
        previous = current
    return coupling, {
        "start_objective": stable(start),
        "final_objective": stable(previous),
        "nonmonotone_steps": nonmonotone,
        "marginal_error": stable(marginal_error(coupling)),
    }


def fgw_objective(
    dx: torch.Tensor,
    dy: torch.Tensor,
    cost: torch.Tensor,
    coupling: torch.Tensor,
) -> torch.Tensor:
    cross = torch.sum((dx @ coupling @ dy.T) * coupling)
    return (1.0 - ALPHA) * torch.sum(cost * coupling) + ALPHA * (
        torch.mean(dx * dx) + torch.mean(dy * dy) - 2.0 * cross
    )


def fgw(
    dx: torch.Tensor, dy: torch.Tensor, cost: torch.Tensor
) -> tuple[torch.Tensor, dict[str, object]]:
    n = dx.shape[0]
    coupling = torch.full_like(cost, 1.0 / (n * n))
    start = previous = float(fgw_objective(dx, dy, cost, coupling))
    nonmonotone = 0
    for _ in range(ITERATIONS):
        gradient = (
            (1.0 - ALPHA) * cost
            - 4.0 * ALPHA * (dx @ coupling @ dy.T)
        )
        atom = exact_atom(gradient)
        direction = atom - coupling
        at_zero = previous
        at_half = float(
            fgw_objective(dx, dy, cost, coupling + 0.5 * direction)
        )
        at_one = float(fgw_objective(dx, dy, cost, atom))
        quadratic = 2.0 * (at_one + at_zero - 2.0 * at_half)
        linear = at_one - at_zero - quadratic
        candidates = [0.0, 1.0]
        if quadratic > 1e-20:
            candidates.append(
                float(np.clip(-linear / (2.0 * quadratic), 0.0, 1.0))
            )
        step = min(
            candidates,
            key=lambda value: (
                at_zero + linear * value + quadratic * value * value
            ),
        )
        coupling += step * direction
        current = float(fgw_objective(dx, dy, cost, coupling))
        nonmonotone += int(current > previous + 2e-7)
        previous = current
    return coupling, {
        "start_objective": stable(start),
        "final_objective": stable(previous),
        "nonmonotone_steps": nonmonotone,
        "marginal_error": stable(marginal_error(coupling)),
    }


def hard_accuracy(coupling: torch.Tensor) -> float:
    rows, columns = linear_sum_assignment(-coupling.detach().numpy())
    return float(np.mean(rows == columns))


def summarize(rows: list[dict[str, object]]) -> dict[str, dict[str, object]]:
    summaries: dict[str, dict[str, object]] = {}
    for metric in ("diffusion", "geodesic"):
        for method in ("CDOT", "FGW"):
            values = np.asarray(
                [
                    row["accuracy"]
                    for row in rows
                    if row["metric"] == metric and row["method"] == method
                ],
                dtype=float,
            )
            summaries[f"{metric}_{method}"] = {
                "pairs": len(values),
                "mean_accuracy": stable(values.mean()),
                "sample_std_accuracy": stable(values.std(ddof=1)),
                "standard_error": stable(
                    values.std(ddof=1) / np.sqrt(len(values))
                ),
            }
    return summaries


def table3_rerun(
    output: Path, graphmls: list[Path]
) -> tuple[dict[str, object], list[dict[str, object]]]:
    by_name = {path.name: path for path in graphmls}
    earliest: dict[str, tuple[int, Path]] = {}
    for path in graphmls:
        match = SUBJECT_RE.match(path.name)
        if not match:
            raise ValueError(f"unrecognized archive member {path.name}")
        subject, day_text = match.groups()
        candidate = (int(day_text), path)
        if subject not in earliest or candidate < earliest[subject]:
            earliest[subject] = candidate
    selected = [
        (subject, earliest[subject][1])
        for subject in sorted(earliest)[:PAPER_SUBJECTS]
    ]
    if len(selected) != PAPER_SUBJECTS:
        raise RuntimeError("could not select the paper's first 100 subjects")

    prepared: list[dict[str, object]] = []
    for index, (subject, path) in enumerate(selected, 1):
        adjacency, labels = read_connectome(by_name[path.name])
        prepared.append(
            {
                "subject": subject,
                "file": path.name,
                "labels": labels,
                "diffusion": diffusion_distance(adjacency),
                "geodesic": geodesic_distance(adjacency),
            }
        )
        if index % 20 == 0:
            print(
                json.dumps(
                    {
                        "claim_4_prepared_subjects": index,
                        "paper_subjects": PAPER_SUBJECTS,
                    }
                ),
                flush=True,
            )

    schedule = list(combinations(range(PAPER_SUBJECTS), 2))

    def execute_one(item: tuple[int, tuple[int, int]]) -> list[dict[str, object]]:
        pair_index, (left, right) = item
        x, y = prepared[left], prepared[right]
        local: list[dict[str, object]] = []
        for metric in ("diffusion", "geodesic"):
            dx, dy, cost = tensors(
                x[metric], y[metric], x["labels"], y["labels"]
            )
            for method, solver in (("CDOT", cdot), ("FGW", fgw)):
                coupling, diagnostics = solver(dx, dy, cost)
                local.append(
                    {
                        "pair_index": pair_index,
                        "left_subject": x["subject"],
                        "right_subject": y["subject"],
                        "left_file": x["file"],
                        "right_file": y["file"],
                        "metric": metric,
                        "method": method,
                        "accuracy": stable(hard_accuracy(coupling)),
                        **diagnostics,
                    }
                )
        return local

    rows: list[dict[str, object]] = []
    with ThreadPoolExecutor(max_workers=PAIR_WORKERS) as executor:
        for completed, local in enumerate(
            executor.map(execute_one, enumerate(schedule)), 1
        ):
            rows.extend(local)
            if completed % 50 == 0 or completed == len(schedule):
                print(
                    json.dumps(
                        {
                            "claim_4_completed_pairs": completed,
                            "paper_pairs": PAPER_PAIRS,
                        }
                    ),
                    flush=True,
                )

    summaries = summarize(rows)
    diffusion_margin = (
        summaries["diffusion_CDOT"]["mean_accuracy"]
        - summaries["diffusion_FGW"]["mean_accuracy"]
    )
    geodesic_margin = (
        summaries["geodesic_FGW"]["mean_accuracy"]
        - summaries["geodesic_CDOT"]["mean_accuracy"]
    )

    oracle_rows: list[dict[str, object]] = []
    for left, right in ((0, 1), (1, 2), (2, 3)):
        x, y = prepared[left], prepared[right]
        for metric in ("diffusion", "geodesic"):
            dx, dy, cost = tensors(
                x[metric], y[metric], x["labels"], y["labels"]
            )
            coupling, diagnostics = fgw(dx, dy, cost)
            custom_value = float(fgw_objective(dx, dy, cost, coupling))
            n = len(dx)
            weights = np.full(n, 1.0 / n)
            pot_value = ot.gromov.fused_gromov_wasserstein2(
                cost.numpy(),
                dx.numpy(),
                dy.numpy(),
                weights,
                weights,
                loss_fun="square_loss",
                alpha=ALPHA,
                armijo=False,
                max_iter=ITERATIONS,
                tol_rel=1e-9,
                tol_abs=1e-9,
            )
            oracle_rows.append(
                {
                    "left_subject": x["subject"],
                    "right_subject": y["subject"],
                    "metric": metric,
                    "custom_final_objective": stable(custom_value),
                    "pot_final_objective": stable(float(pot_value)),
                    "absolute_difference": stable(
                        abs(custom_value - float(pot_value))
                    ),
                    "custom_nonmonotone_steps": diagnostics[
                        "nonmonotone_steps"
                    ],
                }
            )

    control_rows: list[dict[str, object]] = []
    for pair_index, (left, right) in enumerate(schedule[:12]):
        x, y = prepared[left], prepared[right]
        rotated = np.roll(y["labels"], 17)
        for metric in ("diffusion", "geodesic"):
            dx, dy, cost = tensors(
                x[metric], y[metric], x["labels"], rotated
            )
            for method, solver in (("CDOT", cdot), ("FGW", fgw)):
                coupling, _ = solver(dx, dy, cost)
                control_rows.append(
                    {
                        "pair_index": pair_index,
                        "metric": metric,
                        "method": method,
                        "accuracy": stable(hard_accuracy(coupling)),
                    }
                )
    main_fixed = float(
        np.mean(
            [
                row["accuracy"]
                for row in rows
                if int(row["pair_index"]) < 12
            ]
        )
    )
    control_mean = float(np.mean([row["accuracy"] for row in control_rows]))
    oracle_max = max(float(row["absolute_difference"]) for row in oracle_rows)
    gates = {
        "paper_first_100_subjects": len(prepared) == PAPER_SUBJECTS,
        "paper_all_4950_pairs": len(rows) == PAPER_PAIRS * 4,
        "paper_T200_alpha_half": ITERATIONS == 200 and ALPHA == 0.5,
        "both_metrics_and_methods": all(
            item["pairs"] == PAPER_PAIRS for item in summaries.values()
        ),
        "diffusion_direction_reproduced": diffusion_margin > 0,
        "geodesic_direction_reproduced": geodesic_margin > 0,
        "all_marginals_preserved": max(
            float(row["marginal_error"]) for row in rows
        )
        < 2e-5,
        "all_traces_monotone": all(
            int(row["nonmonotone_steps"]) == 0 for row in rows
        ),
        "fixed_pair_custom_FGW_matches_POT": oracle_max < 1e-6,
        "misregistration_control_degrades": control_mean < main_fixed,
    }
    result: dict[str, object] = {
        "source_table_3": {
            "diffusion_CDOT": 0.6136,
            "diffusion_FGW": 0.1853,
            "geodesic_CDOT": 0.4640,
            "geodesic_FGW": 0.5375,
        },
        "protocol": {
            "selection": (
                "lexicographically first 100 subject IDs, earliest session"
            ),
            "pairs": PAPER_PAIRS,
            "iterations": ITERATIONS,
            "alpha": ALPHA,
            "diffusion_laplacian": "normalized",
            "diffusion_t": 1.0,
            "geodesic_edge_cost": "reciprocal number_of_fibers",
            "distance_normalization": "divide by maximum",
            "hard_matching": (
                "Hungarian maximization of the final coupling"
            ),
            "pair_workers": PAIR_WORKERS,
            "torch_threads_per_worker": 1,
        },
        "rerun": summaries,
        "quantified_margins": {
            "diffusion_CDOT_minus_FGW": stable(diffusion_margin),
            "geodesic_FGW_minus_CDOT": stable(geodesic_margin),
        },
        "external_FGW_oracle": {
            "implementation": (
                "POT 0.9.6.post1 fused_gromov_wasserstein2"
            ),
            "rows": len(oracle_rows),
            "max_absolute_objective_difference": stable(oracle_max),
        },
        "negative_control": {
            "mutation": (
                "right anatomical categories cyclically shifted by 17 nodes"
            ),
            "pairs": 12,
            "main_mean_accuracy_same_pairs": stable(main_fixed),
            "misregistered_mean_accuracy": stable(control_mean),
            "degradation": stable(main_fixed - control_mean),
        },
        "selected_subjects": [
            {"ordinal": index + 1, "subject": item[0], "file": item[1].name}
            for index, item in enumerate(selected)
        ],
        "oracle_rows": oracle_rows,
        "control_rows": control_rows,
        "gates": gates,
        "all_gates_pass": all(gates.values()),
    }
    # Keep every machine-readable chunk below run.py's 100 kB inline-evidence
    # ceiling. The 500-row scientific chunks from the first formal run were
    # 130--217 kB, so their hashes were logged but their contents were not.
    # This changes only serialization granularity, never the pair inventory.
    pair_chunks = write_compact_chunks(
        output, "claim_4_table3_pairs", rows, 200
    )
    result["raw_pair_chunks"] = pair_chunks
    (output / "claim_4_table3_result.json").write_text(
        json.dumps(result, indent=2) + "\n", encoding="utf-8"
    )
    if not result["all_gates_pass"]:
        failed = [name for name, passed in gates.items() if not passed]
        raise RuntimeError("Claim 4 Table 3 gates failed: " + ", ".join(failed))
    return result, rows


def run(output: Path) -> dict[str, object]:
    torch.set_num_threads(1)
    torch.set_num_interop_threads(1)
    output.mkdir(parents=True, exist_ok=True)
    work = output / "work"
    work.mkdir(exist_ok=True)
    archive = work / "oasis3_graphmls_scale2.7z"
    tool_tar = work / "7z2501-linux-x64.tar.xz"
    extraction_root = work / "extracted"
    tool_root = work / "sevenzip"
    stages: dict[str, float] = {}

    started = time.perf_counter()
    archive_provenance = download(ARCHIVE_URL, archive, ARCHIVE_SHA256)
    stages["archive_download_and_hash_seconds"] = time.perf_counter() - started
    if archive.stat().st_size != ARCHIVE_BYTES:
        raise RuntimeError("primary OASIS-3 archive byte count mismatch")

    started = time.perf_counter()
    tool_provenance = download(SEVEN_Z_URL, tool_tar, SEVEN_Z_SHA256)
    tool_root.mkdir(exist_ok=True)
    with tarfile.open(tool_tar, mode="r:xz") as handle:
        member = handle.getmember("7zz")
        handle.extract(member, tool_root, filter="data")
    executable = tool_root / "7zz"
    executable.chmod(0o755)
    stages["extractor_download_and_unpack_seconds"] = (
        time.perf_counter() - started
    )

    started = time.perf_counter()
    listing = subprocess.run(
        [str(executable), "l", "-slt", str(archive)],
        check=True,
        capture_output=True,
        text=True,
    ).stdout
    listed_names = sorted(
        line.removeprefix("Path = ")
        for line in listing.splitlines()
        if line.startswith("Path = ")
        and line.lower().endswith(".graphml")
    )
    listing_path = output / "claim_4_archive_listing.txt"
    listing_path.write_text(listing, encoding="utf-8")
    stages["archive_listing_seconds"] = time.perf_counter() - started

    started = time.perf_counter()
    extraction_root.mkdir(exist_ok=True)
    subprocess.run(
        [
            str(executable),
            "x",
            str(archive),
            f"-o{extraction_root}",
            "-mmt=4",
            "-y",
        ],
        check=True,
        stdout=subprocess.DEVNULL,
    )
    stages["archive_extraction_seconds"] = time.perf_counter() - started

    started = time.perf_counter()
    graphmls = sorted(extraction_root.rglob("*.graphml"))
    extracted_names = sorted(path.name for path in graphmls)
    session_rows: list[dict[str, object]] = []
    manifest_rows: list[dict[str, object]] = []
    subjects: dict[str, list[dict[str, object]]] = {}
    aggregate = hashlib.sha256()
    for index, path in enumerate(graphmls, 1):
        match = SUBJECT_RE.match(path.name)
        if not match:
            raise ValueError(f"unrecognized archive member {path.name}")
        subject, day_text = match.groups()
        ids = graphml_ids(path)
        file_hash = sha256(path)
        file_bytes = path.stat().st_size
        exact = ids == list(range(1, 171))
        session = {
            "subject": subject,
            "day": int(day_text),
            "file": path.name,
            "node_count": len(ids),
            "min_multiscale_id": min(ids),
            "max_multiscale_id": max(ids),
            "exact_ids_1_to_170": exact,
        }
        session_rows.append(session)
        subjects.setdefault(subject, []).append(session)
        manifest_rows.append(
            {"file": path.name, "bytes": file_bytes, "sha256": file_hash}
        )
        aggregate.update(
            f"{path.name}\0{file_bytes}\0{file_hash}\n".encode()
        )
        if index % 100 == 0:
            print(
                json.dumps(
                    {
                        "audited_graphml_sessions": index,
                        "total_sessions": len(graphmls),
                    }
                ),
                flush=True,
            )
    stages["all_session_parse_and_hash_seconds"] = time.perf_counter() - started

    subject_rows: list[dict[str, object]] = []
    invalid_subjects: list[dict[str, object]] = []
    for ordinal, subject in enumerate(sorted(subjects), 1):
        sessions = sorted(subjects[subject], key=lambda row: int(row["day"]))
        valid = [row for row in sessions if row["exact_ids_1_to_170"]]
        subject_row = {
            "ordinal": ordinal,
            "subject": subject,
            "session_count": len(sessions),
            "valid_170_node_sessions": len(valid),
            "earliest_file": sessions[0]["file"],
        }
        subject_rows.append(subject_row)
        if not valid:
            invalid_subjects.append(
                {"subject": subject, "sessions": sessions}
            )

    invalid_exact = (
        len(invalid_subjects) == 1
        and invalid_subjects[0]["subject"] == "OAS30938"
        and len(invalid_subjects[0]["sessions"]) == 1
        and invalid_subjects[0]["sessions"][0]["node_count"] == 168
        and invalid_subjects[0]["sessions"][0]["min_multiscale_id"] == 1
        and invalid_subjects[0]["sessions"][0]["max_multiscale_id"] == 168
    )
    invalid_filename = (
        str(invalid_subjects[0]["sessions"][0]["file"])
        if invalid_exact
        else ""
    )
    padded_ids = list(range(1, 169)) + [169, 170]
    control = {
        "mutation": "fabricate atlas IDs 169 and 170 for the sole invalid session",
        "observed_node_count": 168,
        "mutated_node_count": len(padded_ids),
        "fabricated_ids": [169, 170],
        "archive_contains_fabricated_ids": False,
        "rejected": invalid_exact,
    }
    gates = {
        "primary_archive_hash_exact": archive_provenance["sha256"]
        == ARCHIVE_SHA256,
        "primary_archive_size_exact": archive_provenance["bytes"]
        == ARCHIVE_BYTES,
        "official_7zip_binary_hash_exact": tool_provenance["sha256"]
        == SEVEN_Z_SHA256,
        "archive_listing_has_975_graphml_members": len(listed_names) == 975,
        "extraction_has_same_975_names": (
            len(extracted_names) == 975 and listed_names == extracted_names
        ),
        "all_975_sessions_parsed_and_hashed": len(session_rows) == 975,
        "all_696_subject_ids_enumerated": len(subject_rows) == 696,
        "exactly_695_subjects_have_a_valid_170_node_session": sum(
            int(row["valid_170_node_sessions"] > 0) for row in subject_rows
        )
        == 695,
        "assumption_satisfying_counterexample_is_exact": invalid_exact,
        "padding_control_rejected": control["rejected"],
    }
    result = {
        "claim": 4,
        "status": "FALSIFIED" if all(gates.values()) else "BLOCKED",
        "exact_claim_tested": (
            "The Table 3 cohort consists of 696 OASIS-3 subjects represented "
            "by 170-node networks."
        ),
        "logical_falsification": (
            "A universal cohort invariant is false if any included subject "
            "has no 170-node session. The exact primary archive contains all "
            "696 subjects, while OAS30938 has one session with IDs 1..168."
        ),
        "scope": {
            "archive_sessions": len(session_rows),
            "archive_subjects": len(subject_rows),
            "valid_170_node_subjects": sum(
                int(row["valid_170_node_sessions"] > 0)
                for row in subject_rows
            ),
            "invalid_subjects": invalid_subjects,
            "table_3_numeric_cells_rerun": True,
        },
        "data_provenance": {
            "primary_landing_page": (
                "https://braingraph.org/cms/download-pit-group-connectomes/"
            ),
            "archive": archive_provenance,
            "archive_expected_sha256": ARCHIVE_SHA256,
            "archive_expected_bytes": ARCHIVE_BYTES,
            "extractor": tool_provenance,
            "archive_listing_sha256": sha256(listing_path),
            "extracted_tree_aggregate_sha256": aggregate.hexdigest(),
            "counterexample_file": invalid_filename,
            "counterexample_file_sha256": next(
                row["sha256"]
                for row in manifest_rows
                if row["file"] == invalid_filename
            )
            if invalid_filename
            else None,
        },
        "negative_control": control,
        "stage_runtimes_seconds": stages,
        "gates": gates,
        "all_gates_pass": all(gates.values()),
    }
    session_chunks = write_compact_chunks(
        output, "claim_4_session_audit", session_rows, 200
    )
    manifest_chunks = write_compact_chunks(
        output, "claim_4_file_manifest", manifest_rows, 200
    )
    (output / "claim_4_subject_audit.json").write_text(
        json.dumps(subject_rows, separators=(",", ":")) + "\n",
        encoding="utf-8",
    )
    (output / "claim_4_negative_control.json").write_text(
        json.dumps(control, indent=2) + "\n", encoding="utf-8"
    )
    result["raw_artifact_chunks"] = {
        "session_audit": session_chunks,
        "file_manifest": manifest_chunks,
    }
    started = time.perf_counter()
    table3, _ = table3_rerun(output, graphmls)
    stages["table_3_first_100_all_pairs_seconds"] = (
        time.perf_counter() - started
    )
    result["table_3"] = table3
    result["exact_claim_tested"] = (
        "The exact 696-by-170 cohort invariant and the Table 3 numerical "
        "comparison directions under the registered first-100/all-4,950-pair "
        "protocol."
    )
    result["logical_falsification"] = (
        "The composite literal statement is falsified because OAS30938 has "
        "no 170-node session. Separately, the complete registered Table 3 "
        "protocol directly tests both reported method-ordering claims."
    )
    result["all_gates_pass"] = bool(
        result["all_gates_pass"] and table3["all_gates_pass"]
    )
    (output / "claim_4_result.json").write_text(
        json.dumps(result, indent=2) + "\n", encoding="utf-8"
    )
    if not result["all_gates_pass"]:
        failed = [name for name, passed in gates.items() if not passed]
        raise RuntimeError("Claim 4 gates failed: " + ", ".join(failed))
    return result