//! Serving-cache, revision, and telemetry unit tests split from
//! `tests.rs` to keep both under the source-hygiene size target.

#![allow(clippy::expect_used, clippy::unwrap_used)]

use super::*;

#[test]
#[allow(clippy::panic)]
fn hnsw_directory_record_round_trips_a_versioned_locator() {
    let record = HnswDirectoryRecord {
        key_kind: GraphDirectoryKeyKind::Adjacency,
        generation: 7,
        identity: 42,
        ordinal: 2,
        target_page: 19,
        target_slot: 4,
        revision: 11,
    };
    let Ok(decoded) = decode_hnsw_directory_record(&encode_hnsw_directory_record(record)) else {
        panic!("generated directory locator must decode");
    };
    assert_eq!(decoded, record);
}

#[test]
fn hnsw_directory_index_prefers_the_latest_node_locator() {
    let mut directory = HnswDirectoryIndex::default();
    let first = HnswDirectoryRecord {
        key_kind: GraphDirectoryKeyKind::Node,
        generation: 1,
        identity: 42,
        ordinal: 0,
        target_page: 7,
        target_slot: 2,
        revision: 3,
    };
    let newer = HnswDirectoryRecord {
        target_page: 9,
        target_slot: 4,
        revision: 4,
        ..first
    };
    let stale = HnswDirectoryRecord {
        target_page: 11,
        revision: 2,
        ..first
    };

    directory.observe(first);
    directory.observe(newer);
    directory.observe(stale);

    assert_eq!(directory.node(HnswNodeId::new(42)), Some(newer));
}

#[test]
#[should_panic]
fn hnsw_vector_record_rejects_truncated_neighbor_payload() {
    let record = HnswVectorRecord {
        node_id: HnswNodeId::new(1),
        heap_tid: 17,
        vector: valid_test_vector(vec![1.0, 2.5]),
        base_neighbors: vec![HnswNodeId::new(0)],
        layers: Vec::new(),
    };
    let payload = encode_hnsw_vector_record(&record);

    // SAFETY: The pointer is valid, but the length intentionally truncates the
    // encoded neighbor payload to verify defensive decode behavior.
    let _ = unsafe { decode_hnsw_vector_record(payload.as_ptr(), payload.len() - 1) };
}

#[test]
fn hnsw_graph_from_records_uses_persisted_neighbors() -> Result<(), Box<dyn std::error::Error>> {
    let records = vec![
        HnswVectorRecord {
            node_id: HnswNodeId::new(0),
            heap_tid: 10,
            vector: DenseVector::new(vec![0.0, 0.0])?,
            base_neighbors: vec![HnswNodeId::new(1)],
            layers: Vec::new(),
        },
        HnswVectorRecord {
            node_id: HnswNodeId::new(1),
            heap_tid: 20,
            vector: DenseVector::new(vec![1.0, 0.0])?,
            base_neighbors: vec![HnswNodeId::new(0)],
            layers: Vec::new(),
        },
    ];

    let graph = hnsw_graph_from_records_with_config(
        records,
        DistanceMetric::L2,
        test_hnsw_config()?,
        Some(HnswNodeId::new(0)),
    );

    assert_eq!(
        graph.neighbors(HnswNodeId::new(0), LayerIndex::base()),
        Some(&[HnswNodeId::new(1)][..])
    );
    assert_eq!(
        graph.neighbors(HnswNodeId::new(1), LayerIndex::base()),
        Some(&[HnswNodeId::new(0)][..])
    );
    Ok(())
}

#[test]
fn hnsw_unordered_scan_preserves_visible_append_order() -> Result<(), Box<dyn std::error::Error>> {
    let records = vec![
        HnswVectorRecord {
            node_id: HnswNodeId::new(0),
            heap_tid: 10,
            vector: DenseVector::new(vec![100.0, 0.0])?,
            base_neighbors: vec![HnswNodeId::new(1)],
            layers: Vec::new(),
        },
        HnswVectorRecord {
            node_id: HnswNodeId::new(1),
            heap_tid: 20,
            vector: DenseVector::new(vec![90.0, 0.0])?,
            base_neighbors: vec![HnswNodeId::new(0), HnswNodeId::new(2)],
            layers: Vec::new(),
        },
        HnswVectorRecord {
            node_id: HnswNodeId::new(2),
            heap_tid: 30,
            vector: DenseVector::new(vec![0.0, 0.0])?,
            base_neighbors: vec![HnswNodeId::new(1), HnswNodeId::new(3)],
            layers: Vec::new(),
        },
        HnswVectorRecord {
            node_id: HnswNodeId::new(3),
            heap_tid: 40,
            vector: DenseVector::new(vec![1.0, 0.0])?,
            base_neighbors: vec![HnswNodeId::new(2)],
            layers: Vec::new(),
        },
    ];
    let candidates = hnsw_unordered_scan_candidates(records);

    assert_eq!(
        candidates
            .iter()
            .map(|candidate| candidate.heap_tid)
            .collect::<Vec<_>>(),
        vec![10, 20, 30, 40]
    );
    Ok(())
}

