kglite 0.17.10

use super::enumerate_segment_dirs;
use super::segment_subdir;
use crate::datatypes::{DataFrame, Value};
use crate::graph::schema::{EdgeData, InternedKey, NodeData, StringInterner};
use crate::graph::storage::backend::GraphBackend;
use crate::graph::storage::disk::csr::{CsrEdge, DiskNodeSlot, EdgeEndpoints};
use crate::graph::storage::disk::generation::GraphDirectoryLock;
use crate::graph::storage::disk::graph_persist::{concat_segment_csrs, SegmentCsr};
use crate::graph::storage::disk::temp_owner::{TempGraphDir, TrackedOwner};
use crate::graph::storage::mapped::mmap_vec::MmapOrVec;
use crate::graph::storage::{GraphRead, GraphWrite};
use crate::graph::DirGraph;
use petgraph::graph::{EdgeIndex, NodeIndex};
use std::collections::{BTreeMap, HashMap};
use std::sync::Arc;
use tempfile::TempDir;

#[test]
fn pending_edge_failure_is_atomic_and_retryable() {
    use crate::graph::storage::mapped::mmap_vec::{fail_next, FailurePoint};

    let tmp = TempDir::new().unwrap();
    let mut interner = StringInterner::new();
    let mut graph = super::DiskGraph::new_at_path(tmp.path()).unwrap();
    graph.defer_csr = true;
    *graph.pending_edges.get_mut() = MmapOrVec::new();
    let mut edge = || EdgeData::new("LINKS".to_string(), HashMap::new(), &mut interner);

    fail_next(FailurePoint::HeapReserve);
    assert!(graph
        .try_add_pending_edge(NodeIndex::new(0), NodeIndex::new(1), edge())
        .is_err());
    assert_eq!(graph.pending_edges.get_mut().len(), 0);
    assert_eq!(graph.edge_count, 0);
    assert_eq!(graph.next_edge_idx, 0);

    let edge_idx = graph
        .try_add_pending_edge(NodeIndex::new(0), NodeIndex::new(1), edge())
        .unwrap();
    assert_eq!(edge_idx.index(), 0);
    assert_eq!(graph.pending_edges.get_mut().len(), 1);
    assert_eq!(graph.edge_count, 1);
    assert_eq!(graph.next_edge_idx, 1);
}

fn add_docs(graph: &mut DirGraph, ids: &[i64]) {
    let rows = ids
        .iter()
        .map(|id| vec![Value::Int64(*id), Value::String(format!("doc-{id}"))])
        .collect();
    let frame =
        DataFrame::from_cypher_rows(vec!["id".to_string(), "title".to_string()], rows).unwrap();
    crate::graph::mutation::maintain::add_nodes(
        graph,
        frame,
        "Doc".to_string(),
        "id".to_string(),
        Some("title".to_string()),
        None,
    )
    .unwrap();
}

fn one_doc_frame(id: i64) -> DataFrame {
    DataFrame::from_cypher_rows(
        vec!["id".to_string(), "title".to_string()],
        vec![vec![Value::Int64(id), Value::String(format!("doc-{id}"))]],
    )
    .unwrap()
}

/// Byte-for-byte contents of every *data* file under `root`, for asserting that
/// one graph's mutations never reach another's published files.
///
/// Dot-prefixed entries are skipped: `.kglite.lock` (the writer lease) and
/// `.working-*` (a mutation workspace) are coordination state, not graph data,
/// so including them would make a snapshot differ purely because a lease was
/// held. On Windows the read fails outright — `fs2` takes that lease with
/// `LockFileEx`, whose byte-range locks are *mandatory* rather than advisory
/// like `flock`, so it returns ERROR_LOCK_VIOLATION (33). No disk-graph data
/// file starts with a dot.
fn snapshot_files(root: &std::path::Path) -> BTreeMap<String, Vec<u8>> {
    fn collect(root: &std::path::Path, dir: &std::path::Path, out: &mut BTreeMap<String, Vec<u8>>) {
        for entry in std::fs::read_dir(dir).unwrap() {
            let entry = entry.unwrap();
            if entry.file_name().to_string_lossy().starts_with('.') {
                continue;
            }
            let path = entry.path();
            if path.is_dir() {
                collect(root, &path, out);
            } else {
                out.insert(
                    path.strip_prefix(root).unwrap().display().to_string(),
                    std::fs::read(path).unwrap(),
                );
            }
        }
    }

    let mut files = BTreeMap::new();
    collect(root, root, &mut files);
    files
}

fn edge_score(graph: &DirGraph) -> Option<Value> {
    let key = InternedKey::from_str("score");
    // Arena-materializing reads must run under a DiskQueryGuard (the
    // debug assert in materialize_edge enforces the protocol).
    let _guard = graph.graph.begin_query();
    graph
        .graph
        .edge_weight(EdgeIndex::new(0))?
        .properties
        .iter()
        .find_map(|(candidate, value)| (*candidate == key).then(|| value.clone()))
}

#[test]
fn legacy_flat_csr_directory_remains_readable() {
    let tmp = TempDir::new().unwrap();
    let mut interner = StringInterner::new();
    let mut graph = super::DiskGraph::new_at_path(tmp.path()).unwrap();
    graph.defer_csr = true;
    let n0 = graph.add_node(seal_test_node(&mut interner, 0, "Doc"));
    let n1 = graph.add_node(seal_test_node(&mut interner, 1, "Doc"));
    graph.add_edge(n0, n1, seal_test_edge(&mut interner, "LINKS"));
    graph.build_csr_from_pending().unwrap();
    graph.save_to_dir(tmp.path(), &interner).unwrap();
    drop(graph);

    // Recreate the pre-segmentation layout: CSR and auxiliary files at the
    // graph root, with the additive layout-version field set to zero.
    let segment = tmp.path().join("seg_000");
    for entry in std::fs::read_dir(&segment).unwrap() {
        let entry = entry.unwrap();
        std::fs::rename(entry.path(), tmp.path().join(entry.file_name())).unwrap();
    }
    std::fs::remove_dir(&segment).unwrap();
    let metadata_path = tmp.path().join("disk_graph_meta.json");
    let mut metadata: serde_json::Value =
        serde_json::from_slice(&std::fs::read(&metadata_path).unwrap()).unwrap();
    metadata["csr_layout_version"] = serde_json::Value::from(0);
    std::fs::write(
        &metadata_path,
        serde_json::to_vec_pretty(&metadata).unwrap(),
    )
    .unwrap();
    let _ = std::fs::remove_file(tmp.path().join("seg_manifest.json"));

    let mut loaded_interner = StringInterner::new();
    let (loaded, _cache) =
        super::DiskGraph::load_from_dir(tmp.path(), &mut loaded_interner).unwrap();
    assert_eq!(loaded.node_count, 2);
    assert_eq!(loaded.edge_count, 1);
    let start = loaded.out_offsets.get(0) as usize;
    let end = loaded.out_offsets.get(1) as usize;
    assert_eq!(end - start, 1);
    assert_eq!(loaded.out_edges.get(start).peer, 1);
}

#[test]
fn transaction_clone_keeps_published_arrays_mapped() {
    let tmp = TempDir::new().unwrap();
    let mut interner = StringInterner::new();
    let mut graph = super::DiskGraph::new_at_path(tmp.path()).unwrap();
    graph.add_node(NodeData::new(
        Value::Int64(1),
        Value::String("doc-1".into()),
        "Doc".into(),
        HashMap::new(),
        &mut interner,
    ));
    graph.save_to_dir(tmp.path(), &interner).unwrap();

    let mut loaded_interner = StringInterner::new();
    let (loaded, _guard) =
        super::DiskGraph::load_from_dir(tmp.path(), &mut loaded_interner).unwrap();
    assert!(loaded.node_slots.is_mapped());
    let fork = loaded.clone();
    assert!(fork.node_slots.is_mapped());
    assert_eq!(fork.out_offsets.is_mapped(), loaded.out_offsets.is_mapped());
    assert_eq!(fork.out_edges.is_mapped(), loaded.out_edges.is_mapped());
    assert_eq!(fork.node_slots.heap_bytes(), 0);
    assert_eq!(fork.out_edges.heap_bytes(), loaded.out_edges.heap_bytes());
}

#[test]
fn transaction_fork_inherits_lease_but_uses_private_workspace() {
    let tmp = TempDir::new().unwrap();
    let mut parent = super::DiskGraph::new_at_path(tmp.path()).unwrap();
    parent.prepare_mutation().unwrap();
    let parent_dir = parent.active_write_dir().to_path_buf();

    let mut child = parent.clone();
    child.adopt_writer_lineage(&parent);
    child.prepare_mutation().unwrap();

    assert!(child.writer_lock.is_some());
    assert_ne!(parent_dir, child.active_write_dir());
    assert!(parent_dir.exists());
    assert!(child.active_write_dir().exists());
}

/// Once `save_disk` returns, every mapping the writer holds must live inside
/// the generation it just published.
///
/// The mutation workspace — and, for a detached copy, its private root — are
/// removed at the end of `finish_generation`. POSIX keeps a mapping valid after
/// its file is unlinked, so arrays left pointing into the deleted scratch
/// directory kept working and the bug was invisible; Windows refuses to remove
/// a directory that still holds a mapped file, so the scratch roots survived
/// and piled up. Asserting on paths makes the invariant fail loudly on every
/// platform instead of only on the one that complains.
#[test]
fn save_rebases_every_mapping_onto_the_published_generation() {
    let root = TempDir::new().unwrap();
    let root_path = root.path().to_str().unwrap();

    let mut graph = DirGraph::new();
    add_docs(&mut graph, &[1, 2, 3]);
    graph.enable_disk_mode().unwrap();
    graph.save_disk(root_path).unwrap();

    // Mutate after the first save so a mutation workspace exists and the CSR
    // is rebuilt inside it, then publish a second generation over the top.
    add_docs(&mut graph, &[4]);
    graph.save_disk(root_path).unwrap();

    let disk = match &mut graph.graph {
        GraphBackend::Disk(disk) => disk,
        _ => panic!("expected disk backend"),
    };
    let data_dir = disk.data_dir.clone();
    let stray: Vec<_> = disk
        .mapped_file_paths()
        .into_iter()
        .filter(|path| !path.starts_with(&data_dir))
        .collect();
    assert!(
        stray.is_empty(),
        "writer still maps files outside the published generation {}: {stray:?}",
        data_dir.display()
    );
}

