mushroomdb 0.6.9

//! Tests for Task 2: Engine authz core at the write choke point.
//!
//! Decision-table row → test name mapping:
//! CREATE-class:
//!   Row 1 (scope-before-lookup, label not in create_labels):
//!     test_create_scope_denied_empty_store
//!   Row 2 (key exists and visible → DuplicateKey):
//!     test_create_visible_collision
//!   Row 3 (key exists and hidden → not-visible):
//!     test_create_hidden_collision
//!   Row 4 (key absent → proceed):
//!     test_create_allowed
//!
//! UPDATE/DELETE-class (SetProp):
//!   Visible + in update_labels → allowed:
//!     test_update_visible_allowed
//!   Visible + NOT in update_labels → scope-denied:
//!     test_update_scope_denied
//!   Hidden ≡ absent, EXACT-EQUAL error:
//!     test_update_hidden_identical_to_absent
//!
//! DELETE-class (DeleteNode):
//!   Visible + in delete_labels → allowed:
//!     test_delete_node_allowed
//!   Visible + NOT in delete_labels → scope-denied:
//!     test_delete_node_scope_denied
//!
//! DeleteEdge (derived-edge rejection before scope check):
//!   Derived edge, NOT in delete_edge_types → RuleOwned (not scope-denied):
//!     test_delete_edge_derived_before_scope
//!   In delete_edge_types, both endpoints visible → allowed:
//!     test_delete_edge_allowed
//!   Endpoint hidden → endpoint-not-visible:
//!     test_delete_edge_hidden_endpoint
//!   NOT in delete_edge_types → scope-denied:
//!     test_delete_edge_type_not_scoped
//!
//! MERGE:
//!   Neither create nor update scope → scope-denied WITHOUT key lookup:
//!     test_merge_unscoped_no_key_lookup
//!   Key absent + create scope → create arm:
//!     test_merge_create_arm
//!   Key visible + update scope → match arm:
//!     test_merge_match_arm
//!   Key hidden → not-visible:
//!     test_merge_hidden_key
//!   Update-only role: hidden ≡ absent (byte-equal "not visible", §6.1 oracle):
//!     test_merge_update_only_hidden_eq_absent
//!   Create-only role: hidden→not-visible, absent→create (accepted disclosure):
//!     test_merge_create_only_disclosure_pinned
//!
//! EDGE-CREATE:
//!   Both endpoints visible, type in create_edge_types → allowed:
//!     test_edge_create_both_visible
//!   One endpoint hidden → endpoint-not-visible:
//!     test_edge_create_one_hidden
//!   Type not in create_edge_types → scope-denied:
//!     test_edge_create_type_not_scoped
//!
//! InsertEdgeUpsert placeholder:
//!   Endpoint created by prior InsertNode in same batch counts as visible:
//!     test_upsert_placeholder_counts_as_visible
//!
//! Cross-cutting:
//!   Batch atomicity (no WAL frame on deny):
//!     test_batch_atomicity_no_wal_on_deny
//!   Authz before CAS (hidden node → not-visible, never CasConflict):
//!     test_authz_fires_before_cas_would
//!   None authz = full authority (zero-cost bypass):
//!     test_none_authz_full_authority
//!   RenameNode with Some(authz) → endpoint-not-permitted:
//!     test_rename_node_forbidden_for_role
//!   CreateRule with Some(authz) → endpoint-not-permitted:
//!     test_create_rule_forbidden_for_role
//!   Rules fire on role-created nodes; derived edges to hidden neighbors
//!   are masked on read:
//!     test_rules_fire_but_hidden_edges_masked

use core_api::{
    schema::Schema, BatchOp, Direction, GraphDb, GraphError, Predicate, RoleDef, RuleDef, Value,
    WriteScope, MERGE_CREATE_NEEDS_ONE_NAMESPACE,
};
use std::collections::BTreeMap;

// ── helpers ──────────────────────────────────────────────────────────────────

fn tmp(name: &str) -> std::path::PathBuf {
    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .expect("clock")
        .as_nanos();
    let d = std::env::temp_dir().join(format!(
        "graphdb-ws-{}-{}-{}",
        name,
        std::process::id(),
        nanos
    ));
    let _ = std::fs::remove_dir_all(&d);
    d
}

fn no_params() -> BTreeMap<String, Value> {
    BTreeMap::new()
}

fn wal_len(dir: &std::path::Path) -> u64 {
    std::fs::metadata(dir.join("wal.bin"))
        .map(|m| m.len())
        .unwrap_or(0)
}

/// Role with full write scope over "MyLabel" nodes and "KNOWS" edges.
/// Labels visible to the role: ["MyLabel", "Visible"].
/// Hidden label (not in role): "Secret".
fn writer_role() -> RoleDef {
    RoleDef {
        name: "writer".into(),
        keys: vec![],
        labels: vec!["MyLabel".into(), "Visible".into()],
        visible_where: None,
        namespaces: None,
        write: Some(WriteScope {
            create_labels: vec!["MyLabel".into()],
            update_labels: vec!["MyLabel".into(), "Visible".into()],
            delete_labels: vec!["MyLabel".into()],
            create_edge_types: vec!["KNOWS".into()],
            delete_edge_types: vec!["KNOWS".into()],
        }),
    }
}

/// Open a fresh DB, apply the writer role schema.
fn open_with_writer(name: &str) -> (GraphDb<core_storage::fs::RealFs>, std::path::PathBuf) {
    let dir = tmp(name);
    let mut db = GraphDb::open(&dir).unwrap();
    let schema = Schema {
        fulltext: vec![],
        indexes: vec![],
        rules: vec![],
        views: vec![],
        roles: vec![writer_role()],
    };
    db.apply_schema(&schema).unwrap();
    (db, dir)
}

/// Build WriteAuthz for "writer" from the live DB state.
fn writer_authz(db: &mut GraphDb<core_storage::fs::RealFs>) -> core_api::WriteAuthz {
    let roles = db.roles();
    let def = roles.iter().find(|r| r.name == "writer").unwrap();
    let scope = def.write.clone().unwrap();
    let mask = db.mask_for_role("writer").unwrap();
    core_api::WriteAuthz {
        role: "writer".into(),
        scope,
        mask,
    }
}

fn is_role_write_denied(e: &GraphError) -> bool {
    matches!(e, GraphError::RoleWriteDenied { .. })
}

fn denied_reason(e: &GraphError) -> &str {
    match e {
        GraphError::RoleWriteDenied { reason } => reason.as_str(),
        _ => panic!("expected RoleWriteDenied, got {e:?}"),
    }
}

// ── CREATE-class ─────────────────────────────────────────────────────────────

/// Decision table row 4: key absent + label in create_labels → proceed.
#[test]
fn test_create_allowed() {
    let (mut db, _dir) = open_with_writer("create-allowed");
    let authz = writer_authz(&mut db);
    let ops = vec![BatchOp::InsertNode {
        label: "MyLabel".into(),
        key: "n1".into(),
        props: vec![],
    }];
    let (nodes, _) = db.write_batch_authz(Some(&authz), ops).unwrap();
    assert_eq!(nodes, 1);
    assert!(db.has_node("n1"));
}

/// Decision table row 1 (scope-before-lookup): label NOT in create_labels fires
/// even when the store is EMPTY (no key lookup precedes the scope check).
/// This is the structural closure of the §6.2 timing-oracle item.
#[test]
fn test_create_scope_denied_empty_store() {
    let (mut db, _dir) = open_with_writer("create-scope-denied");
    let authz = writer_authz(&mut db);
    // Store is empty — no nodes exist yet.
    assert!(!db.has_node("x"));
    let ops = vec![BatchOp::InsertNode {
        label: "Secret".into(), // NOT in create_labels
        key: "x".into(),
        props: vec![],
    }];
    let err = db.write_batch_authz(Some(&authz), ops).unwrap_err();
    assert!(
        is_role_write_denied(&err),
        "expected RoleWriteDenied, got {err:?}"
    );
    assert!(
        denied_reason(&err).contains("create_labels"),
        "reason should name create_labels: {}",
        denied_reason(&err)
    );
    // Store remains empty: scope denial fired without key lookup.
    assert!(!db.has_node("x"));
}

