algeon-datafusion 0.1.2

//! Registry of cuGraph parity specs shared by tests and oracle-refresh tooling.

use std::{
    fs,
    path::{Path, PathBuf},
};

use serde_json::Value;

use super::CugraphAlgorithm;

/// Repository-relative directory holding the locally prepared cuGraph datasets and manifest.
pub const CUGRAPH_DATA_DIR_FROM_CARGO_MANIFEST: &str = "fixture/graph/cugraph";
/// Manifest file describing locally prepared cuGraph parity datasets.
pub const CUGRAPH_DATA_MANIFEST_PATH: &str = "fixture/graph/cugraph/manifest.json";

/// Comparator family expected for a parity row.
#[doc(hidden)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum ParityComparator {
    /// Exact equality on integer columns after sort-by-vertex.
    ExactInt,
    /// Per-vertex absolute and relative tolerance on floating-point
    /// columns after sort-by-vertex.
    Numeric,
    /// Compare partitions as set-of-sets, ignoring label renaming.
    PartitionEquivalence,
    /// Validate a randomized partition by vertex coverage, a bounded
    /// community count, and modularity relative to the oracle baseline.
    ///
    /// Unlike [`Self::PartitionEquivalence`], this intentionally does not
    /// require a byte-identical partition across cuGraph builds or across
    /// same-seed replays on the same build.
    PartitionQuality,
    /// Seeded random output; sequence + grouping must match plus the
    /// distribution baseline.
    SeededSequence,
    /// Run-order-sensitive layouts compared by structure. Point coordinates are
    /// not checked. The RMS radius stays within `tolerance.rel` of the oracle's,
    /// and the Spearman correlation between layout and graph distances is at
    /// least `1 - tolerance.abs`.
    RankCorrelation,
    /// Edge sets canonicalized as unordered `(min, max)` pairs.
    EdgeSubgraph,
    /// Strict equality on the top-K vertex set when the result is sorted
    /// by descending score. Used with cuGraph-curated reference CSVs
    /// copied under `fixture/graph/cugraph/ref/katz/*.csv` by the local
    /// fixture preparer. Their canonical top-K vertex ordering matches
    /// the upstream cuGraph Python test
    /// (`topKVertices(katz, "katz_centrality", k)`).
    TopKVertices,
    /// Compare `(first, second, similarity)` rows as a multiset.
    ///
    /// Similarity is defined over an explicit input relation: duplicate,
    /// reversed, and self pairs are all semantically observable, so the
    /// normal vertex-order comparators are not valid here.
    SimilarityMultiset,
}

impl ParityComparator {
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::ExactInt => "exact_int",
            Self::Numeric => "numeric",
            Self::PartitionEquivalence => "partition_equivalence",
            Self::PartitionQuality => "partition_quality",
            Self::SeededSequence => "seeded_sequence",
            Self::RankCorrelation => "rank_correlation",
            Self::EdgeSubgraph => "edge_subgraph",
            Self::TopKVertices => "top_k_vertices",
            Self::SimilarityMultiset => "similarity_multiset",
        }
    }
}

/// Source of ground truth for a parity row.
#[doc(hidden)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum ParityTruthSource {
    /// Local oracle JSON generated by `fixture/graph/cugraph_oracle.py`.
    OracleJson,
    /// cuGraph reference CSV copied under `fixture/graph/cugraph/ref/`.
    /// The path is expressed relative to the data root.
    RefCsv { path: &'static str },
    /// A small, hand-derived result document for a contract that is simpler
    /// to audit directly than to delegate to the Python oracle.
    ///
    /// The document contains a `result` object with the same column-array
    /// shape as oracle JSON. It deliberately has no generated-oracle path.
    HandVerified { document_json: &'static str },
}

impl ParityTruthSource {
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::OracleJson => "oracle_json",
            Self::RefCsv { .. } => "ref_csv",
            Self::HandVerified { .. } => "hand_verified",
        }
    }
}

/// Per-row tolerance for floating-point comparators.
///
/// Integer comparators ignore both fields. `PartitionQuality` uses them as
/// absolute and relative modularity tolerances.
#[doc(hidden)]
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct ParityTolerance {
    pub abs: f64,
    pub rel: f64,
}

impl ParityTolerance {
    pub const fn exact() -> Self {
        Self { abs: 0.0, rel: 0.0 }
    }
}

/// One row in the cuGraph parity spec registry.
///
/// Adding an algorithm/dataset combination is one new const entry in
/// [`CUGRAPH_PARITY_SPECS`]. No new file is required. `params_json` is the
/// canonical JSON form of the algorithm parameters; it must equal the
/// `header.params` of the generated oracle JSON.
#[doc(hidden)]
#[derive(Debug, Clone, Copy)]
pub struct ParitySpec {
    /// Stable label used in test output and in plan-display strings.
    pub label: &'static str,
    pub algorithm: CugraphAlgorithm,
    /// Dataset id from local `fixture/graph/cugraph/manifest.json`.
    pub dataset: &'static str,
    /// Canonical JSON parameter object. Empty `{}` is allowed.
    pub params_json: &'static str,
    /// RNG seed; recorded in the oracle JSON header and pinned by the
    /// `SeededSequence` comparator family.
    pub seed: Option<u64>,
    pub truth_source: ParityTruthSource,
    /// Oracle JSON path relative to `fixture/graph/cugraph/oracle/` for
    /// generated-oracle and reference-CSV rows. Hand-verified rows have no
    /// generated oracle, so this is absent.
    pub oracle_relpath: Option<&'static str>,
    pub comparator: ParityComparator,
    pub tolerance: ParityTolerance,
}