/// A clean publish hands the directory back. The engine's `.kglite.lock` is
/// the backstop against lock-free peers, not a permanent claim: once
/// `finish_generation` has rebased every mapping onto the published
/// generation, this handle needs nothing from the directory that a fresh
/// reader would not also take, so holding the lock past that point only
/// blocks other writers for as long as the process happens to live.
#[test]
fn a_published_disk_graph_leaves_its_directory_lockable() {
    let root = TempDir::new().unwrap();
    let root_path = root.path().to_str().unwrap();

    let mut graph = DirGraph::new();
    add_docs(&mut graph, &[1, 2]);
    graph.enable_disk_mode().unwrap();
    graph.save_disk(root_path).unwrap();

    // Mutate after the first save so the lease is genuinely held — and a
    // workspace minted — at the moment the second publish starts.
    add_docs(&mut graph, &[3]);
    assert!(
        GraphDirectoryLock::try_acquire(root.path()).is_err(),
        "a dirty writer must hold the directory against other writers"
    );

    graph.save_disk(root_path).unwrap();

    let lock = GraphDirectoryLock::try_acquire(root.path())
        .expect("a published disk graph must leave its directory lockable");
    drop(lock);
    let disk = match &graph.graph {
        GraphBackend::Disk(disk) => disk,
        _ => panic!("expected disk backend"),
    };
    assert!(
        disk.writer_lock.is_none(),
        "the publish must drop the lease"
    );
    assert!(disk.mutation_workspace.is_none());
}

/// Releasing at publish is only safe because taking is re-entrant: the next
/// mutation acquires the lease again and mints a *fresh* workspace, rather
/// than writing into the one the publish just removed.
#[test]
fn the_next_mutation_re_takes_the_lock_and_mints_a_new_workspace() {
    let root = TempDir::new().unwrap();
    let root_path = root.path().to_str().unwrap();

    let mut graph = DirGraph::new();
    add_docs(&mut graph, &[1, 2]);
    graph.enable_disk_mode().unwrap();
    graph.save_disk(root_path).unwrap();

    add_docs(&mut graph, &[3]);
    let pre_save_workspace = match &graph.graph {
        GraphBackend::Disk(disk) => disk.active_write_dir().to_path_buf(),
        _ => panic!("expected disk backend"),
    };
    graph.save_disk(root_path).unwrap();
    drop(
        GraphDirectoryLock::try_acquire(root.path())
            .expect("the publish releases the directory before the next mutation re-takes it"),
    );

    add_docs(&mut graph, &[4]);
    let disk = match &graph.graph {
        GraphBackend::Disk(disk) => disk,
        _ => panic!("expected disk backend"),
    };
    assert!(
        disk.writer_lock.is_some(),
        "the mutation after a publish must re-take the lease"
    );
    assert!(disk.mutation_workspace.is_some());
    let workspace = disk.active_write_dir().to_path_buf();
    assert_ne!(
        workspace, pre_save_workspace,
        "the removed workspace must not be reused"
    );
    let workspace_root = workspace.parent().unwrap();
    assert_eq!(workspace_root.parent(), Some(root.path()));
    assert!(workspace_root
        .file_name()
        .unwrap()
        .to_string_lossy()
        .starts_with(".working-"));
    assert!(
        GraphDirectoryLock::try_acquire(root.path()).is_err(),
        "the re-taken lease must hold the directory again"
    );
}

/// Two transactions opened on a *clean* graph — one that has published and
/// therefore holds no lease — must not lock each other out. They are one
/// writer lineage; whichever writes first takes the lease and the other finds
/// it through the shared slot. Serialization between them is the commit-time
/// version check, not the directory lock.
#[test]
fn sibling_transactions_after_a_save_share_one_lease() {
    let root = TempDir::new().unwrap();
    let root_path = root.path().to_str().unwrap();

    let mut graph = DirGraph::new();
    add_docs(&mut graph, &[1]);
    graph.enable_disk_mode().unwrap();
    graph.save_disk(root_path).unwrap();

    let parent = match &graph.graph {
        GraphBackend::Disk(disk) => disk,
        _ => panic!("expected disk backend"),
    };
    let mut winner = parent.clone();
    winner.adopt_writer_lineage(parent);
    let mut loser = parent.clone();
    loser.adopt_writer_lineage(parent);

    winner.prepare_mutation().unwrap();
    loser
        .prepare_mutation()
        .expect("a sibling transaction must re-join the lease, not fight it");
    assert!(
        Arc::ptr_eq(
            winner.writer_lock.as_ref().unwrap(),
            loser.writer_lock.as_ref().unwrap()
        ),
        "siblings must share one lease"
    );
    assert_ne!(
        winner.active_write_dir(),
        loser.active_write_dir(),
        "a shared lease still means private workspaces"
    );
}

/// A snapshot taken while the graph was dirty must not lock its own writer
/// out. The copy-on-write fork behind `make_dir_graph_mut_preserving_lineage`
/// leaves the *old* handle holding the lease, so after the live handle
/// publishes and releases its own reference, the OS lock is still held — by
/// this process. Re-acquiring it blindly refuses the writer's own next
/// mutation; the writer must re-join the lease it published under instead.
///
/// Reproduced through the binding as: create disk graph, write, `freeze()`,
/// write, `save()`, write → "already has an active writer".
#[test]
fn a_snapshot_taken_while_dirty_does_not_lock_out_the_writer() {
    let root = TempDir::new().unwrap();
    let root_path = root.path().to_str().unwrap();

    let mut graph = DirGraph::new();
    add_docs(&mut graph, &[1]);
    graph.enable_disk_mode().unwrap();
    graph.save_disk(root_path).unwrap();
    // Dirty when the snapshot is taken: the lease is live and the snapshot
    // inherits the handle that owns it.
    add_docs(&mut graph, &[2]);

    let mut handle = Arc::new(graph);
    let snapshot = Arc::clone(&handle);
    let live = crate::graph::handle::make_dir_graph_mut_preserving_lineage(&mut handle);

    add_docs(live, &[3]);
    live.save_disk(root_path).unwrap();

    add_docs(live, &[4]);
    live.save_disk(root_path).unwrap();

    drop(snapshot);
    let live = crate::graph::handle::make_dir_graph_mut_preserving_lineage(&mut handle);
    add_docs(live, &[5]);
    live.save_disk(root_path).unwrap();
    GraphDirectoryLock::try_acquire(root.path())
        .expect("with the snapshot gone the directory is free again");
}

/// A save releases the directory lease, and the graph's very next mutation
/// re-takes it. Nothing else is writing these directories, so a `WouldBlock`
/// there is the handle losing a race against its own just-released lock.
///
/// The subprocess spawner is the reproducer, not decoration: `flock`
/// ownership belongs to the open file description, a `fork`/`posix_spawn`
/// child inherits a copy of every descriptor and drops the `O_CLOEXEC` ones
/// only at `exec`, and a lease released by closing its descriptor inside that
/// window stays locked until the child gets there. Without a process being
/// spawned alongside, thousands of save/re-take rounds pass and this test
/// proves nothing. Unix-only for the same reason: Windows handles are
/// inherited only when explicitly marked inheritable, which these are not.
#[cfg(unix)]
#[test]
fn a_save_does_not_lock_the_writer_out_of_its_next_mutation() {
    let failures = Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
    let stop = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let spawner = {
        let stop = Arc::clone(&stop);
        std::thread::spawn(move || {
            while !stop.load(std::sync::atomic::Ordering::Relaxed) {
                let mut child = std::process::Command::new("true").spawn().unwrap();
                child.wait().unwrap();
            }
        })
    };
    let mut workers = Vec::new();
    for worker in 0..4 {
        let failures = Arc::clone(&failures);
        workers.push(std::thread::spawn(move || {
            for round in 0..15 {
                let root = TempDir::new().unwrap();
                let mut graph = DirGraph::new();
                add_docs(&mut graph, &[1, 2]);
                graph.enable_disk_mode().unwrap();
                graph.save_disk(root.path().to_str().unwrap()).unwrap();
                let GraphBackend::Disk(disk) = &mut graph.graph else {
                    panic!("expected disk backend");
                };
                if let Err(error) = disk.build_property_index("Doc", "title") {
                    failures
                        .lock()
                        .unwrap()
                        .push(format!("worker {worker} round {round}: {error}"));
                }
            }
        }));
    }
    for worker in workers {
        worker.join().unwrap();
    }
    stop.store(true, std::sync::atomic::Ordering::Relaxed);
    spawner.join().unwrap();
    let failures = failures.lock().unwrap();
    assert!(failures.is_empty(), "{failures:#?}");
}

/// The lease is an `Arc`, and `adopt_writer_lineage` clones it into a
/// transaction fork. A parent's publish therefore drops only *its own*
/// reference: the OS lock stays held for as long as the fork — which may
/// still write into its private workspace — is alive.
#[test]
fn a_live_fork_keeps_the_lock_held_across_the_parents_save() {
    let root = TempDir::new().unwrap();
    let root_path = root.path().to_str().unwrap();

    let mut graph = DirGraph::new();
    add_docs(&mut graph, &[1]);
    graph.enable_disk_mode().unwrap();
    graph.save_disk(root_path).unwrap();
    add_docs(&mut graph, &[2]);

    let mut fork = match &graph.graph {
        GraphBackend::Disk(disk) => {
            let mut fork = (**disk).clone();
            fork.adopt_writer_lineage(disk);
            fork
        }
        _ => panic!("expected disk backend"),
    };
    assert!(fork.writer_lock.is_some(), "the fork inherits the lease");
    fork.prepare_mutation().unwrap();

    graph.save_disk(root_path).unwrap();
    assert!(
        GraphDirectoryLock::try_acquire(root.path()).is_err(),
        "a live fork's cloned lease must keep the directory locked"
    );

    drop(fork);
    GraphDirectoryLock::try_acquire(root.path())
        .expect("the last lease reference dropping must release the directory");
}

