fn create_composite_follow_junction_fixture() {
    Spi::run(
        "CREATE TABLE public.graph_test_junction_pgtest (
                follower TEXT NOT NULL REFERENCES public.graph_test_users_pgtest(id),
                followee  TEXT NOT NULL REFERENCES public.graph_test_users_pgtest(id),
                PRIMARY KEY (follower, followee)
            )",
    )
    .expect("create composite follow junction failed");
    Spi::run(
        "INSERT INTO public.graph_test_junction_pgtest (follower, followee)
             VALUES ('u1', 'u2')",
    )
    .expect("insert composite follow junction failed");
}

#[pg_test]
fn traversal_helpers_edge_path_and_composition_apis_work() {
    reset_and_create_fixtures();
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name']
            )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_friendships_pgtest'::regclass,
                'user_id',
                'graph_test_users_pgtest'::regclass,
                'friend_id',
                'friend'
            )",
    )
    .expect("add friendship edge failed");
    Spi::run("SELECT graph.add_filter_column('graph_test_users_pgtest'::regclass, 'age')")
        .expect("add age filter failed");
    Spi::run("SELECT * FROM graph.build()").expect("build failed");

    let edge_label = Spi::get_one::<String>(
        "SELECT edge_path->>0
             FROM graph.traverse('graph_test_users_pgtest'::regclass, 'u1', 1, hydrate := false)
             WHERE node_id = 'u2'",
    )
    .expect("edge_path traversal failed")
    .expect("edge_path label missing");
    let helper_count = Spi::get_one::<i64>(
        "SELECT count(*)
             FROM graph.traverse(
                'graph_test_users_pgtest'::regclass,
                'u1',
                1,
                filter := graph.all(ARRAY[graph.greater_than('age', 40)]),
                hydrate := false
             )
             WHERE node_id = 'u2'",
    )
    .expect("helper filter traversal failed")
    .unwrap_or(0);
    let multi_start_roots = Spi::get_one::<i64>(
        "SELECT count(DISTINCT root_id)
             FROM graph.traverse(
                ARRAY['graph_test_users_pgtest'::regclass, 'graph_test_users_pgtest'::regclass],
                ARRAY['u1'::text, 'u2'::text],
                0,
                hydrate := false
             )",
    )
    .expect("multi-start traversal failed")
    .unwrap_or(0);
    let traverse_search_count = Spi::get_one::<i64>(
        "SELECT count(*)
             FROM graph.traverse_search(
                'name',
                'Alice',
                table_filter := 'graph_test_users_pgtest'::regclass,
                max_depth := 1,
                hydrate := false
             )
             WHERE root_id = 'u1' AND node_id = 'u2'",
    )
    .expect("traverse_search failed")
    .unwrap_or(0);
    let search_nodes_count = Spi::get_one::<i64>(
        "SELECT count(*)
             FROM graph.search_nodes(
                'name',
                'Alice',
                table_filter := 'graph_test_users_pgtest'::regclass,
                mode := 'exact'
             )
             WHERE node_id = 'u1' AND verified",
    )
    .expect("search_nodes failed")
    .unwrap_or(0);

    assert_eq!(edge_label, "friend");
    assert_eq!(helper_count, 1);
    assert_eq!(multi_start_roots, 2);
    assert_eq!(traverse_search_count, 1);
    assert_eq!(search_nodes_count, 1);
}

#[pg_test]
fn composite_pk_junction_edge_table_builds_and_traverses_when_to_column_is_source_fk() {
    reset_and_create_fixtures();
    create_composite_follow_junction_fixture();

    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name']
            )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_junction_pgtest'::regclass,
                from_column := 'follower',
                to_table := 'graph_test_users_pgtest'::regclass,
                to_column := 'followee',
                label := 'follows',
                bidirectional := true
            )",
    )
    .expect("add junction edge-table edge failed");
    Spi::run("SELECT * FROM graph.build()").expect("build composite junction edge graph failed");

    let outbound = Spi::get_one::<i64>(
        "SELECT count(*)
             FROM graph.traverse(
                'graph_test_users_pgtest'::regclass,
                'u1',
                1,
                edge_types := ARRAY['follows'],
                hydrate := false
             )
             WHERE node_id = 'u2'",
    )
    .expect("outbound junction traversal failed")
    .unwrap_or(0);
    let inbound = Spi::get_one::<i64>(
        "SELECT count(*)
             FROM graph.traverse(
                'graph_test_users_pgtest'::regclass,
                'u2',
                1,
                edge_types := ARRAY['follows'],
                hydrate := false
             )
             WHERE node_id = 'u1'",
    )
    .expect("bidirectional junction traversal failed")
    .unwrap_or(0);

    assert_eq!(outbound, 1);
    assert_eq!(inbound, 1);
}

#[pg_test]
fn mixed_mode_junction_registration_fails_before_build() {
    reset_and_create_fixtures();
    create_composite_follow_junction_fixture();

    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name']
            )",
    )
    .expect("add users table failed");
    let error = sql_error_message(
        "SELECT graph.add_edge(
                'graph_test_junction_pgtest'::regclass,
                from_column := 'follower',
                to_table := 'graph_test_users_pgtest'::regclass,
                to_column := 'id',
                label := 'follows',
                bidirectional := true
            )",
    )
    .expect("mixed-mode add_edge should fail");

    assert!(
        error.contains("edge-table registration")
            && error.contains("graph_test_junction_pgtest")
            && error.contains("to_column 'id'"),
        "unexpected registration error: {error}"
    );
}

