from __future__ import annotations

import fcntl
import json
import math
import os
import random
import shutil
import statistics
import threading
import time
from collections import Counter
from concurrent.futures import ThreadPoolExecutor, as_completed
from contextlib import contextmanager
from copy import deepcopy
from dataclasses import dataclass
from datetime import datetime
from html import escape
from pathlib import Path
from typing import Any, Iterator

from .backends import CodexExecBackend, TranscriptionBackend
from .config import EvaluationProfile
from .diagnostics import detect_cross_page_duplicates
from .owner_review import summarize_uncertainties
from .pipeline import PageRunResult, _resumed_result, codex_version, transcribe_page
from .reconciliation import reconcile_adjacent_pages
from .render import render_page
from .schema_tools import export_model_schema
from .source import SourcePackage
from .util import (
    load_json,
    project_root,
    sha256_bytes,
    sha256_file,
    sha256_json,
    utc_now,
    write_json,
    write_text,
)
from .validation import validate_page, validate_reciprocal_pages


PILOT_RUN_ID = "20260808T201223Z-pilot"
SEQUENTIAL_PILOT_SECONDS_PER_PAGE = 155.181
DEFAULT_QC_SEED = "crowley-pass1-clean-qc-v1"


@dataclass(frozen=True)
class Shard:
    index: int
    start_page: int
    end_page: int
    pending_pages: tuple[int, ...]
    completed_anchors: tuple[int, ...]

    @property
    def shard_id(self) -> str:
        return f"shard-{self.start_page:03d}-{self.end_page:03d}"

    def as_dict(self) -> dict[str, Any]:
        return {
            "index": self.index,
            "shard_id": self.shard_id,
            "start_page": self.start_page,
            "end_page": self.end_page,
            "pending_pages": list(self.pending_pages),
            "completed_anchors": list(self.completed_anchors),
        }


def parse_worker_count(value: int) -> int:
    if isinstance(value, bool) or value < 1:
        raise ValueError("Worker count must be a positive integer")
    if value > 32:
        raise ValueError("Worker count above 32 is not supported by this local scheduler")
    return value


def partition_physical_shards(
    *, total_pages: int, completed_pages: set[int], shard_size: int = 20
) -> list[Shard]:
    if shard_size < 1 or shard_size > 20:
        raise ValueError("Corpus shard size must be between 1 and 20 physical pages")
    shards: list[Shard] = []
    for index, start in enumerate(range(1, total_pages + 1, shard_size), start=1):
        end = min(total_pages, start + shard_size - 1)
        physical = tuple(range(start, end + 1))
        pending = tuple(page for page in physical if page not in completed_pages)
        anchors = tuple(page for page in physical if page in completed_pages)
        if pending:
            shards.append(Shard(index, start, end, pending, anchors))
    return shards


def source_tree_digest(root: Path) -> dict[str, Any]:
    root = root.resolve()
    digest_input = bytearray()
    file_count = 0
    total_bytes = 0
    for path in sorted((path for path in root.rglob("*") if path.is_file())):
        relative = path.relative_to(root).as_posix()
        file_hash = sha256_file(path)
        size = path.stat().st_size
        digest_input.extend(f"{relative}\0{size}\0{file_hash}\n".encode())
        file_count += 1
        total_bytes += size
    return {
        "root": str(root),
        "algorithm": "sha256(relative-path NUL size NUL file-sha256 LF)",
        "sha256": sha256_bytes(bytes(digest_input)),
        "file_count": file_count,
        "total_bytes": total_bytes,
    }


def promote_reviewed_pilot(
    *, source: SourcePackage, owner: str = "John Crowley"
) -> dict[str, Any]:
    """Derive 20 owner-approved editions without changing pilot candidates."""

    root = project_root()
    pilot_root = root / "pilot" / "runs" / PILOT_RUN_ID
    approved_root = root / "edition" / "owner-approved"
    state_root = root / "pilot" / "owner-review" / "states"
    approved: list[int] = []
    existing: list[int] = []
    timestamp = utc_now()
    for page_number in range(41, 61):
        page_id = f"page-{page_number:03d}"
        candidate_path = pilot_root / "candidates" / f"{page_id}.json"
        destination = approved_root / f"{page_id}.json"
        state_path = state_root / f"{page_id}.json"
        if destination.is_file():
            page = load_json(destination)
            report = validate_page(page, source)
            if not report.valid or page["review"]["status"] != "owner_approved":
                raise ValueError(f"Existing pilot approval is invalid: {destination}")
            existing.append(page_number)
            continue
        candidate = load_json(candidate_path)
        candidate_report = validate_page(candidate, source)
        if not candidate_report.valid:
            raise ValueError(
                f"Cannot approve invalid pilot candidate {page_id}: "
                f"{candidate_report.errors}"
            )
        candidate_hash = sha256_file(candidate_path)
        edition = deepcopy(candidate)
        edition["review"] = {
            "status": "owner_approved",
            "authority": "owner_approved_edition",
            "reviewer": owner,
            "reviewed_at": timestamp,
            "notes": [
                "The owner visually reviewed this production-pilot page against the authoritative scan.",
                "Explicit owner action: approve_as_is.",
                f"Approval lineage begins with pilot candidate SHA-256 {candidate_hash}.",
            ],
        }
        edition["provenance"]["steps"].append(
            {
                "kind": "owner_review",
                "actor": owner,
                "artifact": str(candidate_path.relative_to(root)),
                "created_at": timestamp,
                "action": "approve_as_is",
                "input_sha256": candidate_hash,
                "change_categories": [],
            }
        )
        report = validate_page(edition, source)
        if not report.valid:
            raise ValueError(f"Derived pilot approval is invalid: {report.errors}")
        write_json(destination, edition)
        write_json(
            state_path,
            {
                "page_id": page_id,
                "current_status": "owner_approved",
                "events": [
                    {
                        "action": "approve_as_is",
                        "owner": owner,
                        "decided_at": timestamp,
                        "note": "Owner visually reviewed the accepted 20-page production pilot against the authoritative scan.",
                        "candidate_path": str(candidate_path.relative_to(root)),
                        "candidate_sha256": candidate_hash,
                        "approved_path": str(destination.relative_to(root)),
                        "approved_sha256": sha256_json(edition),
                        "change_categories": [],
                        "owner_textual_change": False,
                    }
                ],
            },
        )
        approved.append(page_number)
    return {
        "approved_now": approved,
        "already_approved": existing,
        "owner_approved_pilot_count": len(approved) + len(existing),
    }