#[test]
fn hnsw_unordered_scan_does_not_rank_disconnected_graph() -> Result<(), Box<dyn std::error::Error>>
{
    let records = vec![
        HnswVectorRecord {
            node_id: HnswNodeId::new(0),
            heap_tid: 10,
            vector: DenseVector::new(vec![100.0, 0.0])?,
            base_neighbors: vec![HnswNodeId::new(1)],
            layers: Vec::new(),
        },
        HnswVectorRecord {
            node_id: HnswNodeId::new(1),
            heap_tid: 20,
            vector: DenseVector::new(vec![90.0, 0.0])?,
            base_neighbors: vec![HnswNodeId::new(0)],
            layers: Vec::new(),
        },
        HnswVectorRecord {
            node_id: HnswNodeId::new(2),
            heap_tid: 30,
            vector: DenseVector::new(vec![0.0, 0.0])?,
            base_neighbors: vec![HnswNodeId::new(3)],
            layers: Vec::new(),
        },
        HnswVectorRecord {
            node_id: HnswNodeId::new(3),
            heap_tid: 40,
            vector: DenseVector::new(vec![1.0, 0.0])?,
            base_neighbors: vec![HnswNodeId::new(2)],
            layers: Vec::new(),
        },
    ];
    let candidates = hnsw_unordered_scan_candidates(records);

    assert_eq!(
        candidates
            .iter()
            .map(|candidate| candidate.heap_tid)
            .collect::<Vec<_>>(),
        vec![10, 20, 30, 40]
    );
    Ok(())
}

#[test]
fn hnsw_unordered_scan_does_not_rank_incremental_records() -> Result<(), Box<dyn std::error::Error>>
{
    let records = vec![
        HnswVectorRecord {
            node_id: HnswNodeId::new(0),
            heap_tid: 10,
            vector: DenseVector::new(vec![100.0, 0.0])?,
            base_neighbors: Vec::new(),
            layers: Vec::new(),
        },
        HnswVectorRecord {
            node_id: HnswNodeId::new(1),
            heap_tid: 20,
            vector: DenseVector::new(vec![0.0, 0.0])?,
            base_neighbors: Vec::new(),
            layers: Vec::new(),
        },
    ];
    let candidates = hnsw_unordered_scan_candidates(records);

    assert_eq!(
        candidates
            .iter()
            .map(|candidate| candidate.heap_tid)
            .collect::<Vec<_>>(),
        vec![10, 20]
    );
    Ok(())
}

#[test]
fn hnsw_unordered_scan_does_not_rank_mixed_append_only_records()
-> Result<(), Box<dyn std::error::Error>> {
    let records = vec![
        HnswVectorRecord {
            node_id: HnswNodeId::new(0),
            heap_tid: 10,
            vector: DenseVector::new(vec![100.0, 0.0])?,
            base_neighbors: vec![HnswNodeId::new(1)],
            layers: Vec::new(),
        },
        HnswVectorRecord {
            node_id: HnswNodeId::new(1),
            heap_tid: 20,
            vector: DenseVector::new(vec![90.0, 0.0])?,
            base_neighbors: vec![HnswNodeId::new(0)],
            layers: Vec::new(),
        },
        HnswVectorRecord {
            node_id: HnswNodeId::new(2),
            heap_tid: 30,
            vector: DenseVector::new(vec![0.0, 0.0])?,
            base_neighbors: Vec::new(),
            layers: Vec::new(),
        },
    ];
    let candidates = hnsw_unordered_scan_candidates(records);

    assert_eq!(
        candidates
            .iter()
            .map(|candidate| candidate.heap_tid)
            .collect::<Vec<_>>(),
        vec![10, 20, 30]
    );
    Ok(())
}