#[test]
fn independent_copy_uses_lazy_private_root_and_rebases_on_save() {
    let source = TempDir::new().unwrap();
    let destination = TempDir::new().unwrap();
    let source_path = source.path().to_str().unwrap();
    let destination_path = destination.path().to_str().unwrap();

    let mut writer = DirGraph::new();
    add_docs(&mut writer, &[1, 2]);
    writer.enable_disk_mode().unwrap();
    writer.save_disk(source_path).unwrap();
    let frozen_source = snapshot_files(source.path());

    let mut copy = writer.independent_copy();
    let private_root = match &copy.graph {
        GraphBackend::Disk(disk) => disk.independent_root_path().unwrap().to_path_buf(),
        _ => panic!("expected disk backend"),
    };
    assert!(!private_root.exists(), "copy roots must be lazy");

    add_docs(&mut copy, &[3]);
    assert!(private_root.exists());
    assert_eq!(writer.graph.node_count(), 2);
    assert_eq!(copy.graph.node_count(), 3);
    assert_eq!(
        snapshot_files(source.path()),
        frozen_source,
        "copy mutations must not write into the selected source generation"
    );

    copy.save_disk(destination_path).unwrap();
    assert!(
        !private_root.exists(),
        "save-as must clean the scratch root"
    );
    match &copy.graph {
        GraphBackend::Disk(disk) => assert!(disk.independent_root_path().is_none()),
        _ => panic!("expected disk backend"),
    }

    let source_reader = crate::graph::io::file::load_file(source_path).unwrap();
    let copy_reader = crate::graph::io::file::load_file(destination_path).unwrap();
    assert_eq!(source_reader.graph.node_count(), 2);
    assert_eq!(copy_reader.graph.node_count(), 3);
    assert!(source_reader
        .lookup_by_id_readonly("Doc", &Value::Int64(3))
        .is_none());
    assert!(copy_reader
        .lookup_by_id_readonly("Doc", &Value::Int64(3))
        .is_some());

    // The source retains its own writer lease and can continue independently.
    add_docs(&mut writer, &[4]);
    writer.save_disk(source_path).unwrap();
    let newest_source = crate::graph::io::file::load_file(source_path).unwrap();
    let held_copy = crate::graph::io::file::load_file(destination_path).unwrap();
    assert!(newest_source
        .lookup_by_id_readonly("Doc", &Value::Int64(4))
        .is_some());
    assert!(held_copy
        .lookup_by_id_readonly("Doc", &Value::Int64(4))
        .is_none());
}

#[test]
fn dropping_unsaved_independent_copy_cleans_private_root() {
    let source = TempDir::new().unwrap();
    let mut writer = DirGraph::new();
    add_docs(&mut writer, &[1]);
    writer.enable_disk_mode().unwrap();
    writer.save_disk(source.path().to_str().unwrap()).unwrap();

    let mut copy = writer.independent_copy();
    let private_root = match &copy.graph {
        GraphBackend::Disk(disk) => disk.independent_root_path().unwrap().to_path_buf(),
        _ => panic!("expected disk backend"),
    };
    add_docs(&mut copy, &[2]);
    assert!(private_root.exists());
    drop(copy);
    assert!(!private_root.exists());
}

#[test]
fn independent_copy_retains_unsaved_parent_index_files() {
    let source = TempDir::new().unwrap();
    let destination = TempDir::new().unwrap();
    let mut writer = DirGraph::new();
    let frame = DataFrame::from_cypher_rows(
        vec!["id".into(), "title".into(), "tag".into()],
        vec![
            vec![
                Value::Int64(1),
                Value::String("one".into()),
                Value::String("a".into()),
            ],
            vec![
                Value::Int64(2),
                Value::String("two".into()),
                Value::String("b".into()),
            ],
        ],
    )
    .unwrap();
    crate::graph::mutation::maintain::add_nodes(
        &mut writer,
        frame,
        "Doc".to_string(),
        "id".to_string(),
        Some("title".to_string()),
        None,
    )
    .unwrap();
    writer.enable_disk_mode().unwrap();
    writer.save_disk(source.path().to_str().unwrap()).unwrap();
    match &mut writer.graph {
        GraphBackend::Disk(disk) => {
            assert_eq!(disk.build_property_index("Doc", "tag").unwrap(), 2);
        }
        _ => panic!("expected disk backend"),
    }

    let mut copy = writer.independent_copy();
    drop(writer);
    copy.save_disk(destination.path().to_str().unwrap())
        .unwrap();
    let reloaded = crate::graph::io::file::load_file(destination.path().to_str().unwrap()).unwrap();
    match &reloaded.graph {
        GraphBackend::Disk(disk) => assert!(disk.has_property_index("Doc", "tag")),
        _ => panic!("expected disk backend"),
    }
}

#[test]
fn arc_copy_on_write_retains_disk_writer_lineage() {
    let source = TempDir::new().unwrap();
    let path = source.path().to_str().unwrap();
    let mut writer = DirGraph::new();
    add_docs(&mut writer, &[1]);
    writer.enable_disk_mode().unwrap();
    writer.save_disk(path).unwrap();

    let mut active = Arc::new(writer);
    let held_snapshot = Arc::clone(&active);
    let active_graph = crate::graph::handle::make_dir_graph_mut(&mut active);
    add_docs(active_graph, &[2]);
    assert_eq!(held_snapshot.graph.node_count(), 1);
    assert_eq!(active.graph.node_count(), 2);
    match &active.graph {
        GraphBackend::Disk(disk) => {
            assert!(disk.writer_lock.is_some());
            assert!(disk.mutation_workspace.is_some());
        }
        _ => panic!("expected disk backend"),
    }

    crate::graph::io::file::save_graph(&mut active, path).unwrap();
    assert_eq!(held_snapshot.graph.node_count(), 1);
    let reloaded = crate::graph::io::file::load_file(path).unwrap();
    assert_eq!(reloaded.graph.node_count(), 2);
}

#[test]
fn generation_publish_keeps_held_reader_on_old_snapshot() {
    let target = TempDir::new().unwrap();
    let path = target.path().to_str().unwrap();
    let mut writer = DirGraph::new();
    add_docs(&mut writer, &[1, 2]);
    writer.enable_disk_mode().unwrap();
    writer.save_disk(path).unwrap();
    let first_current = std::fs::read_to_string(target.path().join("CURRENT")).unwrap();
    let first_snapshot = crate::graph::storage::disk::generation::resolve_snapshot(target.path())
        .unwrap()
        .snapshot_dir;
    let frozen_slots = std::fs::read(first_snapshot.join("seg_000/node_slots.bin")).unwrap();
    let held_reader = crate::graph::io::file::load_file(path).unwrap();
    assert_eq!(held_reader.graph.node_count(), 2);

    add_docs(&mut writer, &[3]);
    assert_eq!(
        std::fs::read(first_snapshot.join("seg_000/node_slots.bin")).unwrap(),
        frozen_slots,
        "creating a mutation overlay must not write the selected generation"
    );
    writer.save_disk(path).unwrap();
    let second_current = std::fs::read_to_string(target.path().join("CURRENT")).unwrap();
    assert_ne!(first_current, second_current);

    let newest = crate::graph::io::file::load_file(path).unwrap();
    assert_eq!(newest.graph.node_count(), 3);
    assert_eq!(
        held_reader.graph.node_count(),
        2,
        "a reader that resolved generation 1 must remain on generation 1"
    );
}

#[test]
fn generation_preserves_edge_property_snapshots_and_rebases_writer() {
    let target = TempDir::new().unwrap();
    let path = target.path().to_str().unwrap();
    let mut writer = DirGraph::new();
    add_docs(&mut writer, &[1, 2]);
    writer.enable_disk_mode().unwrap();
    writer.prepare_mutation().unwrap();
    let source = writer
        .lookup_by_id_readonly("Doc", &Value::Int64(1))
        .unwrap();
    let target_node = writer
        .lookup_by_id_readonly("Doc", &Value::Int64(2))
        .unwrap();
    let edge = EdgeData::new(
        "LINKS".to_string(),
        HashMap::from([("score".to_string(), Value::Int64(1))]),
        &mut writer.interner,
    );
    GraphWrite::add_edge(&mut writer.graph, source, target_node, edge);
    writer.save_disk(path).unwrap();

    let first_snapshot = crate::graph::storage::disk::generation::resolve_snapshot(target.path())
        .unwrap()
        .snapshot_dir;
    let frozen_tree = snapshot_files(&first_snapshot);
    let held_reader = crate::graph::io::file::load_file(path).unwrap();
    assert_eq!(edge_score(&held_reader), Some(Value::Int64(1)));

    writer.prepare_mutation().unwrap();
    let edge = GraphWrite::edge_weight_mut(&mut writer.graph, EdgeIndex::new(0)).unwrap();
    edge.properties
        .iter_mut()
        .find(|(key, _)| *key == InternedKey::from_str("score"))
        .unwrap()
        .1 = Value::Int64(2);
    assert_eq!(snapshot_files(&first_snapshot), frozen_tree);

    writer.save_disk(path).unwrap();
    assert_eq!(snapshot_files(&first_snapshot), frozen_tree);
    assert_eq!(edge_score(&writer), Some(Value::Int64(2)));
    let newest = crate::graph::io::file::load_file(path).unwrap();
    assert_eq!(edge_score(&newest), Some(Value::Int64(2)));
    assert_eq!(edge_score(&held_reader), Some(Value::Int64(1)));
}