def owner_approved_pages(source: SourcePackage) -> dict[int, Path]:
    found: dict[int, Path] = {}
    root = project_root() / "edition" / "owner-approved"
    for path in sorted(root.glob("page-*.json")):
        page = load_json(path)
        number = int(page["source"]["pdf_page"])
        if number in found:
            raise ValueError(f"Duplicate owner-approved physical page {number}")
        report = validate_page(page, source)
        if not report.valid:
            raise ValueError(f"Invalid owner-approved page {number}: {report.errors}")
        if page["review"] != {
            **page["review"],
            "status": "owner_approved",
            "authority": "owner_approved_edition",
        }:
            raise ValueError(f"Invalid owner authority on page {number}")
        found[number] = path
    return found


@contextmanager
def page_lock(lock_path: Path) -> Iterator[bool]:
    lock_path.parent.mkdir(parents=True, exist_ok=True)
    descriptor = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o644)
    acquired = False
    try:
        try:
            fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
            acquired = True
        except BlockingIOError:
            acquired = False
        yield acquired
    finally:
        if acquired:
            fcntl.flock(descriptor, fcntl.LOCK_UN)
        os.close(descriptor)


class AdaptiveShardLimiter:
    def __init__(self, workers: int) -> None:
        self.configured = parse_worker_count(workers)
        self.effective = self.configured
        self.active = 0
        self.max_observed = 0
        self.reductions: list[dict[str, Any]] = []
        self._condition = threading.Condition()

    def acquire(self) -> None:
        with self._condition:
            while self.active >= self.effective:
                self._condition.wait()
            self.active += 1
            self.max_observed = max(self.max_observed, self.active)

    def release(self, *, shard_id: str, throttling_events: int) -> None:
        with self._condition:
            self.active -= 1
            if throttling_events >= 2 and self.effective > 1:
                self.effective -= 1
                self.reductions.append(
                    {
                        "shard_id": shard_id,
                        "reason": "repeated infrastructure throttling",
                        "throttling_events": throttling_events,
                        "new_effective_workers": self.effective,
                        "recorded_at": utc_now(),
                    }
                )
            self._condition.notify_all()


def _quarantine(
    run_root: Path, page: int, errors: list[str], category: str
) -> None:
    write_json(
        run_root / "quarantine" / f"page-{page:03d}.json",
        {
            "page": page,
            "state": "quarantined_unresolved_failure",
            "category": category,
            "errors": errors,
            "recorded_at": utc_now(),
            "page_artifacts": str(run_root / "pages" / f"page-{page:03d}"),
        },
    )


def _shard_local_check(
    *, shard: Shard, run_root: Path, baseline: dict[int, Path]
) -> dict[str, Any]:
    pages: dict[int, dict[str, Any]] = {}
    for number in range(shard.start_page, shard.end_page + 1):
        path = baseline.get(number) or (
            run_root / "candidates" / f"page-{number:03d}.json"
        )
        if path.is_file():
            pages[number] = deepcopy(load_json(path))
    pairs: list[dict[str, Any]] = []
    for left_number in sorted(pages):
        if left_number + 1 not in pages:
            continue
        result = reconcile_adjacent_pages(
            pages[left_number], pages[left_number + 1], created_at="2000-01-01T00:00:00Z"
        )
        pairs.append(
            {
                "left_page": left_number,
                "right_page": left_number + 1,
                **result.as_dict(),
            }
        )
    return {
        "pairs": pairs,
        "reciprocal_validation": validate_reciprocal_pages(list(pages.values())).as_dict(),
    }


def _process_shard(
    *,
    shard: Shard,
    source: SourcePackage,
    run_id: str,
    run_root: Path,
    profile: EvaluationProfile,
    backend: TranscriptionBackend,
    baseline: dict[int, Path],
    resume: bool,
    limiter: AdaptiveShardLimiter,
) -> dict[str, Any]:
    limiter.acquire()
    started = time.perf_counter()
    started_at = utc_now()
    results: list[PageRunResult] = []
    orchestration_errors: list[dict[str, Any]] = []
    throttling_events = 0
    try:
        for page in shard.pending_pages:
            try:
                result = _resumed_result(source=source, page=page, run_root=run_root) if resume else None
                if result is None:
                    lock_path = run_root / "locks" / f"page-{page:03d}.lock"
                    with page_lock(lock_path) as acquired:
                        if not acquired:
                            orchestration_errors.append(
                                {
                                    "page": page,
                                    "error": "page lock already held; duplicate work prevented",
                                }
                            )
                            continue
                        result = (
                            _resumed_result(source=source, page=page, run_root=run_root)
                            if resume
                            else None
                        )
                        if result is None:
                            result = transcribe_page(
                                source=source,
                                page=page,
                                run_id=run_id,
                                run_root=run_root,
                                profile=profile,
                                backend=backend,
                            )
                results.append(result)
                throttling_events += sum(
                    event.get("category") == "throttling"
                    for event in result.infrastructure_events
                )
                if result.status != "valid_candidate":
                    _quarantine(run_root, page, result.errors, "transcription")
            except Exception as exc:
                errors = [f"Unhandled page orchestration failure: {type(exc).__name__}: {exc}"]
                orchestration_errors.append({"page": page, "error": errors[0]})
                _quarantine(run_root, page, errors, "orchestration")
        check = _shard_local_check(shard=shard, run_root=run_root, baseline=baseline)
        report = {
            **shard.as_dict(),
            "status": "complete_with_quarantine"
            if any(result.status != "valid_candidate" for result in results)
            or orchestration_errors
            else "complete",
            "started_at": started_at,
            "elapsed_seconds": round(time.perf_counter() - started, 3),
            "results": [result.as_dict() for result in sorted(results, key=lambda item: item.page)],
            "orchestration_errors": sorted(orchestration_errors, key=lambda item: item["page"]),
            "shard_local_reconciliation_check": check,
        }
        write_json(run_root / "shards" / f"{shard.shard_id}.complete.json", report)
        return report
    finally:
        limiter.release(shard_id=shard.shard_id, throttling_events=throttling_events)