#[test]
fn hnsw_graph_from_records_rejects_corrupt_neighbor_ids() -> Result<(), Box<dyn std::error::Error>>
{
    let records = vec![HnswVectorRecord {
        node_id: HnswNodeId::new(0),
        heap_tid: 10,
        vector: valid_test_vector(vec![0.0, 0.0]),
        base_neighbors: vec![HnswNodeId::new(99)],
        layers: Vec::new(),
    }];

    assert!(matches!(
        try_hnsw_graph_from_records_with_config(
            records,
            DistanceMetric::L2,
            test_hnsw_config()?,
            Some(HnswNodeId::new(0)),
        ),
        Err(HnswError::InvalidParameter {
            parameter: "neighbor_id",
            value: 99,
        })
    ));
    Ok(())
}

#[test]
fn tombstone_snapshots_use_unique_traversal_only_point_ids() {
    let first = hnsw_tombstone_record(&HnswVectorRecord {
        node_id: HnswNodeId::new(0),
        heap_tid: 42,
        vector: valid_test_vector(vec![0.0, 0.0]),
        base_neighbors: vec![HnswNodeId::new(1)],
        layers: Vec::new(),
    });
    let second = hnsw_tombstone_record(&HnswVectorRecord {
        node_id: HnswNodeId::new(1),
        heap_tid: 42,
        vector: valid_test_vector(vec![1.0, 1.0]),
        base_neighbors: vec![HnswNodeId::new(0)],
        layers: Vec::new(),
    });

    let first = hnsw_graph_snapshot_from_record(first);
    let second = hnsw_graph_snapshot_from_record(second);

    assert_ne!(first.point_id(), second.point_id());
    assert!(hnsw_point_id_is_tombstoned(first.point_id()));
    assert!(hnsw_point_id_is_tombstoned(second.point_id()));
}

fn test_hnsw_config() -> context_index::Result<HnswConfig> {
    HnswConfig::new(
        context_core::policy::DEFAULT_HNSW_M,
        context_core::policy::DEFAULT_HNSW_EF_CONSTRUCTION,
        context_core::policy::DEFAULT_HNSW_EF_SEARCH,
    )
}

#[test]
fn hnsw_metric_metapage_identity_round_trips_every_dense_metric() -> context_index::Result<()> {
    let config = HnswConfig::new(7, 31, 13)?;
    let cases = [
        (HnswScoreMetric::L2, DistanceMetric::L2, 1_u16),
        (
            HnswScoreMetric::NegativeInnerProduct,
            DistanceMetric::NegativeInnerProduct,
            2,
        ),
        (
            HnswScoreMetric::Cosine,
            DistanceMetric::NegativeInnerProduct,
            3,
        ),
        (HnswScoreMetric::L1, DistanceMetric::L1, 4),
        (HnswScoreMetric::BitHamming, DistanceMetric::Hamming, 5),
        (HnswScoreMetric::BitJaccard, DistanceMetric::Jaccard, 6),
    ];

    for (score_metric, graph_metric, storage_tag) in cases {
        let mut meta = HnswMetaPage::empty();
        meta.record_index_identity(score_metric, config);

        assert_eq!(score_metric.navigation_metric(), graph_metric);
        assert_eq!(score_metric.storage_tag(), storage_tag);
        assert_eq!(
            meta.stored_config(score_metric, 19),
            HnswConfig::new(7, 31, 19)?
        );
    }

    Ok(())
}

#[test]
fn bit_jaccard_orderby_lower_bound_covers_float4_navigation_collisions() {
    // This 8,001-bit pair is valid under the SQL type policy and pins the
    // precision hazard for a future multi-page or bit-native graph record.
    let row_a_exact = 3.0_f64 / 8_000.0;
    let row_b_exact = 3.0_f64 / 8_001.0;
    let row_a_navigation = 1.0_f32 - (7_997.0_f32 / 8_000.0_f32);
    let row_b_navigation = 1.0_f32 - (7_998.0_f32 / 8_001.0_f32);

    assert_eq!(row_a_navigation, row_b_navigation);
    assert!(row_b_exact < row_a_exact);
    let (row_a_bound, row_a_recheck) =
        float8_orderby_distance(HnswScoreMetric::BitJaccard, row_a_navigation);
    let (row_b_bound, row_b_recheck) =
        float8_orderby_distance(HnswScoreMetric::BitJaccard, row_b_navigation);
    assert!(row_a_recheck);
    assert!(row_b_recheck);
    assert!(row_a_bound <= row_a_exact);
    assert!(row_b_bound <= row_b_exact);

    assert_eq!(
        float8_orderby_distance(HnswScoreMetric::L2, row_a_navigation),
        (f64::from(row_a_navigation), false)
    );
}