#[test]
fn generation_round_trips_node_and_edge_add_delete_overlays() {
    let target = TempDir::new().unwrap();
    let path = target.path().to_str().unwrap();
    let mut writer = DirGraph::new();
    add_docs(&mut writer, &[1, 2, 3]);
    writer.enable_disk_mode().unwrap();
    writer.prepare_mutation().unwrap();
    let nodes: Vec<_> = [1, 2, 3]
        .into_iter()
        .map(|id| {
            writer
                .lookup_by_id_readonly("Doc", &Value::Int64(id))
                .unwrap()
        })
        .collect();
    for pair in nodes.windows(2) {
        let edge = EdgeData::new("LINKS".to_string(), HashMap::new(), &mut writer.interner);
        GraphWrite::add_edge(&mut writer.graph, pair[0], pair[1], edge);
    }
    writer.save_disk(path).unwrap();
    let first_snapshot = crate::graph::storage::disk::generation::resolve_snapshot(target.path())
        .unwrap()
        .snapshot_dir;
    let frozen_tree = snapshot_files(&first_snapshot);
    let held_reader = crate::graph::io::file::load_file(path).unwrap();

    writer.prepare_mutation().unwrap();
    GraphWrite::remove_edge(&mut writer.graph, EdgeIndex::new(0)).unwrap();
    crate::graph::mutation::maintain::detach_delete_nodes(
        &mut writer,
        &std::collections::HashSet::from([nodes[2]]),
    );
    add_docs(&mut writer, &[4]);
    let fourth = writer
        .lookup_by_id_readonly("Doc", &Value::Int64(4))
        .unwrap();
    let edge = EdgeData::new("LINKS".to_string(), HashMap::new(), &mut writer.interner);
    GraphWrite::add_edge(&mut writer.graph, nodes[0], fourth, edge);
    assert_eq!(snapshot_files(&first_snapshot), frozen_tree);

    writer.save_disk(path).unwrap();
    assert_eq!(snapshot_files(&first_snapshot), frozen_tree);
    let newest = crate::graph::io::file::load_file(path).unwrap();
    assert_eq!(newest.graph.node_count(), 3);
    assert_eq!(newest.graph.edge_count(), 1);
    let only_edge = newest.graph.edge_references().next().unwrap();
    assert_eq!((only_edge.source(), only_edge.target()), (nodes[0], fourth));
    assert_eq!(held_reader.graph.node_count(), 3);
    assert_eq!(held_reader.graph.edge_count(), 2);
}

/// A refused write must be refused *whole*: the overlay is unchanged, so the
/// caller can retry the same statement once the directory frees up.
///
/// The lease covers the first writer's unpublished window — the interval in
/// which a second writer's publish would land on top of changes the first is
/// still holding — so `first` is deliberately left dirty here. Its `save_disk`
/// then releases the directory and the retry succeeds without `first` being
/// dropped.
#[test]
fn second_disk_writer_is_rejected_before_mutation() {
    let target = TempDir::new().unwrap();
    let path = target.path().to_str().unwrap();
    let mut first = DirGraph::new();
    add_docs(&mut first, &[1]);
    first.enable_disk_mode().unwrap();
    first.save_disk(path).unwrap();

    let loaded = crate::graph::io::file::load_file(path).unwrap();
    let mut second = match std::sync::Arc::try_unwrap(loaded) {
        Ok(graph) => graph,
        Err(_) => panic!("fresh load unexpectedly had another Arc owner"),
    };
    add_docs(&mut first, &[3]);
    let error = crate::graph::mutation::maintain::add_nodes(
        &mut second,
        one_doc_frame(2),
        "Doc".to_string(),
        "id".to_string(),
        Some("title".to_string()),
        None,
    )
    .expect_err("second writer must fail before changing its overlay");
    assert!(error.contains("active writer"), "{error}");
    assert_eq!(second.graph.node_count(), 1);

    first.save_disk(path).unwrap();
    crate::graph::mutation::maintain::add_nodes(
        &mut second,
        one_doc_frame(2),
        "Doc".to_string(),
        "id".to_string(),
        Some("title".to_string()),
        None,
    )
    .unwrap();
    assert_eq!(second.graph.node_count(), 2);
}

#[test]
fn failed_graph_save_withholds_root_metadata_and_stays_retryable() {
    // `target` must be declared before `graph`: reverse-declaration drop
    // order then destroys the graph (and its mappings into `target`) first.
    // `TempGraphDir` asserts that ordering instead of trusting it — on Unix a
    // still-mapped file can be unlinked without complaint.
    let target = TempGraphDir::new();
    let mut graph = TrackedOwner::new("disk-mode DirGraph", DirGraph::new());
    target.watch(&graph);
    graph.enable_disk_mode().unwrap();
    let blocked_generations = target.path().join("generations");
    std::fs::write(&blocked_generations, b"not a directory").unwrap();

    let error = graph
        .save_disk(target.path().to_str().unwrap())
        .expect_err("a blocked generations directory must fail the save");
    assert!(error.contains("Failed to begin disk generation"));
    assert!(!target.path().join("CURRENT").exists());
    assert!(
        matches!(&graph.graph, GraphBackend::Disk(_)),
        "a failed save must not knock the graph out of disk mode"
    );

    std::fs::remove_file(blocked_generations).unwrap();
    graph.save_disk(target.path().to_str().unwrap()).unwrap();
    let snapshot =
        crate::graph::storage::disk::generation::resolve_snapshot(target.path()).unwrap();
    assert!(snapshot.snapshot_dir.join("metadata.json").exists());
}

fn seg(
    node_slots: Vec<DiskNodeSlot>,
    out_offsets: Vec<u64>,
    out_edges: Vec<CsrEdge>,
    in_offsets: Vec<u64>,
    in_edges: Vec<CsrEdge>,
    edge_endpoints: Vec<EdgeEndpoints>,
) -> SegmentCsr {
    SegmentCsr {
        node_slots: from_vec(node_slots),
        out_offsets: from_vec(out_offsets),
        out_edges: from_vec(out_edges),
        in_offsets: from_vec(in_offsets),
        in_edges: from_vec(in_edges),
        edge_endpoints: from_vec(edge_endpoints),
        // Empty in these CSR-only tests; the auxiliary-index tests
        // populate them explicitly.
        conn_type_index_types: MmapOrVec::new(),
        conn_type_index_offsets: MmapOrVec::new(),
        conn_type_index_sources: MmapOrVec::new(),
        peer_count_types: MmapOrVec::new(),
        peer_count_offsets: MmapOrVec::new(),
        peer_count_entries: MmapOrVec::new(),
    }
}

fn from_vec<T: crate::graph::storage::mapped::mmap_vec::MmapPod>(v: Vec<T>) -> MmapOrVec<T> {
    let mut m: MmapOrVec<T> = MmapOrVec::with_capacity(v.len());
    for x in v {
        m.push(x);
    }
    m
}

fn slot(node_type: u64, row_id: u32) -> DiskNodeSlot {
    DiskNodeSlot {
        node_type,
        row_id,
        flags: DiskNodeSlot::ALIVE_BIT,
    }
}

#[test]
fn overlapping_query_guards_keep_materializations_alive() {
    let tmp = TempDir::new().unwrap();
    let mut graph = super::DiskGraph::new_at_path(tmp.path()).unwrap();
    let mut interner = StringInterner::new();
    let a = graph.add_node(seal_test_node(&mut interner, 0, "Item"));
    let b = graph.add_node(seal_test_node(&mut interner, 1, "Item"));
    let edge = graph.add_edge(a, b, seal_test_edge(&mut interner, "LINKS"));

    let first = graph.begin_query();
    assert!(graph.node_weight(a).is_some());
    graph.materialize_edge(edge.index() as u32);
    assert_eq!(graph.node_arena_len(), 1);
    assert_eq!(graph.edge_arena_len(), 1);

    let second = graph.begin_query();
    assert_eq!(graph.active_query_count(), 2);
    assert_eq!(graph.node_arena_len(), 1);
    assert_eq!(graph.edge_arena_len(), 1);

    // A reset from another execution path must not invalidate either guard.
    graph.reset_arenas();
    assert_eq!(graph.node_arena_len(), 1);
    assert_eq!(graph.edge_arena_len(), 1);

    // Nor may a *younger* query's completion reclaim what an older, still
    // running query materialized.
    drop(second);
    assert_eq!(graph.active_query_count(), 1);
    assert_eq!(graph.node_arena_len(), 1);
    assert_eq!(graph.edge_arena_len(), 1);

    // The materializing query finishing does reclaim them — without waiting
    // for the graph to go completely idle, which is what makes the arena
    // bounded under sustained overlapping reads.
    let third = graph.begin_query();
    drop(first);
    assert_eq!(graph.active_query_count(), 1);
    assert_eq!(graph.node_arena_len(), 0);
    assert_eq!(graph.edge_arena_len(), 0);
    drop(third);
}

#[test]
fn sustained_overlapping_reads_do_not_grow_the_arena() {
    const NODES: usize = 100;
    const ROUNDS: usize = 50;

    let tmp = TempDir::new().unwrap();
    let mut graph = super::DiskGraph::new_at_path(tmp.path()).unwrap();
    let mut interner = StringInterner::new();
    for i in 0..NODES {
        graph.add_node(seal_test_node(&mut interner, i as i64, "Item"));
    }

    // Guards overlap by construction: every round opens the next query before
    // closing the previous one, so the active-query count never reaches zero —
    // the state a concurrently served disk graph lives in permanently. Each
    // round's materializations are dead as soon as its own query ends, so a
    // bounded arena must reclaim them without waiting for global quiescence.
    let mut prev = Some(graph.begin_query());
    for _ in 0..ROUNDS {
        let next = graph.begin_query();
        for i in 0..NODES {
            assert!(graph.node_weight(NodeIndex::new(i)).is_some());
        }
        drop(prev.take());
        prev = Some(next);
    }
    let retained = graph.node_arena_len();
    drop(prev);

    let bound = 3 * NODES;
    assert!(
        retained <= bound,
        "node arena retained {retained} records after {ROUNDS} overlapping rounds \
         of {NODES} materializations (bound {bound}) — arena reclamation is not \
         reachable under sustained concurrent reads"
    );
}

// ------------- segment_subdir + enumerate -------------

#[test]
fn segment_subdir_zero_pads_three_digits() {
    assert_eq!(segment_subdir(0), "seg_000");
    assert_eq!(segment_subdir(1), "seg_001");
    assert_eq!(segment_subdir(42), "seg_042");
    assert_eq!(segment_subdir(999), "seg_999");
    // Past 999 the name widens; enumerate sorts by parsed u32 so
    // this still round-trips cleanly.
    assert_eq!(segment_subdir(1234), "seg_1234");
}

#[test]
fn enumerate_segment_dirs_returns_sorted_ids() {
    let tmp = TempDir::new().unwrap();
    for id in [5u32, 0, 2, 17] {
        std::fs::create_dir_all(tmp.path().join(segment_subdir(id))).unwrap();
    }
    let got: Vec<u32> = enumerate_segment_dirs(tmp.path())
        .into_iter()
        .map(|(id, _)| id)
        .collect();
    assert_eq!(got, vec![0, 2, 5, 17]);
}