/// Source-of-truth list of cuGraph parity rows.
///
/// Every SQL-visible algorithm has one GPU parity acceptance row. New
/// algorithms must add a registry row here alongside their
/// [`crate::cugraph_sql::CUGRAPH_OPERATION_REGISTRY`] entry.
#[doc(hidden)]
pub const CUGRAPH_PARITY_SPECS: &[ParitySpec] = &[
    ParitySpec {
        label: "pagerank/karate@alpha=0.85",
        algorithm: CugraphAlgorithm::PageRank,
        dataset: "karate",
        params_json: r#"{"alpha":0.85,"epsilon":1e-05,"max_iterations":100}"#,
        seed: None,
        truth_source: ParityTruthSource::OracleJson,
        oracle_relpath: Some("karate/pagerank__alpha_0_85.json"),
        comparator: ParityComparator::Numeric,
        tolerance: ParityTolerance { abs: 1.0e-5, rel: 1.0e-3 },
    },
    ParitySpec {
        label: "personalized_pagerank/karate@vertex=0",
        algorithm: CugraphAlgorithm::PersonalizedPageRank,
        dataset: "karate",
        params_json: r#"{"alpha":0.85,"epsilon":1e-05,"max_iterations":100,"personalization_table":"ppr_seeds","personalization_vertex_col":"vertex","personalization_value_col":"value","personalization_vertices":[0],"personalization_values":[1.0]}"#,
        seed: None,
        truth_source: ParityTruthSource::OracleJson,
        oracle_relpath: Some("karate/personalized_pagerank__vertex_0.json"),
        comparator: ParityComparator::Numeric,
        tolerance: ParityTolerance { abs: 1.0e-5, rel: 1.0e-3 },
    },
    ParitySpec {
        label: "bfs/karate@src=0",
        algorithm: CugraphAlgorithm::Bfs,
        dataset: "karate",
        params_json: r#"{"source_vertex":0,"depth_limit":null}"#,
        seed: None,
        truth_source: ParityTruthSource::OracleJson,
        oracle_relpath: Some("karate/bfs__source_0.json"),
        comparator: ParityComparator::ExactInt,
        tolerance: ParityTolerance::exact(),
    },
    ParitySpec {
        label: "louvain/karate@default",
        algorithm: CugraphAlgorithm::Louvain,
        dataset: "karate",
        params_json: r#"{"max_level":100,"threshold":1e-07,"resolution":1.0}"#,
        seed: None,
        truth_source: ParityTruthSource::OracleJson,
        oracle_relpath: Some("karate/louvain__resolution_1_0.json"),
        comparator: ParityComparator::PartitionEquivalence,
        tolerance: ParityTolerance::exact(),
    },
    ParitySpec {
        label: "eigenvector_centrality/karate@default",
        algorithm: CugraphAlgorithm::EigenvectorCentrality,
        dataset: "karate",
        // Mirrors `CugraphEigenvectorCentralityParams::default()`. The
        // Python oracle accepts the same keys, so the header round-trip
        // assertion in the parity harness compares matching key sets.
        params_json: r#"{"epsilon":1e-06,"max_iterations":200}"#,
        seed: None,
        truth_source: ParityTruthSource::OracleJson,
        oracle_relpath: Some("karate/eigenvector_centrality__default.json"),
        comparator: ParityComparator::Numeric,
        // Eigenvector centrality is sensitive to convergence tolerance; this
        // row uses the PageRank tolerance (abs=1e-5, rel=1e-3).
        tolerance: ParityTolerance { abs: 1.0e-5, rel: 1.0e-3 },
    },
    ParitySpec {
        label: "hits/karate@default",
        algorithm: CugraphAlgorithm::Hits,
        dataset: "karate",
        // Mirrors `CugraphHitsParams::default()`; epsilon and
        // max_iterations track the local `cugraph` crate defaults; `normalize`
        // stays off so the score range matches the upstream cuGraph
        // sample tests.
        params_json: r#"{"epsilon":1e-05,"max_iterations":100,"normalize":false}"#,
        seed: None,
        truth_source: ParityTruthSource::OracleJson,
        oracle_relpath: Some("karate/hits__default.json"),
        comparator: ParityComparator::Numeric,
        // HITS uses two float score columns; the Numeric comparator compares
        // them column by column under the same abs/rel tolerance as
        // PageRank and eigenvector centrality.
        tolerance: ParityTolerance { abs: 1.0e-5, rel: 1.0e-3 },
    },
    ParitySpec {
        label: "betweenness_centrality/karate@seeds=0,1,2",
        algorithm: CugraphAlgorithm::BetweennessCentrality,
        dataset: "karate",
        params_json: r#"{"seeds":[0,1,2],"normalized":true,"include_endpoints":false,"exact_vertex_threshold":100000}"#,
        seed: None,
        truth_source: ParityTruthSource::OracleJson,
        oracle_relpath: Some("karate/betweenness_centrality__seeds_0_1_2.json"),
        comparator: ParityComparator::Numeric,
        tolerance: ParityTolerance { abs: 1.0e-5, rel: 1.0e-3 },
    },
    ParitySpec {
        label: "edge_betweenness_centrality/karate@seeds=0,1,2",
        algorithm: CugraphAlgorithm::EdgeBetweennessCentrality,
        dataset: "karate",
        params_json: r#"{"seeds":[0,1,2],"normalized":true,"exact_vertex_threshold":100000}"#,
        seed: None,
        truth_source: ParityTruthSource::OracleJson,
        oracle_relpath: Some("karate/edge_betweenness_centrality__seeds_0_1_2.json"),
        comparator: ParityComparator::Numeric,
        tolerance: ParityTolerance { abs: 1.0e-5, rel: 1.0e-3 },
    },
    ParitySpec {
        label: "katz_centrality/karate@ref",
        algorithm: CugraphAlgorithm::KatzCentrality,
        dataset: "karate",
        // cuGraph's upstream Katz reference (`ref/katz/karate.csv`) was
        // generated with the algorithm's documented default options;
        // mirror those defaults so the oracle header round-trip and the
        // RefCsv top-K comparison are consistent.
        params_json: r#"{"alpha":0.01,"beta":1.0,"epsilon":1e-06,"max_iterations":1000}"#,
        seed: None,
        truth_source: ParityTruthSource::RefCsv { path: "ref/katz/karate.csv" },
        // Even with a RefCsv primary truth, keep an oracle JSON slot
        // so callers can cross-check the full per-vertex score table
        // against Python cuGraph once it is available; the parity
        // harness reports the file as "needs refresh" until then.
        oracle_relpath: Some("karate/katz_centrality__default.json"),
        comparator: ParityComparator::TopKVertices,
        // Comparator family is integer-set equality on the top-K
        // vertex list; tolerance fields are unused for `TopKVertices`.
        tolerance: ParityTolerance::exact(),
    },
    ParitySpec {
        label: "sssp/karate@source=0",
        algorithm: CugraphAlgorithm::Sssp,
        dataset: "karate",
        params_json: r#"{"source_vertex":0}"#,
        seed: None,
        truth_source: ParityTruthSource::OracleJson,
        oracle_relpath: Some("karate/sssp__source_0.json"),
        // SSSP distance is Float64; predecessor is Int64. Numeric compares
        // float columns with tolerance and integer columns exactly.
        comparator: ParityComparator::Numeric,
        tolerance: ParityTolerance { abs: 1.0e-9, rel: 1.0e-9 },
    },
    ParitySpec {
        label: "degrees_all/karate@default",
        algorithm: CugraphAlgorithm::DegreesAll,
        dataset: "karate",
        params_json: r#"{}"#,
        seed: None,
        truth_source: ParityTruthSource::OracleJson,
        oracle_relpath: Some("karate/degrees_all__default.json"),
        comparator: ParityComparator::ExactInt,
        tolerance: ParityTolerance::exact(),
    },
    ParitySpec {
        label: "in_degrees_all/karate@default",
        algorithm: CugraphAlgorithm::InDegreesAll,
        dataset: "karate",
        params_json: r#"{}"#,
        seed: None,
        truth_source: ParityTruthSource::OracleJson,
        oracle_relpath: Some("karate/in_degrees_all__default.json"),
        comparator: ParityComparator::ExactInt,
        tolerance: ParityTolerance::exact(),
    },
    ParitySpec {
        label: "out_degrees_all/karate@default",
        algorithm: CugraphAlgorithm::OutDegreesAll,
        dataset: "karate",
        params_json: r#"{}"#,
        seed: None,
        truth_source: ParityTruthSource::OracleJson,
        oracle_relpath: Some("karate/out_degrees_all__default.json"),
        comparator: ParityComparator::ExactInt,
        tolerance: ParityTolerance::exact(),
    },
    ParitySpec {
        // WCC requires a symmetric (undirected) graph. The karate dataset
        // is undirected; the oracle runs with directed=false so reverse
        // edges are added.
        label: "weakly_connected_components/karate@default",
        algorithm: CugraphAlgorithm::WeaklyConnectedComponents,
        dataset: "karate",
        params_json: r#"{}"#,
        seed: None,
        truth_source: ParityTruthSource::OracleJson,
        oracle_relpath: Some("karate/weakly_connected_components__default.json"),
        // Labels are unordered partition identifiers; use
        // partition-equivalence comparison.
        comparator: ParityComparator::PartitionEquivalence,
        tolerance: ParityTolerance::exact(),
    },
    ParitySpec {
        label: "strongly_connected_components/karate@default",
        algorithm: CugraphAlgorithm::StronglyConnectedComponents,
        dataset: "karate",
        params_json: r#"{}"#,
        seed: None,
        truth_source: ParityTruthSource::OracleJson,
        oracle_relpath: Some("karate/strongly_connected_components__default.json"),
        comparator: ParityComparator::PartitionEquivalence,
        tolerance: ParityTolerance::exact(),
    },
    ParitySpec {
        label: "triangle_count_all/karate@default",
        algorithm: CugraphAlgorithm::TriangleCountAll,
        dataset: "karate",
        params_json: r#"{}"#,
        seed: None,
        truth_source: ParityTruthSource::OracleJson,
        oracle_relpath: Some("karate/triangle_count_all__default.json"),
        comparator: ParityComparator::ExactInt,
        tolerance: ParityTolerance::exact(),
    },
    ParitySpec {
        label: "core_number/karate@default",
        algorithm: CugraphAlgorithm::CoreNumber,
        dataset: "karate",
        // Default degree_type is "in_out".
        params_json: r#"{"degree_type":"in_out"}"#,
        seed: None,
        truth_source: ParityTruthSource::OracleJson,
        oracle_relpath: Some("karate/core_number__default.json"),
        comparator: ParityComparator::ExactInt,
        tolerance: ParityTolerance::exact(),
    },
    ParitySpec {
        label: "k_core/karate@k=2",
        algorithm: CugraphAlgorithm::KCore,
        dataset: "karate",
        // Default k=2, degree_type="in_out".
        params_json: r#"{"k":2,"degree_type":"in_out"}"#,
        seed: None,
        truth_source: ParityTruthSource::OracleJson,
        oracle_relpath: Some("karate/k_core__k_2.json"),
        comparator: ParityComparator::EdgeSubgraph,
        tolerance: ParityTolerance::exact(),
    },
    ParitySpec {
        // MST is a weighted algorithm, and the unweighted karate graph does not
        // serve it. Use cuGraph's upstream weighted toy_graph fixture so
        // `refresh_cugraph_oracle --all` can produce a meaningful row.
        label: "minimum_spanning_tree/toy_graph@default",
        algorithm: CugraphAlgorithm::MinimumSpanningTree,
        dataset: "toy_graph",
        params_json: r#"{}"#,
        seed: None,
        truth_source: ParityTruthSource::OracleJson,
        oracle_relpath: Some("toy_graph/minimum_spanning_tree__default.json"),
        comparator: ParityComparator::EdgeSubgraph,
        tolerance: ParityTolerance::exact(),
    },
    ParitySpec {
        label: "leiden/karate@default",
        algorithm: CugraphAlgorithm::Leiden,
        dataset: "karate",
        // seed=0 matches GraphLeidenParams::default() seed.
        params_json: r#"{"max_level":100,"resolution":1.0,"seed":0}"#,
        seed: None,
        truth_source: ParityTruthSource::OracleJson,
        oracle_relpath: Some("karate/leiden__default.json"),
        // Leiden partition labels are arbitrary identifiers. Compare them by
        // partition equivalence; exact label equality does not apply.
        comparator: ParityComparator::PartitionEquivalence,
        tolerance: ParityTolerance::exact(),
    },
    ParitySpec {
        label: "ecg/karate@default",
        algorithm: CugraphAlgorithm::Ecg,
        dataset: "karate",
        // seed=0 matches GraphEcgParams::default() seed.
        params_json: r#"{"min_weight":0.001,"ensemble_size":10,"max_level":100,"threshold":1e-07,"resolution":1.0,"seed":0}"#,
        seed: None,
        truth_source: ParityTruthSource::OracleJson,
        oracle_relpath: Some("karate/ecg__default.json"),
        // ECG runs `ensemble_size` Louvain passes and a final pass, each from
        // random initial community labels. `seed` fixes only those labels, and
        // the ECG contract promises no identical communities across runs. The
        // oracle also comes from another build (pylibcugraph 26.06.00, per its
        // header). The check therefore compares vertex coverage, community
        // count and modularity against the oracle. It does not compare the
        // oracle's exact set-of-sets partition.
        comparator: ParityComparator::PartitionQuality,
        tolerance: ParityTolerance { abs: 5.0e-2, rel: 0.0 },
    },
    ParitySpec {
        label: "spectral_modularity_maximization/karate@default",
        algorithm: CugraphAlgorithm::SpectralModularityMaximization,
        dataset: "karate",
        // seed=0 matches GraphSpectralClusteringParams::default() seed.
        params_json: r#"{"n_clusters":2,"n_eigenvectors":2,"evs_tolerance":0.01,"seed":0}"#,
        seed: None,
        truth_source: ParityTruthSource::OracleJson,
        oracle_relpath: Some("karate/spectral_modularity_maximization__default.json"),
        // Spectral clustering runs k-means over a Lanczos embedding. A
        // fixed seed makes the random initialization repeatable, but GPU
        // parallel reductions accumulate in scheduling order, so
        // same-seed replays may choose an equally valid partition. The
        // Python result serves as a modularity baseline. Its exact
        // set-of-sets partition is not required.
        comparator: ParityComparator::PartitionQuality,
        tolerance: ParityTolerance { abs: 5.0e-2, rel: 0.0 },
    },
    ParitySpec {
        label: "force_atlas2/karate@default",
        algorithm: CugraphAlgorithm::ForceAtlas2,
        dataset: "karate",
        // Mirrors the unweighted default SQL call and GraphForceAtlas2Params::default();
        // pass every option explicitly because Python cuGraph defaults differ
        // for some layout settings while the `cugraph` crate follows libcugraph C API defaults.
        params_json: r#"{"max_iter":500,"outbound_attraction_distribution":false,"lin_log_mode":false,"prevent_overlapping":false,"overlap_scaling_ratio":2.0,"edge_weight_influence":1.0,"jitter_tolerance":1.0,"barnes_hut_optimize":false,"barnes_hut_theta":0.5,"scaling_ratio":2.0,"strong_gravity_mode":false,"gravity":1.0,"verbose":false,"seed":0}"#,
        seed: None,
        truth_source: ParityTruthSource::OracleJson,
        oracle_relpath: Some("karate/force_atlas2__default.json"),
        // cuGraph accumulates ForceAtlas2 forces with float atomics and has
        // no deterministic mode, so coordinates vary between runs under the
        // same seed (`GraphForceAtlas2Params::seed`). Compare the RMS radius
        // within 2% and a layout-to-graph-distance Spearman of at least 0.80.
        comparator: ParityComparator::RankCorrelation,
        tolerance: ParityTolerance { abs: 0.20, rel: 0.02 },
    },
    // Explicit-pair similarity is a relation-preserving contract. The
    // `toy_graph` fixture has reciprocal edges, and these inputs retain a
    // duplicate, reverse, self, and zero-similarity pair. Their small
    // unweighted Jaccard, Sorensen, and overlap results are hand-verifiable
    // directly from the adjacency sets: N(0)={1,2}, N(3)={1,2,5},
    // N(1)={0,2,3,4}, N(4)={1,5}. Unweighted cosine uses the set formula
    // |A∩B|/sqrt(|A||B|) over full neighborhoods, so `(0,3)` scores
    // 2/sqrt(2*3) and `(1,4)` scores zero.
    ParitySpec {
        label: "jaccard/toy_graph@explicit_pairs",
        algorithm: CugraphAlgorithm::Jaccard,
        dataset: "toy_graph",
        params_json: r#"{}"#,
        seed: None,
        truth_source: ParityTruthSource::HandVerified {
            document_json: r#"{"result":{"first":[0,0,3,0,1],"second":[3,3,0,0,4],"similarity":[0.6666666666666666,0.6666666666666666,0.6666666666666666,1.0,0.0]}}"#,
        },
        oracle_relpath: None,
        comparator: ParityComparator::SimilarityMultiset,
        tolerance: ParityTolerance { abs: 1.0e-6, rel: 1.0e-6 },
    },
    ParitySpec {
        label: "sorensen/toy_graph@explicit_pairs",
        algorithm: CugraphAlgorithm::Sorensen,
        dataset: "toy_graph",
        params_json: r#"{}"#,
        seed: None,
        truth_source: ParityTruthSource::HandVerified {
            document_json: r#"{"result":{"first":[0,0,3,0,1],"second":[3,3,0,0,4],"similarity":[0.8,0.8,0.8,1.0,0.0]}}"#,
        },
        oracle_relpath: None,
        comparator: ParityComparator::SimilarityMultiset,
        tolerance: ParityTolerance { abs: 1.0e-6, rel: 1.0e-6 },
    },
    ParitySpec {
        label: "overlap/toy_graph@explicit_pairs",
        algorithm: CugraphAlgorithm::Overlap,
        dataset: "toy_graph",
        params_json: r#"{}"#,
        seed: None,
        truth_source: ParityTruthSource::HandVerified {
            document_json: r#"{"result":{"first":[0,0,3,0,1],"second":[3,3,0,0,4],"similarity":[1.0,1.0,1.0,1.0,0.0]}}"#,
        },
        oracle_relpath: None,
        comparator: ParityComparator::SimilarityMultiset,
        tolerance: ParityTolerance { abs: 1.0e-6, rel: 1.0e-6 },
    },
    ParitySpec {
        label: "cosine/toy_graph@explicit_pairs",
        algorithm: CugraphAlgorithm::Cosine,
        dataset: "toy_graph",
        params_json: r#"{}"#,
        seed: None,
        truth_source: ParityTruthSource::HandVerified {
            document_json: r#"{"result":{"first":[0,0,3,0,1],"second":[3,3,0,0,4],"similarity":[0.816496580927726,0.816496580927726,0.816496580927726,1.0,0.0]}}"#,
        },
        oracle_relpath: None,
        comparator: ParityComparator::SimilarityMultiset,
        tolerance: ParityTolerance { abs: 1.0e-6, rel: 1.0e-6 },
    },
];

