SummerProject2026/scripts/export_dashboard_data.py

"""Publish privacy-reviewed aggregate JSON assets for the dashboard.

The exporter never loads a model or emits row-level predictions. It reads only
the train, tune, and calibration portions of the private feature mart. The 2025
locked-test partition is outside every aggregation query.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import math
import os
import re
import shutil
import sys
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Dict, Iterable, List, Mapping, Optional, Sequence, Tuple

import duckdb


PROJECT_ROOT = Path(__file__).resolve().parents[1]
PUBLIC_DATA_ROOT = (PROJECT_ROOT / "dashboard/public/data").resolve()
SCHEMA_VERSION = "dashboard_data_v1"
ALLOWED_PARTITIONS = ("train", "tune", "calibrate")
PARTITION_ORDER = {name: index for index, name in enumerate(ALLOWED_PARTITIONS)}
LOCKED_PARTITION_ALIASES = ("test", "locked_test", "locked-test")
MIN_SUPPORT = 100
MIN_CLASS_COUNT = 10
SUPPORT_ROUNDING = 100
MAX_SCORECARDS = 200

ROW_ASSET_SCHEMAS = {
    "overview_period_county.json": (
        "year",
        "quarter",
        "public_county",
        "support_rounded",
        "nonpass_rate",
    ),
    "age_risk_curve.json": (
        "age_band",
        "support_rounded",
        "nonpass_rate",
    ),
    "cohort_scorecard.json": (
        "prior_make",
        "prior_model",
        "support_rounded",
        "nonpass_rate",
    ),
    "model_diagnostics.json": (
        "model",
        "partition",
        "average_precision",
        "brier",
        "log_loss",
        "roc_auc",
        "top_10_capture",
    ),
    "coverage_quality.json": (
        "year",
        "source_era",
        "support_rounded",
        "labeled_rate",
        "overall_result_share",
        "utah_obd_proxy_share",
    ),
}
PUBLIC_ASSET_NAMES = tuple(
    sorted(
        list(ROW_ASSET_SCHEMAS)
        + ["filter_catalog.json", "data_manifest.json"]
    )
)
SHA256_MANIFEST_NAME = "sha256_manifest.json"
PUBLIC_OUTPUT_NAMES = frozenset(PUBLIC_ASSET_NAMES + (SHA256_MANIFEST_NAME,))

REQUIRED_MART_COLUMNS = {
    "vehicle_token",
    "is_vin_audit",
    "episode_number",
    "episode_start",
    "first_outcome",
    "target_outcome_label_source",
    "target_nonpass",
    "eligible_returning_target",
    "temporal_partition",
    "public_county",
    "source_era",
    "vehicle_age",
    "last_observed_make",
    "last_observed_model",
}
DENIED_EXACT_KEYS = {
    "credential",
    "credentials",
    "inspector",
    "inspector_id",
    "password",
    "secret",
    "technician",
    "technician_id",
    "vehicle_token",
    "vehicle_bucket",
    "is_vin_audit",
    "episode_number",
    "episode_start",
    "first_outcome",
    "target_nonpass",
    "target_outcome_label_source",
    "outcome_label_source",
    "internal_event_id",
    "vin",
    "plate",
    "zip",
    "zipcode",
    "station",
    "station_id",
    "raw_json",
    "row_prediction",
    "prediction",
    "probability",
    "pass_count",
    "nonpass_count",
    "row_count",
}
DENIED_KEY_FRAGMENTS = (
    "credential",
    "inspector",
    "password",
    "secret",
    "technician",
    "vehicle_token",
    "internal_event",
    "license_plate",
    "exact_timestamp",
    "row_prediction",
    "raw_",
    "label_source",
)
EXACT_TIMESTAMP_PATTERN = re.compile(r"^\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}")
VIN_LIKE_PATTERN = re.compile(r"^[A-HJ-NPR-Z0-9]{17}$")
SAFE_CATEGORY_PATTERN = re.compile(r"^[^\x00-\x1f\x7f]{1,80}$")
SAFE_MODEL_NAME_PATTERN = re.compile(r"^[A-Za-z0-9._-]{1,64}$")


class DashboardExportError(Exception):
    """Base class for expected publication failures."""


class ManifestError(DashboardExportError):
    """Raised when private input provenance cannot be verified."""


class PublicSchemaError(DashboardExportError):
    """Raised when an asset violates the public-data contract."""


@dataclass(frozen=True)
class ModelBundle:
    manifest: Path
    metrics: Path


def parse_args(argv: Optional[Sequence[str]] = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--mart", required=True, type=Path)
    parser.add_argument("--mart-manifest", type=Path)
    parser.add_argument(
        "--model-manifest",
        required=True,
        action="append",
        type=Path,
        help="Private model manifest; repeat once per model bundle.",
    )
    parser.add_argument(
        "--model-metrics",
        required=True,
        action="append",
        type=Path,
        help="Metrics JSON paired by position with --model-manifest.",
    )
    parser.add_argument("--output-dir", type=Path, default=PUBLIC_DATA_ROOT)
    parser.add_argument("--development-preview", action="store_true")
    parser.add_argument("--overwrite", action="store_true")
    return parser.parse_args(argv)


def _reject_json_constant(value: str) -> object:
    raise ManifestError("JSON contains a non-finite numeric constant: " + value)


def _unique_object(pairs: Iterable[Tuple[str, object]]) -> Dict[str, object]:
    result: Dict[str, object] = {}
    for key, value in pairs:
        if key in result:
            raise ManifestError("JSON contains duplicate key: " + key)
        result[key] = value
    return result


def load_json(path: Path) -> object:
    try:
        if path.stat().st_size > 20 * 1024 * 1024:
            raise ManifestError("Refusing an unexpectedly large JSON input")
        return json.loads(
            path.read_text(encoding="utf-8"),
            object_pairs_hook=_unique_object,
            parse_constant=_reject_json_constant,
        )
    except OSError as exc:
        raise ManifestError("Cannot read required JSON input") from exc
    except UnicodeError as exc:
        raise ManifestError("JSON input is not valid UTF-8") from exc
    except json.JSONDecodeError as exc:
        raise ManifestError("JSON input is malformed") from exc


def file_sha256(path: Path) -> str:
    digest = hashlib.sha256()
    try:
        with path.open("rb") as handle:
            for chunk in iter(lambda: handle.read(1024 * 1024), b""):
                digest.update(chunk)
    except OSError as exc:
        raise ManifestError("Cannot read a required private input") from exc
    return digest.hexdigest()


def _require_mapping(value: object, label: str) -> Mapping[str, object]:
    if not isinstance(value, dict):
        raise ManifestError(label + " must be a JSON object")
    return value


def _mart_manifest_path(mart: Path, explicit: Optional[Path]) -> Path:
    if explicit is not None:
        return explicit
    return mart.with_name(mart.name + ".manifest.json")


def validate_mart_manifest(
    mart: Path,
    manifest_path: Path,
    development_preview: bool,
) -> Tuple[Mapping[str, object], bool, str]:
    manifest = _require_mapping(load_json(manifest_path), "Mart manifest")
    if manifest.get("build_kind") != "private_leakage_safe_inspection_feature_mart":
        raise ManifestError("Input is not the approved private feature mart")
    classification = manifest.get("classification")
    if not isinstance(classification, str) or not classification.startswith("private_"):
        raise ManifestError("Feature mart classification is not private")

    expected_digest = manifest.get("parquet_sha256")
    if not isinstance(expected_digest, str) or not re.fullmatch(
        r"[0-9a-f]{64}", expected_digest
    ):
        raise ManifestError("Mart manifest has no valid Parquet digest")
    actual_digest = file_sha256(mart)
    if actual_digest != expected_digest:
        raise ManifestError("Feature mart digest does not match its manifest")

    population_allowed = manifest.get("population_estimate_allowed")
    if not isinstance(population_allowed, bool):
        raise ManifestError("Mart population-estimate flag must be boolean")
    source_kind = manifest.get("source_data_kind")
    if not isinstance(source_kind, str):
        raise ManifestError("Mart source-data kind is missing")
    source_kind_lower = source_kind.lower()
    development_source = (
        not population_allowed
        or "development" in source_kind_lower
        or "sample" in source_kind_lower
    )
    if development_source and not development_preview:
        raise ManifestError(
            "Development-sample marts require --development-preview"
        )
    if not development_preview and not population_allowed:
        raise ManifestError("Population publication requires an approved mart")

    columns = manifest.get("columns")
    if not isinstance(columns, list):
        raise ManifestError("Mart manifest has no column contract")
    declared_names = {
        item.get("name")
        for item in columns
        if isinstance(item, dict) and isinstance(item.get("name"), str)
    }
    missing = sorted(REQUIRED_MART_COLUMNS - declared_names)
    if missing:
        raise ManifestError("Mart manifest is missing required columns")
    return manifest, development_source, actual_digest


def load_model_diagnostics(
    bundles: Sequence[ModelBundle],
    mart_digest: str,
    partition_vehicle_support: Mapping[str, Tuple[int, int]],
) -> Tuple[List[Dict[str, object]], List[Tuple[str, str]]]:
    diagnostics: List[Dict[str, object]] = []
    model_provenance: List[Tuple[str, str]] = []
    seen = set()
    for bundle in bundles:
        manifest = _require_mapping(load_json(bundle.manifest), "Model manifest")
        if manifest.get("locked_test_evaluated") is not False:
            raise ManifestError(
                "Every model manifest must state locked_test_evaluated=false"
            )
        classification = manifest.get("classification")
        if not isinstance(classification, str) or "no_row_predictions" not in classification:
            raise ManifestError("Model artifact classification is not publishable")
        if manifest.get("input_sha256") != mart_digest:
            raise ManifestError("Model metrics were not built from this mart")
        expected_metrics_digest = manifest.get("metrics_sha256")
        if not isinstance(expected_metrics_digest, str) or not re.fullmatch(
            r"[0-9a-f]{64}", expected_metrics_digest
        ):
            raise ManifestError("Model manifest has no valid metrics digest")
        metrics_digest = file_sha256(bundle.metrics)
        if metrics_digest != expected_metrics_digest:
            raise ManifestError("Model metrics digest does not match its manifest")
        model_version = manifest.get("model_version")
        if not isinstance(model_version, str) or not SAFE_MODEL_NAME_PATTERN.fullmatch(
            model_version
        ):
            raise ManifestError("Model manifest has an unsafe model version")
        model_provenance.append((model_version, metrics_digest))

        payload = load_json(bundle.metrics)
        if isinstance(payload, dict):
            payload = payload.get("metrics")
        if not isinstance(payload, list):
            raise ManifestError("Model metrics must be a JSON array")

        for raw in payload:
            if not isinstance(raw, dict):
                raise ManifestError("Each model metric must be an object")
            partition = raw.get("partition")
            if partition not in ALLOWED_PARTITIONS:
                raise ManifestError(
                    "Metrics contain a locked, shadow, or unknown partition"
                )
            if raw.get("cohort") != "non_audit":
                continue
            episodes = raw.get("episodes")
            nonpass = raw.get("nonpass")
            vehicles = raw.get("vehicles")
            if (
                isinstance(episodes, bool)
                or not isinstance(episodes, int)
                or isinstance(nonpass, bool)
                or not isinstance(nonpass, int)
                or isinstance(vehicles, bool)
                or not isinstance(vehicles, int)
                or vehicles < 0
                or vehicles > episodes
                or nonpass < 0
                or nonpass > episodes
            ):
                raise ManifestError(
                    "Model metrics lack valid private suppression counts"
                )
            class_vehicle_support = partition_vehicle_support.get(partition)
            if class_vehicle_support is None or not _publishable(
                episodes,
                nonpass,
                vehicles,
                class_vehicle_support[0],
                class_vehicle_support[1],
            ):
                continue
            model = raw.get("model")
            if not isinstance(model, str) or not SAFE_MODEL_NAME_PATTERN.fullmatch(model):
                raise ManifestError("Model metric has an unsafe model name")

            row: Dict[str, object] = {"model": model, "partition": partition}
            for key in (
                "average_precision",
                "brier",
                "log_loss",
                "roc_auc",
                "top_10_capture",
            ):
                value = raw.get(key)
                if isinstance(value, bool) or not isinstance(value, (int, float)):
                    raise ManifestError("Model metric is missing a numeric value")
                number = float(value)
                if not math.isfinite(number) or number < 0.0:
                    raise ManifestError("Model metric is non-finite or negative")
                if key != "log_loss" and number > 1.0:
                    raise ManifestError("Bounded model metric exceeds one")
                row[key] = round(number, 4)

            identity = (model, partition)
            if identity in seen:
                raise ManifestError("Duplicate model/partition diagnostic")
            seen.add(identity)
            diagnostics.append(row)

    if not diagnostics:
        raise ManifestError("No approved train/tune/calibrate diagnostics found")
    diagnostics.sort(
        key=lambda row: (str(row["model"]), PARTITION_ORDER[str(row["partition"])])
    )
    model_provenance.sort()
    return diagnostics, model_provenance


def _round_support(value: int) -> int:
    if value < MIN_SUPPORT:
        raise PublicSchemaError("Unsuppressed support is below the threshold")
    return int(math.floor((value + SUPPORT_ROUNDING / 2) / SUPPORT_ROUNDING)) * SUPPORT_ROUNDING


def _rate(numerator: int, denominator: int) -> float:
    if denominator <= 0 or numerator < 0 or numerator > denominator:
        raise PublicSchemaError("Invalid aggregate rate inputs")
    return round(numerator / denominator, 3)


def _publishable(
    n: int,
    nonpass: int,
    vehicles: int,
    pass_vehicles: int,
    nonpass_vehicles: int,
) -> bool:
    return (
        n >= MIN_SUPPORT
        and nonpass >= MIN_CLASS_COUNT
        and n - nonpass >= MIN_CLASS_COUNT
        and vehicles >= MIN_SUPPORT
        and pass_vehicles >= MIN_CLASS_COUNT
        and nonpass_vehicles >= MIN_CLASS_COUNT
    )


def _safe_category(value: object, label: str) -> str:
    if not isinstance(value, str):
        raise PublicSchemaError(label + " category is not text")
    normalized = value.strip()
    if not SAFE_CATEGORY_PATTERN.fullmatch(normalized):
        raise PublicSchemaError(label + " category contains unsafe characters")
    compact = re.sub(r"[^A-Za-z0-9]", "", normalized).upper()
    if VIN_LIKE_PATTERN.fullmatch(compact):
        raise PublicSchemaError(label + " category resembles a VIN")
    return normalized


def _create_safe_mart(connection: duckdb.DuckDBPyConnection, mart: Path) -> None:
    description = connection.execute(
        "DESCRIBE SELECT * FROM read_parquet(?)", [str(mart)]
    ).fetchall()
    actual_columns = {str(row[0]) for row in description}
    if not REQUIRED_MART_COLUMNS.issubset(actual_columns):
        raise ManifestError("Parquet schema does not match the mart contract")

    boundary_errors = connection.execute(
        """
        SELECT count(*)
        FROM read_parquet(?)
        WHERE (temporal_partition IN ('train', 'tune', 'calibrate')
               AND episode_start >= TIMESTAMP '2025-01-01')
           OR (temporal_partition IN ('test', 'locked_test', 'locked-test')
               AND episode_start < TIMESTAMP '2025-01-01')
        """,
        [str(mart)],
    ).fetchone()[0]
    if boundary_errors:
        raise ManifestError("Mart partition timestamps violate the locked boundary")

    # The filter is applied before target fields enter this temporary table.
    # No subsequent query can read or evaluate a 2025 target.
    connection.execute(
        """
        CREATE TEMP TABLE safe_mart AS
        SELECT
            vehicle_token,
            is_vin_audit,
            episode_number,
            episode_start,
            first_outcome,
            target_outcome_label_source,
            target_nonpass,
            eligible_returning_target,
            temporal_partition,
            public_county,
            source_era,
            vehicle_age,
            last_observed_make,
            last_observed_model
        FROM read_parquet(?)
        WHERE temporal_partition IN ('train', 'tune', 'calibrate')
          AND episode_start < TIMESTAMP '2025-01-01'
        """,
        [str(mart)],
    )


def _partition_vehicle_support(
    connection: duckdb.DuckDBPyConnection,
) -> Dict[str, Tuple[int, int]]:
    """Return private class-level entity supports for diagnostic suppression."""
    raw = connection.execute(
        """
        SELECT temporal_partition,
               count(DISTINCT vehicle_token) FILTER (
                   WHERE target_nonpass = 0
               )::BIGINT AS pass_vehicles,
               count(DISTINCT vehicle_token) FILTER (
                   WHERE target_nonpass = 1
               )::BIGINT AS nonpass_vehicles
        FROM safe_mart
        WHERE eligible_returning_target
          AND target_nonpass IN (0, 1)
          AND NOT is_vin_audit
        GROUP BY 1
        """
    ).fetchall()
    return {
        str(partition): (int(pass_vehicles), int(nonpass_vehicles))
        for partition, pass_vehicles, nonpass_vehicles in raw
    }


def _overview_rows(connection: duckdb.DuckDBPyConnection) -> List[Dict[str, object]]:
    raw = connection.execute(
        """
        SELECT year(episode_start)::INTEGER AS year,
               quarter(episode_start)::INTEGER AS quarter,
               public_county,
               count(*)::BIGINT AS n,
               sum(target_nonpass)::BIGINT AS nonpass,
               count(DISTINCT vehicle_token)::BIGINT AS vehicles,
               count(DISTINCT vehicle_token) FILTER (
                   WHERE target_nonpass = 0
               )::BIGINT AS pass_vehicles,
               count(DISTINCT vehicle_token) FILTER (
                   WHERE target_nonpass = 1
               )::BIGINT AS nonpass_vehicles
        FROM safe_mart
        WHERE eligible_returning_target
          AND target_nonpass IN (0, 1)
          AND public_county IS NOT NULL
        GROUP BY 1, 2, 3
        ORDER BY 1, 2, 3
        """
    ).fetchall()
    rows = []
    for (
        year,
        quarter,
        county,
        n,
        nonpass,
        vehicles,
        pass_vehicles,
        nonpass_vehicles,
    ) in raw:
        if not _publishable(
            n, nonpass, vehicles, pass_vehicles, nonpass_vehicles
        ):
            continue
        rows.append(
            {
                "year": int(year),
                "quarter": int(quarter),
                "public_county": _safe_category(county, "County"),
                "support_rounded": _round_support(n),
                "nonpass_rate": _rate(nonpass, n),
            }
        )
    return rows


AGE_BAND_SQL = """CASE
    WHEN vehicle_age BETWEEN 0 AND 3 THEN '0-3'
    WHEN vehicle_age BETWEEN 4 AND 7 THEN '4-7'
    WHEN vehicle_age BETWEEN 8 AND 11 THEN '8-11'
    WHEN vehicle_age BETWEEN 12 AND 15 THEN '12-15'
    WHEN vehicle_age BETWEEN 16 AND 20 THEN '16-20'
    WHEN vehicle_age >= 21 THEN '21+'
    ELSE NULL