#[test]
fn enumerate_segment_dirs_skips_non_matching_entries() {
    let tmp = TempDir::new().unwrap();
    std::fs::create_dir_all(tmp.path().join("seg_000")).unwrap();
    std::fs::create_dir_all(tmp.path().join("seg_abc")).unwrap(); // unparsable
    std::fs::create_dir_all(tmp.path().join("not_a_segment")).unwrap();
    // Top-level files must not be mistaken for segments.
    std::fs::write(tmp.path().join("seg_001"), b"not-a-dir").unwrap();
    std::fs::write(tmp.path().join("disk_graph_meta.json"), b"{}").unwrap();

    let got: Vec<u32> = enumerate_segment_dirs(tmp.path())
        .into_iter()
        .map(|(id, _)| id)
        .collect();
    assert_eq!(got, vec![0]);
}

#[test]
fn enumerate_segment_dirs_on_missing_dir_returns_empty() {
    let tmp = TempDir::new().unwrap();
    let missing = tmp.path().join("does-not-exist");
    assert!(enumerate_segment_dirs(&missing).is_empty());
}

#[test]
fn enumerate_segment_dirs_empty_root_returns_empty() {
    let tmp = TempDir::new().unwrap();
    assert!(enumerate_segment_dirs(tmp.path()).is_empty());
}

// ------------- concat_segment_csrs -------------

#[test]
fn concat_empty_input_returns_all_empty() {
    let c = concat_segment_csrs(Vec::new()).unwrap();
    assert_eq!(c.node_slots.len(), 0);
    assert_eq!(c.out_offsets.len(), 0);
    assert_eq!(c.out_edges.len(), 0);
    assert_eq!(c.in_offsets.len(), 0);
    assert_eq!(c.in_edges.len(), 0);
    assert_eq!(c.edge_endpoints.len(), 0);
}

#[test]
fn concat_single_segment_returns_it_unchanged() {
    // Node 0 → node 1 (edge_idx 0). Two nodes, one edge: the minimal
    // shape for comparison — passthrough must not mutate anything.
    let s = seg(
        vec![slot(7, 100), slot(7, 101)],
        vec![0, 1, 1], // out_offsets: node 0 emits edge [0,1), node 1 emits nothing
        vec![CsrEdge {
            peer: 1,
            edge_idx: 0,
        }],
        vec![0, 0, 1], // in_offsets: node 0 receives nothing, node 1 receives [0,1)
        vec![CsrEdge {
            peer: 0,
            edge_idx: 0,
        }],
        vec![EdgeEndpoints {
            source: 0,
            target: 1,
            connection_type: 42,
        }],
    );
    let c = concat_segment_csrs(vec![s]).unwrap();
    assert_eq!(c.node_slots.len(), 2);
    assert_eq!(c.out_offsets.len(), 3);
    assert_eq!(c.out_edges.len(), 1);
    assert_eq!(c.out_edges.get(0).edge_idx, 0);
    assert_eq!(c.edge_endpoints.len(), 1);
    assert_eq!(c.edge_endpoints.get(0).source, 0);
}

#[test]
fn concat_two_segments_stitches_offsets_and_shifts_edge_idx() {
    // Segment 0: 2 nodes, 1 intra-segment edge  0 → 1
    let s0 = seg(
        vec![slot(1, 10), slot(1, 11)],
        vec![0, 1, 1],
        vec![CsrEdge {
            peer: 1,
            edge_idx: 0,
        }],
        vec![0, 0, 1],
        vec![CsrEdge {
            peer: 0,
            edge_idx: 0,
        }],
        vec![EdgeEndpoints {
            source: 0,
            target: 1,
            connection_type: 100,
        }],
    );
    // Segment 1: 2 nodes (global ids 2, 3), 2 intra-segment edges
    // 2 → 3 (segment-local edge_idx 0) and 3 → 2 (segment-local 1).
    let s1 = seg(
        vec![slot(2, 20), slot(2, 21)],
        vec![0, 1, 2],
        vec![
            CsrEdge {
                peer: 3,
                edge_idx: 0,
            },
            CsrEdge {
                peer: 2,
                edge_idx: 1,
            },
        ],
        vec![0, 1, 2],
        vec![
            CsrEdge {
                peer: 3,
                edge_idx: 1,
            },
            CsrEdge {
                peer: 2,
                edge_idx: 0,
            },
        ],
        vec![
            EdgeEndpoints {
                source: 2,
                target: 3,
                connection_type: 200,
            },
            EdgeEndpoints {
                source: 3,
                target: 2,
                connection_type: 201,
            },
        ],
    );

    let c = concat_segment_csrs(vec![s0, s1]).unwrap();

    assert_eq!(c.node_slots.len(), 4);
    assert_eq!(c.out_offsets.len(), 5); // n+1
    assert_eq!(c.in_offsets.len(), 5);
    assert_eq!(c.out_edges.len(), 3);
    assert_eq!(c.in_edges.len(), 3);
    assert_eq!(c.edge_endpoints.len(), 3);

    // Stitched out_offsets: [0, 1, 1, 2, 3] — seg 0 contributes
    // [0,1,1]; seg 1 contributes [+1, +2] atop seg 0's last (=1),
    // so combined ends [..., 2, 3].
    let out_off: Vec<u64> = (0..c.out_offsets.len())
        .map(|i| c.out_offsets.get(i))
        .collect();
    assert_eq!(out_off, vec![0, 1, 1, 2, 3]);

    // Stitched in_offsets: [0, 0, 1, 2, 3] — seg 0 [0,0,1]; seg 1
    // contributes [+1, +2].
    let in_off: Vec<u64> = (0..c.in_offsets.len())
        .map(|i| c.in_offsets.get(i))
        .collect();
    assert_eq!(in_off, vec![0, 0, 1, 2, 3]);

    // out_edges[0] comes from seg 0, edge_idx unchanged (0).
    // out_edges[1..3] come from seg 1, edge_idx shifted by seg 0's
    // edge_endpoints.len() == 1 → (1, 2).
    assert_eq!(c.out_edges.get(0).edge_idx, 0);
    assert_eq!(c.out_edges.get(0).peer, 1);
    assert_eq!(c.out_edges.get(1).edge_idx, 1);
    assert_eq!(c.out_edges.get(1).peer, 3);
    assert_eq!(c.out_edges.get(2).edge_idx, 2);
    assert_eq!(c.out_edges.get(2).peer, 2);

    // in_edges shifts follow the same rule.
    assert_eq!(c.in_edges.get(0).edge_idx, 0); // seg 0, unchanged
    assert_eq!(c.in_edges.get(1).edge_idx, 2); // seg 1, +1
    assert_eq!(c.in_edges.get(2).edge_idx, 1); // seg 1, +1

    // edge_endpoints concat — source/target are global node ids.
    assert_eq!(c.edge_endpoints.get(0).source, 0);
    assert_eq!(c.edge_endpoints.get(0).target, 1);
    assert_eq!(c.edge_endpoints.get(1).source, 2);
    assert_eq!(c.edge_endpoints.get(1).target, 3);
    assert_eq!(c.edge_endpoints.get(2).source, 3);
    assert_eq!(c.edge_endpoints.get(2).target, 2);
}

#[test]
fn concat_three_segments_keeps_offset_chain_consistent() {
    // Three one-node-one-self-edge segments.
    let mk_one_node = |global_id: u32, conn: u64| {
        seg(
            vec![slot(1, 0)],
            vec![0, 1],
            vec![CsrEdge {
                peer: global_id,
                edge_idx: 0,
            }],
            vec![0, 1],
            vec![CsrEdge {
                peer: global_id,
                edge_idx: 0,
            }],
            vec![EdgeEndpoints {
                source: global_id,
                target: global_id,
                connection_type: conn,
            }],
        )
    };
    let c = concat_segment_csrs(vec![
        mk_one_node(0, 10),
        mk_one_node(1, 20),
        mk_one_node(2, 30),
    ])
    .unwrap();

    let out_off: Vec<u64> = (0..c.out_offsets.len())
        .map(|i| c.out_offsets.get(i))
        .collect();
    assert_eq!(out_off, vec![0, 1, 2, 3]);

    // Each out_edges entry's edge_idx should point at its own
    // self-loop's endpoint in the combined array — segment K's
    // endpoint lands at index K.
    assert_eq!(c.out_edges.get(0).edge_idx, 0);
    assert_eq!(c.out_edges.get(1).edge_idx, 1);
    assert_eq!(c.out_edges.get(2).edge_idx, 2);

    // The endpoint at edge_idx K should be the self-loop of node K.
    for k in 0..3 {
        assert_eq!(c.edge_endpoints.get(k).source, k as u32);
        assert_eq!(c.edge_endpoints.get(k).target, k as u32);
    }
}

#[test]
fn concat_handles_edgeless_segment() {
    let s0 = seg(
        vec![slot(1, 0)],
        vec![0, 1],
        vec![CsrEdge {
            peer: 0,
            edge_idx: 0,
        }],
        vec![0, 1],
        vec![CsrEdge {
            peer: 0,
            edge_idx: 0,
        }],
        vec![EdgeEndpoints {
            source: 0,
            target: 0,
            connection_type: 1,
        }],
    );
    // Middle segment: nodes but no edges, e.g. one freshly created.
    let s_empty = seg(
        vec![slot(1, 1)],
        vec![0, 0],
        Vec::new(),
        vec![0, 0],
        Vec::new(),
        Vec::new(),
    );
    let s1 = seg(
        vec![slot(1, 2)],
        vec![0, 1],
        vec![CsrEdge {
            peer: 2,
            edge_idx: 0,
        }],
        vec![0, 1],
        vec![CsrEdge {
            peer: 2,
            edge_idx: 0,
        }],
        vec![EdgeEndpoints {
            source: 2,
            target: 2,
            connection_type: 3,
        }],
    );
    let c = concat_segment_csrs(vec![s0, s_empty, s1]).unwrap();
    let out_off: Vec<u64> = (0..c.out_offsets.len())
        .map(|i| c.out_offsets.get(i))
        .collect();
    // 3 nodes total; middle node contributes no edges.
    assert_eq!(out_off, vec![0, 1, 1, 2]);
    assert_eq!(c.out_edges.len(), 2);
    // Segment 2 (index 2 in the input) had endpoint_base of
    // s0.edge_endpoints.len() + s_empty.edge_endpoints.len() == 1,
    // so its self-loop's edge_idx should now be 1.
    assert_eq!(c.out_edges.get(1).edge_idx, 1);
}