/// Manifest entry for one locally prepared cuGraph dataset.
#[doc(hidden)]
#[derive(Debug, Clone)]
pub struct ParityDataset {
    pub id: String,
    pub path: String,
    pub format: String,
    pub edge_columns: Vec<String>,
    pub vertices: u64,
    pub edges: u64,
    pub directed: bool,
    pub weighted: bool,
    pub sha256: String,
}

/// Manifest entry for one reference output (cuGraph-curated CSV).
#[doc(hidden)]
#[derive(Debug, Clone)]
pub struct ParityReference {
    pub id: String,
    pub path: String,
    pub format: String,
    pub columns: Vec<String>,
    pub dataset: String,
    pub sha256: String,
}

/// Parsed view of locally generated `fixture/graph/cugraph/manifest.json`.
#[doc(hidden)]
#[derive(Debug, Clone)]
pub struct ParityManifest {
    pub source_repo: String,
    pub license: String,
    pub datasets: Vec<ParityDataset>,
    pub references: Vec<ParityReference>,
}

impl ParityManifest {
    pub fn dataset(&self, id: &str) -> Option<&ParityDataset> {
        self.datasets.iter().find(|d| d.id == id)
    }
}

/// Errors raised while loading and validating the parity manifest.
#[doc(hidden)]
#[derive(Debug)]
pub enum ParityManifestError {
    MissingManifest { path: PathBuf, source: std::io::Error },
    InvalidManifest { path: PathBuf, source: serde_json::Error },
}