def run_corpus_transcription(
    *,
    source: SourcePackage,
    profile: EvaluationProfile,
    workers: int,
    run_id: str,
    resume: bool = False,
    shard_size: int = 20,
    backend: TranscriptionBackend | None = None,
) -> dict[str, Any]:
    workers = parse_worker_count(workers)
    run_root = project_root() / "corpus" / "runs" / run_id
    run_root.mkdir(parents=True, exist_ok=True)
    before_path = run_root / "source-integrity-before.json"
    if before_path.is_file():
        source_before = load_json(before_path)
    else:
        source_before = source_tree_digest(source.root)
        source_before["recorded_at"] = utc_now()
        write_json(before_path, source_before)
    schema_path = run_root / "input-schema.json"
    if not schema_path.is_file():
        export_model_schema(schema_path)
    specification_path = run_root / "input-transcription-specification.md"
    if not specification_path.is_file():
        write_text(
            specification_path,
            (project_root() / "prompts" / "transcribe-page.md").read_text(encoding="utf-8"),
        )
    baseline = owner_approved_pages(source)
    shards = partition_physical_shards(
        total_pages=source.page_count,
        completed_pages=set(baseline),
        shard_size=shard_size,
    )
    actual_backend = backend or CodexExecBackend()
    wall_started = time.perf_counter()
    started_at = utc_now()
    manifest_path = run_root / "run-manifest.json"
    prior = load_json(manifest_path) if resume and manifest_path.is_file() else {}
    manifest = {
        "run_id": run_id,
        "status": "running",
        "created_at": prior.get("created_at", started_at),
        "started_at": started_at,
        "resume": resume,
        "resume_count": int(prior.get("resume_count", 0)) + (1 if resume else 0),
        "source_root": str(source.root),
        "source_tree_sha256_before": source_before["sha256"],
        "source_pdf_sha256": source.lock["source_sha256"],
        "page_count": source.page_count,
        "owner_approved_baseline_pages": sorted(baseline),
        "owner_approved_baseline_count": len(baseline),
        "new_work_pages": sorted(page for shard in shards for page in shard.pending_pages),
        "configured_workers": workers,
        "shard_size": shard_size,
        "shards": [shard.as_dict() for shard in shards],
        "backend": actual_backend.name,
        "model": profile.model,
        "reasoning_effort": profile.reasoning_effort,
        "codex_version": codex_version() if actual_backend.name == "codex-exec" else None,
        "input_schema_sha256": sha256_file(schema_path),
        "transcription_specification_sha256": sha256_file(specification_path),
    }
    write_json(manifest_path, manifest)
    limiter = AdaptiveShardLimiter(workers)
    reports: list[dict[str, Any]] = []
    with ThreadPoolExecutor(max_workers=workers, thread_name_prefix="crowley-shard") as pool:
        futures = {
            pool.submit(
                _process_shard,
                shard=shard,
                source=source,
                run_id=run_id,
                run_root=run_root,
                profile=profile,
                backend=actual_backend,
                baseline=baseline,
                resume=resume,
                limiter=limiter,
            ): shard
            for shard in shards
        }
        for future in as_completed(futures):
            shard = futures[future]
            try:
                reports.append(future.result())
            except Exception as exc:
                failure = {
                    **shard.as_dict(),
                    "status": "worker_failure",
                    "elapsed_seconds": None,
                    "results": [],
                    "orchestration_errors": [
                        {
                            "page": None,
                            "error": f"Shard worker failure: {type(exc).__name__}: {exc}",
                        }
                    ],
                }
                write_json(
                    run_root / "shards" / f"{shard.shard_id}.complete.json",
                    failure,
                )
                reports.append(failure)
    completed_at = utc_now()
    wall_elapsed = time.perf_counter() - wall_started
    source_after = source_tree_digest(source.root)
    source_after["recorded_at"] = completed_at
    write_json(run_root / "source-integrity-after-model.json", source_after)
    if source_after["sha256"] != source_before["sha256"]:
        manifest["status"] = "source_integrity_failure"
        manifest["source_tree_sha256_after_model"] = source_after["sha256"]
        write_json(manifest_path, manifest)
        raise RuntimeError("Source-tree digest changed during corpus model execution")
    reports.sort(key=lambda item: item["start_page"])
    manifest.update(
        {
            "status": "model_execution_complete",
            "completed_at": completed_at,
            "wall_clock_elapsed_seconds": round(wall_elapsed, 3),
            "maximum_observed_active_workers": limiter.max_observed,
            "final_effective_workers": limiter.effective,
            "concurrency_reductions": limiter.reductions,
            "source_tree_sha256_after_model": source_after["sha256"],
            "shard_reports": [
                str(
                    Path("shards")
                    / f"shard-{report['start_page']:03d}-{report['end_page']:03d}.complete.json"
                )
                for report in reports
            ],
        }
    )
    write_json(manifest_path, manifest)
    return finalize_corpus(run_root=run_root, source=source)


def _walk_dicts(node: Any) -> Iterator[dict[str, Any]]:
    if isinstance(node, dict):
        yield node
        for value in node.values():
            yield from _walk_dicts(value)
    elif isinstance(node, list):
        for value in node:
            yield from _walk_dicts(value)