#[test]
fn hnsw_metric_graph_reload_uses_selected_metric() -> context_index::Result<()> {
    let records = vec![
        HnswVectorRecord {
            node_id: HnswNodeId::new(0),
            heap_tid: 10,
            vector: DenseVector::new(vec![1.0, 0.0]).map_err(HnswError::from)?,
            base_neighbors: vec![HnswNodeId::new(1)],
            layers: vec![vec![HnswNodeId::new(1)]],
        },
        HnswVectorRecord {
            node_id: HnswNodeId::new(1),
            heap_tid: 20,
            vector: DenseVector::new(vec![3.0, 0.0]).map_err(HnswError::from)?,
            base_neighbors: vec![HnswNodeId::new(0)],
            layers: vec![vec![HnswNodeId::new(0)]],
        },
    ];
    let graph = try_hnsw_graph_from_records_with_config(
        records,
        DistanceMetric::NegativeInnerProduct,
        HnswConfig::new(2, 4, 4)?,
        Some(HnswNodeId::new(0)),
    )?;
    let query = DenseVector::new(vec![1.0, 0.0]).map_err(HnswError::from)?;
    let results = graph.search(&query, SearchLimit::new(2).map_err(HnswError::from)?)?;

    assert_eq!(
        results
            .iter()
            .map(|result| result.point_id().get())
            .collect::<Vec<_>>(),
        vec![20, 10]
    );
    Ok(())
}

#[allow(
    clippy::expect_used,
    reason = "test fixture vectors are fixed finite non-empty values"
)]
fn valid_test_vector(values: Vec<f32>) -> DenseVector {
    DenseVector::new(values).expect("fixed test vector should be valid")
}

#[test]
fn serving_stats_record_builds_and_reuses() {
    let baseline = hnsw_serving_stats_snapshot();
    record_hnsw_pack_build(1024, 7);
    record_hnsw_pack_reuse();
    record_hnsw_pack_reuse();
    record_hnsw_pack_build(2048, 3);
    record_hnsw_generation_bytes(3072);
    let stats = hnsw_serving_stats_snapshot();
    assert_eq!(stats.pack_builds, baseline.pack_builds + 2);
    assert_eq!(stats.pack_reuses, baseline.pack_reuses + 2);
    assert_eq!(stats.last_pack_bytes, 3072);
    assert_eq!(stats.last_pack_millis, 3);
    assert_eq!(stats.total_pack_millis, baseline.total_pack_millis + 10);
}

#[test]
fn serving_stats_saturate_instead_of_overflowing() {
    HNSW_SERVING_STATS.with(|stats| {
        stats.set(HnswServingStats {
            pack_builds: u64::MAX,
            pack_reuses: u64::MAX,
            last_pack_bytes: 0,
            last_pack_millis: 0,
            total_pack_millis: u64::MAX,
            shared_attaches: 0,
            shared_publishes: 0,
            shared_publish_skips: 0,
            mapped_attaches: 0,
            mapped_publishes: 0,
            mapped_publish_skips: 0,
            page_native_fallbacks: 0,
            delta_segment_records: 0,
            delta_segment_scans: 0,
            segment_rotations: 0,
            multi_segment_scans: 0,
            parallel_segment_scans: 0,
            serial_segment_degradations: 0,
            max_segments_observed: 0,
            ..HnswServingStats::default()
        });
    });
    record_hnsw_pack_build(1, 1);
    record_hnsw_pack_reuse();
    let stats = hnsw_serving_stats_snapshot();
    assert_eq!(stats.pack_builds, u64::MAX);
    assert_eq!(stats.pack_reuses, u64::MAX);
    assert_eq!(stats.total_pack_millis, u64::MAX);
    HNSW_SERVING_STATS.with(|stats| stats.set(HnswServingStats::default()));
}