impl std::fmt::Display for ParityManifestError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::MissingManifest { path, source } => {
                write!(f, "cuGraph manifest missing at {}: {source}", path.display())
            },
            Self::InvalidManifest { path, source } => {
                write!(f, "cuGraph manifest at {} could not be parsed: {source}", path.display())
            },
        }
    }
}

impl std::error::Error for ParityManifestError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::MissingManifest { source, .. } => Some(source),
            Self::InvalidManifest { source, .. } => Some(source),
        }
    }
}

/// Resolve the cuGraph data root.
///
/// The harness honors `ALGEON_CUGRAPH_DATA_ROOT` so refresh tooling can point at a
/// staging copy; default is the local `fixture/graph/cugraph` directory
/// relative to the workspace root determined by `CARGO_MANIFEST_DIR`.
pub fn data_root() -> PathBuf {
    if let Ok(env) = std::env::var("ALGEON_CUGRAPH_DATA_ROOT") {
        return PathBuf::from(env);
    }
    // The adapter manifest is under crates/<name>, two levels below the repository root.
    let repository = Path::new(env!("CARGO_MANIFEST_DIR"))
        .ancestors()
        .nth(2)
        .expect("adapter manifest must be two levels below the repository root");
    repository.join(CUGRAPH_DATA_DIR_FROM_CARGO_MANIFEST)
}