#[pg_test]
fn multi_start_traverse_applies_limit_after_global_merge() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();

    let (row_count, first_root) = Spi::connect(|client| {
        let result = client
            .select(
                "SELECT count(*)::bigint, min(root_id)
                     FROM graph.traverse(
                        ARRAY[
                            'graph_test_users_pgtest'::regclass,
                            'graph_test_users_pgtest'::regclass
                        ],
                        ARRAY['u1'::text, 'u2'::text],
                        0,
                        hydrate := false,
                        max_rows := 1,
                        row_offset := 0
                     )",
                None,
                &[],
            )
            .expect("multi-start limited traversal failed");
        let row = result.first();
        Ok::<_, pgrx::spi::Error>((
            row.get::<i64>(1)?.unwrap_or_default(),
            row.get::<String>(2)?.unwrap_or_default(),
        ))
    })
    .expect("multi-start limited traversal read failed");

    assert_eq!(row_count, 1);
    assert_eq!(first_root, "u1");
}

#[pg_test]
fn multi_start_traverse_honors_node_global_uniqueness() {
    reset_and_create_fixtures();
    Spi::run(
        "INSERT INTO public.graph_test_users_pgtest (id, name, age)
             VALUES ('u3', 'Carol', 29)",
    )
    .expect("insert converging user failed");
    Spi::run(
        "INSERT INTO public.graph_test_friendships_pgtest (id, user_id, friend_id)
             VALUES ('f2', 'u1', 'u3'), ('f3', 'u2', 'u3')",
    )
    .expect("insert converging friendships failed");
    Spi::run("UPDATE public.graph_test_users_pgtest SET name = 'Root' WHERE id IN ('u1', 'u2')")
        .expect("set searchable roots failed");
    build_friendship_fixture_graph();

    let counts = Spi::connect(|client| {
        let per_root = client
            .select(
                "SELECT count(*)::bigint
                     FROM graph.traverse(
                        ARRAY[
                            'graph_test_users_pgtest'::regclass,
                            'graph_test_users_pgtest'::regclass
                        ],
                        ARRAY['u1'::text, 'u2'::text],
                        max_depth := 1,
                        uniqueness := 'node_per_root',
                        include_start := false,
                        hydrate := false
                     )
                     WHERE node_id = 'u3'",
                None,
                &[],
            )
            .expect("node_per_root traversal failed")
            .first()
            .get::<i64>(1)?
            .unwrap_or_default();
        let global = client
            .select(
                "SELECT count(*)::bigint
                     FROM graph.traverse(
                        ARRAY[
                            'graph_test_users_pgtest'::regclass,
                            'graph_test_users_pgtest'::regclass
                        ],
                        ARRAY['u1'::text, 'u2'::text],
                        max_depth := 1,
                        uniqueness := 'node_global',
                        include_start := false,
                        hydrate := false
                     )
                     WHERE node_id = 'u3'",
                None,
                &[],
            )
            .expect("node_global traversal failed")
            .first()
            .get::<i64>(1)?
            .unwrap_or_default();
        let default_global = client
            .select(
                "SELECT count(*)::bigint
                     FROM graph.traverse(
                        ARRAY[
                            'graph_test_users_pgtest'::regclass,
                            'graph_test_users_pgtest'::regclass
                        ],
                        ARRAY['u1'::text, 'u2'::text],
                        max_depth := 1,
                        include_start := false,
                        hydrate := false
                     )
                     WHERE node_id = 'u3'",
                None,
                &[],
            )
            .expect("default traversal failed")
            .first()
            .get::<i64>(1)?
            .unwrap_or_default();
        let search_per_root = client
            .select(
                "SELECT count(*)::bigint
                     FROM graph.traverse_search(
                        'name',
                        'Root',
                        table_filter := 'graph_test_users_pgtest'::regclass,
                        search_mode := 'exact',
                        max_depth := 1,
                        uniqueness := 'node_per_root',
                        include_start := false,
                        hydrate := false
                     )
                     WHERE node_id = 'u3'",
                None,
                &[],
            )
            .expect("node_per_root traverse_search failed")
            .first()
            .get::<i64>(1)?
            .unwrap_or_default();
        let search_global = client
            .select(
                "SELECT count(*)::bigint
                     FROM graph.traverse_search(
                        'name',
                        'Root',
                        table_filter := 'graph_test_users_pgtest'::regclass,
                        search_mode := 'exact',
                        max_depth := 1,
                        uniqueness := 'node_global',
                        include_start := false,
                        hydrate := false
                     )
                     WHERE node_id = 'u3'",
                None,
                &[],
            )
            .expect("node_global traverse_search failed")
            .first()
            .get::<i64>(1)?
            .unwrap_or_default();
        Ok::<_, pgrx::spi::Error>((
            per_root,
            global,
            default_global,
            search_per_root,
            search_global,
        ))
    })
    .expect("multi-start uniqueness counts failed");
    let (
        per_root_count,
        global_count,
        default_count,
        search_per_root_count,
        search_global_count,
    ) = counts;

    assert_eq!(per_root_count, 2);
    assert_eq!(global_count, 1);
    assert_eq!(default_count, 1);
    assert_eq!(search_per_root_count, 2);
    assert_eq!(search_global_count, 1);
}