#[test]
fn build_profile_records_latest_build_phases() {
    record_hnsw_build_profile(HnswBuildProfile {
        tuples: 10,
        graph_millis: 100,
        write_millis: 40,
    });
    record_hnsw_build_profile(HnswBuildProfile {
        tuples: 20,
        graph_millis: 7,
        write_millis: 3,
    });
    let profile = hnsw_build_profile_snapshot();
    assert_eq!(profile.tuples, 20);
    assert_eq!(profile.graph_millis, 7);
    assert_eq!(profile.write_millis, 3);
}

#[test]
fn directory_index_keeps_the_latest_revision_per_node() {
    let mut directory = HnswDirectoryIndex::default();
    let record = |revision: u64| HnswDirectoryRecord {
        key_kind: GraphDirectoryKeyKind::Node,
        generation: 1,
        identity: 7,
        ordinal: 0,
        target_page: 0,
        target_slot: 0,
        revision,
    };
    directory.observe(record(5));
    directory.observe(record(2));
    directory.observe(record(9));
    assert_eq!(
        directory
            .node(HnswNodeId::new(7))
            .map(|record| record.revision),
        Some(9)
    );
}

#[test]
fn rotated_residuals_drop_live_history_and_delta_owned_tombstones() {
    let records = vec![
        context_storage::DeltaRecord::live(10, vec![1.0, 2.0]).unwrap(),
        context_storage::DeltaRecord::tombstone(10),
        context_storage::DeltaRecord::tombstone(20),
        context_storage::DeltaRecord::live(30, vec![3.0, 4.0]).unwrap(),
    ];
    let delta_owned = BTreeSet::from([10_u64, 30_u64]);
    let residual = residual_tombstones(&records, Some(&delta_owned));
    assert_eq!(
        residual
            .iter()
            .map(|record| record.heap_tid)
            .collect::<Vec<_>>(),
        vec![20]
    );
    assert!(
        residual
            .iter()
            .all(|record| record.kind == DeltaRecordKind::Tombstone)
    );
}

#[test]
fn bounded_compaction_projection_counts_overlapping_working_sets() {
    let config = HnswConfig::new(16, 64, 40).unwrap();
    let one = bounded_compaction_working_set_bytes(1, 0, 384, config).unwrap();
    let many = bounded_compaction_working_set_bytes(1_000_000, 500, 384, config).unwrap();
    assert!(one >= 384 * size_of::<f32>() * 3);
    assert!(many > 4 * 1024 * 1024 * 1024);
}

#[test]
fn parallel_segment_projection_accounts_for_every_extent_before_loading() {
    let mut meta = HnswMetaPage::empty();
    meta.dimensions = 8;
    meta.hnsw_m = 16;
    for (index, start_block) in [1_u64, 4, 7].into_iter().enumerate() {
        meta.segments[index] = HnswSegmentMeta {
            segment_id: index as u64 + 1,
            generation: 1,
            start_block,
            end_block: start_block + 2,
            graph_nodes: 1,
            entry_node_id: 0,
            mutation_generation: u64::MAX,
            mutation_start_block: u64::MAX,
            mutation_end_block: u64::MAX,
            mutation_record_count: 0,
        };
    }
    meta.segment_count = 3;

    let projected =
        projected_parallel_segment_bytes(meta).expect("the parallel projection should fit");
    assert!(
        projected >= 3 * 4 * pg_sys::BLCKSZ as u64,
        "the projection must retain every pack while charging decoded extent overlap"
    );

    meta.segments[2].end_block = u64::MAX;
    assert_eq!(
        projected_parallel_segment_bytes(meta),
        None,
        "projection overflow must deny parallel admission"
    );
}

#[test]
fn typed_hnsw_tiny_port_memory_degrades_parallel_segments_to_serial() {
    let mut meta = HnswMetaPage::empty();
    meta.dimensions = 8;
    meta.hnsw_m = 16;
    for (index, start_block) in [1_u64, 4, 7].into_iter().enumerate() {
        meta.segments[index] = HnswSegmentMeta {
            segment_id: index as u64 + 1,
            generation: 1,
            start_block,
            end_block: start_block + 2,
            graph_nodes: 32,
            entry_node_id: 0,
            mutation_generation: u64::MAX,
            mutation_start_block: u64::MAX,
            mutation_end_block: u64::MAX,
            mutation_record_count: 0,
        };
    }
    meta.segment_count = 3;
    let serial = projected_serial_segment_scan_bytes(meta, 10, 40, 0, false)
        .expect("serial projection should fit");
    let parallel = projected_parallel_segment_scan_bytes(meta, 10, 40, 0, false)
        .expect("parallel projection should fit");
    assert!(parallel > serial);

    assert!(serial_segment_memory_admitted(serial, serial));
    assert!(
        !parallel_segment_projection_admitted(meta, parallel, u64::MAX, serial, 2),
        "a typed query memory cap between the serial and parallel peaks must force serial serving"
    );
}