def _page_qa(page: dict[str, Any], status: dict[str, Any] | None) -> dict[str, Any]:
    nodes = list(_walk_dicts(page["blocks"]))
    types = Counter(str(node.get("type")) for node in nodes if node.get("type"))
    mediums = Counter(str(node.get("medium")) for node in nodes if node.get("medium"))
    evidence_statuses = Counter(
        str(node["evidence"]["status"])
        for node in nodes
        if isinstance(node.get("evidence"), dict)
        and node["evidence"].get("status")
    )
    date_statuses = Counter()
    for block in page["blocks"]:
        for date_value in (
            block.get("date"),
            block.get("start_date") if block["type"] == "gap_notice" else None,
            block.get("end_date") if block["type"] == "gap_notice" else None,
        ):
            if date_value:
                date_statuses[date_value["normalization_status"]] += 1
    flags: list[str] = []
    if status and status.get("repair_used"):
        flags.append("repaired")
    for kind in ("unclear", "illegible", "addition", "correction", "deletion"):
        if types[kind]:
            flags.append(kind)
    if mediums["handwritten"] or mediums["mixed"]:
        flags.append("handwriting")
    graphical_note_terms = (
        "non-textual",
        "diagram",
        "drawing",
        "doodle",
        "handwritten mark",
        "handwritten star",
    )
    legacy_graphical_notes = sum(
        any(term in note["text"].casefold() for term in graphical_note_terms)
        for note in page["editorial_notes"]
    )
    if types["graphical_feature"] or legacy_graphical_notes:
        flags.append("graphical")
    for kind in ("table", "list", "letter"):
        if types[kind]:
            flags.append(kind)
    if any(date_statuses[kind] for kind in ("inferred", "ambiguous", "invalid")):
        flags.append("date_normalization")
    if any(
        value == "uncertain"
        for block in page["blocks"]
        for value in (
            block["continuation"]["from_previous"],
            block["continuation"]["to_next"],
        )
    ) or "uncertain" in page["boundary"].values():
        flags.append("continuation_uncertain")
    return {
        "flags": sorted(set(flags)),
        "uncertainties": summarize_uncertainties(page),
        "type_counts": dict(sorted(types.items())),
        "medium_counts": dict(sorted(mediums.items())),
        "evidence_status_counts": dict(sorted(evidence_statuses.items())),
        "date_statuses": dict(sorted(date_statuses.items())),
        "legacy_graphical_note_count": legacy_graphical_notes,
    }


def _status_for_page(run_root: Path, page: int) -> dict[str, Any] | None:
    path = run_root / "pages" / f"page-{page:03d}" / "status.json"
    return load_json(path) if path.is_file() else None


def _backend_results(run_root: Path) -> list[dict[str, Any]]:
    results: list[dict[str, Any]] = []
    for path in sorted((run_root / "pages").glob("page-*/attempt-*/backend-result.json")):
        value = load_json(path)
        value["artifact"] = str(path.relative_to(run_root))
        results.append(value)
    return results


def _repair_failure_categories(
    run_root: Path, statuses: list[dict[str, Any]]
) -> dict[str, dict[str, Any]]:
    categories: dict[str, dict[str, Any]] = {}
    for status in statuses:
        if not status.get("repair_used"):
            continue
        page = int(status["page"])
        prompt = (
            run_root
            / "pages"
            / f"page-{page:03d}"
            / "attempt-02"
            / "prompt.md"
        )
        messages = [
            line.removeprefix("- ").strip()
            for line in prompt.read_text(encoding="utf-8").splitlines()
            if line.startswith(("- canonical:", "- schema:"))
        ] if prompt.is_file() else []
        if not messages:
            category = "uncategorized_validation_repair"
        elif all("block_end before the last line" in item for item in messages):
            category = "premature_internal_block_end"
        elif any(item.startswith("schema:") for item in messages):
            category = "schema_validation"
        else:
            category = "other_canonical_invariant"
        entry = categories.setdefault(category, {"count": 0, "pages": []})
        entry["count"] += 1
        entry["pages"].append(page)
    return dict(sorted(categories.items()))