/// Load and validate the manifest at `data_root/manifest.json`.
pub fn load_manifest(data_root: &Path) -> Result<ParityManifest, ParityManifestError> {
    let manifest_path = data_root.join("manifest.json");
    let raw = fs::read_to_string(&manifest_path)
        .map_err(|source| ParityManifestError::MissingManifest { path: manifest_path.clone(), source })?;
    let value: Value = serde_json::from_str(&raw)
        .map_err(|source| ParityManifestError::InvalidManifest { path: manifest_path.clone(), source })?;
    let datasets = value
        .get("datasets")
        .and_then(|v| v.as_array())
        .into_iter()
        .flatten()
        .map(parity_dataset_from_json)
        .collect::<Vec<_>>();
    let references = value
        .get("references")
        .and_then(|v| v.as_array())
        .into_iter()
        .flatten()
        .map(parity_reference_from_json)
        .collect::<Vec<_>>();
    Ok(ParityManifest {
        source_repo: value.get("source_repo").and_then(|v| v.as_str()).unwrap_or("").to_string(),
        license: value.get("license").and_then(|v| v.as_str()).unwrap_or("").to_string(),
        datasets,
        references,
    })
}

fn parity_dataset_from_json(v: &Value) -> ParityDataset {
    ParityDataset {
        id: v.get("id").and_then(|v| v.as_str()).unwrap_or("").to_string(),
        path: v.get("path").and_then(|v| v.as_str()).unwrap_or("").to_string(),
        format: v.get("format").and_then(|v| v.as_str()).unwrap_or("").to_string(),
        edge_columns: v
            .get("edge_columns")
            .and_then(|v| v.as_array())
            .into_iter()
            .flatten()
            .filter_map(|s| s.as_str().map(str::to_string))
            .collect(),
        vertices: v.get("vertices").and_then(|v| v.as_u64()).unwrap_or(0),
        edges: v.get("edges").and_then(|v| v.as_u64()).unwrap_or(0),
        directed: v.get("directed").and_then(|v| v.as_bool()).unwrap_or(false),
        weighted: v.get("weighted").and_then(|v| v.as_bool()).unwrap_or(false),
        sha256: v.get("sha256").and_then(|v| v.as_str()).unwrap_or("").to_string(),
    }
}