END"""
AGE_BAND_ORDER = {name: index for index, name in enumerate(
    ("0-3", "4-7", "8-11", "12-15", "16-20", "21+")
)}


def _age_risk_rows(connection: duckdb.DuckDBPyConnection) -> List[Dict[str, object]]:
    raw = connection.execute(
        f"""
        SELECT {AGE_BAND_SQL} AS age_band,
               count(*)::BIGINT AS n,
               sum(target_nonpass)::BIGINT AS nonpass,
               count(DISTINCT vehicle_token)::BIGINT AS vehicles,
               count(DISTINCT vehicle_token) FILTER (
                   WHERE target_nonpass = 0
               )::BIGINT AS pass_vehicles,
               count(DISTINCT vehicle_token) FILTER (
                   WHERE target_nonpass = 1
               )::BIGINT AS nonpass_vehicles
        FROM safe_mart
        WHERE eligible_returning_target AND target_nonpass IN (0, 1)
        GROUP BY 1
        ORDER BY 1
        """
    ).fetchall()
    rows = []
    for (
        age_band,
        n,
        nonpass,
        vehicles,
        pass_vehicles,
        nonpass_vehicles,
    ) in raw:
        if age_band is None or not _publishable(
            n, nonpass, vehicles, pass_vehicles, nonpass_vehicles
        ):
            continue
        if age_band not in AGE_BAND_ORDER:
            raise PublicSchemaError("Unexpected age band")
        rows.append(
            {
                "age_band": age_band,
                "support_rounded": _round_support(n),
                "nonpass_rate": _rate(nonpass, n),
            }
        )
    rows.sort(key=lambda row: AGE_BAND_ORDER[str(row["age_band"])])
    return rows


def _scorecard_rows(connection: duckdb.DuckDBPyConnection) -> List[Dict[str, object]]:
    raw = connection.execute(
        """
        SELECT last_observed_make, last_observed_model,
               count(*)::BIGINT AS n,
               sum(target_nonpass)::BIGINT AS nonpass,
               count(DISTINCT vehicle_token)::BIGINT AS vehicles,
               count(DISTINCT vehicle_token) FILTER (
                   WHERE target_nonpass = 0
               )::BIGINT AS pass_vehicles,
               count(DISTINCT vehicle_token) FILTER (
                   WHERE target_nonpass = 1
               )::BIGINT AS nonpass_vehicles
        FROM safe_mart
        WHERE eligible_returning_target
          AND target_nonpass IN (0, 1)
          AND last_observed_make IS NOT NULL
          AND last_observed_model IS NOT NULL
        GROUP BY 1, 2
        HAVING count(*) >= 100
           AND sum(target_nonpass) >= 10
           AND count(*) - sum(target_nonpass) >= 10
           AND count(DISTINCT vehicle_token) >= 100
           AND count(DISTINCT vehicle_token) FILTER (
               WHERE target_nonpass = 0
           ) >= 10
           AND count(DISTINCT vehicle_token) FILTER (
               WHERE target_nonpass = 1
           ) >= 10
        ORDER BY n DESC, last_observed_make, last_observed_model
        LIMIT 200
        """
    ).fetchall()
    rows = []
    for (
        make,
        model,
        n,
        nonpass,
        vehicles,
        pass_vehicles,
        nonpass_vehicles,
    ) in raw:
        if not _publishable(
            n, nonpass, vehicles, pass_vehicles, nonpass_vehicles
        ):
            continue
        try:
            safe_make = _safe_category(make, "Make")
            safe_model = _safe_category(model, "Model")
        except PublicSchemaError:
            continue
        rows.append(
            {
                "prior_make": safe_make,
                "prior_model": safe_model,
                "support_rounded": _round_support(n),
                "nonpass_rate": _rate(nonpass, n),
            }
        )
    rows.sort(key=lambda row: (str(row["prior_make"]), str(row["prior_model"])))
    return rows


def _coverage_rows(connection: duckdb.DuckDBPyConnection) -> List[Dict[str, object]]:
    raw = connection.execute(
        """
        SELECT year(episode_start)::INTEGER AS year,
               source_era,
               count(*)::BIGINT AS n,
               count(*) FILTER (
                   WHERE first_outcome IN ('pass','fail','reject','abort')
               )::BIGINT AS labeled,
               count(*) FILTER (WHERE first_outcome = 'pass')::BIGINT AS pass,
               count(*) FILTER (
                   WHERE first_outcome IN ('fail','reject','abort')
               )::BIGINT AS nonpass,
               count(*) FILTER (
                   WHERE target_outcome_label_source = 'overall_result'
               )::BIGINT AS overall_labels,
               count(*) FILTER (
                   WHERE target_outcome_label_source = 'utah_obd_proxy'
               )::BIGINT AS proxy_labels,
               count(DISTINCT vehicle_token)::BIGINT AS vehicles,
               count(DISTINCT vehicle_token) FILTER (
                   WHERE first_outcome = 'pass'
               )::BIGINT AS pass_vehicles,
               count(DISTINCT vehicle_token) FILTER (
                   WHERE first_outcome IN ('fail','reject','abort')
               )::BIGINT AS nonpass_vehicles
        FROM safe_mart
        WHERE episode_number > 1 AND source_era IS NOT NULL
        GROUP BY 1, 2
        ORDER BY 1, 2
        """
    ).fetchall()
    rows = []
    for (
        year,
        source,
        n,
        labeled,
        passes,
        nonpass,
        overall,
        proxy,
        vehicles,
        pass_vehicles,
        nonpass_vehicles,
    ) in raw:
        if not _publishable(
            n, nonpass, vehicles, pass_vehicles, nonpass_vehicles
        ) or passes < MIN_CLASS_COUNT:
            continue
        if labeled != passes + nonpass or labeled != overall + proxy:
            raise PublicSchemaError("Label-source quality totals are inconsistent")
        rows.append(
            {
                "year": int(year),
                "source_era": _safe_category(source, "Source era"),
                "support_rounded": _round_support(n),
                "labeled_rate": _rate(labeled, n),
                "overall_result_share": _rate(overall, labeled),
                "utah_obd_proxy_share": _rate(proxy, labeled),
            }
        )
    return rows


def _base_asset(
    development_preview: bool,
    population_estimate_allowed: bool,
) -> Dict[str, object]:
    return {
        "schema_version": SCHEMA_VERSION,
        "development_preview": development_preview,
        "population_estimate_allowed": population_estimate_allowed,
    }


def _release_id(
    mart_digest: str,
    model_provenance: Sequence[Tuple[str, str]],
) -> str:
    material = {
        "schema_version": SCHEMA_VERSION,
        "mart_sha256": mart_digest,
        "models": [
            {"model_version": version, "metrics_sha256": metrics_digest}
            for version, metrics_digest in sorted(model_provenance)
        ],
    }
    encoded = json.dumps(
        material,
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=True,
        allow_nan=False,
    ).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def build_assets(
    mart: Path,
    mart_manifest_path: Path,
    bundles: Sequence[ModelBundle],
    development_preview: bool,
) -> Dict[str, object]:
    mart_manifest, development_source, mart_digest = validate_mart_manifest(
        mart, mart_manifest_path, development_preview
    )
    population_allowed = bool(
        mart_manifest.get("population_estimate_allowed")
    ) and not development_preview and not development_source

    connection = duckdb.connect(":memory:")
    try:
        _create_safe_mart(connection, mart)
        diagnostics, model_provenance = load_model_diagnostics(
            bundles,
            mart_digest,
            _partition_vehicle_support(connection),
        )
        overview = _overview_rows(connection)
        age_risk = _age_risk_rows(connection)
        scorecards = _scorecard_rows(connection)
        coverage = _coverage_rows(connection)
    finally:
        connection.close()
    if not overview:
        raise PublicSchemaError("No overview cells survive publication suppression")

    base = _base_asset(development_preview, population_allowed)
    assets: Dict[str, object] = {
        "overview_period_county.json": {**base, "rows": overview},
        "age_risk_curve.json": {**base, "rows": age_risk},
        "cohort_scorecard.json": {**base, "rows": scorecards},
        "model_diagnostics.json": {**base, "rows": diagnostics},
        "coverage_quality.json": {**base, "rows": coverage},
    }

    counties = sorted({str(row["public_county"]) for row in overview})
    periods_by_year: Dict[int, set] = {}
    for row in overview:
        periods_by_year.setdefault(int(row["year"]), set()).add(int(row["quarter"]))
    periods = [
        {"year": year, "quarters": sorted(quarters)}
        for year, quarters in sorted(periods_by_year.items())
    ]
    make_models = [
        {"prior_make": row["prior_make"], "prior_model": row["prior_model"]}
        for row in scorecards
    ]
    model_names = sorted({str(row["model"]) for row in diagnostics})
    assets["filter_catalog.json"] = {
        **base,
        "public_counties": counties,
        "age_bands": [str(row["age_band"]) for row in age_risk],
        "prior_make_models": make_models,
        "periods": periods,
        "models": model_names,
        "partitions": list(ALLOWED_PARTITIONS),
    }

    source_years = [int(row["year"]) for row in overview]
    model_versions = sorted({version for version, _ in model_provenance})
    assets["data_manifest.json"] = {
        **base,
        "release_id": _release_id(mart_digest, model_provenance),
        "model_versions": model_versions,
        "data_scope": {
            "first_year": min(source_years),
            "last_year": max(source_years),
            "partitions": list(ALLOWED_PARTITIONS),
            "model_names": model_names,
        },
        "definitions": {
            "target": "first-attempt next-episode binary non-pass rate",
            "episode_gap_days": int(mart_manifest.get("episode_gap_days", 30)),
            "support_rounding": SUPPORT_ROUNDING,
            "suppression_min_support": MIN_SUPPORT,
            "suppression_min_pass": MIN_CLASS_COUNT,
            "suppression_min_nonpass": MIN_CLASS_COUNT,
            "suppression_min_distinct_vehicles": MIN_SUPPORT,
            "suppression_min_distinct_pass_vehicles": MIN_CLASS_COUNT,
            "suppression_min_distinct_nonpass_vehicles": MIN_CLASS_COUNT,
            "locked_test_metrics_published": False,
        },
        "assets": sorted(
            name for name in PUBLIC_ASSET_NAMES if name != "data_manifest.json"
        ),
    }

    for name, asset in assets.items():
        validate_public_asset(name, asset)
    return assets


def _validate_flags(asset: Mapping[str, object]) -> None:
    if asset.get("schema_version") != SCHEMA_VERSION:
        raise PublicSchemaError("Asset schema version is invalid")
    if not isinstance(asset.get("development_preview"), bool):
        raise PublicSchemaError("Preview flag must be boolean")
    if not isinstance(asset.get("population_estimate_allowed"), bool):
        raise PublicSchemaError("Population flag must be boolean")
    if asset["development_preview"] and asset["population_estimate_allowed"]:
        raise PublicSchemaError("Preview assets cannot permit population estimates")


def _scan_public_value(value: object, path: str = "root") -> None:
    if isinstance(value, dict):
        for key, child in value.items():
            normalized = key.lower()
            if normalized in DENIED_EXACT_KEYS or any(
                fragment in normalized for fragment in DENIED_KEY_FRAGMENTS
            ):
                raise PublicSchemaError("Denied public key at " + path)
            _scan_public_value(child, path + "." + key)
    elif isinstance(value, list):
        for index, child in enumerate(value):
            _scan_public_value(child, path + "[{}]".format(index))
    elif isinstance(value, str):
        if EXACT_TIMESTAMP_PATTERN.match(value):
            raise PublicSchemaError("Exact timestamp in public asset")
    elif isinstance(value, float):
        if not math.isfinite(value):
            raise PublicSchemaError("Non-finite public number")
    elif value is not None and not isinstance(value, (bool, int)):
        raise PublicSchemaError("Unsupported public JSON value")


def _validate_rate(value: object, label: str) -> None:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise PublicSchemaError(label + " must be numeric")
    number = float(value)
    if not math.isfinite(number) or not 0.0 <= number <= 1.0:
        raise PublicSchemaError(label + " is outside [0,1]")


def validate_public_asset(name: str, asset: object) -> None:
    if not isinstance(asset, dict):
        raise PublicSchemaError("Public asset must be an object")
    _validate_flags(asset)
    _scan_public_value(asset)

    if name in ROW_ASSET_SCHEMAS:
        if set(asset) != {
            "schema_version",
            "development_preview",
            "population_estimate_allowed",
            "rows",
        }:
            raise PublicSchemaError("Row asset has unexpected top-level keys")
        rows = asset.get("rows")
        if not isinstance(rows, list):
            raise PublicSchemaError("Row asset rows must be an array")
        required = set(ROW_ASSET_SCHEMAS[name])
        for row in rows:
            if not isinstance(row, dict) or set(row) != required:
                raise PublicSchemaError("Row does not match its public schema")
            if "support_rounded" in row:
                support = row["support_rounded"]
                if (
                    isinstance(support, bool)
                    or not isinstance(support, int)
                    or support < MIN_SUPPORT
                    or support % SUPPORT_ROUNDING
                ):
                    raise PublicSchemaError("Published support is not safely rounded")
            for key, value in row.items():
                if key.endswith("_rate") or key.endswith("_share") or key in {
                    "average_precision",
                    "brier",
                    "roc_auc",
                    "top_10_capture",
                }:
                    _validate_rate(value, key)
                if key == "log_loss":
                    if (
                        isinstance(value, bool)
                        or not isinstance(value, (int, float))
                        or not math.isfinite(float(value))
                        or float(value) < 0.0
                    ):
                        raise PublicSchemaError("Invalid public log loss")
        return

    if name == "filter_catalog.json":
        expected = {
            "schema_version",
            "development_preview",
            "population_estimate_allowed",
            "public_counties",
            "age_bands",
            "prior_make_models",
            "periods",
            "models",
            "partitions",
        }
        if set(asset) != expected:
            raise PublicSchemaError("Filter catalog schema is invalid")
        if asset["partitions"] != list(ALLOWED_PARTITIONS):
            raise PublicSchemaError("Filter catalog exposes an invalid partition")
        return

    if name == "data_manifest.json":
        expected = {
            "schema_version",
            "development_preview",
            "population_estimate_allowed",
            "release_id",
            "model_versions",
            "data_scope",
            "definitions",
            "assets",
        }
        if set(asset) != expected:
            raise PublicSchemaError("Data manifest schema is invalid")
        definitions = asset.get("definitions")
        if not isinstance(definitions, dict) or definitions.get(
            "locked_test_metrics_published"
        ) is not False:
            raise PublicSchemaError("Data manifest does not lock 2025 metrics")
        release_id = asset.get("release_id")
        if not isinstance(release_id, str) or not re.fullmatch(
            r"[0-9a-f]{64}", release_id
        ):
            raise PublicSchemaError("Data manifest release ID is invalid")
        model_versions = asset.get("model_versions")
        if (
            not isinstance(model_versions, list)
            or not model_versions
            or model_versions != sorted(set(model_versions))
            or any(
                not isinstance(version, str)
                or not SAFE_MODEL_NAME_PATTERN.fullmatch(version)
                for version in model_versions
            )
        ):
            raise PublicSchemaError("Data manifest model versions are invalid")
        return
    raise PublicSchemaError("Unexpected public asset name")


def _json_bytes(value: object) -> bytes:
    return (
        json.dumps(
            value,
            sort_keys=True,
            separators=(",", ":"),
            ensure_ascii=True,
            allow_nan=False,
        )
        + "\n"
    ).encode("utf-8")


def _validate_output_dir(output_dir: Path) -> Path:
    resolved = output_dir.resolve()
    if not resolved.is_relative_to(PUBLIC_DATA_ROOT):
        raise PublicSchemaError(
            "Dashboard assets must remain under dashboard/public/data"
        )
    return resolved


def _validate_existing_output_contract(
    output_dir: Path,
    expected_names: frozenset,
    require_complete: bool = False,
) -> Dict[str, Path]:
    if not output_dir.exists():
        return {}
    if not output_dir.is_dir():
        raise PublicSchemaError("Dashboard data output is not a directory")

    entries = {entry.name: entry for entry in output_dir.iterdir()}
    unexpected = sorted(set(entries) - expected_names)
    invalid_types = sorted(
        name
        for name, entry in entries.items()
        if entry.is_symlink() or not entry.is_file()
    )
    if unexpected or invalid_types:
        raise PublicSchemaError(
            "Dashboard data directory contains unexpected content"
        )
    if require_complete and set(entries) != expected_names:
        raise PublicSchemaError(
            "Dashboard data directory does not match the exact public contract"
        )
    return entries


def publish_assets(
    assets: Mapping[str, object],
    output_dir: Path,
    overwrite: bool,
) -> Dict[str, Path]:
    output_dir = _validate_output_dir(output_dir)
    if set(assets) != set(PUBLIC_ASSET_NAMES):
        raise PublicSchemaError("Asset set does not match the public contract")

    serialized = {name: _json_bytes(assets[name]) for name in sorted(assets)}
    first_asset = next(iter(assets.values()))
    if not isinstance(first_asset, dict):
        raise PublicSchemaError("Invalid public asset")
    sha_manifest = {
        "schema_version": SCHEMA_VERSION,
        "development_preview": first_asset["development_preview"],
        "population_estimate_allowed": first_asset["population_estimate_allowed"],
        "files": [
            {
                "name": name,
                "sha256": hashlib.sha256(serialized[name]).hexdigest(),
            }
            for name in sorted(serialized)
        ],
    }
    _validate_flags(sha_manifest)
    _scan_public_value(sha_manifest)
    serialized[SHA256_MANIFEST_NAME] = _json_bytes(sha_manifest)

    targets = {name: output_dir / name for name in serialized}
    if set(targets) != PUBLIC_OUTPUT_NAMES:
        raise PublicSchemaError("Output file set does not match the public contract")
    existing = _validate_existing_output_contract(
        output_dir, PUBLIC_OUTPUT_NAMES
    )
    if not overwrite and existing:
        raise FileExistsError("Refusing to overwrite existing dashboard assets")

    os.umask(0o022)
    output_dir.parent.mkdir(parents=True, exist_ok=True)
    staging_root = Path(
        tempfile.mkdtemp(prefix=".dashboard-data-", dir=str(output_dir.parent))
    )
    try:
        for name, content in serialized.items():
            staged = staging_root / name
            with staged.open("xb") as handle:
                handle.write(content)
                handle.flush()
                os.fsync(handle.fileno())
            staged.chmod(0o644)

        output_dir.mkdir(parents=True, exist_ok=True)
        existing = _validate_existing_output_contract(
            output_dir, PUBLIC_OUTPUT_NAMES
        )
        if not overwrite and existing:
            raise FileExistsError("Dashboard assets appeared during publication")
        # Publish the checksum manifest last; consumers can treat it as the
        # commit marker for the individually atomic file replacements.
        for name in sorted(serialized):
            if name == SHA256_MANIFEST_NAME:
                continue
            os.replace(staging_root / name, targets[name])
        os.replace(
            staging_root / SHA256_MANIFEST_NAME,
            targets[SHA256_MANIFEST_NAME],
        )
        _validate_existing_output_contract(
            output_dir, PUBLIC_OUTPUT_NAMES, require_complete=True
        )
    finally:
        shutil.rmtree(staging_root, ignore_errors=True)
    return targets


def export_dashboard_data(
    mart: Path,
    mart_manifest: Path,
    bundles: Sequence[ModelBundle],
    output_dir: Path,
    development_preview: bool = False,
    overwrite: bool = False,
) -> Dict[str, Path]:
    assets = build_assets(
        mart=mart,
        mart_manifest_path=mart_manifest,
        bundles=bundles,
        development_preview=development_preview,
    )
    return publish_assets(assets, output_dir, overwrite)


def main(argv: Optional[Sequence[str]] = None) -> int:
    args = parse_args(argv)
    if len(args.model_manifest) != len(args.model_metrics):
        print(
            "Configuration error: pair each model manifest with one metrics file",
            file=sys.stderr,
        )
        return 2
    bundles = [
        ModelBundle(manifest=manifest, metrics=metrics)
        for manifest, metrics in zip(args.model_manifest, args.model_metrics)
    ]
    try:
        paths = export_dashboard_data(
            mart=args.mart,
            mart_manifest=_mart_manifest_path(args.mart, args.mart_manifest),
            bundles=bundles,
            output_dir=args.output_dir,
            development_preview=args.development_preview,
            overwrite=args.overwrite,
        )
    except (DashboardExportError, FileExistsError) as exc:
        print("Dashboard export refused: " + str(exc), file=sys.stderr)
        return 2
    print("Published {} sanitized dashboard assets".format(len(paths)))
    print("Public data directory: " + str(args.output_dir.resolve()))
    return 0


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