def finalize_corpus(
    *, run_root: Path, source: SourcePackage, qc_seed: str = DEFAULT_QC_SEED
) -> dict[str, Any]:
    manifest_path = run_root / "run-manifest.json"
    manifest = load_json(manifest_path)
    if manifest["status"] not in {"model_execution_complete", "complete", "partial_failure"}:
        raise ValueError("Model execution must be stopped before corpus finalization")
    baseline = owner_approved_pages(source)
    pages: dict[int, dict[str, Any]] = {}
    states: dict[int, dict[str, Any]] = {}
    for number in range(1, source.page_count + 1):
        baseline_path = baseline.get(number)
        candidate_path = run_root / "candidates" / f"page-{number:03d}.json"
        if baseline_path:
            pages[number] = deepcopy(load_json(baseline_path))
            states[number] = {
                "state": "valid_page_edition",
                "origin": "owner_approved_baseline",
                "path": str(baseline_path),
            }
        elif candidate_path.is_file():
            pages[number] = load_json(candidate_path)
            states[number] = {
                "state": "valid_page_edition",
                "origin": "new_model_candidate",
                "path": str(candidate_path),
            }
        else:
            quarantine_path = run_root / "quarantine" / f"page-{number:03d}.json"
            if not quarantine_path.is_file():
                _quarantine(run_root, number, ["No valid PageEdition exists"], "missing")
            states[number] = {
                "state": "quarantined_unresolved_failure",
                "origin": "quarantine",
                "path": str(quarantine_path),
            }
    reconciliation_pairs: list[dict[str, Any]] = []
    reconciliation_time = manifest.get("completed_at") or utc_now()
    for left_number in range(1, source.page_count):
        if left_number not in pages or left_number + 1 not in pages:
            continue
        result = reconcile_adjacent_pages(
            pages[left_number],
            pages[left_number + 1],
            created_at=reconciliation_time,
        )
        reconciliation_pairs.append(
            {
                "left_page": left_number,
                "right_page": left_number + 1,
                **result.as_dict(),
            }
        )
    canonical_root = run_root / "canonical" / "pages"
    validation_results: list[dict[str, Any]] = []
    invalid_pages: list[int] = []
    page_records: list[dict[str, Any]] = []
    for number, page in sorted(pages.items()):
        path = canonical_root / f"page-{number:03d}.json"
        write_json(path, page)
        validation = validate_page(page, source)
        if not validation.valid:
            invalid_pages.append(number)
            states[number] = {
                "state": "quarantined_unresolved_failure",
                "origin": "post_reconciliation_validation",
                "path": str(path),
                "errors": validation.errors,
            }
        validation_results.append(
            {"page": number, "path": str(path), **validation.as_dict()}
        )
        status = _status_for_page(run_root, number)
        qa = _page_qa(page, status)
        if not validation.valid:
            qa["flags"].append("quarantined")
        page_records.append(
            {
                "page": number,
                "page_id": page["page_id"],
                "state": states[number]["state"],
                "origin": states[number]["origin"],
                "page_role": page["page_role"],
                "review": page["review"],
                "provenance": {
                    "model_run_id": page["provenance"]["model_run_id"],
                    "model": page["provenance"]["model"],
                    "reasoning_effort": page["provenance"]["reasoning_effort"],
                },
                "repair_used": bool(status and status.get("repair_used")),
                "cache_hit": bool(status and status.get("cache_hit")),
                "infrastructure_retries": int(
                    (status or {}).get("infrastructure_retries", 0)
                ),
                "validation": validation.as_dict(),
                "qa": qa,
                "canonical_json": str(path),
                "source_image": str(source.image_path(number)),
                "run_artifacts": str(run_root / "pages" / f"page-{number:03d}"),
            }
        )
    reciprocal = validate_reciprocal_pages(list(pages.values()))
    duplicates = detect_cross_page_duplicates(pages)
    write_json(
        run_root / "reconciliation.json",
        {"pairs": reconciliation_pairs, "validation": reciprocal.as_dict()},
    )
    write_json(
        run_root / "cross-page-duplicates.json",
        {"threshold_words": 8, "findings": duplicates},
    )
    write_json(run_root / "validation.json", {"pages": validation_results})
    unresolved_pages: set[int] = set()
    for pair in reconciliation_pairs:
        if pair["unresolved_left"]:
            unresolved_pages.add(pair["left_page"])
        if pair["unresolved_right"]:
            unresolved_pages.add(pair["right_page"])
    duplicate_pages = {
        number
        for finding in duplicates
        for number in (finding["left_page"], finding["right_page"])
    }
    for record in page_records:
        if record["page"] in unresolved_pages:
            record["qa"]["flags"].append("continuation_unresolved")
        if record["page"] in duplicate_pages:
            record["qa"]["flags"].append("duplicate_boundary_text")
        record["qa"]["flags"] = sorted(set(record["qa"]["flags"]))
    for number, state in sorted(states.items()):
        if number in pages:
            continue
        quarantine = load_json(Path(state["path"]))
        page_records.append(
            {
                "page": number,
                "page_id": f"page-{number:03d}",
                "state": state["state"],
                "origin": state["origin"],
                "page_role": "quarantined",
                "review": {
                    "status": "model_candidate",
                    "authority": "diagnostic_only",
                    "reviewer": None,
                    "reviewed_at": None,
                    "notes": ["No valid PageEdition was produced."],
                },
                "provenance": {
                    "model_run_id": manifest["run_id"],
                    "model": manifest["model"],
                    "reasoning_effort": manifest["reasoning_effort"],
                },
                "repair_used": False,
                "cache_hit": False,
                "infrastructure_retries": 0,
                "validation": {
                    "valid": False,
                    "errors": quarantine["errors"],
                    "warnings": [],
                },
                "qa": {
                    "flags": ["quarantined"],
                    "uncertainties": quarantine["errors"],
                    "type_counts": {},
                    "medium_counts": {},
                    "date_statuses": {},
                    "legacy_graphical_note_count": 0,
                },
                "canonical_json": None,
                "source_image": str(source.image_path(number)),
                "run_artifacts": str(run_root / "pages" / f"page-{number:03d}"),
                "quarantine_json": state["path"],
            }
        )
    page_records.sort(key=lambda item: item["page"])
    targeted = [
        record["page"]
        for record in page_records
        if record["qa"]["flags"] or not record["validation"]["valid"]
    ]
    newly_processed = sorted(set(range(1, source.page_count + 1)) - set(baseline))
    clean_new = [
        record["page"]
        for record in page_records
        if record["page"] in newly_processed and not record["qa"]["flags"]
    ]
    sample_size = min(len(clean_new), max(10, math.ceil(len(clean_new) * 0.05)))
    random_qc = sorted(random.Random(qc_seed).sample(clean_new, sample_size)) if sample_size else []
    for record in page_records:
        record["qa"]["targeted_review"] = record["page"] in targeted
        record["qa"]["random_qc"] = record["page"] in random_qc
        if record["page"] in random_qc:
            record["qa"]["flags"].append("random_qc")
        if record["qa"]["targeted_review"] or record["qa"]["random_qc"]:
            record["qa"]["flags"].append("needs_review")
        record["qa"]["flags"] = sorted(set(record["qa"]["flags"]))
    quarantine_pages = sorted(
        number
        for number, state in states.items()
        if state["state"] == "quarantined_unresolved_failure"
    )
    backend_results = _backend_results(run_root)
    invocation_times = [
        float(result["elapsed_seconds"])
        for result in backend_results
        if result.get("elapsed_seconds") is not None
    ]
    status_values = [
        status
        for number in newly_processed
        if (status := _status_for_page(run_root, number)) is not None
    ]
    block_counts: Counter[str] = Counter()
    role_counts: Counter[str] = Counter()
    profile_counts: Counter[str] = Counter()
    flag_counts: Counter[str] = Counter()
    evidence_status_counts: Counter[str] = Counter()
    continuation_counts: Counter[str] = Counter()
    for number, page in sorted(pages.items()):
        role_counts[page["page_role"]] += 1
        profile_counts[
            f"{page['provenance']['model']}:{page['provenance']['reasoning_effort']}"
        ] += 1
        for block in page["blocks"]:
            block_counts[block["type"]] += 1
            if block["continuation"]["next"]:
                continuation_counts[block["type"]] += 1
        page_qa = _page_qa(page, _status_for_page(run_root, number))
        flag_counts.update(page_qa["flags"])
        evidence_status_counts.update(page_qa["evidence_status_counts"])
    infrastructure_events = [
        event
        for status in status_values
        for event in status.get("infrastructure_events", [])
    ]
    wall_elapsed = float(manifest.get("wall_clock_elapsed_seconds", 0.0))
    summed_invocation = sum(invocation_times)
    prior_baseline_estimate = len(newly_processed) * SEQUENTIAL_PILOT_SECONDS_PER_PAGE
    report = {
        "created_at": utc_now(),
        "run_id": manifest["run_id"],
        "warning": "Model candidates are not owner-approved unless their review metadata explicitly says so; no accuracy percentage is calculated.",
        "coverage": {
            "total_physical_pages": source.page_count,
            "represented_states": len(states),
            "valid_page_editions": len(pages) - len(invalid_pages),
            "owner_approved_baseline_count": len(baseline),
            "owner_approved_baseline_pages": sorted(baseline),
            "newly_processed_count": len(newly_processed),
            "first_pass_successes": sum(
                status.get("status") == "valid_candidate"
                and not status.get("repair_used")
                for status in status_values
            ),
            "repaired_successes": sum(
                status.get("status") == "valid_candidate" and status.get("repair_used")
                for status in status_values
            ),
            "quarantined_pages": quarantine_pages,
            "page_role_counts": dict(sorted(role_counts.items())),
        },
        "structure": {
            "block_type_counts": dict(sorted(block_counts.items())),
            "graphical_non_textual_feature_count": block_counts["graphical_feature"]
            + sum(record["qa"]["legacy_graphical_note_count"] for record in page_records),
            "explicit_graphical_feature_blocks": block_counts["graphical_feature"],
            "legacy_graphical_editorial_notes": sum(
                record["qa"]["legacy_graphical_note_count"] for record in page_records
            ),
            "book_segments": sorted({page["segment"]["id"] for page in pages.values()}),
            "continuation_links_by_block_type": dict(sorted(continuation_counts.items())),
            "continuation_link_count": sum(continuation_counts.values()),
            "unresolved_continuation_pairs": [
                {
                    "left_page": pair["left_page"],
                    "right_page": pair["right_page"],
                    "unresolved_left": pair["unresolved_left"],
                    "unresolved_right": pair["unresolved_right"],
                }
                for pair in reconciliation_pairs
                if pair["unresolved_left"] or pair["unresolved_right"]
            ],
            "duplicate_boundary_findings": duplicates,
        },
        "qa": {
            "flag_counts": dict(sorted(flag_counts.items())),
            "evidence_alignment_status_counts": dict(
                sorted(evidence_status_counts.items())
            ),
            "repair_failure_categories": _repair_failure_categories(
                run_root, status_values
            ),
            "targeted_review_pages": sorted(targeted),
            "targeted_review_count": len(targeted),
            "random_qc": {
                "seed": qc_seed,
                "method": "Python random.Random(seed), sorted sample of max(10, ceil(5%)) otherwise-clean new pages",
                "eligible_clean_new_pages": len(clean_new),
                "pages": random_qc,
            },
            "invalid_pages": invalid_pages,
            "reciprocal_validation": reciprocal.as_dict(),
            "observed_failure_patterns": {
                "ocr_tsv": [
                    "TSV evidence alignment is secondary; partial or absent alignment is expected for corrected, handwritten, unclear, and non-textual source material.",
                    "QA remains sensitive to line-break hyphens, lexical hyphens, small source marks, and tabular spacing that conventional OCR can lose or flatten.",
                ],
                "model": [
                    "Categorized validation-repair causes are reported in repair_failure_categories.",
                    "Unresolved continuations, duplicate boundary text, and quarantined pages are reported separately and are never silently repaired.",
                ],
            },
        },
        "execution": {
            "backend": manifest["backend"],
            "configured_workers": manifest["configured_workers"],
            "maximum_observed_active_workers": manifest.get(
                "maximum_observed_active_workers"
            ),
            "final_effective_workers": manifest.get("final_effective_workers"),
            "concurrency_reductions": manifest.get("concurrency_reductions", []),
            "shards": manifest["shards"],
            "model_profile_page_counts": dict(sorted(profile_counts.items())),
            "codex_version": manifest.get("codex_version"),
            "model_invocations": len(backend_results),
            "transcription_repair_invocations": sum(
                status.get("repair_used") for status in status_values
            ),
            "infrastructure_retries": sum(
                status.get("infrastructure_retries", 0) for status in status_values
            ),
            "infrastructure_failure_events": infrastructure_events,
            "throttling_events": sum(
                event.get("category") == "throttling" for event in infrastructure_events
            ),
            "cache_hits": sum(status.get("cache_hit", False) for status in status_values),
            "cache_misses": sum(not status.get("cache_hit", False) for status in status_values),
            "resumed_pages": sum(status.get("resumed", False) for status in status_values),
            "summed_invocation_runtime_seconds": round(summed_invocation, 3),
            "wall_clock_elapsed_seconds": wall_elapsed,
            "mean_invocation_seconds": round(statistics.mean(invocation_times), 3)
            if invocation_times
            else None,
            "median_invocation_seconds": round(statistics.median(invocation_times), 3)
            if invocation_times
            else None,
            "parallel_effectiveness": {
                "pilot_sequential_baseline_seconds_per_page": SEQUENTIAL_PILOT_SECONDS_PER_PAGE,
                "baseline_estimated_sequential_seconds": round(prior_baseline_estimate, 3),
                "estimated_speedup_vs_pilot_baseline": round(
                    prior_baseline_estimate / wall_elapsed, 2
                )
                if wall_elapsed
                else None,
                "observed_invocation_concurrency_ratio": round(
                    summed_invocation / wall_elapsed, 2
                )
                if wall_elapsed
                else None,
            },
        },
        "source_integrity": {
            "before": load_json(run_root / "source-integrity-before.json"),
            "after_model": load_json(run_root / "source-integrity-after-model.json"),
            "after_final": load_json(run_root / "source-integrity-after-final.json")
            if (run_root / "source-integrity-after-final.json").is_file()
            else load_json(run_root / "source-integrity-after-model.json"),
        },
        "pages": sorted(page_records, key=lambda item: item["page"]),
    }
    report["source_integrity"]["matches"] = (
        report["source_integrity"]["before"]["sha256"]
        == report["source_integrity"]["after_final"]["sha256"]
    )
    report_root = run_root / "report"
    write_json(report_root / "corpus-report.json", report)
    write_text(report_root / "corpus-report.md", render_corpus_report(report))
    write_json(
        report_root / "qa-queue.json",
        {
            "targeted_review_pages": sorted(targeted),
            "random_qc": report["qa"]["random_qc"],
            "pages": [
                record
                for record in report["pages"]
                if record["qa"]["targeted_review"] or record["qa"]["random_qc"]
            ],
        },
    )
    build_corpus_review(run_root=run_root, source=source, report=report)
    manifest.update(
        {
            "status": "complete" if not quarantine_pages and not invalid_pages and reciprocal.valid else "partial_failure",
            "finalized_at": utc_now(),
            "coverage_state_count": len(states),
            "valid_page_editions": report["coverage"]["valid_page_editions"],
            "quarantined_pages": quarantine_pages,
            "corpus_report": str(report_root / "corpus-report.json"),
            "review_index": str(run_root / "review" / "index.html"),
        }
    )
    write_json(manifest_path, manifest)
    try:
        run_root.resolve().relative_to((project_root() / "corpus" / "runs").resolve())
    except ValueError:
        pass
    else:
        write_json(
            project_root() / "corpus" / "latest-run.json",
            {"run_id": manifest["run_id"], "status": manifest["status"]},
        )
    return report