/// Decision table row 3: key exists and HIDDEN → 403 target-not-visible.
#[test]
fn test_create_hidden_collision() {
    let (mut db, _dir) = open_with_writer("create-hidden-collision");
    // Insert a "Secret" node as admin (no authz).
    db.insert_node("Secret", "secret_key", vec![]).unwrap();
    let authz = writer_authz(&mut db);
    // Try to create "MyLabel" node with the same key.
    let ops = vec![BatchOp::InsertNode {
        label: "MyLabel".into(),
        key: "secret_key".into(),
        props: vec![],
    }];
    let err = db.write_batch_authz(Some(&authz), ops).unwrap_err();
    assert!(
        is_role_write_denied(&err),
        "expected RoleWriteDenied, got {err:?}"
    );
    assert_eq!(
        denied_reason(&err),
        "role-bound token: target node not visible",
        "hidden key must return not-visible, not DuplicateKey"
    );
}

/// Decision table row 2: key exists and VISIBLE → DuplicateKey (existing behavior).
#[test]
fn test_create_visible_collision() {
    let (mut db, _dir) = open_with_writer("create-visible-collision");
    // Insert visible node as admin.
    db.insert_node("MyLabel", "alice", vec![]).unwrap();
    let authz = writer_authz(&mut db);
    let ops = vec![BatchOp::InsertNode {
        label: "MyLabel".into(),
        key: "alice".into(),
        props: vec![],
    }];
    let err = db.write_batch_authz(Some(&authz), ops).unwrap_err();
    assert!(
        matches!(err, GraphError::DuplicateKey { .. }),
        "visible collision must return DuplicateKey, got {err:?}"
    );
}

// ── UPDATE-class ──────────────────────────────────────────────────────────────

/// Visible node + label in update_labels → SetProp succeeds.
#[test]
fn test_update_visible_allowed() {
    let (mut db, _dir) = open_with_writer("update-allowed");
    db.insert_node("MyLabel", "alice", vec![]).unwrap();
    let authz = writer_authz(&mut db);
    let ops = vec![BatchOp::SetProp {
        key: "alice".into(),
        field: "name".into(),
        value: Value::Str("Alice".into()),
    }];
    db.write_batch_authz(Some(&authz), ops).unwrap();
    assert_eq!(
        db.get_prop("alice", "name"),
        Some(Value::Str("Alice".into()))
    );
}

/// Visible node + label NOT in update_labels → scope-denied.
#[test]
fn test_update_scope_denied() {
    // "Visible" is in role's read labels and update_labels, but NOT update_labels for... wait.
    // writer_role has update_labels: ["MyLabel", "Visible"]. Let me use a label only in read.
    // Actually let me use a separate role for this test.
    let dir = tmp("update-scope-denied");
    let mut db = GraphDb::open(&dir).unwrap();
    let schema = Schema {
        fulltext: vec![],
        indexes: vec![],
        rules: vec![],
        views: vec![],
        roles: vec![RoleDef {
            name: "reader_writer".into(),
            keys: vec![],
            labels: vec!["MyLabel".into(), "ReadOnly".into()],
            visible_where: None,
            namespaces: None,
            write: Some(WriteScope {
                create_labels: vec!["MyLabel".into()],
                update_labels: vec!["MyLabel".into()], // ReadOnly NOT in update_labels
                delete_labels: vec![],
                create_edge_types: vec![],
                delete_edge_types: vec![],
            }),
        }],
    };
    db.apply_schema(&schema).unwrap();
    // Insert a ReadOnly-labeled node as admin.
    db.insert_node("ReadOnly", "ro_node", vec![]).unwrap();
    // Build authz.
    let roles = db.roles();
    let def = roles.iter().find(|r| r.name == "reader_writer").unwrap();
    let scope = def.write.clone().unwrap();
    let mask = db.mask_for_role("reader_writer").unwrap();
    let authz = core_api::WriteAuthz {
        role: "reader_writer".into(),
        scope,
        mask,
    };
    let ops = vec![BatchOp::SetProp {
        key: "ro_node".into(),
        field: "x".into(),
        value: Value::Int(1),
    }];
    let err = db.write_batch_authz(Some(&authz), ops).unwrap_err();
    assert!(
        is_role_write_denied(&err),
        "expected RoleWriteDenied, got {err:?}"
    );
    assert!(
        denied_reason(&err).contains("update_labels"),
        "reason should name update_labels: {}",
        denied_reason(&err)
    );
}

/// Hidden node and absent node return EXACT-EQUAL errors (spec §3.1).
#[test]
fn test_update_hidden_identical_to_absent() {
    let (mut db, _dir) = open_with_writer("update-hidden-absent");
    // Insert a hidden node (label not in role's read labels).
    db.insert_node("Secret", "hidden_node", vec![]).unwrap();
    let authz = writer_authz(&mut db);

    let hidden_ops = vec![BatchOp::SetProp {
        key: "hidden_node".into(),
        field: "x".into(),
        value: Value::Int(1),
    }];
    let absent_ops = vec![BatchOp::SetProp {
        key: "nonexistent".into(), // does not exist at all
        field: "x".into(),
        value: Value::Int(1),
    }];

    let err_hidden = db.write_batch_authz(Some(&authz), hidden_ops).unwrap_err();
    let err_absent = db.write_batch_authz(Some(&authz), absent_ops).unwrap_err();

    // EXACT equality: same error variant, same reason string.
    assert_eq!(
        denied_reason(&err_hidden),
        denied_reason(&err_absent),
        "hidden and absent must produce identical error messages (spec §3.1)"
    );
    assert_eq!(
        denied_reason(&err_hidden),
        "role-bound token: target node not visible"
    );
}

// ── DELETE-class: DeleteNode ──────────────────────────────────────────────────

/// Visible node + label in delete_labels → DeleteNode succeeds.
#[test]
fn test_delete_node_allowed() {
    let (mut db, _dir) = open_with_writer("delete-node-allowed");
    db.insert_node("MyLabel", "del_me", vec![]).unwrap();
    let authz = writer_authz(&mut db);
    let ops = vec![BatchOp::DeleteNode {
        key: "del_me".into(),
    }];
    db.write_batch_authz(Some(&authz), ops).unwrap();
    assert!(!db.has_node("del_me"));
}

/// Visible node + label NOT in delete_labels → scope-denied.
#[test]
fn test_delete_node_scope_denied() {
    let (mut db, _dir) = open_with_writer("delete-node-scope-denied");
    // "Visible" is in role labels but NOT in delete_labels (which only has "MyLabel").
    db.insert_node("Visible", "vis_node", vec![]).unwrap();
    let authz = writer_authz(&mut db);
    let ops = vec![BatchOp::DeleteNode {
        key: "vis_node".into(),
    }];
    let err = db.write_batch_authz(Some(&authz), ops).unwrap_err();
    assert!(
        is_role_write_denied(&err),
        "expected RoleWriteDenied, got {err:?}"
    );
    assert!(
        denied_reason(&err).contains("delete_labels"),
        "reason should name delete_labels: {}",
        denied_reason(&err)
    );
}

// ── DELETE-class: DeleteEdge (derived-edge rejection order) ──────────────────

fn derived_rule() -> RuleDef {
    RuleDef {
        name: "sim".into(),
        src_label: "MyLabel".into(),
        dst_label: "MyLabel".into(),
        predicate: Predicate::Overlap {
            field: "tags".into(),
            min: 0.5,
        },
        edge_type: "SIMILAR".into(),
        weight_prop: None,
        max_edges: None,
        approximate: false,
        via_label: None,
        via_edge: None,
        via_dir: None,
        namespace: None,
    }
}