fn parity_reference_from_json(v: &Value) -> ParityReference {
    ParityReference {
        id: v.get("id").and_then(|v| v.as_str()).unwrap_or("").to_string(),
        path: v.get("path").and_then(|v| v.as_str()).unwrap_or("").to_string(),
        format: v.get("format").and_then(|v| v.as_str()).unwrap_or("").to_string(),
        columns: v
            .get("columns")
            .and_then(|v| v.as_array())
            .into_iter()
            .flatten()
            .filter_map(|s| s.as_str().map(str::to_string))
            .collect(),
        dataset: v.get("dataset").and_then(|v| v.as_str()).unwrap_or("").to_string(),
        sha256: v.get("sha256").and_then(|v| v.as_str()).unwrap_or("").to_string(),
    }
}

/// Path to the generated oracle JSON under `data_root` for one parity row,
/// when the row has a generated-oracle contract.
pub fn oracle_path(data_root: &Path, spec: &ParitySpec) -> Option<PathBuf> {
    spec.oracle_relpath.map(|path| data_root.join("oracle").join(path))
}

/// Errors raised by [`load_ref_csv_vertex_list`].
#[doc(hidden)]
#[derive(Debug)]
pub enum RefCsvError {
    /// The reference CSV could not be opened.
    Missing { path: PathBuf, source: std::io::Error },
    /// A line in the reference CSV was not parseable as an integer
    /// vertex id.
    InvalidVertex { path: PathBuf, line: usize, value: String },
}