#[pg_test]
fn format_path_formats_traversal_path_and_edge_path() {
    let formatted = Spi::get_one::<String>(
        "SELECT graph.format_path(
                '[
                    {\"table\":\"users\",\"id\":\"u1\"},
                    {\"table\":\"users\",\"id\":\"u2\"},
                    {\"table\":\"companies\",\"id\":\"c1\"}
                ]'::jsonb,
                '[\"friend\", \"officer_of\"]'::jsonb
             )",
    )
    .expect("format_path failed")
    .expect("format_path returned null");
    let custom_separator = Spi::get_one::<String>(
        "SELECT graph.format_path(
                '[
                    {\"table\":\"users\",\"id\":\"u1\"},
                    {\"table\":\"users\",\"id\":\"u2\"}
                ]'::jsonb,
                '[\"friend\"]'::jsonb,
                E'\n'
             )",
    )
    .expect("format_path with custom separator failed")
    .expect("format_path returned null");
    let root = Spi::get_one::<String>(
        "SELECT graph.format_path(
                '[{\"table\":\"users\",\"id\":\"u1\"}]'::jsonb,
                '[]'::jsonb
             )",
    )
    .expect("format_path root failed")
    .expect("format_path returned null");

    assert_eq!(
        formatted,
        "users:u1 --friend--> users:u2 | users:u2 --officer_of--> companies:c1"
    );
    assert_eq!(custom_separator, "users:u1 --friend--> users:u2");
    assert_eq!(root, "");
}

#[pg_test]
fn edge_label_column_drives_shortest_path_and_traverse_labels() {
    reset_and_create_fixtures();
    Spi::run("ALTER TABLE public.graph_test_friendships_pgtest ADD COLUMN rel_type TEXT")
        .expect("add rel_type failed");
    Spi::run(
        "UPDATE public.graph_test_friendships_pgtest
             SET rel_type = 'knows'",
    )
    .expect("set rel_type failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name']
            )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_friendships_pgtest'::regclass,
                'user_id',
                'graph_test_users_pgtest'::regclass,
                'friend_id',
                'relationship',
                label_column := 'rel_type'
            )",
    )
    .expect("add dynamic-label edge failed");
    Spi::run("SELECT * FROM graph.build()").expect("build failed");

    let edge_label = Spi::get_one::<String>(
        "SELECT edge_label
             FROM graph.shortest_path(
                'graph_test_users_pgtest'::regclass,
                'u1',
                'graph_test_users_pgtest'::regclass,
                'u2',
                5
             )
             WHERE step = 1",
    )
    .expect("shortest_path label failed")
    .expect("edge_label missing");
    let traverse_label = Spi::get_one::<String>(
        "SELECT edge_path->>0
             FROM graph.traverse('graph_test_users_pgtest'::regclass, 'u1', 1, hydrate := false)
             WHERE node_id = 'u2'",
    )
    .expect("traverse label failed")
    .expect("edge_path label missing");

    assert_eq!(edge_label, "knows");
    assert_eq!(traverse_label, "knows");
}