/// Derived edge + NOT in delete_edge_types → RuleOwned fires BEFORE scope-denied.
/// The order: derived-edge rejection precedes delete_edge_types check (spec §3.5).
#[test]
fn test_delete_edge_derived_before_scope() {
    let dir = tmp("delete-edge-derived-order");
    let mut db = GraphDb::open(&dir).unwrap();
    let schema = Schema {
        fulltext: vec![],
        indexes: vec![],
        rules: vec![derived_rule()],
        views: vec![],
        roles: vec![writer_role()],
    };
    db.apply_schema(&schema).unwrap();
    // Insert two MyLabel nodes with overlapping tags → rule derives SIMILAR edge.
    db.insert_node(
        "MyLabel",
        "a",
        vec![("tags".into(), Value::List(vec![Value::Str("x".into())]))],
    )
    .unwrap();
    db.insert_node(
        "MyLabel",
        "b",
        vec![("tags".into(), Value::List(vec![Value::Str("x".into())]))],
    )
    .unwrap();
    // SIMILAR is derived but NOT in delete_edge_types (only KNOWS is).
    let authz = writer_authz(&mut db);
    let ops = vec![BatchOp::DeleteEdge {
        edge_type: "SIMILAR".into(),
        src_key: "a".into(),
        dst_key: "b".into(),
    }];
    let err = db.write_batch_authz(Some(&authz), ops).unwrap_err();
    // Must be RuleOwned (derived-edge check fires first), NOT RoleWriteDenied scope.
    assert!(
        matches!(err, GraphError::RuleOwned { .. }),
        "derived-edge rejection must precede scope check; got {err:?}"
    );
}

/// In delete_edge_types, both endpoints visible → DeleteEdge succeeds.
#[test]
fn test_delete_edge_allowed() {
    let (mut db, _dir) = open_with_writer("delete-edge-allowed");
    db.insert_node("MyLabel", "a", vec![]).unwrap();
    db.insert_node("MyLabel", "b", vec![]).unwrap();
    db.insert_edge("KNOWS", "a", "b").unwrap();
    let authz = writer_authz(&mut db);
    let ops = vec![BatchOp::DeleteEdge {
        edge_type: "KNOWS".into(),
        src_key: "a".into(),
        dst_key: "b".into(),
    }];
    db.write_batch_authz(Some(&authz), ops).unwrap();
    let neighbors = db
        .neighbors("a", "KNOWS", Direction::Out)
        .unwrap_or_default();
    assert!(
        !neighbors.contains(&"b".to_string()),
        "edge should be deleted"
    );
}

/// Edge with one hidden endpoint → endpoint-not-visible.
#[test]
fn test_delete_edge_hidden_endpoint() {
    let (mut db, _dir) = open_with_writer("delete-edge-hidden-ep");
    db.insert_node("MyLabel", "a", vec![]).unwrap();
    db.insert_node("Secret", "secret_b", vec![]).unwrap();
    // Insert the edge as admin (no authz).
    db.insert_edge("KNOWS", "a", "secret_b").unwrap();
    let authz = writer_authz(&mut db);
    let ops = vec![BatchOp::DeleteEdge {
        edge_type: "KNOWS".into(),
        src_key: "a".into(),
        dst_key: "secret_b".into(),
    }];
    let err = db.write_batch_authz(Some(&authz), ops).unwrap_err();
    assert!(
        is_role_write_denied(&err),
        "expected RoleWriteDenied, got {err:?}"
    );
    assert_eq!(
        denied_reason(&err),
        "role-bound token: edge endpoint not visible"
    );
}

/// Edge type NOT in delete_edge_types → scope-denied.
#[test]
fn test_delete_edge_type_not_scoped() {
    let (mut db, _dir) = open_with_writer("delete-edge-unscoped");
    db.insert_node("MyLabel", "a", vec![]).unwrap();
    db.insert_node("MyLabel", "b", vec![]).unwrap();
    db.insert_edge("UNSCOPED_TYPE", "a", "b").unwrap();
    let authz = writer_authz(&mut db);
    let ops = vec![BatchOp::DeleteEdge {
        edge_type: "UNSCOPED_TYPE".into(),
        src_key: "a".into(),
        dst_key: "b".into(),
    }];
    let err = db.write_batch_authz(Some(&authz), ops).unwrap_err();
    assert!(
        is_role_write_denied(&err),
        "expected RoleWriteDenied, got {err:?}"
    );
    assert!(
        denied_reason(&err).contains("delete_edge_types"),
        "reason should name delete_edge_types: {}",
        denied_reason(&err)
    );
}

// ── MERGE (via query_write_authz) ─────────────────────────────────────────────

/// MERGE scope precondition: neither create nor update scope for label → 403
/// WITHOUT a key lookup (timing-oracle closure, spec §6.2).
#[test]
fn test_merge_unscoped_no_key_lookup() {
    let (mut db, _dir) = open_with_writer("merge-unscoped");
    // The role has no create_labels or update_labels for "Secret".
    let err = db
        .query_write_authz("writer", "MERGE (n:Secret {id: 'x'})", &no_params())
        .unwrap_err();
    assert!(
        is_role_write_denied(&err),
        "expected RoleWriteDenied for unscoped MERGE, got {err:?}"
    );
    assert_eq!(
        denied_reason(&err),
        "role-bound token: label 'Secret' not in write scope (create_labels)",
        "unscoped MERGE must return exact §4.3 reason string"
    );
    // Key "x" must not exist (no key lookup before scope denial).
    assert!(
        !db.has_node("x"),
        "scope denial must fire before key lookup"
    );
}

/// MERGE create arm: key absent + label in create_labels → creates node.
#[test]
fn test_merge_create_arm() {
    let (mut db, _dir) = open_with_writer("merge-create");
    let result = db
        .query_write_authz("writer", "MERGE (n:MyLabel {id: 'new_node'})", &no_params())
        .unwrap();
    assert!(
        db.has_node("new_node"),
        "MERGE create arm must create the node"
    );
    let _ = result;
}

/// MERGE match arm: key visible + label in update_labels → updates props.
#[test]
fn test_merge_match_arm() {
    let (mut db, _dir) = open_with_writer("merge-match");
    db.insert_node("MyLabel", "existing", vec![]).unwrap();
    db.query_write_authz(
        "writer",
        "MERGE (n:MyLabel {id: 'existing'}) ON MATCH SET n.updated = 1",
        &no_params(),
    )
    .unwrap();
    assert_eq!(
        db.get_prop("existing", "updated"),
        Some(Value::Int(1)),
        "MERGE match arm must update the property"
    );
}

/// MERGE hidden key: key exists but hidden → not-visible.
#[test]
fn test_merge_hidden_key() {
    let (mut db, _dir) = open_with_writer("merge-hidden");
    // Insert a hidden node with label "MyLabel" but key that will collide.
    // Wait — for MERGE, the node's LABEL is the one in MERGE stmt. If we do
    // MERGE (n:MyLabel {id: 'hidden_key'}), and hidden_key exists as "Secret",
    // then the stored node has a different label than the MERGE target.
    // The MERGE just uses the id key; the label in MERGE is the declared label.
    // For the hidden test, we need a node with the MERGE key hidden.
    // Insert a "Secret"-labeled node with key "hidden_key" (hidden from role).
    db.insert_node("Secret", "hidden_key", vec![]).unwrap();
    let err = db
        .query_write_authz(
            "writer",
            "MERGE (n:MyLabel {id: 'hidden_key'})",
            &no_params(),
        )
        .unwrap_err();
    assert!(
        is_role_write_denied(&err),
        "MERGE on hidden key must return RoleWriteDenied, got {err:?}"
    );
    assert_eq!(
        denied_reason(&err),
        "role-bound token: target node not visible"
    );
}