// ------------- seal_to_new_segment round-trip -------------

fn seal_test_node(interner: &mut StringInterner, id: i64, ntype: &str) -> NodeData {
    NodeData::new(
        Value::Int64(id),
        Value::String(format!("n{id}")),
        ntype.to_string(),
        std::collections::HashMap::new(),
        interner,
    )
}

fn seal_test_edge(interner: &mut StringInterner, ct: &str) -> EdgeData {
    EdgeData::new(ct.to_string(), std::collections::HashMap::new(), interner)
}

#[test]
fn seal_rejects_when_nothing_to_seal() {
    let tmp = TempDir::new().unwrap();
    let mut interner = StringInterner::new();
    let mut dg = super::DiskGraph::new_at_path(tmp.path()).unwrap();
    dg.defer_csr = true;
    let _n0 = dg.add_node(seal_test_node(&mut interner, 0, "A"));
    dg.build_csr_from_pending().unwrap();
    dg.save_to_dir(tmp.path(), &interner).unwrap();
    // save_to_dir set sealed_nodes_bound = node_count, so tail is empty.
    let err = dg.seal_to_new_segment(tmp.path()).unwrap_err();
    assert!(err.to_string().contains("nothing to seal"));
}

#[test]
fn seal_accepts_cross_segment_edges_via_full_range() {
    // Cross-segment overflow is legal: the sealed segment writes
    // full-range out_offsets (indexed by global node id) so an edge
    // from a seg_0 source into a tail target — or between two seg_0
    // sources — is reachable after reload.
    let tmp = TempDir::new().unwrap();
    let mut interner = StringInterner::new();
    let mut dg = super::DiskGraph::new_at_path(tmp.path()).unwrap();
    dg.defer_csr = true;
    let n0 = dg.add_node(seal_test_node(&mut interner, 0, "A"));
    let n1 = dg.add_node(seal_test_node(&mut interner, 1, "A"));
    dg.add_edge(n0, n1, seal_test_edge(&mut interner, "T"));
    dg.build_csr_from_pending().unwrap();
    dg.save_to_dir(tmp.path(), &interner).unwrap();

    let n2 = dg.add_node(seal_test_node(&mut interner, 2, "A"));
    dg.add_edge(n0, n2, seal_test_edge(&mut interner, "T"));

    let seg_id = dg.seal_to_new_segment(tmp.path()).unwrap();
    assert_eq!(seg_id, 1);

    // Full-range so concat can locate the n0→n2 edge.
    let out_offsets_size = std::fs::metadata(tmp.path().join("seg_001/out_offsets.bin"))
        .unwrap()
        .len();
    assert_eq!(
        out_offsets_size,
        (3 + 1) * 8,
        "seg_001 must be full-range (4 u64 offsets covering 3 global nodes)"
    );

    drop(dg);
    let mut interner2 = StringInterner::new();
    let (reloaded, _tmp_zst) = super::DiskGraph::load_from_dir(tmp.path(), &mut interner2).unwrap();
    let start = reloaded.out_offsets.get(0) as usize;
    let end = reloaded.out_offsets.get(1) as usize;
    let peers: Vec<u32> = (start..end)
        .map(|i| reloaded.out_edges.get(i).peer)
        .collect();
    assert!(peers.contains(&1), "missing seg_0 edge n0→n1");
    assert!(peers.contains(&2), "missing seg_1 cross-segment edge n0→n2");
}

#[test]
fn seal_round_trip_basic_reads() {
    // Build seg_0: 3 nodes of type A with one edge between them.
    let tmp = TempDir::new().unwrap();
    let mut interner = StringInterner::new();
    let mut dg = super::DiskGraph::new_at_path(tmp.path()).unwrap();
    dg.defer_csr = true;
    let n0 = dg.add_node(seal_test_node(&mut interner, 0, "A"));
    let n1 = dg.add_node(seal_test_node(&mut interner, 1, "A"));
    let _n2 = dg.add_node(seal_test_node(&mut interner, 2, "A"));
    dg.add_edge(n0, n1, seal_test_edge(&mut interner, "T"));
    dg.build_csr_from_pending().unwrap();
    dg.save_to_dir(tmp.path(), &interner).unwrap();

    assert_eq!(dg.node_count, 3);
    assert_eq!(dg.sealed_nodes_bound, 3);

    // Both endpoints of the new B→B edge are strictly above the
    // watermark, so the seal constraint holds.
    let n3 = dg.add_node(seal_test_node(&mut interner, 3, "B"));
    let n4 = dg.add_node(seal_test_node(&mut interner, 4, "B"));
    dg.add_edge(n3, n4, seal_test_edge(&mut interner, "U"));

    let pre_edge_count = dg.edge_count;
    let pre_node_count = dg.node_count;
    assert_eq!(pre_node_count, 5);
    assert_eq!(pre_edge_count, 2);

    let seg_id = dg.seal_to_new_segment(tmp.path()).unwrap();
    assert_eq!(seg_id, 1);
    assert_eq!(dg.sealed_nodes_bound, 5);
    assert!(dg.overflow_out.is_empty());
    assert!(dg.overflow_in.is_empty());

    let seg1 = tmp.path().join("seg_001");
    for name in [
        "node_slots.bin",
        "out_offsets.bin",
        "out_edges.bin",
        "in_offsets.bin",
        "in_edges.bin",
        "edge_endpoints.bin",
    ] {
        assert!(seg1.join(name).exists(), "missing {name}");
    }
    let manifest = super::super::segment_summary::SegmentManifest::load_from(tmp.path()).unwrap();
    assert_eq!(manifest.len(), 2);
    assert_eq!(manifest.segments[1].segment_id, 1);
    assert_eq!(manifest.segments[1].node_id_lo, 3);
    assert_eq!(manifest.segments[1].node_id_hi, 5);
    assert_eq!(manifest.segments[1].edge_count, 1);

    // Reload exercises the concat read path.
    drop(dg);
    let mut interner2 = StringInterner::new();
    let (reloaded, _tmp_zst) = super::DiskGraph::load_from_dir(tmp.path(), &mut interner2).unwrap();

    assert_eq!(reloaded.node_count, pre_node_count);
    assert_eq!(reloaded.edge_count, pre_edge_count);
    assert_eq!(reloaded.sealed_nodes_bound, 5);

    // Untyped outgoing edges for node 3 should be exactly 1 (the
    // n3 → n4 edge). This verifies the concat stitched the
    // out_offsets correctly for the sealed tail.
    let n3_idx = 3usize;
    let start = reloaded.out_offsets.get(n3_idx) as usize;
    let end = reloaded.out_offsets.get(n3_idx + 1) as usize;
    assert_eq!(end - start, 1, "expected 1 outgoing edge for seg_1 node 3");
    let e = reloaded.out_edges.get(start);
    assert_eq!(e.peer, 4);
    // Combined edge_idx = segment-local 0 + endpoint_base (= seg_0's 1)
    // = 1. Verifies the concat shift lands at the right slot.
    assert_eq!(e.edge_idx, 1);
    // edge_endpoints at that global index should hold the original
    // global node ids.
    let ep = reloaded.edge_endpoints.get(1);
    assert_eq!(ep.source, 3);
    assert_eq!(ep.target, 4);

    // And seg_0's original edge (0 → 1) is still present at
    // combined edge_idx 0.
    let ep0 = reloaded.edge_endpoints.get(0);
    assert_eq!(ep0.source, 0);
    assert_eq!(ep0.target, 1);
}

#[test]
fn seal_round_trip_auxiliary_indexes() {
    // conn_type_index_*, peer_count_* and edge_properties must all
    // survive the seal → reload round-trip.
    let tmp = TempDir::new().unwrap();
    let mut interner = StringInterner::new();
    let mut dg = super::DiskGraph::new_at_path(tmp.path()).unwrap();
    dg.defer_csr = true;
    // seg_0: 3 nodes of type A with one T edge (0 → 1).
    let n0 = dg.add_node(seal_test_node(&mut interner, 0, "A"));
    let n1 = dg.add_node(seal_test_node(&mut interner, 1, "A"));
    let _n2 = dg.add_node(seal_test_node(&mut interner, 2, "A"));
    dg.add_edge(n0, n1, seal_test_edge(&mut interner, "T"));
    dg.build_csr_from_pending().unwrap();
    dg.save_to_dir(tmp.path(), &interner).unwrap();

    // Tail: 2 nodes of type B, 2 U-edges — one intra-tail (3→4), one
    // self-loop on 3. Attach a property on the self-loop to verify
    // edge_properties flushing on seal.
    let n3 = dg.add_node(seal_test_node(&mut interner, 3, "B"));
    let n4 = dg.add_node(seal_test_node(&mut interner, 4, "B"));
    dg.add_edge(n3, n4, seal_test_edge(&mut interner, "U"));
    let self_loop = {
        let weight_key = interner.get_or_intern("weight");
        let ed = crate::graph::schema::EdgeData {
            connection_type: interner.get_or_intern("U"),
            properties: vec![(weight_key, Value::Float64(2.5))],
        };
        dg.add_edge(n3, n3, ed)
    };

    dg.seal_to_new_segment(tmp.path()).unwrap();

    let seg1 = tmp.path().join("seg_001");
    for name in [
        "conn_type_index_types.bin",
        "conn_type_index_offsets.bin",
        "conn_type_index_sources.bin",
        "peer_count_types.bin",
        "peer_count_offsets.bin",
        "peer_count_entries.bin",
    ] {
        assert!(seg1.join(name).exists(), "sealed segment missing {name}");
    }

    drop(dg);
    let mut interner2 = StringInterner::new();
    let (reloaded, _tmp_zst) = super::DiskGraph::load_from_dir(tmp.path(), &mut interner2).unwrap();

    // `concat_segment_csrs::merge_conn_type_index` must cover both
    // T (from seg_0) and U (from seg_1).
    let t_key = interner2.get_or_intern("T").as_u64();
    let u_key = interner2.get_or_intern("U").as_u64();
    let cti_types: Vec<u64> = (0..reloaded.conn_type_index_types.len())
        .map(|i| reloaded.conn_type_index_types.get(i))
        .collect();
    assert!(
        cti_types.contains(&t_key),
        "T missing from merged conn_type_index"
    );
    assert!(
        cti_types.contains(&u_key),
        "U missing from merged conn_type_index"
    );

    // peer_count histogram for U should report node 4 as a target
    // once (from n3 → n4) and node 3 as a target once (self-loop).
    let u_counts = reloaded
        .lookup_peer_counts(u_key)
        .expect("U histogram present");
    assert_eq!(u_counts.get(&4), Some(&1));
    assert_eq!(u_counts.get(&3), Some(&1));

    // And T's histogram still has the seg_0 entry intact.
    let t_counts = reloaded
        .lookup_peer_counts(t_key)
        .expect("T histogram present");
    assert_eq!(t_counts.get(&1), Some(&1));

    // Edge properties on the self-loop should survive — combined
    // edge_idx matches the original global assignment
    // (seg_0's 1 edge + seg_1's local index), which equals the
    // self_loop edge_index we captured pre-seal.
    let weight_key = interner2.get_or_intern("weight");
    let weight = reloaded
        .edge_properties
        .get(self_loop.index() as u32)
        .expect("self_loop has props");
    let (k, v) = &weight.as_ref()[0];
    assert_eq!(*k, weight_key);
    assert_eq!(*v, Value::Float64(2.5));
}