def render_corpus_report(report: dict[str, Any]) -> str:
    coverage = report["coverage"]
    structure = report["structure"]
    qa = report["qa"]
    execution = report["execution"]
    shard_ranges = [
        f"{item['start_page']}-{item['end_page']}" for item in execution["shards"]
    ]
    lines = [
        "# Crowley diary corpus-wide PASS 1 report",
        "",
        report["warning"],
        "",
        "## Coverage",
        "",
        f"- Physical pages: {coverage['total_physical_pages']}",
        f"- Valid PageEditions: {coverage['valid_page_editions']}",
        f"- Owner-approved baseline: {coverage['owner_approved_baseline_count']}",
        f"- Newly processed: {coverage['newly_processed_count']}",
        f"- First-pass successes: {coverage['first_pass_successes']}",
        f"- Repaired successes: {coverage['repaired_successes']}",
        f"- Quarantined: {coverage['quarantined_pages']}",
        f"- Page roles: {coverage['page_role_counts']}",
        "",
        "## Structure and QA",
        "",
        f"- Block types: {structure['block_type_counts']}",
        f"- Graphical/non-textual features: {structure['graphical_non_textual_feature_count']}",
        f"- Continuation links: {structure['continuation_link_count']} ({structure['continuation_links_by_block_type']})",
        f"- Unresolved continuation boundaries: {len(structure['unresolved_continuation_pairs'])}",
        f"- Duplicate boundary findings: {len(structure['duplicate_boundary_findings'])}",
        f"- QA flags: {qa['flag_counts']}",
        f"- TSV evidence alignment statuses: {qa['evidence_alignment_status_counts']}",
        f"- Validation-repair causes: {qa['repair_failure_categories']}",
        f"- Targeted-review pages: {qa['targeted_review_count']}",
        f"- Random clean-page QC seed: `{qa['random_qc']['seed']}`",
        f"- Random clean-page QC pages: {qa['random_qc']['pages']}",
        f"- Recurring OCR-sensitive patterns: {qa['observed_failure_patterns']['ocr_tsv']}",
        f"- Recurring model/validation patterns: {qa['observed_failure_patterns']['model']}",
        "",
        "## Parallel execution",
        "",
        f"- Configured / maximum observed workers: {execution['configured_workers']} / {execution['maximum_observed_active_workers']}",
        f"- Shards: {shard_ranges}",
        f"- Model invocations: {execution['model_invocations']}",
        f"- Transcription repairs: {execution['transcription_repair_invocations']}",
        f"- Infrastructure retries / throttling events: {execution['infrastructure_retries']} / {execution['throttling_events']}",
        f"- Cache hits / misses: {execution['cache_hits']} / {execution['cache_misses']}",
        f"- Summed invocation runtime: {execution['summed_invocation_runtime_seconds']} seconds",
        f"- Wall-clock corpus runtime: {execution['wall_clock_elapsed_seconds']} seconds",
        f"- Mean / median invocation: {execution['mean_invocation_seconds']} / {execution['median_invocation_seconds']} seconds",
        f"- Estimated speedup vs the 155.181 s/page pilot baseline: {execution['parallel_effectiveness']['estimated_speedup_vs_pilot_baseline']}×",
        f"- Concurrency reductions: {execution['concurrency_reductions']}",
        "",
        "## Source integrity",
        "",
        f"- Before: `{report['source_integrity']['before']['sha256']}`",
        f"- After model execution: `{report['source_integrity']['after_model']['sha256']}`",
        f"- After final resumption and aggregation: `{report['source_integrity']['after_final']['sha256']}`",
        f"- Match: {report['source_integrity']['matches']}",
        "",
        "The corpus review index is `review/index.html`; the prioritized queue is `report/qa-queue.json`.",
        "",
    ]
    return "\n".join(lines)