/// Update-only role (create_labels empty, update_labels=["MyLabel"]):
/// MERGE on a hidden key → "target node not visible".
/// MERGE on an absent key → "target node not visible".
/// The two bodies must be BYTE-EQUAL (spec §6.1 "confirm existence of hidden
/// nodes: No"; closes the update-only MERGE existence oracle).
#[test]
fn test_merge_update_only_hidden_eq_absent() {
    let dir = tmp("merge-update-only");
    let mut db = GraphDb::open(&dir).unwrap();
    let schema = Schema {
        fulltext: vec![],
        indexes: vec![],
        rules: vec![],
        views: vec![],
        roles: vec![RoleDef {
            name: "updater".into(),
            keys: vec![],
            labels: vec!["MyLabel".into()],
            visible_where: None,
            namespaces: None,
            write: Some(WriteScope {
                create_labels: vec![], // no create scope
                update_labels: vec!["MyLabel".into()],
                delete_labels: vec![],
                create_edge_types: vec![],
                delete_edge_types: vec![],
            }),
        }],
    };
    db.apply_schema(&schema).unwrap();

    // Insert a node hidden from the updater role.
    db.insert_node("Secret", "hidden_node", vec![]).unwrap();

    // MERGE on hidden key.
    let err_hidden = db
        .query_write_authz(
            "updater",
            "MERGE (n:MyLabel {id: 'hidden_node'})",
            &no_params(),
        )
        .unwrap_err();
    // MERGE on absent key (store has no "absent_node").
    let err_absent = db
        .query_write_authz(
            "updater",
            "MERGE (n:MyLabel {id: 'absent_node'})",
            &no_params(),
        )
        .unwrap_err();

    // Both must be RoleWriteDenied.
    assert!(is_role_write_denied(&err_hidden), "hidden: {err_hidden:?}");
    assert!(is_role_write_denied(&err_absent), "absent: {err_absent:?}");

    // Bodies must be byte-identical (no oracle).
    assert_eq!(
        denied_reason(&err_hidden),
        denied_reason(&err_absent),
        "update-only role: hidden and absent MERGE bodies must be equal"
    );
    assert_eq!(
        denied_reason(&err_hidden),
        "role-bound token: target node not visible"
    );

    // Confirm neither node was created.
    assert!(
        !db.has_node("absent_node"),
        "absent node must not be created by update-only MERGE"
    );
}

/// Create+update role: absent key → create arm (unchanged; pin against over-correction).
/// This is the accepted structural key-existence disclosure (§THREAT-MODEL):
/// hidden→not-visible, absent→create arm with create scope.
#[test]
fn test_merge_create_only_disclosure_pinned() {
    // create-only role: create_labels=["MyLabel"], update_labels=[].
    // hidden → not-visible; absent → create arm (accepted disclosure).
    let dir = tmp("merge-create-only-disclosure");
    let mut db = GraphDb::open(&dir).unwrap();
    let schema = Schema {
        fulltext: vec![],
        indexes: vec![],
        rules: vec![],
        views: vec![],
        roles: vec![RoleDef {
            name: "creator".into(),
            keys: vec![],
            labels: vec!["MyLabel".into()],
            visible_where: None,
            namespaces: None,
            write: Some(WriteScope {
                create_labels: vec!["MyLabel".into()],
                update_labels: vec![],
                delete_labels: vec![],
                create_edge_types: vec![],
                delete_edge_types: vec![],
            }),
        }],
    };
    db.apply_schema(&schema).unwrap();

    // Hidden key → not-visible (not a new DuplicateKey disclosure).
    db.insert_node("Secret", "hidden_key", vec![]).unwrap();
    let err_hidden = db
        .query_write_authz(
            "creator",
            "MERGE (n:MyLabel {id: 'hidden_key'})",
            &no_params(),
        )
        .unwrap_err();
    assert_eq!(
        denied_reason(&err_hidden),
        "role-bound token: target node not visible",
        "create-only: hidden key must be not-visible"
    );

    // Absent key → create arm proceeds (the structural disclosure: key-existence
    // via collision, accepted in the threat model).
    db.query_write_authz(
        "creator",
        "MERGE (n:MyLabel {id: 'new_node'})",
        &no_params(),
    )
    .unwrap();
    assert!(
        db.has_node("new_node"),
        "create-only role: absent key must create the node"
    );
}

/// MERGE ON CREATE SET under a scoped role: InsertNode + SetProp arrive in the
/// same batch.  The SetProp must see the batch-created node as Visible and must
/// succeed without checking update_labels (ruling §3.5: batch-created nodes are
/// updatable by the same batch that created them).
#[test]
fn test_merge_on_create_set_with_role_authz() {
    // Writer role: create_labels=["MyLabel"], update_labels=["MyLabel","Visible"].
    // Confirm ON CREATE SET succeeds (create + update in scope).
    let (mut db, _dir) = open_with_writer("merge-on-create-set");
    db.query_write_authz(
        "writer",
        "MERGE (n:MyLabel {id: 'new_mc'}) ON CREATE SET n.created = 1",
        &no_params(),
    )
    .unwrap();
    assert!(
        db.has_node("new_mc"),
        "MERGE ON CREATE SET must create the node"
    );
    assert_eq!(
        db.get_prop("new_mc", "created"),
        Some(Value::Int(1)),
        "ON CREATE SET property must be applied"
    );

    // Create-only role: create_labels=["MyLabel"], update_labels=[].
    // ON CREATE SET on the batch-created node must STILL succeed (ruling §3.5).
    let dir2 = tmp("merge-on-create-set-create-only");
    let mut db2 = GraphDb::open(&dir2).unwrap();
    let schema = core_api::schema::Schema {
        fulltext: vec![],
        indexes: vec![],
        rules: vec![],
        views: vec![],
        roles: vec![RoleDef {
            name: "creator".into(),
            keys: vec![],
            labels: vec!["MyLabel".into()],
            visible_where: None,
            namespaces: None,
            write: Some(WriteScope {
                create_labels: vec!["MyLabel".into()],
                update_labels: vec![], // empty — no update scope
                delete_labels: vec![],
                create_edge_types: vec![],
                delete_edge_types: vec![],
            }),
        }],
    };
    db2.apply_schema(&schema).unwrap();
    // ON CREATE SET on a batch-created node bypasses update_labels (ruling).
    db2.query_write_authz(
        "creator",
        "MERGE (n:MyLabel {id: 'creator_node'}) ON CREATE SET n.x = 42",
        &no_params(),
    )
    .unwrap();
    assert!(db2.has_node("creator_node"));
    assert_eq!(
        db2.get_prop("creator_node", "x"),
        Some(Value::Int(42)),
        "batch-created node updatable by same batch regardless of update_labels"
    );
}

// ── EDGE-CREATE ───────────────────────────────────────────────────────────────

/// Both endpoints visible + type in create_edge_types → InsertEdge succeeds.
#[test]
fn test_edge_create_both_visible() {
    let (mut db, _dir) = open_with_writer("edge-create-both-vis");
    db.insert_node("MyLabel", "a", vec![]).unwrap();
    db.insert_node("MyLabel", "b", vec![]).unwrap();
    let authz = writer_authz(&mut db);
    let ops = vec![BatchOp::InsertEdge {
        edge_type: "KNOWS".into(),
        src_key: "a".into(),
        dst_key: "b".into(),
    }];
    db.write_batch_authz(Some(&authz), ops).unwrap();
    let neighbors = db
        .neighbors("a", "KNOWS", Direction::Out)
        .unwrap_or_default();
    assert!(neighbors.contains(&"b".to_string()), "edge should exist");
}

/// One endpoint hidden → edge endpoint not visible.
#[test]
fn test_edge_create_one_hidden() {
    let (mut db, _dir) = open_with_writer("edge-create-hidden-ep");
    db.insert_node("MyLabel", "a", vec![]).unwrap();
    db.insert_node("Secret", "hidden_b", vec![]).unwrap();
    let authz = writer_authz(&mut db);
    let ops = vec![BatchOp::InsertEdge {
        edge_type: "KNOWS".into(),
        src_key: "a".into(),
        dst_key: "hidden_b".into(),
    }];
    let err = db.write_batch_authz(Some(&authz), ops).unwrap_err();
    assert!(
        is_role_write_denied(&err),
        "expected RoleWriteDenied, got {err:?}"
    );
    assert_eq!(
        denied_reason(&err),
        "role-bound token: edge endpoint not visible"
    );
}