#[pg_test]
fn edge_and_weighted_path_v1_acceptance() {
    reset_and_create_fixtures();
    Spi::run(
        "CREATE TABLE public.graph_test_weighted_nodes_pgtest (
                id   TEXT PRIMARY KEY,
                name TEXT NOT NULL
            )",
    )
    .expect("create weighted nodes failed");
    Spi::run(
        "CREATE TABLE public.graph_test_weighted_edges_pgtest (
                id       TEXT PRIMARY KEY,
                src      TEXT NOT NULL,
                dst      TEXT NOT NULL,
                cost     INT NOT NULL,
                rel_type TEXT
            )",
    )
    .expect("create weighted edges failed");
    Spi::run(
        "INSERT INTO public.graph_test_weighted_nodes_pgtest (id, name) VALUES
                ('a', 'A'),
                ('b', 'B'),
                ('c', 'C'),
                ('d', 'D')",
    )
    .expect("insert weighted nodes failed");
    Spi::run(
        "INSERT INTO public.graph_test_weighted_edges_pgtest (id, src, dst, cost, rel_type)
             VALUES
                ('e1', 'a', 'b', 5, 'expensive'),
                ('e2', 'b', 'd', 5, NULL),
                ('e3', 'a', 'c', 1, ''),
                ('e4', 'c', 'd', 1, 'cheap')",
    )
    .expect("insert weighted edges failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_weighted_nodes_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name']
            )",
    )
    .expect("add weighted node table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_weighted_edges_pgtest'::regclass,
                from_column := 'src',
                to_table := 'graph_test_weighted_nodes_pgtest'::regclass,
                to_column := 'dst',
                label := 'route',
                bidirectional := false,
                weight_column := 'cost',
                label_column := 'rel_type'
            )",
    )
    .expect("add weighted edge table failed");
    Spi::run("SELECT * FROM graph.build()").expect("build failed");

    let weighted = Spi::get_one::<String>(
        "SELECT string_agg(node_id, '->' ORDER BY step) || ':' || max(total_cost)::text
             FROM graph.weighted_shortest_path(
                'graph_test_weighted_nodes_pgtest'::regclass,
                'a',
                'graph_test_weighted_nodes_pgtest'::regclass,
                'd'
             )",
    )
    .expect("weighted shortest path failed")
    .expect("weighted shortest path returned no rows");
    let weighted_steps = Spi::get_one::<String>(
        "SELECT string_agg(
                    step::text || ':' || node_id || ':' ||
                    coalesce(edge_label, '<start>') || ':' ||
                    coalesce(edge_weight::text, '<start>') || ':' ||
                    step_cost::text || ':' || total_cost::text,
                    ',' ORDER BY step
                )
             FROM graph.weighted_shortest_path(
                'graph_test_weighted_nodes_pgtest'::regclass,
                'a',
                'graph_test_weighted_nodes_pgtest'::regclass,
                'd'
             )",
    )
    .expect("weighted path step detail failed")
    .expect("weighted path step detail missing");
    let weighted_table_identity_rows = Spi::get_one::<i64>(
        "SELECT count(*)
             FROM graph.weighted_shortest_path(
                'graph_test_weighted_nodes_pgtest'::regclass,
                'a',
                'graph_test_weighted_nodes_pgtest'::regclass,
                'd'
             )
             WHERE node_table = 'graph_test_weighted_nodes_pgtest'::regclass::oid
               AND node_table_name = 'graph_test_weighted_nodes_pgtest'::regclass::text",
    )
    .expect("weighted path table identity check failed")
    .unwrap_or(0);
    let fallback_label = Spi::get_one::<String>(
        "SELECT edge_path->>0
             FROM graph.traverse(
                'graph_test_weighted_nodes_pgtest'::regclass,
                'a',
                1,
                direction := 'out',
                hydrate := false
             )
             WHERE node_id = 'c'",
    )
    .expect("fallback label traverse failed")
    .expect("fallback label missing");
    let null_fallback_label = Spi::get_one::<String>(
        "SELECT edge_path->>0
             FROM graph.traverse(
                'graph_test_weighted_nodes_pgtest'::regclass,
                'b',
                1,
                direction := 'out',
                hydrate := false
             )
             WHERE node_id = 'd'",
    )
    .expect("null fallback label traverse failed")
    .expect("null fallback label missing");
    let reverse_out_count = Spi::get_one::<i64>(
        "SELECT count(*)
             FROM graph.traverse(
                'graph_test_weighted_nodes_pgtest'::regclass,
                'd',
                1,
                direction := 'out',
                hydrate := false
             )
             WHERE node_id IN ('b', 'c')",
    )
    .expect("unidirectional reverse traverse failed")
    .unwrap_or(-1);
    let inbound_count = Spi::get_one::<i64>(
        "SELECT count(*)
             FROM graph.traverse(
                'graph_test_weighted_nodes_pgtest'::regclass,
                'd',
                1,
                direction := 'in',
                hydrate := false
             )
             WHERE node_id IN ('b', 'c')",
    )
    .expect("inbound unidirectional traverse failed")
    .unwrap_or(-1);
    let no_weight_param = Spi::get_one::<bool>(
        "SELECT bool_and(pg_get_function_arguments(p.oid) NOT LIKE '%weight%')
             FROM pg_proc p
             JOIN pg_namespace n ON n.oid = p.pronamespace
             WHERE n.nspname = 'graph'
               AND p.proname = 'weighted_shortest_path'",
    )
    .expect("weighted signature inspection failed")
    .unwrap_or(false);
    let weighted_shape = Spi::get_one::<bool>(
        "SELECT bool_and(pg_get_function_result(p.oid) =
                    'TABLE(step integer, node_table oid, node_table_name text, node_id text, edge_label text, edge_weight bigint, step_cost bigint, total_cost bigint)'
                )
             FROM pg_proc p
             JOIN pg_namespace n ON n.oid = p.pronamespace
             WHERE n.nspname = 'graph'
               AND p.proname = 'weighted_shortest_path'",
    )
    .expect("weighted result shape inspection failed")
    .unwrap_or(false);
    let no_weighted_path_rows = Spi::get_one::<i64>(
        "SELECT count(*)
             FROM graph.weighted_shortest_path(
                'graph_test_weighted_nodes_pgtest'::regclass,
                'd',
                'graph_test_weighted_nodes_pgtest'::regclass,
                'a'
             )",
    )
    .expect("weighted empty path inspection failed")
    .unwrap_or(-1);
    let typed_unweighted = Spi::get_one::<String>(
        "SELECT string_agg(node_id, '->' ORDER BY step)
           FROM graph.shortest_path(
             'graph_test_weighted_nodes_pgtest'::regclass, 'a',
             'graph_test_weighted_nodes_pgtest'::regclass, 'd',
             5, false, ARRAY['expensive', 'route']::text[]
           )",
    )
    .expect("typed unweighted path failed")
    .unwrap_or_default();
    let typed_weighted = Spi::get_one::<String>(
        "SELECT string_agg(node_id, '->' ORDER BY step)
           FROM graph.weighted_shortest_path(
             'graph_test_weighted_nodes_pgtest'::regclass, 'a',
             'graph_test_weighted_nodes_pgtest'::regclass, 'd',
             ARRAY['expensive', 'route']::text[]
           )",
    )
    .expect("typed weighted path failed")
    .unwrap_or_default();
    let recursive_typed = Spi::get_one::<String>(
        "WITH RECURSIVE paths(node_id, path, depth) AS (
             SELECT 'a'::text, ARRAY['a']::text[], 0
             UNION ALL
             SELECT e.dst, p.path || e.dst, p.depth + 1
             FROM paths p
             JOIN public.graph_test_weighted_edges_pgtest e ON e.src = p.node_id
             WHERE coalesce(nullif(e.rel_type, ''), 'route') = ANY(ARRAY['expensive', 'route']::text[])
               AND NOT e.dst = ANY(p.path)
               AND p.depth < 5
           )
           SELECT array_to_string(path, '->')
           FROM paths
           WHERE node_id = 'd'
           ORDER BY depth
           LIMIT 1",
    )
    .expect("typed recursive SQL comparison failed")
    .unwrap_or_default();
    let empty_typed_rows = Spi::get_one::<i64>(
        "SELECT count(*)
           FROM graph.shortest_path(
             'graph_test_weighted_nodes_pgtest'::regclass, 'a',
             'graph_test_weighted_nodes_pgtest'::regclass, 'd',
             5, false, ARRAY[]::text[]
           )",
    )
    .expect("empty typed path failed")
    .unwrap_or(-1);
    let overloads_present = Spi::get_one::<bool>(
        "SELECT to_regprocedure('graph.shortest_path(oid,text,oid,text,integer,boolean)') IS NOT NULL
            AND to_regprocedure('graph.shortest_path(oid,text,oid,text,integer,boolean,text[])') IS NOT NULL
            AND to_regprocedure('graph.weighted_shortest_path(oid,text,oid,text)') IS NOT NULL
            AND to_regprocedure('graph.weighted_shortest_path(oid,text,oid,text,text[])') IS NOT NULL",
    )
    .expect("shortest-path overload inspection failed")
    .unwrap_or(false);
    let legacy_null_rows = Spi::get_one::<i64>(
        "SELECT count(*)
           FROM graph.shortest_path(
             'graph_test_weighted_nodes_pgtest'::regclass, 'a',
             'graph_test_weighted_nodes_pgtest'::regclass, 'd',
             NULL
           )",
    )
    .expect("legacy NULL max_depth call became ambiguous")
    .unwrap_or(-1);

    assert_eq!(weighted, "a->c->d:2");
    assert_eq!(weighted_steps, "0:a:<start>:<start>:0:2,1:c:route:1:1:2,2:d:cheap:1:2:2");
    assert_eq!(weighted_table_identity_rows, 3);
    assert_eq!(fallback_label, "route");
    assert_eq!(null_fallback_label, "route");
    assert_eq!(reverse_out_count, 0);
    assert_eq!(inbound_count, 2);
    assert!(no_weight_param);
    assert!(weighted_shape);
    assert_eq!(no_weighted_path_rows, 0);
    assert_eq!(typed_unweighted, "a->b->d");
    assert_eq!(typed_weighted, "a->b->d");
    assert_eq!(recursive_typed, typed_unweighted);
    assert_eq!(empty_typed_rows, 0);
    assert!(overloads_present);
    assert_eq!(legacy_null_rows, 0);
}