CORPUS_REVIEW_CSS = """body{font:15px/1.45 system-ui,sans-serif;margin:0;background:#f7f3ea;color:#25231f}header,main{padding:1rem 2rem}header{background:#26362f;color:#fff}header a{color:#bfe1ff}a{color:#17609a}.warning{background:#fff0bd;color:#4b3300;padding:.7rem;border-left:5px solid #c78a00}.filters{display:flex;gap:.45rem;flex-wrap:wrap;margin:1rem 0}.filters button{padding:.4rem .65rem}.page-grid{display:grid;grid-template-columns:minmax(350px,45vw) minmax(0,1fr);gap:1.4rem}.scan{position:sticky;top:0;align-self:start}.scan img{width:100%;background:#fff}.panels{min-width:0}.renderings{display:grid;grid-template-columns:1fr 1fr;gap:.8rem}pre{white-space:pre-wrap;overflow-wrap:anywhere;max-height:34rem;overflow:auto;background:#fff;border:1px solid #bbb;padding:.8rem}table{width:100%;border-collapse:collapse;background:#fff}th,td{border:1px solid #bbb;padding:.45rem;text-align:left;vertical-align:top}th{background:#e5e0d5;position:sticky;top:0}.tag{display:inline-block;background:#e4ece7;border-radius:999px;padding:.15rem .45rem;margin:.1rem}nav{display:flex;justify-content:space-between}@media(max-width:900px){header,main{padding:1rem}.page-grid,.renderings{grid-template-columns:1fr}.scan{position:static}}"""