/// Edge type NOT in create_edge_types → scope-denied (before endpoint lookup).
#[test]
fn test_edge_create_type_not_scoped() {
    let (mut db, _dir) = open_with_writer("edge-create-unscoped");
    db.insert_node("MyLabel", "a", vec![]).unwrap();
    db.insert_node("MyLabel", "b", vec![]).unwrap();
    let authz = writer_authz(&mut db);
    let ops = vec![BatchOp::InsertEdge {
        edge_type: "UNSCOPED".into(), // not in create_edge_types
        src_key: "a".into(),
        dst_key: "b".into(),
    }];
    let err = db.write_batch_authz(Some(&authz), ops).unwrap_err();
    assert!(
        is_role_write_denied(&err),
        "expected RoleWriteDenied, got {err:?}"
    );
    assert!(
        denied_reason(&err).contains("create_edge_types"),
        "reason should name create_edge_types: {}",
        denied_reason(&err)
    );
}

// ── InsertEdgeUpsert: placeholder counts as visible ───────────────────────────

/// A placeholder endpoint created by an earlier InsertNode in the SAME batch
/// counts as visible for InsertEdgeUpsert endpoint visibility check.
#[test]
fn test_upsert_placeholder_counts_as_visible() {
    let (mut db, _dir) = open_with_writer("upsert-placeholder");
    let authz = writer_authz(&mut db);
    // Both src and dst don't exist yet. We create them via InsertNode first
    // in the same batch, then insert the edge via InsertEdgeUpsert.
    let ops = vec![
        BatchOp::InsertNode {
            label: "MyLabel".into(),
            key: "new_src".into(),
            props: vec![],
        },
        BatchOp::InsertNode {
            label: "MyLabel".into(),
            key: "new_dst".into(),
            props: vec![],
        },
        BatchOp::InsertEdge {
            edge_type: "KNOWS".into(),
            src_key: "new_src".into(),
            dst_key: "new_dst".into(),
        },
    ];
    db.write_batch_authz(Some(&authz), ops).unwrap();
    assert!(db.has_node("new_src"));
    assert!(db.has_node("new_dst"));
    let neighbors = db
        .neighbors("new_src", "KNOWS", Direction::Out)
        .unwrap_or_default();
    assert!(
        neighbors.contains(&"new_dst".to_string()),
        "edge should exist"
    );
}

/// InsertEdgeUpsert: placeholder endpoint created by same batch is visible.
#[test]
fn test_upsert_direct_placeholder_visible() {
    let (mut db, _dir) = open_with_writer("upsert-direct-placeholder");
    let authz = writer_authz(&mut db);
    // Use InsertEdgeUpsert directly — both endpoints absent, will be created
    // with placeholder_label = "MyLabel" which IS in create_labels.
    let ops = vec![BatchOp::InsertEdgeUpsert {
        edge_type: "KNOWS".into(),
        src_key: "upsert_src".into(),
        dst_key: "upsert_dst".into(),
        placeholder_label: "MyLabel".into(),
    }];
    db.write_batch_authz(Some(&authz), ops).unwrap();
    assert!(db.has_node("upsert_src"));
    assert!(db.has_node("upsert_dst"));
    let neighbors = db
        .neighbors("upsert_src", "KNOWS", Direction::Out)
        .unwrap_or_default();
    assert!(
        neighbors.contains(&"upsert_dst".to_string()),
        "edge should exist"
    );
}

/// InsertEdgeUpsert with placeholder_label NOT in create_labels → scope-denied.
#[test]
fn test_upsert_placeholder_not_in_create_labels() {
    let (mut db, _dir) = open_with_writer("upsert-unscoped-placeholder");
    let authz = writer_authz(&mut db);
    let ops = vec![BatchOp::InsertEdgeUpsert {
        edge_type: "KNOWS".into(),
        src_key: "x".into(),
        dst_key: "y".into(),
        placeholder_label: "Secret".into(), // NOT in create_labels
    }];
    let err = db.write_batch_authz(Some(&authz), ops).unwrap_err();
    assert!(
        is_role_write_denied(&err),
        "expected RoleWriteDenied, got {err:?}"
    );
}

// ── Cross-cutting ─────────────────────────────────────────────────────────────

/// Batch atomicity: if any op denies, NO WAL frame is written.
#[test]
fn test_batch_atomicity_no_wal_on_deny() {
    let (mut db, dir) = open_with_writer("batch-atomic");
    db.insert_node("MyLabel", "existing", vec![]).unwrap();
    let seq_before = db.commit_seq();
    let wal_before = wal_len(&dir);

    let authz = writer_authz(&mut db);
    // op1 is allowed, op2 is denied → entire batch fails, no WAL frame.
    let ops = vec![
        BatchOp::InsertNode {
            label: "MyLabel".into(),
            key: "new_node".into(),
            props: vec![],
        },
        BatchOp::InsertNode {
            label: "Secret".into(), // NOT in create_labels → denied
            key: "bad_node".into(),
            props: vec![],
        },
    ];
    let err = db.write_batch_authz(Some(&authz), ops).unwrap_err();
    assert!(is_role_write_denied(&err));
    // Neither node was inserted.
    assert!(
        !db.has_node("new_node"),
        "first op must not be applied on deny"
    );
    assert!(!db.has_node("bad_node"));
    // commit_seq unchanged → no WAL frame was written.
    assert_eq!(
        db.commit_seq(),
        seq_before,
        "commit_seq must not advance on deny"
    );
    let wal_after = wal_len(&dir);
    assert_eq!(wal_after, wal_before, "WAL must not grow on deny");
}

/// Authz fires before CAS would: hidden node → not-visible, not CasConflict.
///
/// The authz pre-check runs BEFORE CAS preconditions are evaluated (spec §5),
/// so a hidden node always returns RoleWriteDenied, never CasConflict.
#[test]
fn test_authz_fires_before_cas_would() {
    let (mut db, _dir) = open_with_writer("authz-before-cas");
    // Insert a hidden node (label not in role's read scope).
    db.insert_node("Secret", "hidden", vec![]).unwrap();
    let authz = writer_authz(&mut db);
    // Attempt SetProp on the hidden node.
    // If authz ran AFTER CAS: we'd potentially see CasConflict or KeyNotFound.
    // If authz runs FIRST: we see RoleWriteDenied (not-visible).
    let ops = vec![BatchOp::SetProp {
        key: "hidden".into(),
        field: "x".into(),
        value: Value::Int(1),
    }];
    let err = db.write_batch_authz(Some(&authz), ops).unwrap_err();
    assert!(
        matches!(err, GraphError::RoleWriteDenied { .. }),
        "authz must fire before CAS; expected RoleWriteDenied, got {err:?}"
    );
    assert_eq!(
        denied_reason(&err),
        "role-bound token: target node not visible"
    );
}

/// None authz = full authority: write_batch_authz(None, ops) behaves identically
/// to write_batch.
#[test]
fn test_none_authz_full_authority() {
    let (mut db, _dir) = open_with_writer("none-authz");
    // With None, "Secret" label is allowed (full authority).
    let ops = vec![BatchOp::InsertNode {
        label: "Secret".into(),
        key: "sec_node".into(),
        props: vec![],
    }];
    db.write_batch_authz(None, ops).unwrap();
    assert!(
        db.has_node("sec_node"),
        "None authz must bypass all role checks"
    );
}

/// RenameNode op with Some(authz) → endpoint-not-permitted (defense in depth).
#[test]
fn test_rename_node_forbidden_for_role() {
    let (mut db, _dir) = open_with_writer("rename-forbidden");
    db.insert_node("MyLabel", "old_key", vec![]).unwrap();
    let authz = writer_authz(&mut db);
    let ops = vec![BatchOp::RenameNode {
        old_key: "old_key".into(),
        new_key: "new_key".into(),
    }];
    let err = db.write_batch_authz(Some(&authz), ops).unwrap_err();
    assert!(
        is_role_write_denied(&err),
        "expected RoleWriteDenied, got {err:?}"
    );
    assert_eq!(
        denied_reason(&err),
        "role-bound token: this endpoint is not permitted"
    );
}