impl std::fmt::Display for RefCsvError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Missing { path, source } => {
                write!(f, "reference CSV missing at {}: {source}", path.display())
            },
            Self::InvalidVertex { path, line, value } => {
                write!(f, "reference CSV at {} line {line} has non-integer vertex {value:?}", path.display())
            },
        }
    }
}

impl std::error::Error for RefCsvError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Missing { source, .. } => Some(source),
            Self::InvalidVertex { .. } => None,
        }
    }
}

/// Load a cuGraph-curated reference CSV that lists one vertex id per
/// line, ordered by descending score.
///
/// Used by parity rows whose [`ParityTruthSource`] is
/// [`ParityTruthSource::RefCsv`] (e.g. `ref/katz/karate.csv`). The file
/// format mirrors the upstream cuGraph Python test convention
/// (`topKVertices(katz, "katz_centrality", k)` writes one vertex id
/// per line, top-down). Blank lines are ignored so a CSV that ends
/// with a trailing newline still parses cleanly.
pub fn load_ref_csv_vertex_list(data_root: &Path, relpath: &str) -> Result<Vec<i64>, RefCsvError> {
    let path = data_root.join(relpath);
    let raw = fs::read_to_string(&path).map_err(|source| RefCsvError::Missing { path: path.clone(), source })?;
    let mut vertices = Vec::new();
    for (idx, line) in raw.lines().enumerate() {
        let trimmed = line.trim();
        if trimmed.is_empty() {
            continue;
        }
        let value = trimmed.parse::<i64>().map_err(|_| RefCsvError::InvalidVertex {
            path: path.clone(),
            line: idx + 1,
            value: trimmed.to_string(),
        })?;
        vertices.push(value);
    }
    Ok(vertices)
}

/// Outcome of loading the oracle JSON for one parity row.
#[doc(hidden)]
#[derive(Debug)]
pub enum OracleLookup {
    /// The oracle JSON is present on disk and its header matches the
    /// parity spec.
    Present { document: Value, path: PathBuf },
    /// The oracle JSON file does not exist yet. Callers can render the
    /// structured refresh hint; the GPU acceptance run treats it as a failure.
    Missing { path: PathBuf },
    /// The oracle JSON exists but disagrees with the parity spec.
    Mismatch { path: PathBuf, reason: String },
    /// The oracle JSON could not be parsed.
    Invalid { path: PathBuf, reason: String },
    /// The row owns a hand-verified result and has no generated oracle.
    NotApplicable,
}

/// Load + cross-check the oracle JSON header for a parity row.
///
/// Returns [`OracleLookup::Missing`] when the file is absent so callers
/// can degrade gracefully. A present file that cannot be read or parsed
/// produces `Invalid`, and one that disagrees with the spec produces `Mismatch`.
///
/// When `manifest` is `Some`, the oracle's `header.dataset_sha256` is
/// cross-checked against the manifest entry for `spec.dataset`. This
/// catches the case where a local dataset is updated (manifest
/// SHA-256 rotated) but a stale oracle JSON survives on disk.
pub fn load_oracle(data_root: &Path, spec: &ParitySpec, manifest: Option<&ParityManifest>) -> OracleLookup {
    let Some(path) = oracle_path(data_root, spec) else {
        return OracleLookup::NotApplicable;
    };
    let raw = match fs::read_to_string(&path) {
        Ok(raw) => raw,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return OracleLookup::Missing { path },
        Err(error) => return OracleLookup::Invalid { path, reason: format!("read failed: {error}") },
    };
    let document: Value = match serde_json::from_str(&raw) {
        Ok(value) => value,
        Err(error) => return OracleLookup::Invalid { path, reason: format!("invalid JSON: {error}") },
    };
    if let Err(reason) = check_oracle_header(&document, spec, manifest) {
        return OracleLookup::Mismatch { path, reason };
    }
    OracleLookup::Present { document, path }
}