def build_corpus_review(
    *, run_root: Path, source: SourcePackage, report: dict[str, Any]
) -> None:
    review_root = run_root / "review"
    write_text(review_root / "assets" / "review.css", CORPUS_REVIEW_CSS)
    (review_root / "images").mkdir(parents=True, exist_ok=True)
    (review_root / "pages").mkdir(parents=True, exist_ok=True)
    rows: list[str] = []
    records = report["pages"]
    for position, record in enumerate(records):
        number = record["page"]
        page_id = record["page_id"]
        image_link = review_root / "images" / f"{page_id}.png"
        if not image_link.exists() and not image_link.is_symlink():
            image_link.symlink_to(source.image_path(number))
        page = (
            load_json(Path(record["canonical_json"]))
            if record["canonical_json"]
            else None
        )
        reading = (
            render_page(page, diplomatic=False)
            if page
            else "[No valid PageEdition; see quarantine errors.]"
        )
        diplomatic = (
            render_page(page, diplomatic=True)
            if page
            else "[No valid PageEdition; see quarantine errors.]"
        )
        previous_page = records[position - 1]["page"] if position else None
        next_page = records[position + 1]["page"] if position + 1 < len(records) else None
        navigation = (
            f'<a href="page-{previous_page:03d}.html">← page {previous_page}</a>'
            if previous_page
            else "<span></span>"
        )
        navigation += ' <a href="../index.html">index</a> '
        navigation += (
            f'<a href="page-{next_page:03d}.html">page {next_page} →</a>'
            if next_page
            else "<span></span>"
        )
        flags = record["qa"]["flags"]
        warnings = [*record["validation"]["warnings"]]
        json_panel = (
            f'<details><summary>PageEdition JSON</summary><p><a href="../../canonical/pages/{page_id}.json">open JSON</a></p><pre>{escape(json.dumps(page, ensure_ascii=False, indent=2))}</pre></details>'
            if page
            else f'<details open><summary>Quarantine record</summary><p><a href="../../quarantine/{page_id}.json">open quarantine JSON</a></p><pre>{escape(json.dumps(record["validation"], ensure_ascii=False, indent=2))}</pre></details>'
        )
        provenance_panel = (
            json.dumps(page["provenance"], ensure_ascii=False, indent=2)
            if page
            else json.dumps(record["provenance"], ensure_ascii=False, indent=2)
        )
        page_html = f"""<!doctype html><html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width"><title>{page_id} corpus review</title><link rel="stylesheet" href="../assets/review.css"></head><body><header><nav>{navigation}</nav><h1>{page_id}</h1><p class="warning">The scan is authoritative. This engineering review page does not change approval status.</p><p>{''.join(f'<span class="tag">{escape(flag)}</span>' for flag in flags) or 'clean / unflagged'}</p></header><main class="page-grid"><section class="scan"><h2>Authoritative scan</h2><a href="../images/{page_id}.png"><img src="../images/{page_id}.png" alt="Scan {page_id}"></a></section><section class="panels"><p><strong>Role:</strong> {escape(record['page_role'])} · <strong>Authority:</strong> {escape(record['review']['authority'])} · <strong>Status:</strong> {escape(record['review']['status'])} · <strong>Repair:</strong> {'yes' if record['repair_used'] else 'no'}</p><h2>Uncertainties</h2><ul>{''.join(f'<li>{escape(item)}</li>' for item in record['qa']['uncertainties']) or '<li>None recorded</li>'}</ul><h2>Warnings</h2><ul>{''.join(f'<li>{escape(item)}</li>' for item in warnings) or '<li>None recorded</li>'}</ul><div class="renderings"><section><h2>Reading</h2><pre>{escape(reading)}</pre></section><section><h2>Diplomatic</h2><pre>{escape(diplomatic)}</pre></section></div>{json_panel}<h2>Provenance</h2><pre>{escape(provenance_panel)}</pre><p><a href="../../pages/{page_id}/">new-run artifacts (when present)</a></p></section></main></body></html>"""
        write_text(review_root / "pages" / f"{page_id}.html", page_html)
        row_flags = " ".join(flags) or "clean"
        rows.append(
            f'<tr data-flags="{escape(row_flags)}"><td><a href="pages/{page_id}.html">{number}</a></td><td>{escape(record["page_role"])}</td><td>{escape(record["review"]["status"])}</td><td>{"yes" if record["repair_used"] else "no"}</td><td>{len(record["qa"]["uncertainties"])}</td><td>{escape(", ".join(flags) or "clean / unflagged")}</td></tr>'
        )
    filters = [
        "all",
        "needs_review",
        "repaired",
        "handwriting",
        "graphical",
        "unclear",
        "illegible",
        "table",
        "list",
        "letter",
        "date_normalization",
        "addition",
        "correction",
        "deletion",
        "quarantined",
        "random_qc",
        "clean",
    ]
    index = f"""<!doctype html><html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width"><title>Crowley corpus PASS 1 review</title><link rel="stylesheet" href="assets/review.css"></head><body><header><h1>Crowley diary corpus PASS 1 review</h1><p class="warning">The scan is authoritative. Model candidates are not owner-approved unless explicitly labeled.</p></header><main><p><a href="../report/corpus-report.md">corpus report</a> · <a href="../report/qa-queue.json">prioritized QA queue</a></p><div class="filters">{''.join(f'<button data-filter="{value}">{value}</button>' for value in filters)}</div><table><thead><tr><th>Page</th><th>Role</th><th>Review status</th><th>Repair</th><th>Uncertainties</th><th>QA flags</th></tr></thead><tbody>{''.join(rows)}</tbody></table></main><script>for(const button of document.querySelectorAll('button[data-filter]'))button.addEventListener('click',()=>{{const value=button.dataset.filter;for(const row of document.querySelectorAll('tbody tr')){{const flags=row.dataset.flags.split(' ');row.hidden=value!=='all'&&!flags.includes(value)}}}});</script></body></html>"""
    write_text(review_root / "index.html", index)
    write_json(review_root / "index.json", {"pages": records})