/// CreateRule op with Some(authz) → endpoint-not-permitted (defense in depth).
#[test]
fn test_create_rule_forbidden_for_role() {
    let (mut db, _dir) = open_with_writer("create-rule-forbidden");
    let authz = writer_authz(&mut db);
    let rule = RuleDef {
        name: "test_rule".into(),
        src_label: "MyLabel".into(),
        dst_label: "MyLabel".into(),
        predicate: Predicate::KeyMatch { field: "x".into() },
        edge_type: "KNOWS".into(),
        weight_prop: None,
        max_edges: None,
        approximate: false,
        via_label: None,
        via_edge: None,
        via_dir: None,
        namespace: None,
    };
    let ops = vec![BatchOp::CreateRule(rule)];
    let err = db.write_batch_authz(Some(&authz), ops).unwrap_err();
    assert!(
        is_role_write_denied(&err),
        "expected RoleWriteDenied, got {err:?}"
    );
    assert_eq!(
        denied_reason(&err),
        "role-bound token: this endpoint is not permitted"
    );
}

/// Rules fire on role-created nodes (DB authority, spec §3.5).
/// Derived edges to hidden neighbors exist in the DB but are masked on read.
#[test]
fn test_rules_fire_but_hidden_edges_masked() {
    let dir = tmp("rules-fire-hidden-masked");
    let mut db = GraphDb::open(&dir).unwrap();
    // Rule: SIMILAR edges between MyLabel nodes sharing a tag.
    let schema = Schema {
        fulltext: vec![],
        indexes: vec![],
        rules: vec![derived_rule()],
        views: vec![],
        roles: vec![writer_role()],
    };
    db.apply_schema(&schema).unwrap();

    // Admin inserts a hidden node with overlapping tag.
    db.insert_node(
        "Secret",
        "hidden_similar",
        vec![("tags".into(), Value::List(vec![Value::Str("xyz".into())]))],
    )
    .unwrap();

    // Role creates a MyLabel node with the same overlapping tag.
    let authz = writer_authz(&mut db);
    let ops = vec![BatchOp::InsertNode {
        label: "MyLabel".into(),
        key: "role_node".into(),
        props: vec![("tags".into(), Value::List(vec![Value::Str("xyz".into())]))],
    }];
    db.write_batch_authz(Some(&authz), ops).unwrap();

    // Rules ran with DB authority: SIMILAR edge may exist between role_node and hidden_similar.
    // But the role's read mask must NOT expose that edge.
    let mask = db.mask_for_role("writer").unwrap();
    let masked_edges = db.node_edges_masked("role_node", &mask).unwrap();
    assert!(
        !masked_edges
            .iter()
            .any(|e| e.src_key == "hidden_similar" || e.dst_key == "hidden_similar"),
        "hidden neighbor must not appear in masked edge list; edges: {masked_edges:?}"
    );
}

/// Regression: delete+recreate of the same key in one batch must NOT let a
/// following SetProp bypass update_labels via the batch_created fast-path.
///
/// The batch_created guard only admits genuinely new keys (absent from the
/// snapshot at authz-check time).  A pre-existing visible key is a potential
/// DuplicateKey, not a real creation, so it must NOT enter batch_created.
#[test]
fn test_delete_recreate_setprop_respects_update_labels() {
    // Role: can delete+create "MyLabel" but has EMPTY update_labels.
    let dir = tmp("delete-recreate-setprop");
    let mut db = GraphDb::open(&dir).unwrap();
    let schema = core_api::schema::Schema {
        fulltext: vec![],
        indexes: vec![],
        rules: vec![],
        views: vec![],
        roles: vec![RoleDef {
            name: "delcreate".into(),
            keys: vec![],
            labels: vec!["MyLabel".into()],
            visible_where: None,
            namespaces: None,
            write: Some(WriteScope {
                create_labels: vec!["MyLabel".into()],
                update_labels: vec![], // intentionally empty
                delete_labels: vec!["MyLabel".into()],
                create_edge_types: vec![],
                delete_edge_types: vec![],
            }),
        }],
    };
    db.apply_schema(&schema).unwrap();
    db.insert_node("MyLabel", "existing", vec![]).unwrap();

    let roles = db.roles();
    let def = roles.iter().find(|r| r.name == "delcreate").unwrap();
    let authz = core_api::WriteAuthz {
        role: "delcreate".into(),
        scope: def.write.clone().unwrap(),
        mask: db.mask_for_role("delcreate").unwrap(),
    };

    // [DeleteNode, InsertNode, SetProp] — SetProp must be denied because
    // "existing" was pre-existing at authz-check time (update_labels is empty).
    let ops = vec![
        BatchOp::DeleteNode {
            key: "existing".into(),
        },
        BatchOp::InsertNode {
            label: "MyLabel".into(),
            key: "existing".into(),
            props: vec![],
        },
        BatchOp::SetProp {
            key: "existing".into(),
            field: "x".into(),
            value: Value::Int(1),
        },
    ];
    let err = db.write_batch_authz(Some(&authz), ops).unwrap_err();
    assert!(
        is_role_write_denied(&err),
        "SetProp after delete+recreate must be denied when update_labels is empty; got {err:?}"
    );
    assert!(
        denied_reason(&err).contains("update_labels"),
        "reason must name update_labels: {}",
        denied_reason(&err)
    );
}

// ── MERGE RETURN: read-after-write (M1 fix verification) ─────────────────────

/// MERGE create arm with RETURN yields the created node (read-after-write fix).
///
/// Before the fix, the mask was resolved before the node existed, so the RETURN
/// clause returned 0 rows. After the fix (mask re-resolved after batch.commit()),
/// the created node is visible and RETURN yields exactly 1 row.
#[test]
fn test_merge_create_return_yields_node() {
    let (mut db, _dir) = open_with_writer("merge-create-return");
    let rs = db
        .query_write_authz(
            "writer",
            "MERGE (n:MyLabel {id: 'new_m1'}) ON CREATE SET n.x = 1 RETURN n",
            &no_params(),
        )
        .unwrap();
    assert_eq!(
        rs.len(),
        1,
        "MERGE create arm RETURN must yield exactly 1 row (read-after-write)"
    );
    assert_eq!(
        rs.get(0, "n"),
        Some(&Value::Str("new_m1".into())),
        "RETURN n must project the created node key"
    );
    assert_eq!(
        db.get_prop("new_m1", "x"),
        Some(Value::Int(1)),
        "ON CREATE SET must be applied"
    );
}

/// MERGE match arm with RETURN is unaffected by the M1 mask refresh.
///
/// The refresh fires only in the !existed (CREATE) branch. MATCH arm RETURN
/// must still project the matched node correctly (no regression).
#[test]
fn test_merge_match_return_unaffected() {
    let (mut db, _dir) = open_with_writer("merge-match-return");
    db.insert_node("MyLabel", "existing_m1", vec![]).unwrap();
    let rs = db
        .query_write_authz(
            "writer",
            "MERGE (n:MyLabel {id: 'existing_m1'}) RETURN n",
            &no_params(),
        )
        .unwrap();
    assert_eq!(
        rs.len(),
        1,
        "MERGE match arm RETURN must yield exactly 1 row"
    );
    assert_eq!(
        rs.get(0, "n"),
        Some(&Value::Str("existing_m1".into())),
        "RETURN n must project the matched node key"
    );
}