#[pg_test]
fn traversal_and_shortest_path_use_layered_manifest_snapshot() {
    build_persisted_mutable_friendship_graph();
    publish_friendship_segment_and_reload("f-layered", "u2", "u1");

    let traversal_count = Spi::get_one::<i64>(
        "SELECT count(*)
             FROM graph.traverse(
                'graph_test_users_pgtest'::regclass,
                'u1',
                1,
                edge_types := ARRAY['friend'],
                hydrate := false
             )
             WHERE node_id = 'u2'",
    )
    .expect("layered traversal query failed")
    .unwrap_or(0);
    let shortest = Spi::get_one::<String>(
        "SELECT string_agg(node_id, '->' ORDER BY step)
             FROM graph.shortest_path(
                'graph_test_users_pgtest'::regclass,
                'u1',
                'graph_test_users_pgtest'::regclass,
                'u2',
                5,
                hydrate := false
             )",
    )
    .expect("layered shortest path query failed")
    .unwrap_or_default();

    assert_eq!(traversal_count, 1);
    assert_eq!(shortest, "u1->u2");
}

#[pg_test]
fn weighted_shortest_path_uses_layered_manifest_snapshot() {
    reset_and_create_fixtures();
    Spi::run("SET graph.sync_mode = 'trigger'").expect("set trigger sync failed");
    Spi::run("SET graph.persist_on_build = on").expect("enable persist_on_build failed");
    Spi::run("SET graph.mutable_enabled = on").expect("enable mutable projection failed");
    Spi::run("SET graph.query_freshness = 'off'").expect("disable automatic query sync failed");
    Spi::run(
        "CREATE TABLE public.graph_test_weighted_nodes_pgtest (
                id   TEXT PRIMARY KEY,
                name TEXT NOT NULL,
                parent_id TEXT NULL REFERENCES public.graph_test_weighted_nodes_pgtest(id),
                cost INT NOT NULL DEFAULT 1
            )",
    )
    .expect("create weighted nodes failed");
    Spi::run(
        "INSERT INTO public.graph_test_weighted_nodes_pgtest (id, name) VALUES
                ('a', 'A'), ('b', 'B'), ('d', 'D')",
    )
    .expect("insert weighted nodes failed");
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_weighted_nodes_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name', 'parent_id', 'cost']
            )",
    )
    .expect("add weighted nodes failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_weighted_nodes_pgtest'::regclass,
                from_column := 'parent_id',
                to_table := 'graph_test_weighted_nodes_pgtest'::regclass,
                to_column := 'id',
                label := 'route',
                bidirectional := false,
                weight_column := 'cost'
            )",
    )
    .expect("add weighted edge failed");
    Spi::run("SELECT * FROM graph.build(mode := 'mutable_overlay')")
        .expect("build persisted mutable weighted graph failed");
    Spi::run(
        "UPDATE public.graph_test_weighted_nodes_pgtest
             SET parent_id = 'd', cost = 2
             WHERE id = 'a'",
    )
    .expect("update layered weighted edge failed");
    let segments = Spi::get_one::<i64>("SELECT segments_published FROM graph.ingest_projection()")
        .expect("ingest weighted projection failed")
        .unwrap_or(0);
    assert!(segments > 0, "ingest_projection must publish a weighted segment");
    Spi::run("SET graph.auto_load = on").expect("enable auto_load failed");
    super::ENGINE.with(|engine| {
        *engine.borrow_mut() = super::engine::Engine::new();
    });

    let weighted = Spi::get_one::<String>(
        "SELECT string_agg(node_id, '->' ORDER BY step) || ':' || max(total_cost)::text
             FROM graph.weighted_shortest_path(
                'graph_test_weighted_nodes_pgtest'::regclass,
                'a',
                'graph_test_weighted_nodes_pgtest'::regclass,
                'd'
             )",
    )
    .expect("layered weighted shortest path failed")
    .unwrap_or_default();

    assert_eq!(weighted, "a->d:2");
}