#[test]
fn save_to_dir_auto_wires_seal_when_tail_is_clean() {
    // A second save after a clean-tail workload must dispatch to
    // `seal_to_new_segment`, not the compact-and-rewrite path.
    let tmp = TempDir::new().unwrap();
    let mut interner = StringInterner::new();
    let mut dg = super::DiskGraph::new_at_path(tmp.path()).unwrap();
    dg.defer_csr = true;
    let n0 = dg.add_node(seal_test_node(&mut interner, 0, "A"));
    let n1 = dg.add_node(seal_test_node(&mut interner, 1, "A"));
    dg.add_edge(n0, n1, seal_test_edge(&mut interner, "T"));
    dg.build_csr_from_pending().unwrap();
    dg.save_to_dir(tmp.path(), &interner).unwrap();

    assert!(tmp.path().join("seg_000").exists());
    assert!(!tmp.path().join("seg_001").exists());
    assert_eq!(dg.sealed_nodes_bound, 2);

    // Tail: 2 new nodes, 1 intra-tail edge.
    let n2 = dg.add_node(seal_test_node(&mut interner, 2, "B"));
    let n3 = dg.add_node(seal_test_node(&mut interner, 3, "B"));
    dg.add_edge(n2, n3, seal_test_edge(&mut interner, "U"));

    dg.save_to_dir(tmp.path(), &interner).unwrap();

    assert!(
        tmp.path().join("seg_001").exists(),
        "the auto-wired seal should have produced seg_001/"
    );
    assert_eq!(dg.sealed_nodes_bound, 4);

    let manifest = super::super::segment_summary::SegmentManifest::load_from(tmp.path()).unwrap();
    assert_eq!(manifest.len(), 2, "manifest should have 2 segments");

    drop(dg);
    let mut interner2 = StringInterner::new();
    let (reloaded, _tmp_zst) = super::DiskGraph::load_from_dir(tmp.path(), &mut interner2).unwrap();
    assert_eq!(reloaded.node_count, 4);
    assert_eq!(reloaded.edge_count, 2);
}

#[test]
fn save_to_dir_seals_cross_segment_overflow_as_full_range() {
    // Cross-segment overflow (an old→new edge) is handled by the
    // full-range seal, not the compact fallback.
    let tmp = TempDir::new().unwrap();
    let mut interner = StringInterner::new();
    let mut dg = super::DiskGraph::new_at_path(tmp.path()).unwrap();
    dg.defer_csr = true;
    let n0 = dg.add_node(seal_test_node(&mut interner, 0, "A"));
    let n1 = dg.add_node(seal_test_node(&mut interner, 1, "A"));
    dg.add_edge(n0, n1, seal_test_edge(&mut interner, "T"));
    dg.build_csr_from_pending().unwrap();
    dg.save_to_dir(tmp.path(), &interner).unwrap();

    // Cross-segment edge (old n0 → new n2) + a purely-tail edge.
    let n2 = dg.add_node(seal_test_node(&mut interner, 2, "B"));
    let n3 = dg.add_node(seal_test_node(&mut interner, 3, "B"));
    dg.add_edge(n0, n2, seal_test_edge(&mut interner, "T"));
    dg.add_edge(n2, n3, seal_test_edge(&mut interner, "U"));

    dg.save_to_dir(tmp.path(), &interner).unwrap();
    assert!(
        tmp.path().join("seg_001").exists(),
        "save_to_dir should seal cross-segment overflow"
    );

    drop(dg);
    let mut interner2 = StringInterner::new();
    let (reloaded, _tmp_zst) = super::DiskGraph::load_from_dir(tmp.path(), &mut interner2).unwrap();
    assert_eq!(reloaded.node_count, 4);
    assert_eq!(reloaded.edge_count, 3);
}

#[test]
fn conn_type_index_sources_are_global_after_segment_local_seal() {
    // Regression test for the 0.8.11 seal-path bug where the
    // merged `conn_type_index_sources` stored segment-local
    // indices (0..tail_len) instead of global node ids for
    // segment-local segments. Symptom: post-reload,
    // `MATCH (a)-[:T]->(b) RETURN a.id, b.id` returned no rows
    // even though `count(*)` reported the right number — the
    // enumeration looked up `out_offsets[0]` (empty, Person
    // range) instead of `out_offsets[tail_lo]` (the actual
    // TestHuman range).
    let tmp = TempDir::new().unwrap();
    let mut interner = StringInterner::new();
    let mut dg = super::DiskGraph::new_at_path(tmp.path()).unwrap();
    // Stay in the default `defer_csr = false` mode — mirrors the
    // path Python mutations take (the streaming subset save is the
    // only caller that flips defer_csr on, and its save_disk owns
    // the matching CSR build).
    // seg_0: 5 nodes, no edges.
    for i in 0..5 {
        dg.add_node(seal_test_node(&mut interner, i, "A"));
    }
    dg.save_to_dir(tmp.path(), &interner).unwrap();

    // Tail: 3 nodes + 3 intra-tail edges → segment-local seal.
    // Edges go into overflow (defer_csr is false).
    let n5 = dg.add_node(seal_test_node(&mut interner, 5, "B"));
    let n6 = dg.add_node(seal_test_node(&mut interner, 6, "B"));
    let n7 = dg.add_node(seal_test_node(&mut interner, 7, "B"));
    dg.add_edge(n5, n6, seal_test_edge(&mut interner, "T"));
    dg.add_edge(n6, n7, seal_test_edge(&mut interner, "T"));
    dg.add_edge(n7, n5, seal_test_edge(&mut interner, "T"));
    dg.save_to_dir(tmp.path(), &interner).unwrap();

    drop(dg);
    let mut interner2 = StringInterner::new();
    let (reloaded, _tmp_zst) = super::DiskGraph::load_from_dir(tmp.path(), &mut interner2).unwrap();

    let t_key = interner2.get_or_intern("T").as_u64();
    let sources = reloaded
        .sources_for_conn_type(t_key)
        .expect("T index should exist");
    let mut sources = sources;
    sources.sort_unstable();
    assert_eq!(
        sources,
        vec![5u32, 6, 7],
        "segment-local seal's conn_type_index_sources must be shifted \
         by node_lo on merge (regression from 0.8.11 pre-fix)"
    );
}

#[test]
fn compact_rewrite_after_seal_cleans_stale_segs_and_persists_heap_arrays() {
    // Regression test for the 0.8.11 seal-path bugs C and D. Flow
    // that used to fail:
    //   1. Build + save       → seg_000
    //   2. Add nodes + edges  → overflow
    //   3. Save               → seal creates seg_001; reconcile_seg0_csr
    //                           replaces self.{node_slots, …} with
    //                           `MmapOrVec::Heap` copies
    //   4. Add more overflow edges between *existing* nodes (no new
    //                           nodes, so `sealed_nodes_bound ==
    //                           node_count` → falls to compact-rewrite)
    //   5. Save again         → compact-rewrite path
    //
    // Before the fixes, step 5 (a) left seg_001 on disk so reload's
    // `enumerate_segment_dirs` double-counted, and (b) relied on mmap
    // persistence for the core arrays — which was impossible after
    // reconcile made them heap-backed — so the on-disk node_slots /
    // edge_endpoints files stayed at the pre-seal trimmed sizes and
    // reload errored with "File too small".
    let tmp = TempDir::new().unwrap();
    let mut interner = StringInterner::new();
    let mut dg = super::DiskGraph::new_at_path(tmp.path()).unwrap();
    // seg_0: 5 nodes.
    for i in 0..5 {
        dg.add_node(seal_test_node(&mut interner, i, "A"));
    }
    dg.save_to_dir(tmp.path(), &interner).unwrap();

    // Tail: add 3 nodes + intra-tail edges.
    let n5 = dg.add_node(seal_test_node(&mut interner, 5, "B"));
    let n6 = dg.add_node(seal_test_node(&mut interner, 6, "B"));
    let n7 = dg.add_node(seal_test_node(&mut interner, 7, "B"));
    dg.add_edge(n5, n6, seal_test_edge(&mut interner, "T"));
    dg.add_edge(n6, n7, seal_test_edge(&mut interner, "T"));
    dg.save_to_dir(tmp.path(), &interner).unwrap(); // seal → seg_001

    assert!(
        tmp.path().join("seg_001").exists(),
        "the seal should have produced seg_001"
    );

    // Now add *edges* between existing nodes only — `sealed_nodes_bound
    // == node_count` so this next save will take the compact-rewrite
    // path, not another seal.
    let n0 = NodeIndex::new(0);
    dg.add_edge(n0, n5, seal_test_edge(&mut interner, "T"));
    dg.save_to_dir(tmp.path(), &interner).unwrap();

    // Fix C: stale seg_001 must be removed.
    assert!(
        !tmp.path().join("seg_001").exists(),
        "compact-rewrite must clean up stale seg_NNN dirs"
    );

    // Fix D: reload must succeed — the heap-backed arrays from
    // reconcile must have been explicitly persisted by save_to_file.
    drop(dg);
    let mut interner2 = StringInterner::new();
    let (reloaded, _tmp_zst) = super::DiskGraph::load_from_dir(tmp.path(), &mut interner2).unwrap();
    assert_eq!(reloaded.node_count, 8, "all 8 nodes must survive");
    assert_eq!(reloaded.edge_count, 3, "3 T-edges must survive");
}