/// No-widening invariant: after MERGE-create, the mask refresh does NOT let the
/// role see nodes outside its declared read labels.
///
/// Security proof: create_labels ⊆ labels (enforced at apply_schema), so
/// mask_for_role resolves only over the role's declared labels. A "Secret" node
/// never enters the mask regardless of how many MERGE-creates occur.
#[test]
fn test_merge_create_no_mask_widening() {
    let (mut db, _dir) = open_with_writer("merge-no-widen");
    // Admin inserts a node with label "Secret" (not in writer's read labels).
    db.insert_node("Secret", "hidden_pre", vec![]).unwrap();

    // Writer MERGE-creates a new MyLabel node with RETURN (exercises the refresh path).
    let rs = db
        .query_write_authz(
            "writer",
            "MERGE (n:MyLabel {id: 'new_no_widen'}) RETURN n",
            &no_params(),
        )
        .unwrap();
    // M1 fix: create arm RETURN yields exactly the created node (via the refreshed
    // internal mask), not the hidden node or an empty result.
    assert_eq!(rs.len(), 1, "create arm RETURN must yield 1 row");
    assert_eq!(
        rs.get(0, "n"),
        Some(&Value::Str("new_no_widen".into())),
        "RETURN must project the created node key, not hidden_pre or nothing"
    );

    // No-widening: the role's mask must still exclude "Secret"-labeled nodes.
    // Obtain a fresh mask and run a masked read for the hidden node.
    let mask = db.mask_for_role("writer").unwrap();
    let hidden_rs = db
        .query_masked(
            "MATCH (n:Secret {id: 'hidden_pre'}) RETURN n",
            &no_params(),
            &mask,
        )
        .unwrap();
    assert_eq!(
        hidden_rs.len(),
        0,
        "mask refresh must NOT widen to Secret label; hidden_pre must remain invisible"
    );
}

// ── M2: DETACH DELETE cascade is mask-independent ────────────────────────────

/// Confirm DETACH DELETE cascades ALL incident edges regardless of caller mask,
/// leaving no orphaned edges to hidden neighbors.
///
/// Setup: node A (MyLabel, visible) has KNOWS edges to B (MyLabel, visible) and
/// C (Secret, hidden). Role DETACH-DELETEs A. After deletion:
/// - A is gone; B and C survive
/// - No edges remain incident to A (deleted) in the topology — including the
///   edge to the hidden neighbor C, which the cascade removes without consulting
///   the mask (topology integrity requires unconditional removal).
#[test]
fn test_detach_delete_cascades_hidden_edges() {
    let (mut db, _dir) = open_with_writer("detach-delete-cascade");
    // Admin setup: nodes and edges including one to a hidden neighbor.
    db.insert_node("MyLabel", "del_a", vec![]).unwrap();
    db.insert_node("MyLabel", "surv_b", vec![]).unwrap();
    db.insert_node("Secret", "hidden_c", vec![]).unwrap();
    db.insert_edge("KNOWS", "del_a", "surv_b").unwrap();
    db.insert_edge("KNOWS", "del_a", "hidden_c").unwrap();

    // Role-scoped DETACH DELETE: writer has delete_labels=["MyLabel"].
    db.query_write_authz(
        "writer",
        "MATCH (n:MyLabel {id: 'del_a'}) DETACH DELETE n",
        &no_params(),
    )
    .unwrap();

    // del_a is gone.
    assert!(!db.has_node("del_a"), "del_a must be deleted");
    // Surviving nodes intact.
    assert!(db.has_node("surv_b"), "surv_b must survive");
    assert!(db.has_node("hidden_c"), "hidden_c must survive");

    // No orphan edges: verify from surviving neighbors' perspectives.
    // surv_b must have no incoming KNOWS from del_a.
    let b_in = db
        .neighbors("surv_b", "KNOWS", Direction::In)
        .unwrap_or_default();
    assert!(
        !b_in.contains(&"del_a".to_string()),
        "edge del_a→surv_b must be cascade-deleted; b_in={b_in:?}"
    );
    // hidden_c must have no incoming KNOWS from del_a (mask-independent cascade).
    let c_in = db
        .neighbors("hidden_c", "KNOWS", Direction::In)
        .unwrap_or_default();
    assert!(
        !c_in.contains(&"del_a".to_string()),
        "edge del_a→hidden_c must be cascade-deleted regardless of mask; c_in={c_in:?}"
    );
}

// ── Predicate masks × write scopes ───────────────────────────────────────────

/// A `visible_where` predicate narrows what a write-scoped role may mutate:
/// a node of an allowed label that fails the predicate is hidden, and hidden
/// is exactly as unwritable as absent.
#[test]
fn test_predicate_narrows_the_write_target_set() {
    let dir = tmp("predicate-write-scope");
    let mut db = GraphDb::open(&dir).unwrap();
    db.insert_node(
        "MyLabel",
        "open",
        vec![("status".into(), Value::Str("published".into()))],
    )
    .unwrap();
    db.insert_node(
        "MyLabel",
        "closed",
        vec![("status".into(), Value::Str("draft".into()))],
    )
    .unwrap();
    db.apply_schema(&Schema {
        roles: vec![RoleDef {
            visible_where: Some(core_api::PropPredicate {
                field: "status".into(),
                eq: None,
                in_: Some(vec![Value::Str("published".into())]),
            }),
            ..writer_role()
        }],
        ..Default::default()
    })
    .unwrap();

    let authz = {
        let roles = db.roles();
        let def = roles.iter().find(|r| r.name == "writer").unwrap();
        core_api::WriteAuthz {
            role: "writer".into(),
            scope: def.write.clone().unwrap(),
            mask: db.mask_for_role("writer").unwrap(),
        }
    };
    assert_eq!(authz.mask.len(), 1, "only the published node is in scope");

    // The published node is writable.
    db.write_batch_authz(
        Some(&authz),
        vec![BatchOp::SetProp {
            key: "open".into(),
            field: "note".into(),
            value: Value::Str("ok".into()),
        }],
    )
    .expect("a visible node of an update label is writable");

    // The draft is not — and is refused exactly as an absent key is.
    let hidden = db
        .write_batch_authz(
            Some(&authz),
            vec![BatchOp::SetProp {
                key: "closed".into(),
                field: "note".into(),
                value: Value::Str("no".into()),
            }],
        )
        .expect_err("a node failing the predicate must not be writable");
    let absent = db
        .write_batch_authz(
            Some(&authz),
            vec![BatchOp::SetProp {
                key: "never-existed".into(),
                field: "note".into(),
                value: Value::Str("no".into()),
            }],
        )
        .expect_err("an absent key must not be writable either");
    assert!(is_role_write_denied(&hidden));
    assert_eq!(
        denied_reason(&hidden),
        denied_reason(&absent),
        "hidden-by-predicate must be byte-equal to absent — no existence oracle"
    );
}

// ── Namespace binding on the create gate (v0.6.6 §7) ─────────────────────────

/// `writer`, bound to namespace `x`, with the same write scope.
fn tenant_writer_role() -> RoleDef {
    RoleDef {
        namespaces: Some(vec!["x".into()]),
        ..writer_role()
    }
}

fn open_with_tenant_writer(name: &str) -> (GraphDb<core_storage::fs::RealFs>, std::path::PathBuf) {
    let dir = tmp(name);
    let mut db = GraphDb::open(&dir).unwrap();
    db.apply_schema(&Schema {
        roles: vec![tenant_writer_role()],
        ..Default::default()
    })
    .unwrap();
    (db, dir)
}

fn ns_prop(n: &str) -> (String, Value) {
    (core_api::NS_PROP.to_string(), Value::Str(n.to_string()))
}