fn check_oracle_header(document: &Value, spec: &ParitySpec, manifest: Option<&ParityManifest>) -> Result<(), String> {
    let header = document.get("header").ok_or_else(|| "oracle JSON is missing a header object".to_string())?;
    let dataset = header.get("dataset").and_then(|v| v.as_str()).unwrap_or("");
    if dataset != spec.dataset {
        return Err(format!("oracle header dataset='{dataset}' but parity spec expects '{}'", spec.dataset));
    }
    let algorithm = header.get("algorithm").and_then(|v| v.as_str()).unwrap_or("");
    let expected_algorithm = spec.algorithm.native_algorithm().as_str();
    if algorithm != expected_algorithm {
        return Err(format!("oracle header algorithm='{algorithm}' but parity spec expects '{expected_algorithm}'"));
    }
    let params = header.get("params").cloned().unwrap_or(Value::Null);
    let expected_params: Value = serde_json::from_str(spec.params_json)
        .map_err(|error| format!("parity spec params_json is not valid JSON: {error}"))?;
    if !json_equal_numerically(&params, &expected_params) {
        return Err(format!("oracle header params {params} but parity spec expects {expected_params}"));
    }
    if let Some(manifest) = manifest {
        let dataset_entry = manifest.dataset(spec.dataset).ok_or_else(|| {
            format!(
                "parity spec dataset '{}' is not present in the manifest; refresh the dataset list before validating the oracle",
                spec.dataset,
            )
        })?;
        let header_sha = header.get("dataset_sha256").and_then(|v| v.as_str()).unwrap_or("");
        if header_sha != dataset_entry.sha256 {
            return Err(format!(
                "oracle header dataset_sha256='{header_sha}' but manifest expects '{}' for dataset '{}'; \
                 the local dataset bytes changed since this oracle was generated; refresh it",
                dataset_entry.sha256, spec.dataset,
            ));
        }
    }
    Ok(())
}

/// JSON equality that treats `1.0` and `1` as the same value to avoid
/// false positives across Python json (which writes `1.0`) and Rust
/// `serde_json` (which round-trips integers when possible).
fn json_equal_numerically(left: &Value, right: &Value) -> bool {
    match (left, right) {
        (Value::Number(a), Value::Number(b)) => match (a.as_f64(), b.as_f64()) {
            (Some(x), Some(y)) => x == y,
            _ => a == b,
        },
        (Value::Array(a), Value::Array(b)) => {
            a.len() == b.len() && a.iter().zip(b).all(|(l, r)| json_equal_numerically(l, r))
        },
        (Value::Object(a), Value::Object(b)) => {
            a.len() == b.len() && a.iter().all(|(k, v)| b.get(k).is_some_and(|w| json_equal_numerically(v, w)))
        },
        _ => left == right,
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn parity_specs_have_unique_oracle_paths() {
        let mut seen = std::collections::BTreeSet::new();
        for spec in CUGRAPH_PARITY_SPECS {
            let Some(path) = spec.oracle_relpath else {
                continue;
            };
            assert!(seen.insert(path), "oracle path {} appears twice in parity specs", path);
        }
    }

    #[test]
    fn json_equal_numerically_treats_integer_and_float_as_equivalent() {
        assert!(json_equal_numerically(
            &serde_json::json!({"alpha": 0.85, "max_iterations": 100}),
            &serde_json::json!({"alpha": 0.85, "max_iterations": 100.0}),
        ));
        assert!(!json_equal_numerically(&serde_json::json!({"alpha": 0.85}), &serde_json::json!({"alpha": 0.86}),));
    }

    #[test]
    fn ref_csv_loader_reads_vertex_list_top_to_bottom() {
        let root = data_root();
        let vertices = load_ref_csv_vertex_list(&root, "ref/katz/karate.csv")
            .expect("local Katz reference CSV must parse cleanly");
        // The cuGraph karate top-K starts at vertex 33 by descending
        // Katz score; the exact ordering is what the upstream cuGraph
        // test asserts against, so pin both the first vertex and that
        // all entries fall in the karate graph's vertex range so an
        // accidental file mutation surfaces here.
        assert!(!vertices.is_empty(), "karate Katz reference must list at least one vertex");
        assert_eq!(vertices.first().copied(), Some(33), "first ref vertex is the top-Katz vertex");
        assert!(vertices.iter().all(|v| (0..=33).contains(v)), "karate vertices are 0..=33; got {vertices:?}");
    }

    #[test]
    fn ref_csv_loader_reports_unparseable_line() {
        let tmp_dir = std::env::temp_dir();
        let file_name = format!("cugraph_parity_invalid_{}.csv", std::process::id());
        let tmp = tmp_dir.join(&file_name);
        std::fs::write(&tmp, "33\nnot-an-int\n0\n").unwrap();
        let err = load_ref_csv_vertex_list(&tmp_dir, &file_name)
            .expect_err("non-integer vertex must surface as InvalidVertex");
        match err {
            RefCsvError::InvalidVertex { line, value, .. } => {
                assert_eq!(line, 2);
                assert_eq!(value, "not-an-int");
            },
            other => panic!("expected InvalidVertex, got {other:?}"),
        }
        let _ = std::fs::remove_file(tmp);
    }
}