// ---------- enable_disk_mode's conversion ----------

#[test]
fn conversion_streams_edge_properties_into_the_mapped_base() {
    // `from_stable_digraph` is what `enable_disk_mode()` runs. It used to
    // clone every property-bearing edge's `Vec<(InternedKey, Value)>` into a
    // heap overlay — ~175 B/edge of pure duplication, since the petgraph it
    // copies from is dropped moments later. The blob is now written as the
    // edges are walked, and the store comes back mapping it.
    let tmp = TempDir::new().unwrap();
    let mut interner = StringInterner::new();
    let mut source: petgraph::stable_graph::StableDiGraph<NodeData, EdgeData> =
        petgraph::stable_graph::StableDiGraph::new();
    let nodes: Vec<NodeIndex> = (0..4)
        .map(|i| source.add_node(seal_test_node(&mut interner, i, "Item")))
        .collect();
    let links = interner.get_or_intern("LINKS");
    let weight = interner.get_or_intern("weight");
    let note = interner.get_or_intern("note");
    // Edge 1 has no properties: the sparsity encoding and the trailing pad
    // are both exercised.
    source.add_edge(
        nodes[0],
        nodes[1],
        EdgeData {
            connection_type: links,
            properties: vec![
                (weight, Value::Float64(2.5)),
                (note, Value::String("first".to_string())),
            ],
        },
    );
    source.add_edge(nodes[1], nodes[2], seal_test_edge(&mut interner, "LINKS"));
    source.add_edge(
        nodes[2],
        nodes[3],
        EdgeData {
            connection_type: links,
            properties: vec![(weight, Value::Float64(-0.75))],
        },
    );

    let mut dg = super::DiskGraph::from_stable_digraph(&mut source, tmp.path()).unwrap();

    assert_eq!(
        dg.edge_property_overlay_len(),
        0,
        "the conversion must not leave edge properties on the heap"
    );
    let seg = tmp.path().join(segment_subdir(0));
    let offsets =
        std::fs::metadata(seg.join(crate::graph::storage::disk::edge_properties::OFFSETS_FILE))
            .unwrap();
    assert_eq!(
        offsets.len(),
        4 * std::mem::size_of::<u64>() as u64,
        "one offset per edge plus the trailing total"
    );
    assert!(
        std::fs::metadata(seg.join(crate::graph::storage::disk::edge_properties::HEAP_FILE))
            .unwrap()
            .len()
            > 0
    );

    assert_eq!(
        dg.edge_properties_at(0).unwrap().as_ref(),
        &[
            (weight, Value::Float64(2.5)),
            (note, Value::String("first".to_string())),
        ]
    );
    assert!(dg.edge_properties_at(1).is_none());
    assert_eq!(
        dg.edge_properties_at(2).unwrap().as_ref(),
        &[(weight, Value::Float64(-0.75))]
    );
    assert!(dg.edge_properties_at(3).is_none());

    // A write after the conversion takes the store's normal overlay path and
    // wins over the streamed base.
    dg.edge_properties
        .insert(0, vec![(weight, Value::Float64(9.0))]);
    assert_eq!(
        dg.edge_properties_at(0).unwrap().as_ref(),
        &[(weight, Value::Float64(9.0))]
    );
    assert_eq!(dg.edge_property_overlay_len(), 1);
}

#[test]
fn conversion_without_edge_properties_writes_no_blob() {
    let tmp = TempDir::new().unwrap();
    let mut interner = StringInterner::new();
    let mut source: petgraph::stable_graph::StableDiGraph<NodeData, EdgeData> =
        petgraph::stable_graph::StableDiGraph::new();
    let a = source.add_node(seal_test_node(&mut interner, 0, "Item"));
    let b = source.add_node(seal_test_node(&mut interner, 1, "Item"));
    source.add_edge(a, b, seal_test_edge(&mut interner, "LINKS"));

    let dg = super::DiskGraph::from_stable_digraph(&mut source, tmp.path()).unwrap();

    let seg = tmp.path().join(segment_subdir(0));
    assert_eq!(
        std::fs::metadata(seg.join(crate::graph::storage::disk::edge_properties::OFFSETS_FILE))
            .unwrap()
            .len(),
        0,
        "a property-less graph writes the zero-length representation"
    );
    assert_eq!(dg.edge_property_overlay_len(), 0);
    assert!(dg.edge_properties_at(0).is_none());
}

// ---------- enable_disk_mode's published form ----------

/// `enable_disk_mode_at` must leave the same end state the two-call
/// `enable_disk_mode()` + `save_disk(dir)` sequence does — a published
/// generation the live handle maps — while keeping the conversion's scratch
/// inside the destination and removing it once the publish has rebased.
///
/// The path-form's reason to exist is that scratch placement: the pathless
/// conversion materializes the whole CSR under the system temp directory, so a
/// graph too large for `/tmp` (or a RAM-backed `tmpfs`) failed there rather
/// than on the filesystem the caller pointed at.
#[test]
fn enable_disk_mode_at_publishes_and_leaves_no_scratch_behind() {
    let root = TempDir::new().unwrap();
    let root_path = root.path().to_str().unwrap();

    let mut graph = DirGraph::new();
    add_docs(&mut graph, &[1, 2, 3]);
    graph.enable_disk_mode_at(root_path).unwrap();

    assert!(
        root.path().join("CURRENT").is_file(),
        "no generation was published"
    );
    let stray: Vec<String> = std::fs::read_dir(root.path())
        .unwrap()
        .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned())
        .filter(|name| name.starts_with(".converting-"))
        .collect();
    assert!(
        stray.is_empty(),
        "conversion scratch survived the publish: {stray:?}"
    );

    let disk = match &mut graph.graph {
        GraphBackend::Disk(disk) => disk,
        _ => panic!("expected disk backend"),
    };
    let data_dir = disk.data_dir.clone();
    assert!(
        data_dir.starts_with(root.path()),
        "the live handle must write inside the destination, not {}",
        data_dir.display()
    );
    let outside: Vec<_> = disk
        .mapped_file_paths()
        .into_iter()
        .filter(|path| !path.starts_with(&data_dir))
        .collect();
    assert!(
        outside.is_empty(),
        "writer still maps files outside the published generation: {outside:?}"
    );

    let reader = crate::graph::io::file::load_file(root_path).unwrap();
    assert_eq!(reader.graph.node_count(), 3);
}

/// A failed publish leaves the graph converted onto scratch that drop still
/// owns — the same guarantee the pathless form gives.
#[test]
fn a_failed_publish_registers_its_scratch_for_cleanup() {
    let root = TempDir::new().unwrap();
    // A file where the destination directory must be: `GenerationTxn::begin`
    // cannot create `generations/` under it, so the publish fails after the
    // conversion has already run.
    let target = root.path().join("occupied");
    std::fs::write(&target, b"not a directory").unwrap();

    let mut graph = DirGraph::new();
    add_docs(&mut graph, &[1, 2]);
    assert!(graph.enable_disk_mode_at(target.to_str().unwrap()).is_err());

    let registered = graph.temp_dirs.lock().unwrap().clone();
    assert!(
        registered.iter().any(|dir| dir.starts_with(&target)),
        "the conversion scratch must stay registered for drop-time cleanup: {registered:?}"
    );
}

/// `enable_disk_mode` converts a petgraph whose edges are already present, so
/// the whole edge set lands in the CSR with no `conn_type_index_*` and no
/// overflow. `for_each_edge_of_conn_type` must still visit those edges —
/// treating the absent index as "no sources of this type" makes every
/// consumer (fused aggregates, `describe` connection sampling) silently
/// report an empty edge set.
#[test]
fn for_each_edge_of_conn_type_visits_csr_edges_without_a_conn_type_index() {
    let mut graph = DirGraph::new();
    add_docs(&mut graph, &[1, 2, 3]);
    let links = DataFrame::from_cypher_rows(
        vec!["src".to_string(), "tgt".to_string()],
        vec![
            vec![Value::Int64(1), Value::Int64(3)],
            vec![Value::Int64(2), Value::Int64(3)],
        ],
    )
    .unwrap();
    crate::graph::mutation::maintain::add_connections(
        &mut graph,
        links,
        "LINKS".to_string(),
        "Doc".to_string(),
        "src".to_string(),
        "Doc".to_string(),
        "tgt".to_string(),
        None,
        None,
        None,
    )
    .unwrap();
    graph.enable_disk_mode().unwrap();

    let disk = graph.graph.as_disk().expect("disk mode");
    assert!(
        disk.conn_type_index_types.is_empty(),
        "the conversion builds no conn-type index, or this test asserts nothing"
    );
    assert!(
        disk.overflow_out.is_empty(),
        "converted edges live in the CSR, not overflow, or this test asserts nothing"
    );

    let conn = InternedKey::from_str("LINKS").as_u64();
    let mut seen: Vec<(usize, usize)> = Vec::new();
    disk.for_each_edge_of_conn_type(conn, |src, tgt, _edge_idx| {
        seen.push((src.index(), tgt.index()));
        true
    });
    seen.sort_unstable();
    assert_eq!(
        seen.len(),
        2,
        "both LINKS edges must be visited without a conn-type index, got {seen:?}"
    );

    // A non-existent type must still yield nothing.
    let absent = InternedKey::from_str("NO_SUCH_TYPE").as_u64();
    let mut count = 0usize;
    disk.for_each_edge_of_conn_type(absent, |_, _, _| {
        count += 1;
        true
    });
    assert_eq!(count, 0, "an unmatched conn type must visit no edges");

    // Early stop still works.
    let mut visited = 0usize;
    disk.for_each_edge_of_conn_type(conn, |_, _, _| {
        visited += 1;
        false
    });
    assert_eq!(
        visited, 1,
        "returning false must stop after the first match"
    );
}