/// A role bound to a namespace may only create inside it. The never-widen rule
/// is about what a write makes visible to *any* party: a node the writer could
/// never read back is a write into somebody else's tenancy.
#[test]
fn test_create_outside_the_roles_namespace_denied() {
    let (mut db, _dir) = open_with_tenant_writer("create-foreign-ns");
    let authz = writer_authz(&mut db);

    // Inside the role's namespace: allowed.
    db.write_batch_authz(
        Some(&authz),
        vec![BatchOp::InsertNode {
            label: "MyLabel".into(),
            key: "mine".into(),
            props: vec![ns_prop("x")],
        }],
    )
    .unwrap();
    assert_eq!(db.namespace_of("mine").as_deref(), Some("x"));

    // Another tenant's namespace: denied, and nothing is written.
    let authz = writer_authz(&mut db);
    let err = db
        .write_batch_authz(
            Some(&authz),
            vec![BatchOp::InsertNode {
                label: "MyLabel".into(),
                key: "theirs".into(),
                props: vec![ns_prop("y")],
            }],
        )
        .unwrap_err();
    assert!(is_role_write_denied(&err), "got {err:?}");
    assert_eq!(
        denied_reason(&err),
        "role-bound token: namespace 'y' not in the role's namespaces"
    );
    assert!(!db.has_node("theirs"));

    // The default namespace is a namespace like any other: an `ns`-less create
    // by an x-bound role lands in `default` and is denied too.
    let authz = writer_authz(&mut db);
    let err = db
        .write_batch_authz(
            Some(&authz),
            vec![BatchOp::InsertNode {
                label: "MyLabel".into(),
                key: "bare".into(),
                props: vec![],
            }],
        )
        .unwrap_err();
    assert_eq!(
        denied_reason(&err),
        "role-bound token: namespace 'default' not in the role's namespaces"
    );
    assert!(!db.has_node("bare"));

    // An upsert-edge whose placeholder endpoint would be created lands in
    // `default` too, so it is refused on the same ground — but with the
    // endpoint-visibility wording, which is what keeps hidden and absent
    // indistinguishable there. See
    // `test_upsert_placeholder_hidden_equals_absent_for_a_namespaced_role`.
    let authz = writer_authz(&mut db);
    let err = db
        .write_batch_authz(
            Some(&authz),
            vec![BatchOp::InsertEdgeUpsert {
                edge_type: "KNOWS".into(),
                src_key: "mine".into(),
                dst_key: "ghost".into(),
                placeholder_label: "MyLabel".into(),
            }],
        )
        .unwrap_err();
    assert_eq!(
        denied_reason(&err),
        "role-bound token: edge endpoint not visible"
    );
    assert!(!db.has_node("ghost"));
}

/// The same gate on the Cypher path: `CREATE` and the node `MERGE` creates both
/// arrive as `BatchOp::InsertNode`.
#[test]
fn test_cypher_create_outside_the_roles_namespace_denied() {
    let (mut db, _dir) = open_with_tenant_writer("cypher-foreign-ns");

    let err = db
        .query_write_authz(
            "writer",
            "CREATE (n:MyLabel {id: 'theirs', ns: 'y'})",
            &no_params(),
        )
        .unwrap_err();
    assert_eq!(
        denied_reason(&err),
        "role-bound token: namespace 'y' not in the role's namespaces"
    );
    assert!(!db.has_node("theirs"));

    // MERGE naming a foreign `ns` is the same op, so the same refusal.
    let err = db
        .query_write_authz(
            "writer",
            "MERGE (n:MyLabel {id: 'merged', ns: 'y'})",
            &no_params(),
        )
        .unwrap_err();
    assert_eq!(
        denied_reason(&err),
        "role-bound token: namespace 'y' not in the role's namespaces"
    );
    assert!(!db.has_node("merged"));

    // Inside the namespace the same statement succeeds.
    db.query_write_authz(
        "writer",
        "CREATE (n:MyLabel {id: 'mine', ns: 'x'})",
        &no_params(),
    )
    .unwrap();
    assert_eq!(db.namespace_of("mine").as_deref(), Some("x"));
}

/// A role with no namespace binding is unchanged: it creates wherever it likes,
/// exactly as it did before namespaces existed.
#[test]
fn test_unscoped_role_may_create_in_any_namespace() {
    let (mut db, _dir) = open_with_writer("create-unscoped-ns");
    let authz = writer_authz(&mut db);
    db.write_batch_authz(
        Some(&authz),
        vec![
            BatchOp::InsertNode {
                label: "MyLabel".into(),
                key: "bare".into(),
                props: vec![],
            },
            BatchOp::InsertNode {
                label: "MyLabel".into(),
                key: "tenanted".into(),
                props: vec![ns_prop("y")],
            },
        ],
    )
    .unwrap();
    assert_eq!(db.namespace_of("bare").as_deref(), Some("default"));
    assert_eq!(db.namespace_of("tenanted").as_deref(), Some("y"));
}

/// Hidden ≡ absent for the upsert placeholder gate, byte for byte.
///
/// The namespace refusal fires only for an endpoint that does **not** exist and
/// the visibility refusal only for one that does, so two different strings would
/// turn `POST /edges/upsert` into an existence oracle: ask for an upsert and read
/// off whether the key is taken.
#[test]
fn test_upsert_placeholder_hidden_equals_absent_for_a_namespaced_role() {
    let (mut db, _dir) = open_with_tenant_writer("upsert-oracle");
    // A visible endpoint inside the role's namespace to anchor the edge.
    db.insert_node("MyLabel", "mine", vec![ns_prop("x")])
        .unwrap();
    // A node the role cannot see at all (wrong label, and in another namespace).
    db.insert_node("Secret", "hidden", vec![ns_prop("y")])
        .unwrap();

    let upsert = |key: &str| BatchOp::InsertEdgeUpsert {
        edge_type: "KNOWS".into(),
        src_key: "mine".into(),
        dst_key: key.into(),
        placeholder_label: "MyLabel".into(),
    };

    let authz = writer_authz(&mut db);
    let hidden_err = db
        .write_batch_authz(Some(&authz), vec![upsert("hidden")])
        .unwrap_err();
    let authz = writer_authz(&mut db);
    let absent_err = db
        .write_batch_authz(Some(&authz), vec![upsert("nobody")])
        .unwrap_err();

    assert_eq!(
        denied_reason(&hidden_err),
        denied_reason(&absent_err),
        "hidden and absent must be indistinguishable"
    );
    assert_eq!(
        denied_reason(&absent_err),
        "role-bound token: edge endpoint not visible"
    );
    assert!(!db.has_node("nobody"), "nothing was created");
}

/// A role bound to two namespaces cannot MERGE-create unless the pattern names
/// one. The match arm still works on a node the role can see.
#[test]
fn test_merge_create_denied_for_a_namespaced_role() {
    let dir = tmp("merge-ns-role");
    let mut db = GraphDb::open(&dir).unwrap();
    db.apply_schema(&Schema {
        roles: vec![RoleDef {
            namespaces: Some(vec!["x".into(), "y".into()]),
            ..writer_role()
        }],
        ..Default::default()
    })
    .unwrap();
    db.insert_node("MyLabel", "mine", vec![ns_prop("x")])
        .unwrap();

    let err = db
        .query_write_authz("writer", "MERGE (n:MyLabel {id: 'fresh'})", &no_params())
        .unwrap_err();
    assert_eq!(
        denied_reason(&err),
        MERGE_CREATE_NEEDS_ONE_NAMESPACE,
        "a two-namespace role must name one namespace to MERGE-create"
    );
    assert!(!db.has_node("fresh"));

    // The match arm is unaffected: the node is already in the role's mask.
    db.query_write_authz(
        "writer",
        "MERGE (n:MyLabel {id: 'mine'}) ON MATCH SET n.seen = 1",
        &no_params(),
    )
    .unwrap();
    assert_eq!(db.get_prop("mine", "seen"), Some(Value::Int(1)));
    assert_eq!(db.namespace_of("mine").as_deref(), Some("x"));
}

/// A props list naming `ns` twice is refused on the role path too, before the
/// namespace gate can read one of the two entries.
#[test]
fn test_duplicate_ns_refused_on_the_role_path() {
    let (mut db, _dir) = open_with_tenant_writer("dup-ns-role");
    let authz = writer_authz(&mut db);
    let err = db
        .write_batch_authz(
            Some(&authz),
            vec![BatchOp::InsertNode {
                label: "MyLabel".into(),
                key: "two".into(),
                props: vec![ns_prop("x"), ns_prop("y")],
            }],
        )
        .unwrap_err();
    assert!(
        err.to_string().contains("given more than once"),
        "expected the duplicate-ns refusal, got {err:?}"
    );
    assert!(!db.has_node("two"));
}