#[pg_test]
fn traverse_accepts_dfs_out_and_returns_path_coordinates() {
    reset_and_create_fixtures();
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name']
            )",
    )
    .expect("add users table failed");
    Spi::run(
        "SELECT graph.add_edge(
                'graph_test_friendships_pgtest'::regclass,
                'user_id',
                'graph_test_users_pgtest'::regclass,
                'friend_id',
                'friend'
            )",
    )
    .expect("add friendship edge failed");
    Spi::run("SELECT * FROM graph.build()").expect("build failed");

    let path_id = Spi::get_one::<String>(
        "SELECT path->0->>'id'
             FROM graph.traverse(
                'graph_test_users_pgtest'::regclass,
                'u1',
                1,
                direction := 'out',
                strategy := 'dfs',
                hydrate := false
             )
             WHERE node_id = 'u2'",
    )
    .expect("dfs traverse failed")
    .expect("missing first path id");
    let table_name = Spi::get_one::<String>(
        "SELECT path->0->>'table'
             FROM graph.traverse(
                'graph_test_users_pgtest'::regclass,
                'u1',
                1,
                direction := 'out',
                strategy := 'dfs',
                hydrate := false
             )
             WHERE node_id = 'u2'",
    )
    .expect("dfs path table failed")
    .expect("missing first path table");

    assert_eq!(path_id, "u1");
    assert!(table_name.ends_with("graph_test_users_pgtest"));
}

#[cfg(feature = "development")]
#[pg_test]
fn dfs_lazy_matches_eager_rls_out_in_any_and_hidden_intermediate() {
    build_workflow_rls_fixture();
    Spi::run("SET ROLE graph_workflow_rls_reader").expect("set DFS reader failed");
    for direction in ["out", "in", "any"] {
        let query = format!(
            "SELECT COALESCE(jsonb_agg(to_jsonb(t) ORDER BY depth, node_id), '[]'::jsonb)
               FROM graph.traverse(
                 'graph_test_users_pgtest'::regclass, 'u1', 3,
                 strategy := 'dfs', direction := '{direction}', hydrate := false
               ) AS t"
        );
        Spi::run("SELECT graph._test_set_visibility_strategy('eager')").unwrap();
        let eager = workflow_json(&query);
        Spi::run("SELECT graph._test_set_visibility_strategy('lazy')").unwrap();
        let lazy = workflow_json(&query);
        assert_eq!(lazy.0, eager.0, "DFS {direction} parity");
        assert!(lazy.0.to_string().contains("u2"));
        assert!(!lazy.0.to_string().contains("u3"));
        assert!(
            !lazy.0.to_string().contains("u4"),
            "the visible descendant behind hidden u3 must remain unreachable"
        );
    }
    Spi::run("RESET ROLE; SELECT graph._test_set_visibility_strategy('auto')").unwrap();
}