#[test]
fn parallel_projection_charges_overlay_and_chronological_merge_peak() {
    let mut meta = HnswMetaPage::empty();
    meta.dimensions = 8;
    meta.hnsw_m = 16;
    for (index, start_block) in [1_u64, 4, 7].into_iter().enumerate() {
        meta.segments[index] = HnswSegmentMeta {
            segment_id: index as u64 + 1,
            generation: 1,
            start_block,
            end_block: start_block + 2,
            graph_nodes: 32,
            entry_node_id: 0,
            mutation_generation: 1,
            mutation_start_block: start_block + 2,
            mutation_end_block: start_block + 3,
            mutation_record_count: 8,
        };
    }
    meta.segment_count = 3;
    let without_overlay = {
        let mut clean = meta;
        for index in 0..3 {
            clean.segments[index].mutation_generation = u64::MAX;
            clean.segments[index].mutation_start_block = u64::MAX;
            clean.segments[index].mutation_end_block = u64::MAX;
            clean.segments[index].mutation_record_count = 0;
        }
        projected_parallel_segment_scan_bytes(clean, 34, 40, 0, false)
            .expect("clean parallel projection should fit")
    };
    let with_overlay = projected_parallel_segment_scan_bytes(meta, 34, 40, 0, false)
        .expect("overlay parallel projection should fit");
    assert!(with_overlay > without_overlay);
    assert!(
        !parallel_segment_projection_admitted(meta, with_overlay, u64::MAX, with_overlay - 1, 2,),
        "the query memory ceiling must include retained overlay and merge state"
    );
    assert!(parallel_segment_projection_admitted(
        meta,
        with_overlay,
        u64::MAX,
        with_overlay,
        2,
    ));
}

#[test]
fn parallel_overlay_degrades_before_consuming_a_tight_comparison_budget() {
    let mut meta = HnswMetaPage::empty();
    for (index, start_block) in [1_u64, 4, 7].into_iter().enumerate() {
        meta.segments[index] = HnswSegmentMeta {
            segment_id: index as u64 + 1,
            generation: 1,
            start_block,
            end_block: start_block + 2,
            graph_nodes: 32,
            entry_node_id: 0,
            mutation_generation: 1,
            mutation_start_block: start_block + 2,
            mutation_end_block: start_block + 3,
            mutation_record_count: 8,
        };
    }
    meta.segment_count = 3;
    let expanded_limit = SearchLimit::new(34).expect("expanded parallel limit");
    let exact_minimum = meta.segments().len() * expanded_limit.get();
    let below = HnswComparisonBudget::new(exact_minimum - 1);
    assert!(!parallel_segment_comparisons_admitted(
        meta,
        expanded_limit,
        &below,
    ));
    let exact = HnswComparisonBudget::new(exact_minimum);
    assert!(parallel_segment_comparisons_admitted(
        meta,
        expanded_limit,
        &exact,
    ));
}

#[test]
fn serial_traversal_projection_charges_filter_and_retirement_masks() {
    let mut meta = HnswMetaPage::empty();
    meta.dimensions = 8;
    meta.hnsw_m = 16;
    meta.segments[0] = HnswSegmentMeta {
        segment_id: 1,
        generation: 1,
        start_block: 1,
        end_block: 3,
        graph_nodes: 1_000,
        entry_node_id: 0,
        mutation_generation: 1,
        mutation_start_block: 3,
        mutation_end_block: 4,
        mutation_record_count: 25,
    };
    meta.segment_count = 1;
    meta.delta_generation = 2;
    meta.delta_start_block = 4;
    meta.delta_end_block = 5;
    meta.delta_record_count = 25;

    let unfiltered = projected_serial_segment_scan_bytes(meta, 20, 40, 0, false)
        .expect("unfiltered serial projection should fit");
    let filtered = projected_serial_segment_scan_bytes(meta, 20, 40, 100, true)
        .expect("filtered serial projection should fit");
    assert!(filtered > unfiltered);
    assert!(!serial_segment_memory_admitted(filtered, filtered - 1));
    assert!(serial_segment_memory_admitted(filtered, filtered));
}