#[cfg(feature = "development")]
#[pg_test]
fn dfs_lazy_multiseed_matches_eager_and_selects_lazy() {
    build_workflow_rls_fixture();
    Spi::run("SET ROLE graph_workflow_rls_reader").expect("set DFS reader failed");
    let query = "SELECT COALESCE(
          jsonb_agg(to_jsonb(t) ORDER BY root_id, depth, node_id),
          '[]'::jsonb
        )
        FROM graph.traverse(
          ARRAY[
            'graph_test_users_pgtest'::regclass::oid,
            'graph_test_users_pgtest'::regclass::oid
          ],
          ARRAY['u1'::text, 'u2'::text],
          2,
          strategy := 'dfs', direction := 'out', hydrate := false
        ) AS t";
    Spi::run("SELECT graph._test_set_visibility_strategy('eager')").unwrap();
    let eager = workflow_json(query);
    Spi::run("SELECT graph._test_set_visibility_strategy('lazy')").unwrap();
    let lazy = workflow_json(query);
    let metrics = workflow_json("SELECT graph._test_visibility_metrics()");
    Spi::run("RESET ROLE; SELECT graph._test_set_visibility_strategy('auto')").unwrap();

    assert_eq!(lazy.0, eager.0);
    assert_eq!(metrics.0["strategy"].as_str(), Some("lazy"));
    assert!(metrics.0["spi_calls"].as_u64().unwrap_or_default() > 0);
}

#[cfg(feature = "development")]
#[pg_test]
fn dfs_lazy_traverse_search_matches_eager_and_selects_lazy() {
    build_workflow_rls_fixture();
    Spi::run("SET ROLE graph_workflow_rls_reader").expect("set DFS reader failed");
    let query = "SELECT COALESCE(
          jsonb_agg(to_jsonb(t) ORDER BY root_id, depth, node_id),
          '[]'::jsonb
        )
        FROM graph.traverse_search(
          'name', 'Alice',
          table_filter := 'graph_test_users_pgtest'::regclass,
          search_mode := 'exact', search_max_rows := 2,
          max_depth := 2, strategy := 'dfs', direction := 'out', hydrate := false
        ) AS t";
    Spi::run("SELECT graph._test_set_visibility_strategy('eager')").unwrap();
    let eager = workflow_json(query);
    Spi::run("SELECT graph._test_set_visibility_strategy('lazy')").unwrap();
    let lazy = workflow_json(query);
    let metrics = workflow_json("SELECT graph._test_visibility_metrics()");
    Spi::run("RESET ROLE; SELECT graph._test_set_visibility_strategy('auto')").unwrap();

    assert_eq!(lazy.0, eager.0);
    assert_eq!(metrics.0["strategy"].as_str(), Some("lazy"));
    assert!(metrics.0["spi_calls"].as_u64().unwrap_or_default() > 0);
}

#[cfg(feature = "development")]
#[pg_test]
fn dfs_lazy_relationship_rls_blocks_hidden_parallel_edge() {
    build_workflow_rls_fixture();
    Spi::run(
        "ALTER TABLE public.graph_test_friendships_pgtest ENABLE ROW LEVEL SECURITY;
         CREATE POLICY graph_dfs_all_nodes
           ON public.graph_test_users_pgtest FOR SELECT
           TO graph_workflow_rls_reader USING (true);
         CREATE POLICY graph_dfs_visible_edges
           ON public.graph_test_friendships_pgtest FOR SELECT
           TO graph_workflow_rls_reader USING (id = 'f4');
         SET ROLE graph_workflow_rls_reader;
         SELECT graph._test_set_visibility_strategy('lazy')",
    )
    .expect("configure DFS relationship RLS failed");
    let count = Spi::get_one::<i64>(
        "SELECT count(*) FROM graph.traverse(
           'graph_test_users_pgtest'::regclass, 'u1', 1,
           strategy := 'dfs', direction := 'out', include_start := false, hydrate := false
         )",
    )
    .expect("DFS relationship RLS failed")
    .unwrap_or_default();
    Spi::run(
        "RESET ROLE;
         DROP POLICY graph_dfs_visible_edges ON public.graph_test_friendships_pgtest;
         CREATE POLICY graph_dfs_visible_edges
           ON public.graph_test_friendships_pgtest FOR SELECT
           TO graph_workflow_rls_reader USING (false);
         SET ROLE graph_workflow_rls_reader",
    )
    .expect("hide every DFS relationship failed");
    let all_hidden_count = Spi::get_one::<i64>(
        "SELECT count(*) FROM graph.traverse(
           'graph_test_users_pgtest'::regclass, 'u1', 1,
           strategy := 'dfs', direction := 'out', include_start := false, hydrate := false
         )",
    )
    .expect("all-hidden DFS relationship RLS failed")
    .unwrap_or_default();
    Spi::run("RESET ROLE; SELECT graph._test_set_visibility_strategy('auto')").unwrap();
    assert_eq!(count, 1);
    assert_eq!(all_hidden_count, 0);
}

#[cfg(feature = "development")]
#[pg_test]
fn dfs_lazy_cancellation_and_policy_error_drop_state_then_retry() {
    build_workflow_rls_fixture();
    Spi::run(
        "SET ROLE graph_workflow_rls_reader;
         SELECT graph._test_set_visibility_strategy('lazy');
         SELECT graph._test_arm_lazy_visibility_cancel()",
    )
    .unwrap();
    let cancelled = workflow_cancellation(
        "SELECT * FROM graph.traverse(
           'graph_test_users_pgtest'::regclass, 'u1', 2,
           strategy := 'dfs', direction := 'out', include_start := false, hydrate := false
         )",
    );
    let retry = Spi::get_one::<i64>(
        "SELECT count(*) FROM graph.traverse(
           'graph_test_users_pgtest'::regclass, 'u1', 2,
           strategy := 'dfs', direction := 'out', include_start := false, hydrate := false
         )",
    )
    .expect("DFS retry failed")
    .unwrap_or_default();
    Spi::run(
        "RESET ROLE;
         DROP POLICY graph_workflow_visible_nodes ON public.graph_test_users_pgtest;
         CREATE FUNCTION public.graph_dfs_error_policy()
           RETURNS boolean LANGUAGE plpgsql VOLATILE AS $$
           BEGIN PERFORM 1 / 0; RETURN true; END $$;
         CREATE POLICY graph_dfs_error_policy
           ON public.graph_test_users_pgtest
           TO graph_workflow_rls_reader
           USING (public.graph_dfs_error_policy());
         GRANT EXECUTE ON FUNCTION public.graph_dfs_error_policy()
           TO graph_workflow_rls_reader;
         SET ROLE graph_workflow_rls_reader",
    )
    .expect("configure DFS policy error failed");
    let policy_rejected = std::sync::atomic::AtomicBool::new(false);
    pgrx::pg_sys::PgTryBuilder::new(std::panic::AssertUnwindSafe(|| {
        Spi::run(
            "SELECT * FROM graph.traverse(
               'graph_test_users_pgtest'::regclass, 'u1', 2,
               strategy := 'dfs', direction := 'out', hydrate := false
             )",
        )
        .expect("DFS policy error unexpectedly returned");
    }))
    .catch_when(pgrx::PgSqlErrorCode::ERRCODE_DIVISION_BY_ZERO, |_| {
        policy_rejected.store(true, std::sync::atomic::Ordering::Relaxed);
    })
    .execute();
    Spi::run(
        "RESET ROLE;
         DROP POLICY graph_dfs_error_policy ON public.graph_test_users_pgtest;
         CREATE POLICY graph_dfs_error_policy
           ON public.graph_test_users_pgtest
           TO graph_workflow_rls_reader USING (id <> 'u3');
         SET ROLE graph_workflow_rls_reader",
    )
    .expect("replace DFS policy error failed");
    let policy_retry = Spi::get_one::<i64>(
        "SELECT count(*) FROM graph.traverse(
           'graph_test_users_pgtest'::regclass, 'u1', 2,
           strategy := 'dfs', direction := 'out', include_start := false,
           hydrate := false
         )",
    )
    .expect("DFS policy-error retry failed")
    .unwrap_or_default();
    Spi::run("RESET ROLE; SELECT graph._test_set_visibility_strategy('auto')").unwrap();
    assert!(cancelled);
    assert!(policy_rejected.load(std::sync::atomic::Ordering::Relaxed));
    assert_eq!(retry, 1);
    assert_eq!(policy_retry, 1);
}

#[cfg(feature = "development")]
#[pg_test]
fn dfs_no_rls_fast_path_has_zero_visibility_spi() {
    reset_and_create_fixtures();
    build_friendship_fixture_graph();
    Spi::run("SELECT graph._test_set_visibility_strategy('auto')").unwrap();
    let count = Spi::get_one::<i64>(
        "SELECT count(*) FROM graph.traverse(
           'graph_test_users_pgtest'::regclass, 'u1', 1,
           strategy := 'dfs', direction := 'out', include_start := false, hydrate := false
         )",
    )
    .expect("no-RLS DFS failed")
    .unwrap_or_default();
    let metrics = workflow_json("SELECT graph._test_visibility_metrics()");
    assert_eq!(count, 1);
    assert_eq!(metrics.0["spi_calls"].as_u64(), Some(0));
}

#[pg_test]
fn traverse_accepts_in_direction_and_rejects_weighted_strategy() {
    reset_and_create_fixtures();
    Spi::run(
        "SELECT graph.add_table(
                'graph_test_users_pgtest'::regclass,
                id_column := 'id',
                columns := ARRAY['name']
            )",
    )
    .expect("add users table failed");
    Spi::run("SELECT * FROM graph.build()").expect("build failed");

    let in_result = super::validate_traverse_options("in", None, "bfs", "node_global");
    let weighted_result = super::validate_traverse_options("any", None, "weighted", "node_global");
    let invalid_uniqueness = super::validate_traverse_options("any", None, "bfs", "node_local");

    assert!(in_result.is_ok());
    assert!(weighted_result.is_err());
    assert!(invalid_uniqueness.is_err());
}

#[pg_test]
fn aggregation_traversal_direction_both_maps_to_any() {
    reset_and_create_fixtures();
    let node_ref = Spi::get_one::<String>(
        "SELECT graph.node_ref_string('graph_test_users_pgtest'::regclass, 'u1')",
    )
    .expect("node_ref_string failed")
    .expect("node_ref_string returned null");
    let request = super::sql_aggregation::parse_aggregation_traversal_request(&serde_json::json!({
        "starts": [node_ref],
        "direction": "both"
    }))
    .expect("parse aggregation traversal request failed");

    assert_eq!(request.direction, super::types::TraversalDirection::Any);
}