#[test]
fn large_single_segment_page_traversal_fits_sixteen_mebibytes() {
    let mut meta = HnswMetaPage::empty();
    meta.dimensions = 128;
    meta.hnsw_m = 16;
    meta.segments[0] = HnswSegmentMeta {
        segment_id: 1,
        generation: 1,
        start_block: 1,
        end_block: 4_096,
        graph_nodes: 100_000,
        entry_node_id: 0,
        mutation_generation: u64::MAX,
        mutation_start_block: u64::MAX,
        mutation_end_block: u64::MAX,
        mutation_record_count: 0,
    };
    meta.segment_count = 1;

    let projected = projected_serial_segment_scan_bytes(meta, 10, 40, 0, false)
        .expect("large serial projection should fit");
    assert!(
        projected <= 16 * 1024 * 1024,
        "page-native traversal should charge its bounded search state, not the full persisted graph: {projected}"
    );
}

#[test]
fn pq_projection_accounts_for_codes_codebook_scorer_and_boundary_budget() {
    let mut meta = HnswMetaPage::empty();
    meta.dimensions = 128;
    meta.hnsw_m = 16;
    meta.quantization_mode = options::HNSW_QUANTIZATION_PQ_U16;
    meta.pq_subvector_dimensions = 8;
    let segment = HnswSegmentMeta {
        segment_id: 1,
        generation: 1,
        start_block: 1,
        end_block: 257,
        graph_nodes: 100_000,
        entry_node_id: 0,
        mutation_generation: u64::MAX,
        mutation_start_block: u64::MAX,
        mutation_end_block: u64::MAX,
        mutation_record_count: 0,
    };
    let projected =
        projected_packed_segment_bytes(meta, segment).expect("valid PQ projection should fit u64");
    let codec = projected_hnsw_codec_bytes(meta, segment.graph_nodes)
        .expect("valid PQ codec projection should fit u64");
    let extent = (segment.end_block - segment.start_block) * pg_sys::BLCKSZ as u64;
    assert!(
        codec > segment.graph_nodes * 16,
        "PQ projection must include more than code rows"
    );
    assert!(
        projected > extent * 4 + codec * 2,
        "projection must charge decoded containers, publication scratch, and duplicate codec state"
    );
    assert!(hnsw_packed_projection_admitted(projected, projected));
    assert!(!hnsw_packed_projection_admitted(projected, projected - 1));
}

#[test]
fn metapage_rejects_graph_mutation_and_active_extent_overlap() {
    let mut meta = HnswMetaPage::empty();
    meta.open_delta_region(10);
    meta.delta_end_block = 12;
    meta.delta_record_count = 1;
    meta.segments[0] = HnswSegmentMeta {
        segment_id: 1,
        generation: 1,
        start_block: 1,
        end_block: 11,
        graph_nodes: 1,
        entry_node_id: 0,
        mutation_generation: u64::MAX,
        mutation_start_block: u64::MAX,
        mutation_end_block: u64::MAX,
        mutation_record_count: 0,
    };
    meta.segment_count = 1;
    assert!(!meta.is_valid());

    meta.delta_start_block = 20;
    meta.delta_end_block = 21;
    meta.segments[0].mutation_generation = 1;
    meta.segments[0].mutation_start_block = 10;
    meta.segments[0].mutation_end_block = 12;
    meta.segments[0].mutation_record_count = 1;
    meta.segments[0].end_block = 11;
    assert!(!meta.is_valid());
}

#[test]
fn metapage_requires_exact_frozen_mutation_record_metadata() {
    let mut meta = HnswMetaPage::empty();
    meta.open_delta_region(20);
    meta.segments[0] = HnswSegmentMeta {
        segment_id: 1,
        generation: 1,
        start_block: 1,
        end_block: 10,
        graph_nodes: 1,
        entry_node_id: 0,
        mutation_generation: 1,
        mutation_start_block: 10,
        mutation_end_block: 11,
        mutation_record_count: 1,
    };
    meta.segment_count = 1;
    assert!(meta.is_valid());
    meta.segments[0].mutation_record_count = 0;
    assert!(!meta.is_valid());
}

#[test]
fn metapage_rejects_nonempty_active_delta_with_empty_extent() {
    let mut meta = HnswMetaPage::empty();
    meta.open_delta_region(20);
    meta.delta_record_count = 1;
    assert!(!meta.is_valid());
}
