use super::{
    AedbConfig, AedbError, AedbInstance, CallerContext, ColumnDef, ColumnType, ConsistencyMode,
    DdlOperation, DurabilityMode, Expr, IdempotencyKey, Mutation, Permission, Query, QueryError,
    QueryOptions, ReadAssertion, ResourceType, RestrictiveSqlAdapter, Row, SYSTEM_CALLER_ID,
    TransactionEnvelope, Value, WriteClass, WriteIntent, create_table,
};
use crate::commit::tx::ReadSet;
use crate::commit::validation::CompareOp;
use std::fs;
use std::ops::Bound;
use std::path::Path;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tempfile::tempdir;

#[tokio::test]
async fn permissions_enforced_at_preflight_commit_and_query() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        owner_id: None,
        if_not_exists: false,
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "users".into(),
        columns: vec![
            ColumnDef {
                name: "id".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "name".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
        ],
        primary_key: vec!["id".into()],
    }))
    .await
    .expect("table");

    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        actor_id: None,
        delegable: false,
        caller_id: "alice".into(),
        permission: Permission::TableWrite {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "users".into(),
        },
    }))
    .await
    .expect("grant write");
    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        actor_id: None,
        delegable: false,
        caller_id: "alice".into(),
        permission: Permission::TableRead {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "users".into(),
        },
    }))
    .await
    .expect("grant read");

    let caller = CallerContext::new("alice");
    let insert = Mutation::Upsert {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "users".into(),
        primary_key: vec![Value::Integer(1)],
        row: Row {
            values: vec![Value::Integer(1), Value::Text("alice".into())],
        },
    };
    db.preflight_as(&caller, insert.clone())
        .await
        .expect("preflight allowed");
    db.commit_as(caller.clone(), insert)
        .await
        .expect("commit allowed");

    db.commit(Mutation::Ddl(DdlOperation::RevokePermission {
        actor_id: None,
        caller_id: "alice".into(),
        permission: Permission::TableWrite {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "users".into(),
        },
    }))
    .await
    .expect("revoke");

    let denied = db
        .preflight_as(
            &caller,
            Mutation::Upsert {
                project_id: "p".into(),
                scope_id: "app".into(),
                table_name: "users".into(),
                primary_key: vec![Value::Integer(2)],
                row: Row {
                    values: vec![Value::Integer(2), Value::Text("x".into())],
                },
            },
        )
        .await
        .expect_err("preflight denied");
    assert!(matches!(denied, AedbError::PermissionDenied(_)));

    let q = db
        .query_with_options_as(
            Some(&caller),
            "p",
            "app",
            Query::select(&["*"]).from("users").limit(10),
            QueryOptions::default(),
        )
        .await
        .expect("query with read perm");
    assert_eq!(q.rows.len(), 1);
}

#[tokio::test]
async fn join_queries_require_permissions_for_all_referenced_tables() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project p");
    db.create_project("_global").await.expect("project global");

    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        owner_id: None,
        if_not_exists: false,
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "users".into(),
        columns: vec![
            ColumnDef {
                name: "id".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "name".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
        ],
        primary_key: vec!["id".into()],
    }))
    .await
    .expect("users table");
    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        owner_id: None,
        if_not_exists: false,
        project_id: "_global".into(),
        scope_id: "app".into(),
        table_name: "profiles".into(),
        columns: vec![
            ColumnDef {
                name: "id".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "email".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
        ],
        primary_key: vec!["id".into()],
    }))
    .await
    .expect("profiles table");

    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        actor_id: None,
        delegable: false,
        caller_id: "alice".into(),
        permission: Permission::TableRead {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "users".into(),
        },
    }))
    .await
    .expect("grant base read");
    let caller = CallerContext::new("alice");
    let query = Query::select(&["u.id"])
        .from("users")
        .alias("u")
        .left_join("_global.profiles", "u.id", "id")
        .with_last_join_alias("g")
        .limit(10);
    let denied = db
        .query_with_options_as(
            Some(&caller),
            "p",
            "app",
            query.clone(),
            QueryOptions::default(),
        )
        .await
        .expect_err("join should require global table read");
    assert!(matches!(denied, QueryError::PermissionDenied { .. }));

    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        actor_id: None,
        delegable: false,
        caller_id: "alice".into(),
        permission: Permission::TableRead {
            project_id: "_global".into(),
            scope_id: "app".into(),
            table_name: "profiles".into(),
        },
    }))
    .await
    .expect("grant global read");
    let _ = db
        .query_with_options_as(Some(&caller), "p", "app", query, QueryOptions::default())
        .await
        .expect("join allowed");
}

#[tokio::test]
async fn permission_revocation_between_preflight_and_commit_is_caught() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        owner_id: None,
        if_not_exists: false,
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "users".into(),
        columns: vec![
            ColumnDef {
                name: "id".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "name".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
        ],
        primary_key: vec!["id".into()],
    }))
    .await
    .expect("table");
    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        actor_id: None,
        delegable: false,
        caller_id: "alice".into(),
        permission: Permission::TableWrite {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "users".into(),
        },
    }))
    .await
    .expect("grant");
    let caller = CallerContext::new("alice");
    let mutation = Mutation::Upsert {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "users".into(),
        primary_key: vec![Value::Integer(1)],
        row: Row {
            values: vec![Value::Integer(1), Value::Text("alice".into())],
        },
    };
    db.preflight_as(&caller, mutation.clone())
        .await
        .expect("preflight ok");
    db.commit(Mutation::Ddl(DdlOperation::RevokePermission {
        actor_id: None,
        caller_id: "alice".into(),
        permission: Permission::TableWrite {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "users".into(),
        },
    }))
    .await
    .expect("revoke");
    let err = db
        .commit_as(caller, mutation)
        .await
        .expect_err("denied at commit");
    assert!(matches!(err, AedbError::PermissionDenied(_)));
}

#[tokio::test]
async fn kv_query_apis_enforce_permissions_and_paginate() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");

    let caller = CallerContext::new("alice");
    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        actor_id: None,
        delegable: false,
        caller_id: "alice".into(),
        permission: Permission::KvRead {
            project_id: "p".into(),
            scope_id: Some("app".into()),
            prefix: None,
        },
    }))
    .await
    .expect("grant kv read");
    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        actor_id: None,
        delegable: false,
        caller_id: "alice".into(),
        permission: Permission::KvWrite {
            project_id: "p".into(),
            scope_id: Some("app".into()),
            prefix: None,
        },
    }))
    .await
    .expect("grant kv write");

    for (k, v) in [("user:1", "alice"), ("user:2", "bob"), ("user:3", "carol")] {
        db.commit_as(
            caller.clone(),
            Mutation::KvSet {
                project_id: "p".into(),
                scope_id: "app".into(),
                key: k.as_bytes().to_vec(),
                value: v.as_bytes().to_vec(),
            },
        )
        .await
        .expect("kv set");
    }

    let one = db
        .kv_get("p", "app", b"user:1", ConsistencyMode::AtLatest, &caller)
        .await
        .expect("kv_get")
        .expect("exists");
    assert_eq!(one.value, b"alice".to_vec());

    let first = db
        .kv_scan_prefix(
            "p",
            "app",
            b"user:",
            2,
            None,
            ConsistencyMode::AtLatest,
            &caller,
        )
        .await
        .expect("scan 1");
    assert_eq!(first.entries.len(), 2);
    assert!(first.truncated);

    let second = db
        .kv_scan_prefix(
            "p",
            "app",
            b"user:",
            2,
            first.cursor.clone(),
            ConsistencyMode::AtLatest,
            &caller,
        )
        .await
        .expect("scan 2");
    assert_eq!(second.entries.len(), 1);
    assert!(!second.truncated);

    let ranged = db
        .kv_scan_range(
            "p",
            "app",
            Bound::Included(b"user:1".to_vec()),
            Bound::Included(b"user:2".to_vec()),
            10,
            None,
            ConsistencyMode::AtLatest,
            &caller,
        )
        .await
        .expect("scan range");
    assert_eq!(ranged.entries.len(), 2);
}

#[tokio::test]
async fn kv_permissions_scope_reads_and_writes_with_prefix_filters() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    let caller = CallerContext::new("alice");

    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        actor_id: None,
        delegable: false,
        caller_id: "alice".into(),
        permission: Permission::KvRead {
            project_id: "p".into(),
            scope_id: Some("app".into()),
            prefix: Some(b"bank:user:42:".to_vec()),
        },
    }))
    .await
    .expect("grant kv prefix read");
    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        actor_id: None,
        delegable: false,
        caller_id: "alice".into(),
        permission: Permission::KvWrite {
            project_id: "p".into(),
            scope_id: Some("app".into()),
            prefix: Some(b"bank:user:42:".to_vec()),
        },
    }))
    .await
    .expect("grant kv prefix write");

    db.commit_as(
        caller.clone(),
        Mutation::KvSet {
            project_id: "p".into(),
            scope_id: "app".into(),
            key: b"bank:user:42:balance".to_vec(),
            value: b"100".to_vec(),
        },
    )
    .await
    .expect("allowed prefix write");

    let err = db
        .commit_as(
            caller.clone(),
            Mutation::KvSet {
                project_id: "p".into(),
                scope_id: "app".into(),
                key: b"bank:user:99:balance".to_vec(),
                value: b"100".to_vec(),
            },
        )
        .await
        .expect_err("write outside prefix must fail");
    assert!(matches!(err, AedbError::PermissionDenied(_)));

    db.commit(Mutation::KvSet {
        project_id: "p".into(),
        scope_id: "app".into(),
        key: b"bank:user:99:balance".to_vec(),
        value: b"77".to_vec(),
    })
    .await
    .expect("seed other key");

    let scan = db
        .kv_scan_prefix(
            "p",
            "app",
            b"bank:user:",
            10,
            None,
            ConsistencyMode::AtLatest,
            &caller,
        )
        .await
        .expect("scan with broad prefix");
    assert_eq!(scan.entries.len(), 1);
    assert_eq!(scan.entries[0].0, b"bank:user:42:balance".to_vec());
}

#[tokio::test]
async fn scoped_admin_permissions_bound_grant_revoke() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    db.create_scope("p", "s1").await.expect("scope s1");
    db.create_scope("p", "s2").await.expect("scope s2");

    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        actor_id: None,
        delegable: false,
        caller_id: "owner".into(),
        permission: Permission::ScopeAdmin {
            project_id: "p".into(),
            scope_id: "s1".into(),
        },
    }))
    .await
    .expect("grant scope admin");

    let owner = CallerContext::new("owner");
    db.commit_as(
        owner.clone(),
        Mutation::Ddl(DdlOperation::GrantPermission {
            actor_id: None,
            delegable: false,
            caller_id: "alice".into(),
            permission: Permission::KvRead {
                project_id: "p".into(),
                scope_id: Some("s1".into()),
                prefix: None,
            },
        }),
    )
    .await
    .expect("scope admin can grant in own scope");

    let denied = db
        .commit_as(
            owner.clone(),
            Mutation::Ddl(DdlOperation::GrantPermission {
                actor_id: None,
                delegable: false,
                caller_id: "alice".into(),
                permission: Permission::KvRead {
                    project_id: "p".into(),
                    scope_id: Some("s2".into()),
                    prefix: None,
                },
            }),
        )
        .await
        .expect_err("scope admin cannot grant in other scope");
    assert!(matches!(denied, AedbError::PermissionDenied(_)));

    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        actor_id: None,
        delegable: false,
        caller_id: "proj".into(),
        permission: Permission::ProjectAdmin {
            project_id: "p".into(),
        },
    }))
    .await
    .expect("grant project admin");
    let proj = CallerContext::new("proj");
    db.commit_as(
        proj,
        Mutation::Ddl(DdlOperation::GrantPermission {
            actor_id: None,
            delegable: false,
            caller_id: "alice".into(),
            permission: Permission::KvRead {
                project_id: "p".into(),
                scope_id: Some("s2".into()),
                prefix: None,
            },
        }),
    )
    .await
    .expect("project admin can grant across project scopes");
}

#[tokio::test]
async fn project_owner_and_delegable_grants_control_authz_delegation() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");

    db.commit(Mutation::Ddl(DdlOperation::CreateProject {
        project_id: "p".into(),
        owner_id: Some("owner".into()),
        if_not_exists: true,
    }))
    .await
    .expect("owner creates project");
    let owner = CallerContext::new("owner");

    let attacker = CallerContext::new("mallory");
    let denied = db
        .commit_as(
            attacker,
            Mutation::Ddl(DdlOperation::GrantPermission {
                caller_id: "alice".into(),
                permission: Permission::KvRead {
                    project_id: "p".into(),
                    scope_id: Some("app".into()),
                    prefix: None,
                },
                actor_id: None,
                delegable: false,
            }),
        )
        .await
        .expect_err("non-owner cannot grant");
    assert!(matches!(denied, AedbError::PermissionDenied(_)));

    db.commit_as(
        owner,
        Mutation::Ddl(DdlOperation::GrantPermission {
            caller_id: "alice".into(),
            permission: Permission::KvRead {
                project_id: "p".into(),
                scope_id: Some("app".into()),
                prefix: None,
            },
            actor_id: None,
            delegable: true,
        }),
    )
    .await
    .expect("owner grants delegable read");
    let (_, catalog, _) = db.executor.snapshot_state().await;
    assert!(catalog.has_delegable_grant(
        "alice",
        &Permission::KvRead {
            project_id: "p".into(),
            scope_id: Some("app".into()),
            prefix: None,
        }
    ));

    let alice = CallerContext::new("alice");
    db.commit_as(
        alice.clone(),
        Mutation::Ddl(DdlOperation::GrantPermission {
            caller_id: "bob".into(),
            permission: Permission::KvRead {
                project_id: "p".into(),
                scope_id: Some("app".into()),
                prefix: None,
            },
            actor_id: None,
            delegable: false,
        }),
    )
    .await
    .expect("delegable grant allows onward delegation");

    let denied = db
        .commit_as(
            alice,
            Mutation::Ddl(DdlOperation::GrantPermission {
                caller_id: "bob".into(),
                permission: Permission::KvWrite {
                    project_id: "p".into(),
                    scope_id: Some("app".into()),
                    prefix: None,
                },
                actor_id: None,
                delegable: false,
            }),
        )
        .await
        .expect_err("delegable grant is permission-specific");
    assert!(matches!(denied, AedbError::PermissionDenied(_)));
}

#[tokio::test]
async fn permission_denied_messages_do_not_leak_table_names() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("prod").await.expect("project");
    create_table(
        &db,
        "prod",
        "app",
        "users",
        vec![ColumnDef {
            name: "id".into(),
            col_type: ColumnType::Integer,
            nullable: false,
        }],
        vec!["id"],
    )
    .await;

    let err = db
        .commit_as(
            CallerContext::new("mallory"),
            Mutation::Upsert {
                project_id: "prod".into(),
                scope_id: "app".into(),
                table_name: "users".into(),
                primary_key: vec![Value::Integer(1)],
                row: Row::from_values(vec![Value::Integer(1)]),
            },
        )
        .await
        .expect_err("permission denied");
    let rendered = err.to_string();
    assert!(!rendered.contains("users"));
    assert!(!rendered.contains("prod"));
    assert!(!rendered.contains("app"));
}

#[tokio::test]
async fn kv_permission_overlap_revoke_precedence_is_explicit() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    db.create_scope("p", "private")
        .await
        .expect("private scope");

    db.commit(Mutation::KvSet {
        project_id: "p".into(),
        scope_id: "app".into(),
        key: b"user:1".to_vec(),
        value: b"alice".to_vec(),
    })
    .await
    .expect("seed app user");
    db.commit(Mutation::KvSet {
        project_id: "p".into(),
        scope_id: "app".into(),
        key: b"order:1".to_vec(),
        value: b"100".to_vec(),
    })
    .await
    .expect("seed app order");
    db.commit(Mutation::KvSet {
        project_id: "p".into(),
        scope_id: "private".into(),
        key: b"user:1".to_vec(),
        value: b"secret".to_vec(),
    })
    .await
    .expect("seed private user");

    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        caller_id: "alice".into(),
        permission: Permission::KvRead {
            project_id: "p".into(),
            scope_id: None,
            prefix: None,
        },
        actor_id: None,
        delegable: false,
    }))
    .await
    .expect("grant project wide read");
    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        caller_id: "alice".into(),
        permission: Permission::KvRead {
            project_id: "p".into(),
            scope_id: Some("app".into()),
            prefix: Some(b"user:".to_vec()),
        },
        actor_id: None,
        delegable: false,
    }))
    .await
    .expect("grant narrow read");
    db.commit(Mutation::Ddl(DdlOperation::RevokePermission {
        caller_id: "alice".into(),
        permission: Permission::KvRead {
            project_id: "p".into(),
            scope_id: None,
            prefix: None,
        },
        actor_id: None,
    }))
    .await
    .expect("revoke broad read");

    let alice = CallerContext::new("alice");
    let allowed = db
        .kv_get("p", "app", b"user:1", ConsistencyMode::AtLatest, &alice)
        .await
        .expect("narrow grant still applies");
    assert!(allowed.is_some());

    let denied_key = db
        .kv_get("p", "app", b"order:1", ConsistencyMode::AtLatest, &alice)
        .await
        .expect_err("revoke broad grant should deny non-matching key");
    assert!(matches!(denied_key, QueryError::PermissionDenied { .. }));

    let denied_scope = db
        .kv_get("p", "private", b"user:1", ConsistencyMode::AtLatest, &alice)
        .await
        .expect_err("narrow app grant should not spill into private scope");
    assert!(matches!(denied_scope, QueryError::PermissionDenied { .. }));
}

#[tokio::test]
async fn grant_metadata_tracks_actor_and_system_source() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");

    db.commit(Mutation::Ddl(DdlOperation::CreateProject {
        project_id: "p".into(),
        owner_id: Some("owner".into()),
        if_not_exists: true,
    }))
    .await
    .expect("project");

    let owner = CallerContext::new("owner");
    db.commit_as(
        owner.clone(),
        Mutation::Ddl(DdlOperation::GrantPermission {
            caller_id: "bob".into(),
            permission: Permission::KvRead {
                project_id: "p".into(),
                scope_id: Some("app".into()),
                prefix: None,
            },
            actor_id: Some("mallory".into()),
            delegable: true,
        }),
    )
    .await
    .expect("owner grant with explicit actor");

    db.commit_as(
        owner,
        Mutation::Ddl(DdlOperation::GrantPermission {
            caller_id: "carol".into(),
            permission: Permission::KvRead {
                project_id: "p".into(),
                scope_id: Some("app".into()),
                prefix: None,
            },
            actor_id: None,
            delegable: false,
        }),
    )
    .await
    .expect("owner grant with implicit actor");

    let (_, catalog, _) = db.executor.snapshot_state().await;
    let bob_meta = catalog
        .permission_grants
        .get(&(
            "bob".to_string(),
            Permission::KvRead {
                project_id: "p".into(),
                scope_id: Some("app".into()),
                prefix: None,
            },
        ))
        .expect("bob metadata");
    assert_eq!(bob_meta.granted_by, "owner");
    assert!(bob_meta.delegable);

    let carol_meta = catalog
        .permission_grants
        .get(&(
            "carol".to_string(),
            Permission::KvRead {
                project_id: "p".into(),
                scope_id: Some("app".into()),
                prefix: None,
            },
        ))
        .expect("carol metadata");
    assert_eq!(carol_meta.granted_by, "owner");
    assert!(!carol_meta.delegable);
}

#[tokio::test]
async fn grant_and_revoke_authorization_matrix_is_enforced() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");

    db.commit(Mutation::Ddl(DdlOperation::CreateProject {
        project_id: "p".into(),
        owner_id: Some("owner".into()),
        if_not_exists: true,
    }))
    .await
    .expect("project");
    db.create_scope("p", "s1").await.expect("scope");
    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        project_id: "p".into(),
        scope_id: "s1".into(),
        table_name: "events".into(),
        owner_id: Some("table_owner".into()),
        if_not_exists: false,
        columns: vec![ColumnDef {
            name: "id".into(),
            col_type: ColumnType::Integer,
            nullable: false,
        }],
        primary_key: vec!["id".into()],
    }))
    .await
    .expect("table");
    db.commit(Mutation::Upsert {
        project_id: "p".into(),
        scope_id: "s1".into(),
        table_name: "events".into(),
        primary_key: vec![Value::Integer(1)],
        row: Row {
            values: vec![Value::Integer(1)],
        },
    })
    .await
    .expect("seed row");

    let denied = db
        .commit_as(
            CallerContext::new("mallory"),
            Mutation::Ddl(DdlOperation::GrantPermission {
                caller_id: "pa".into(),
                permission: Permission::ProjectAdmin {
                    project_id: "p".into(),
                },
                actor_id: None,
                delegable: false,
            }),
        )
        .await
        .expect_err("non-owner cannot grant project admin");
    assert!(matches!(denied, AedbError::PermissionDenied(_)));

    db.commit_as(
        CallerContext::new("owner"),
        Mutation::Ddl(DdlOperation::GrantPermission {
            caller_id: "pa".into(),
            permission: Permission::ProjectAdmin {
                project_id: "p".into(),
            },
            actor_id: None,
            delegable: false,
        }),
    )
    .await
    .expect("owner grants project admin");

    db.commit_as(
        CallerContext::new("pa"),
        Mutation::Ddl(DdlOperation::GrantPermission {
            caller_id: "sa".into(),
            permission: Permission::ScopeAdmin {
                project_id: "p".into(),
                scope_id: "s1".into(),
            },
            actor_id: None,
            delegable: false,
        }),
    )
    .await
    .expect("project admin grants scope admin");

    let denied = db
        .commit_as(
            CallerContext::new("sa"),
            Mutation::Ddl(DdlOperation::GrantPermission {
                caller_id: "other".into(),
                permission: Permission::ProjectAdmin {
                    project_id: "p".into(),
                },
                actor_id: None,
                delegable: false,
            }),
        )
        .await
        .expect_err("scope admin cannot grant project admin");
    assert!(matches!(denied, AedbError::PermissionDenied(_)));

    db.commit_as(
        CallerContext::new("table_owner"),
        Mutation::Ddl(DdlOperation::GrantPermission {
            caller_id: "reader".into(),
            permission: Permission::TableRead {
                project_id: "p".into(),
                scope_id: "s1".into(),
                table_name: "events".into(),
            },
            actor_id: None,
            delegable: false,
        }),
    )
    .await
    .expect("table owner grants table read");

    let reader = CallerContext::new("reader");
    let query = db
        .query_with_options_as(
            Some(&reader),
            "p",
            "s1",
            Query::select(&["id"]).from("events").limit(10),
            QueryOptions::default(),
        )
        .await
        .expect("reader query");
    assert_eq!(query.rows.len(), 1);

    let denied = db
        .commit_as(
            CallerContext::new("mallory"),
            Mutation::Ddl(DdlOperation::RevokePermission {
                caller_id: "reader".into(),
                permission: Permission::TableRead {
                    project_id: "p".into(),
                    scope_id: "s1".into(),
                    table_name: "events".into(),
                },
                actor_id: None,
            }),
        )
        .await
        .expect_err("non-admin/non-owner cannot revoke");
    assert!(matches!(denied, AedbError::PermissionDenied(_)));

    db.commit_as(
        CallerContext::new("pa"),
        Mutation::Ddl(DdlOperation::RevokePermission {
            caller_id: "reader".into(),
            permission: Permission::TableRead {
                project_id: "p".into(),
                scope_id: "s1".into(),
                table_name: "events".into(),
            },
            actor_id: None,
        }),
    )
    .await
    .expect("project admin revokes table read");

    let denied_query = db
        .query_with_options_as(
            Some(&reader),
            "p",
            "s1",
            Query::select(&["id"]).from("events").limit(10),
            QueryOptions::default(),
        )
        .await
        .expect_err("reader should lose access after revoke");
    assert!(matches!(denied_query, QueryError::PermissionDenied { .. }));
}

#[tokio::test]
async fn ownership_transfer_requires_owner_or_global_admin() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.commit(Mutation::Ddl(DdlOperation::CreateProject {
        project_id: "p".into(),
        owner_id: Some("owner".into()),
        if_not_exists: true,
    }))
    .await
    .expect("create project with owner");

    let mallory = CallerContext::new("mallory");
    let denied = db
        .commit_as(
            mallory,
            Mutation::Ddl(DdlOperation::TransferOwnership {
                resource_type: ResourceType::Project,
                project_id: "p".into(),
                scope_id: None,
                table_name: None,
                new_owner_id: "next".into(),
                actor_id: None,
            }),
        )
        .await
        .expect_err("non-owner transfer must fail");
    assert!(matches!(denied, AedbError::PermissionDenied(_)));

    let owner = CallerContext::new("owner");
    db.commit_as(
        owner,
        Mutation::Ddl(DdlOperation::TransferOwnership {
            resource_type: ResourceType::Project,
            project_id: "p".into(),
            scope_id: None,
            table_name: None,
            new_owner_id: "next".into(),
            actor_id: None,
        }),
    )
    .await
    .expect("owner transfer");

    let (_, catalog, _) = db.executor.snapshot_state().await;
    assert_eq!(
        catalog
            .projects
            .get("p")
            .and_then(|p| p.owner_id.as_deref()),
        Some("next")
    );
}

#[tokio::test]
async fn authz_grant_revoke_events_are_persisted_in_system_audit_table() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.commit(Mutation::Ddl(DdlOperation::CreateProject {
        project_id: "p".into(),
        owner_id: Some("owner".into()),
        if_not_exists: true,
    }))
    .await
    .expect("project");
    let owner = CallerContext::new("owner");
    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "users".into(),
        owner_id: None,
        if_not_exists: false,
        columns: vec![
            ColumnDef {
                name: "id".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "owner".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
        ],
        primary_key: vec!["id".into()],
    }))
    .await
    .expect("users table");

    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        caller_id: "alice".into(),
        permission: Permission::KvRead {
            project_id: "p".into(),
            scope_id: Some("app".into()),
            prefix: None,
        },
        actor_id: None,
        delegable: false,
    }))
    .await
    .expect("grant");
    db.commit(Mutation::Ddl(DdlOperation::RevokePermission {
        caller_id: "alice".into(),
        permission: Permission::KvRead {
            project_id: "p".into(),
            scope_id: Some("app".into()),
            prefix: None,
        },
        actor_id: None,
    }))
    .await
    .expect("revoke");
    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        caller_id: "owner".into(),
        permission: Permission::TableDdl {
            project_id: "p".into(),
        },
        actor_id: None,
        delegable: false,
    }))
    .await
    .expect("grant owner ddl");
    db.commit_as(
        owner.clone(),
        Mutation::Ddl(DdlOperation::SetReadPolicy {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "users".into(),
            predicate: Expr::Eq("owner".into(), Value::Text("$caller_id".into())),
            actor_id: None,
        }),
    )
    .await
    .expect("set policy");
    db.commit_as(
        owner.clone(),
        Mutation::Ddl(DdlOperation::ClearReadPolicy {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "users".into(),
            actor_id: None,
        }),
    )
    .await
    .expect("clear policy");
    db.commit_as(
        owner,
        Mutation::Ddl(DdlOperation::TransferOwnership {
            resource_type: ResourceType::Project,
            project_id: "p".into(),
            scope_id: None,
            table_name: None,
            new_owner_id: "next".into(),
            actor_id: None,
        }),
    )
    .await
    .expect("transfer policy");

    let audit = db
        .query_no_auth(
            "_system",
            "app",
            Query::select(&["*"]).from("authz_audit").limit(20),
            QueryOptions::default(),
        )
        .await
        .expect("audit query");
    assert!(audit.rows.len() >= 2, "expected at least grant+revoke rows");
    let actions: Vec<String> = audit
        .rows
        .iter()
        .filter_map(|r| match r.values.get(2) {
            Some(Value::Text(v)) => Some(v.to_string()),
            _ => None,
        })
        .collect();
    assert!(actions.iter().any(|a| a == "grant"));
    assert!(actions.iter().any(|a| a == "revoke"));
    assert!(actions.iter().any(|a| a == "set_read_policy"));
    assert!(actions.iter().any(|a| a == "clear_read_policy"));
    assert!(actions.iter().any(|a| a == "transfer_ownership"));

    let owner_actor_actions: Vec<String> = audit
        .rows
        .iter()
        .filter_map(|r| match (r.values.get(2), r.values.get(3)) {
            (Some(Value::Text(action)), Some(Value::Text(actor))) if actor.as_str() == "owner" => {
                Some(action.to_string())
            }
            _ => None,
        })
        .collect();
    assert!(owner_actor_actions.iter().any(|a| a == "set_read_policy"));
    assert!(owner_actor_actions.iter().any(|a| a == "clear_read_policy"));
    assert!(
        owner_actor_actions
            .iter()
            .any(|a| a == "transfer_ownership")
    );
}

#[tokio::test]
async fn read_policies_filter_rows_for_caller_and_can_be_cleared() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "users".into(),
        owner_id: None,
        if_not_exists: false,
        columns: vec![
            ColumnDef {
                name: "id".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "owner".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
        ],
        primary_key: vec!["id".into()],
    }))
    .await
    .expect("table");
    db.commit(Mutation::Upsert {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "users".into(),
        primary_key: vec![Value::Integer(1)],
        row: Row {
            values: vec![Value::Integer(1), Value::Text("reader".into())],
        },
    })
    .await
    .expect("row 1");
    db.commit(Mutation::Upsert {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "users".into(),
        primary_key: vec![Value::Integer(2)],
        row: Row {
            values: vec![Value::Integer(2), Value::Text("bob".into())],
        },
    })
    .await
    .expect("row 2");
    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        caller_id: "reader".into(),
        permission: Permission::TableRead {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "users".into(),
        },
        actor_id: None,
        delegable: false,
    }))
    .await
    .expect("grant read");
    db.set_read_policy(
        "p",
        "app",
        "users",
        Expr::Eq("owner".into(), Value::Text("$caller_id".into())),
    )
    .await
    .expect("set read policy");

    let reader = CallerContext::new("reader");
    let filtered = db
        .query_with_options_as(
            Some(&reader),
            "p",
            "app",
            Query::select(&["*"]).from("users").limit(10),
            QueryOptions::default(),
        )
        .await
        .expect("filtered query");
    assert_eq!(filtered.rows.len(), 1);

    db.clear_read_policy("p", "app", "users")
        .await
        .expect("clear read policy");
    let unfiltered = db
        .query_with_options_as(
            Some(&reader),
            "p",
            "app",
            Query::select(&["*"]).from("users").limit(10),
            QueryOptions::default(),
        )
        .await
        .expect("unfiltered query");
    assert_eq!(unfiltered.rows.len(), 2);
}

#[tokio::test]
async fn read_policy_like_patterns_substitute_caller_id() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "users".into(),
        owner_id: None,
        if_not_exists: false,
        columns: vec![
            ColumnDef {
                name: "id".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "owner".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
        ],
        primary_key: vec!["id".into()],
    }))
    .await
    .expect("table");
    for (id, owner) in [
        (1_i64, "reader-alpha"),
        (2, "reader-beta"),
        (3, "other-alpha"),
    ] {
        db.commit(Mutation::Upsert {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "users".into(),
            primary_key: vec![Value::Integer(id)],
            row: Row::from_values(vec![Value::Integer(id), Value::Text(owner.into())]),
        })
        .await
        .expect("seed row");
    }
    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        caller_id: "reader".into(),
        permission: Permission::TableRead {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "users".into(),
        },
        actor_id: None,
        delegable: false,
    }))
    .await
    .expect("grant read");
    db.set_read_policy(
        "p",
        "app",
        "users",
        Expr::Like("owner".into(), "$caller_id%".into()),
    )
    .await
    .expect("set read policy");

    let rows = db
        .query_with_options_as(
            Some(&CallerContext::new("reader")),
            "p",
            "app",
            Query::select(&["*"]).from("users").limit(10),
            QueryOptions::default(),
        )
        .await
        .expect("filtered query")
        .rows;

    assert_eq!(rows.len(), 2);
}

#[tokio::test]
async fn read_policy_applies_to_joined_tables() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "users".into(),
        owner_id: None,
        if_not_exists: false,
        columns: vec![ColumnDef {
            name: "id".into(),
            col_type: ColumnType::Integer,
            nullable: false,
        }],
        primary_key: vec!["id".into()],
    }))
    .await
    .expect("users");
    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "profiles".into(),
        owner_id: None,
        if_not_exists: false,
        columns: vec![
            ColumnDef {
                name: "user_id".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "owner".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
        ],
        primary_key: vec!["user_id".into()],
    }))
    .await
    .expect("profiles");

    for id in [1, 2] {
        db.commit(Mutation::Upsert {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "users".into(),
            primary_key: vec![Value::Integer(id)],
            row: Row {
                values: vec![Value::Integer(id)],
            },
        })
        .await
        .expect("seed user");
    }
    db.commit(Mutation::Upsert {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "profiles".into(),
        primary_key: vec![Value::Integer(1)],
        row: Row {
            values: vec![Value::Integer(1), Value::Text("reader".into())],
        },
    })
    .await
    .expect("profile 1");
    db.commit(Mutation::Upsert {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "profiles".into(),
        primary_key: vec![Value::Integer(2)],
        row: Row {
            values: vec![Value::Integer(2), Value::Text("bob".into())],
        },
    })
    .await
    .expect("profile 2");

    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        caller_id: "reader".into(),
        permission: Permission::TableRead {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "users".into(),
        },
        actor_id: None,
        delegable: false,
    }))
    .await
    .expect("grant users read");
    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        caller_id: "reader".into(),
        permission: Permission::TableRead {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "profiles".into(),
        },
        actor_id: None,
        delegable: false,
    }))
    .await
    .expect("grant profiles read");
    db.set_read_policy(
        "p",
        "app",
        "profiles",
        Expr::Eq("owner".into(), Value::Text("$caller_id".into())),
    )
    .await
    .expect("set join policy");

    let reader = CallerContext::new("reader");
    let result = db
        .query_with_options_as(
            Some(&reader),
            "p",
            "app",
            Query::select(&["*"])
                .from("users")
                .alias("u")
                .inner_join("profiles", "u.id", "user_id")
                .with_last_join_alias("p"),
            QueryOptions::default(),
        )
        .await
        .expect("join query");
    assert_eq!(result.rows.len(), 1);
}

#[tokio::test]
async fn table_policy_bypass_permission_skips_row_policy_filtering() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "users".into(),
        owner_id: None,
        if_not_exists: false,
        columns: vec![
            ColumnDef {
                name: "id".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "owner".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
        ],
        primary_key: vec!["id".into()],
    }))
    .await
    .expect("users");
    db.commit(Mutation::Upsert {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "users".into(),
        primary_key: vec![Value::Integer(1)],
        row: Row {
            values: vec![Value::Integer(1), Value::Text("reader".into())],
        },
    })
    .await
    .expect("seed user 1");
    db.commit(Mutation::Upsert {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "users".into(),
        primary_key: vec![Value::Integer(2)],
        row: Row {
            values: vec![Value::Integer(2), Value::Text("bob".into())],
        },
    })
    .await
    .expect("seed user 2");
    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        caller_id: "reader".into(),
        permission: Permission::TableRead {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "users".into(),
        },
        actor_id: None,
        delegable: false,
    }))
    .await
    .expect("grant table read");
    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        caller_id: "reader".into(),
        permission: Permission::PolicyBypass {
            project_id: "p".into(),
            table_name: Some("users".into()),
        },
        actor_id: None,
        delegable: false,
    }))
    .await
    .expect("grant policy bypass");
    db.set_read_policy(
        "p",
        "app",
        "users",
        Expr::Eq("owner".into(), Value::Text("$caller_id".into())),
    )
    .await
    .expect("set policy");

    let reader = CallerContext::new("reader");
    let result = db
        .query_with_options_as(
            Some(&reader),
            "p",
            "app",
            Query::select(&["*"]).from("users").limit(10),
            QueryOptions::default(),
        )
        .await
        .expect("query with policy bypass");
    assert_eq!(result.rows.len(), 2);
}

#[tokio::test]
async fn project_policy_bypass_permission_skips_joined_table_policies() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "users".into(),
        owner_id: None,
        if_not_exists: false,
        columns: vec![
            ColumnDef {
                name: "id".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "owner".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
        ],
        primary_key: vec!["id".into()],
    }))
    .await
    .expect("users");
    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "profiles".into(),
        owner_id: None,
        if_not_exists: false,
        columns: vec![
            ColumnDef {
                name: "user_id".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "owner".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
        ],
        primary_key: vec!["user_id".into()],
    }))
    .await
    .expect("profiles");

    for (id, owner) in [(1, "reader"), (2, "bob")] {
        db.commit(Mutation::Upsert {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "users".into(),
            primary_key: vec![Value::Integer(id)],
            row: Row {
                values: vec![Value::Integer(id), Value::Text(owner.into())],
            },
        })
        .await
        .expect("seed user");
        db.commit(Mutation::Upsert {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "profiles".into(),
            primary_key: vec![Value::Integer(id)],
            row: Row {
                values: vec![Value::Integer(id), Value::Text(owner.into())],
            },
        })
        .await
        .expect("seed profile");
    }

    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        caller_id: "reader".into(),
        permission: Permission::TableRead {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "users".into(),
        },
        actor_id: None,
        delegable: false,
    }))
    .await
    .expect("grant users read");
    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        caller_id: "reader".into(),
        permission: Permission::TableRead {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "profiles".into(),
        },
        actor_id: None,
        delegable: false,
    }))
    .await
    .expect("grant profiles read");
    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        caller_id: "reader".into(),
        permission: Permission::PolicyBypass {
            project_id: "p".into(),
            table_name: None,
        },
        actor_id: None,
        delegable: false,
    }))
    .await
    .expect("grant project policy bypass");
    db.set_read_policy(
        "p",
        "app",
        "users",
        Expr::Eq("owner".into(), Value::Text("$caller_id".into())),
    )
    .await
    .expect("set users policy");
    db.set_read_policy(
        "p",
        "app",
        "profiles",
        Expr::Eq("owner".into(), Value::Text("$caller_id".into())),
    )
    .await
    .expect("set profiles policy");

    let reader = CallerContext::new("reader");
    let result = db
        .query_with_options_as(
            Some(&reader),
            "p",
            "app",
            Query::select(&["*"])
                .from("users")
                .alias("u")
                .inner_join("profiles", "u.id", "user_id")
                .with_last_join_alias("pr")
                .limit(10),
            QueryOptions::default(),
        )
        .await
        .expect("query with project policy bypass");
    assert_eq!(result.rows.len(), 2);
}

#[tokio::test]
async fn join_query_rejects_duplicate_aliases_to_prevent_policy_binding_ambiguity() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "users".into(),
        owner_id: None,
        if_not_exists: false,
        columns: vec![
            ColumnDef {
                name: "id".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "owner".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
        ],
        primary_key: vec!["id".into()],
    }))
    .await
    .expect("users");
    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "profiles".into(),
        owner_id: None,
        if_not_exists: false,
        columns: vec![
            ColumnDef {
                name: "user_id".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "owner".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
        ],
        primary_key: vec!["user_id".into()],
    }))
    .await
    .expect("profiles");

    db.commit(Mutation::Upsert {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "users".into(),
        primary_key: vec![Value::Integer(1)],
        row: Row {
            values: vec![Value::Integer(1), Value::Text("reader".into())],
        },
    })
    .await
    .expect("seed user");
    db.commit(Mutation::Upsert {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "profiles".into(),
        primary_key: vec![Value::Integer(1)],
        row: Row {
            values: vec![Value::Integer(1), Value::Text("bob".into())],
        },
    })
    .await
    .expect("seed profile");

    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        caller_id: "reader".into(),
        permission: Permission::TableRead {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "users".into(),
        },
        actor_id: None,
        delegable: false,
    }))
    .await
    .expect("grant users read");
    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        caller_id: "reader".into(),
        permission: Permission::TableRead {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "profiles".into(),
        },
        actor_id: None,
        delegable: false,
    }))
    .await
    .expect("grant profiles read");
    db.set_read_policy(
        "p",
        "app",
        "profiles",
        Expr::Eq("owner".into(), Value::Text("$caller_id".into())),
    )
    .await
    .expect("set policy");

    let reader = CallerContext::new("reader");
    let err = db
        .query_with_options_as(
            Some(&reader),
            "p",
            "app",
            Query::select(&["*"])
                .from("users")
                .alias("u")
                .inner_join("profiles", "u.id", "user_id")
                .with_last_join_alias("u"),
            QueryOptions::default(),
        )
        .await
        .expect_err("duplicate aliases should be rejected");
    assert!(matches!(err, QueryError::InvalidQuery { .. }));
}

#[tokio::test]
async fn secure_profile_requires_hardened_storage_settings() {
    let dir = tempdir().expect("temp");
    let weak = AedbConfig::default();
    let err = AedbInstance::open_secure(weak, dir.path())
        .err()
        .expect("must reject weak secure profile");
    assert!(matches!(err, AedbError::InvalidConfig { .. }));

    let hardened = AedbConfig::production([9u8; 32]);
    AedbInstance::open_secure(hardened, dir.path()).expect("open secure");
}

#[tokio::test]
async fn secure_profile_rejects_batch_durability() {
    let dir = tempdir().expect("temp");
    let mut weak = AedbConfig::production([9u8; 32]);
    weak.durability_mode = DurabilityMode::Batch;
    let err = AedbInstance::open_secure(weak, dir.path())
        .err()
        .expect("secure profile must reject batch durability");
    assert!(matches!(err, AedbError::InvalidConfig { .. }));
}

#[tokio::test]
async fn secure_profile_rejects_short_hmac_key() {
    let dir = tempdir().expect("temp");
    let weak = AedbConfig::default()
        .with_hmac_key(vec![1, 2, 3, 4, 5, 6, 7, 8])
        .with_checkpoint_key([3u8; 32]);
    let err = AedbInstance::open_secure(weak, dir.path())
        .err()
        .expect("short hmac key must be rejected");
    assert!(matches!(err, AedbError::InvalidConfig { .. }));
}

#[tokio::test]
async fn secure_mode_requires_authenticated_apis() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_production(AedbConfig::production([7u8; 32]), dir.path())
        .expect("open secure");

    let err = db
        .commit(Mutation::Ddl(DdlOperation::CreateProject {
            owner_id: None,
            if_not_exists: true,
            project_id: "p".into(),
        }))
        .await
        .expect_err("anonymous commit must fail");
    assert!(matches!(err, AedbError::PermissionDenied(_)));

    db.commit_as(
        CallerContext::system_internal(),
        Mutation::Ddl(DdlOperation::CreateProject {
            owner_id: None,
            if_not_exists: true,
            project_id: "p".into(),
        }),
    )
    .await
    .expect("system commit");

    let spoof_err = db
        .commit_as(
            CallerContext::new(SYSTEM_CALLER_ID),
            Mutation::Ddl(DdlOperation::CreateProject {
                owner_id: None,
                if_not_exists: true,
                project_id: "q".into(),
            }),
        )
        .await
        .expect_err("spoofed system caller must fail");
    assert!(matches!(spoof_err, AedbError::PermissionDenied(_)));

    let qerr = db
        .query_no_auth(
            "p",
            "app",
            Query::select(&["*"]).from("users").limit(1),
            QueryOptions::default(),
        )
        .await
        .expect_err("anonymous query must fail");
    assert!(matches!(qerr, QueryError::PermissionDenied { .. }));

    let qerr_direct = db
        .query_with_options_as(
            None,
            "p",
            "app",
            Query::select(&["*"]).from("users").limit(1),
            QueryOptions::default(),
        )
        .await
        .expect_err("anonymous query_with_options_as must fail");
    assert!(matches!(qerr_direct, QueryError::PermissionDenied { .. }));

    let explain_err = db
        .explain_query_as(
            None,
            "p",
            "app",
            Query::select(&["*"]).from("users").limit(1),
            QueryOptions::default(),
        )
        .await
        .expect_err("anonymous explain_query_as must fail");
    assert!(matches!(explain_err, QueryError::PermissionDenied { .. }));
}

#[tokio::test]
async fn commit_as_rejects_empty_caller_id() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    let err = db
        .commit_as(
            CallerContext::new("   "),
            Mutation::Ddl(DdlOperation::CreateProject {
                owner_id: None,
                if_not_exists: true,
                project_id: "p".into(),
            }),
        )
        .await
        .expect_err("empty caller id should be rejected");
    assert!(matches!(err, AedbError::PermissionDenied(_)));
}

#[tokio::test]
async fn query_as_rejects_empty_caller_id() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    let caller = CallerContext::new("");
    let err = db
        .query_with_options_as(
            Some(&caller),
            "p",
            "app",
            Query::select(&["*"]).from("authz_audit").limit(1),
            QueryOptions::default(),
        )
        .await
        .expect_err("empty caller id should be rejected");
    assert!(matches!(err, QueryError::PermissionDenied { .. }));
}

#[tokio::test]
async fn query_with_options_as_rejects_reserved_system_caller() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_production(AedbConfig::production([8u8; 32]), dir.path())
        .expect("open secure");
    db.commit_as(
        CallerContext::system_internal(),
        Mutation::Ddl(DdlOperation::CreateProject {
            owner_id: None,
            if_not_exists: true,
            project_id: "p".into(),
        }),
    )
    .await
    .expect("create project");

    let caller = CallerContext::new("system");
    let err = db
        .query_with_options_as(
            Some(&caller),
            "p",
            "app",
            Query::select(&["*"]).from("__system_authz"),
            QueryOptions::default(),
        )
        .await
        .expect_err("reserved system caller should be rejected");
    assert!(matches!(err, QueryError::PermissionDenied { .. }));
}

#[tokio::test]
async fn query_no_auth_in_secure_mode_returns_structured_error() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_production(AedbConfig::production([6u8; 32]), dir.path())
        .expect("open secure");
    let err = db
        .query_no_auth(
            "p",
            "app",
            Query::select(&["*"]).from("__system_authz"),
            QueryOptions::default(),
        )
        .await
        .expect_err("secure mode should reject query_no_auth");
    assert!(matches!(err, QueryError::PermissionDenied { .. }));
}

#[tokio::test]
async fn kv_no_auth_apis_in_secure_mode_return_structured_error() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_production(AedbConfig::production([4u8; 32]), dir.path())
        .expect("open secure");

    let get_err = db
        .kv_get_no_auth("p", "app", b"k", ConsistencyMode::AtLatest)
        .await
        .expect_err("secure mode should reject kv_get_no_auth");
    assert!(matches!(get_err, QueryError::PermissionDenied { .. }));

    let scan_err = db
        .kv_scan_prefix_no_auth("p", "app", b"k", 10, ConsistencyMode::AtLatest)
        .await
        .expect_err("secure mode should reject kv_scan_prefix_no_auth");
    assert!(matches!(scan_err, QueryError::PermissionDenied { .. }));
}

#[tokio::test]
async fn postflight_assertions_require_read_permission_in_secure_mode() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_production(AedbConfig::production([8u8; 32]), dir.path())
        .expect("open secure");
    let system = CallerContext::system_internal();
    db.commit_as(
        system.clone(),
        Mutation::Ddl(DdlOperation::CreateProject {
            owner_id: None,
            if_not_exists: true,
            project_id: "p".into(),
        }),
    )
    .await
    .expect("create project");
    db.commit_as(
        system,
        Mutation::KvSet {
            project_id: "p".into(),
            scope_id: "app".into(),
            key: b"secret".to_vec(),
            value: b"not-for-mallory".to_vec(),
        },
    )
    .await
    .expect("seed secret");

    let err = db
        .commit_as(
            CallerContext::new("mallory"),
            Mutation::PostflightCheck {
                assertions: vec![ReadAssertion::KeyCompare {
                    project_id: "p".into(),
                    scope_id: "app".into(),
                    key: b"secret".to_vec(),
                    op: CompareOp::Eq,
                    threshold: b"guess".to_vec(),
                }],
            },
        )
        .await
        .expect_err("postflight assertion without read permission must fail before evaluation");
    assert!(matches!(err, AedbError::PermissionDenied(_)));
}

#[tokio::test]
async fn envelope_assertions_require_read_permission_in_secure_mode() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_production(AedbConfig::production([12u8; 32]), dir.path())
        .expect("open secure");
    let system = CallerContext::system_internal();
    db.commit_as(
        system.clone(),
        Mutation::Ddl(DdlOperation::CreateProject {
            owner_id: None,
            if_not_exists: true,
            project_id: "p".into(),
        }),
    )
    .await
    .expect("create project");
    db.commit_as(
        system,
        Mutation::KvSet {
            project_id: "p".into(),
            scope_id: "app".into(),
            key: b"secret".to_vec(),
            value: b"not-for-mallory".to_vec(),
        },
    )
    .await
    .expect("seed secret");

    let err = db
        .commit_envelope(TransactionEnvelope {
            caller: Some(CallerContext::new("mallory")),
            idempotency_key: None,
            write_class: WriteClass::Standard,
            assertions: vec![ReadAssertion::KeyCompare {
                project_id: "p".into(),
                scope_id: "app".into(),
                key: b"secret".to_vec(),
                op: CompareOp::Eq,
                threshold: b"guess".to_vec(),
            }],
            read_set: ReadSet::default(),
            write_intent: WriteIntent {
                mutations: Vec::new(),
            },
            base_seq: 0,
        })
        .await
        .expect_err("envelope assertion without read permission must fail before evaluation");
    assert!(matches!(err, AedbError::PermissionDenied(_)));
}

#[tokio::test]
async fn async_index_query_requires_table_read_permission() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_production(AedbConfig::production([10u8; 32]), dir.path())
        .expect("open secure");
    let system = CallerContext::system_internal();
    db.commit_as(
        system.clone(),
        Mutation::Ddl(DdlOperation::CreateProject {
            owner_id: None,
            if_not_exists: true,
            project_id: "p".into(),
        }),
    )
    .await
    .expect("create project");
    db.commit_as(
        system.clone(),
        Mutation::Ddl(DdlOperation::CreateScope {
            project_id: "p".into(),
            scope_id: "app".into(),
            owner_id: None,
            if_not_exists: true,
        }),
    )
    .await
    .expect("create scope");
    db.commit_as(
        system.clone(),
        Mutation::Ddl(DdlOperation::CreateTable {
            owner_id: None,
            if_not_exists: false,
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "users".into(),
            columns: vec![
                ColumnDef {
                    name: "id".into(),
                    col_type: ColumnType::Integer,
                    nullable: false,
                },
                ColumnDef {
                    name: "name".into(),
                    col_type: ColumnType::Text,
                    nullable: false,
                },
            ],
            primary_key: vec!["id".into()],
        }),
    )
    .await
    .expect("create table");
    db.commit_as(
        system.clone(),
        Mutation::Ddl(DdlOperation::CreateAsyncIndex {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "users".into(),
            index_name: "users_view".into(),
            if_not_exists: false,
            projected_columns: vec!["id".into(), "name".into()],
        }),
    )
    .await
    .expect("create async index");
    db.commit_as(
        system.clone(),
        Mutation::Ddl(DdlOperation::GrantPermission {
            caller_id: "alice".into(),
            permission: Permission::IndexRead {
                project_id: "p".into(),
                scope_id: "app".into(),
                table_name: "users".into(),
                index_name: "users_view".into(),
            },
            actor_id: Some("system".into()),
            delegable: false,
        }),
    )
    .await
    .expect("grant index read");

    let err = db
        .query_with_options_as(
            Some(&CallerContext::new("alice")),
            "p",
            "app",
            Query::select(&["*"]).from("users"),
            QueryOptions {
                async_index: Some("users_view".into()),
                allow_full_scan: true,
                ..QueryOptions::default()
            },
        )
        .await
        .expect_err("index read alone must not disclose async projection rows");
    assert!(matches!(err, QueryError::PermissionDenied { .. }));
}

#[tokio::test]
async fn no_auth_read_apis_work_in_non_secure_mode() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");

    db.create_project("p").await.expect("project");
    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        owner_id: None,
        if_not_exists: false,
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "cfg".into(),
        columns: vec![
            ColumnDef {
                name: "id".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "name".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
        ],
        primary_key: vec!["id".into()],
    }))
    .await
    .expect("create cfg table");
    db.commit(Mutation::Upsert {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "cfg".into(),
        primary_key: vec![Value::Integer(1)],
        row: Row {
            values: vec![Value::Integer(1), Value::Text("bootstrap".into())],
        },
    })
    .await
    .expect("upsert cfg row");
    db.commit(Mutation::KvSet {
        project_id: "p".into(),
        scope_id: "app".into(),
        key: b"config:bootstrap".to_vec(),
        value: b"v1".to_vec(),
    })
    .await
    .expect("set kv");

    let got = db
        .kv_get_no_auth("p", "app", b"config:bootstrap", ConsistencyMode::AtLatest)
        .await
        .expect("kv_get_no_auth")
        .expect("entry");
    assert_eq!(got.value, b"v1".to_vec());

    let scanned = db
        .kv_scan_prefix_no_auth("p", "app", b"config:", 10, ConsistencyMode::AtLatest)
        .await
        .expect("scan no auth");
    assert_eq!(scanned.len(), 1);

    let query_res = db
        .query_no_auth(
            "p",
            "app",
            Query::select(&["id", "name"]).from("cfg").limit(10),
            QueryOptions::default(),
        )
        .await
        .expect("query_no_auth");
    assert_eq!(query_res.rows.len(), 1);
}

#[tokio::test]
async fn secure_multi_agent_user_perspective_invariants_hold() {
    fn u256_be(v: u64) -> [u8; 32] {
        let mut out = [0u8; 32];
        out[24..].copy_from_slice(&v.to_be_bytes());
        out
    }

    #[allow(clippy::too_many_arguments)]
    fn req(
        instrument: &str,
        owner: &str,
        cid: String,
        side: crate::order_book::OrderSide,
        tif: crate::order_book::TimeInForce,
        price: i64,
        qty: u64,
        nonce: u64,
    ) -> crate::order_book::OrderRequest {
        crate::order_book::OrderRequest {
            instrument: instrument.to_string(),
            client_order_id: cid,
            side,
            order_type: crate::order_book::OrderType::Limit,
            time_in_force: tif,
            exec_instructions: crate::order_book::ExecInstruction(0),
            self_trade_prevention: crate::order_book::SelfTradePrevention::None,
            price_ticks: price,
            qty_be: u256_be(qty),
            owner: owner.to_string(),
            account: None,
            nonce,
            price_limit_ticks: None,
        }
    }

    #[derive(Debug, Default)]
    struct AgentMetrics {
        primary_attempted: usize,
        primary_accepted: usize,
        primary_rejected: usize,
        lifecycle_attempted: usize,
        lifecycle_accepted: usize,
        lifecycle_rejected: usize,
        own_read_checks: usize,
    }

    let dir = tempdir().expect("temp");
    let db = Arc::new(
        AedbInstance::open_secure(AedbConfig::production([6u8; 32]), dir.path())
            .expect("open secure"),
    );
    let system = CallerContext::system_internal();
    db.commit_as(
        system.clone(),
        Mutation::Ddl(DdlOperation::CreateProject {
            owner_id: None,
            if_not_exists: true,
            project_id: "p".into(),
        }),
    )
    .await
    .expect("create project");

    let agents: Vec<String> = (0..8).map(|i| format!("agent_{i}")).collect();
    for a in &agents {
        db.commit_as(
            system.clone(),
            Mutation::Ddl(DdlOperation::GrantPermission {
                caller_id: a.clone(),
                permission: Permission::ProjectAdmin {
                    project_id: "p".into(),
                },
                actor_id: Some("system".into()),
                delegable: false,
            }),
        )
        .await
        .expect("grant project admin");
    }

    db.order_book_set_instrument_config_as(
        system.clone(),
        "p",
        "app",
        "BTC-USD",
        crate::order_book::InstrumentConfig {
            instrument: "BTC-USD".into(),
            tick_size: 1,
            lot_size_be: u256_be(1),
            min_price_ticks: 1,
            max_price_ticks: 1_000_000,
            market_order_price_band: Some(50),
            halted: false,
            balance_config: None,
        },
    )
    .await
    .expect("instrument config");

    for i in 0..16u64 {
        db.order_book_new_as(
            system.clone(),
            "p",
            "app",
            req(
                "BTC-USD",
                &format!("seed_ask_{i}"),
                format!("seed-ask-{i}"),
                crate::order_book::OrderSide::Ask,
                crate::order_book::TimeInForce::Gtc,
                1_000 + i as i64,
                10,
                1,
            ),
        )
        .await
        .expect("seed ask");
        db.order_book_new_as(
            system.clone(),
            "p",
            "app",
            req(
                "BTC-USD",
                &format!("seed_bid_{i}"),
                format!("seed-bid-{i}"),
                crate::order_book::OrderSide::Bid,
                crate::order_book::TimeInForce::Gtc,
                999 - i as i64,
                10,
                1,
            ),
        )
        .await
        .expect("seed bid");
    }

    let mut anchors: Vec<(String, u64)> = Vec::with_capacity(agents.len());
    for (idx, agent_id) in agents.iter().enumerate() {
        let caller = CallerContext::new(agent_id.clone());
        let anchor_cid = format!("anchor-{agent_id}");
        db.order_book_new_as(
            caller.clone(),
            "p",
            "app",
            req(
                "BTC-USD",
                agent_id,
                anchor_cid.clone(),
                if idx % 2 == 0 {
                    crate::order_book::OrderSide::Bid
                } else {
                    crate::order_book::OrderSide::Ask
                },
                crate::order_book::TimeInForce::Gtc,
                if idx % 2 == 0 { 980 } else { 1_020 },
                2,
                1,
            ),
        )
        .await
        .expect("anchor order");
        let own_open = db
            .open_orders(
                "p",
                "app",
                "BTC-USD",
                agent_id,
                ConsistencyMode::AtLatest,
                &caller,
            )
            .await
            .expect("own open orders");
        let anchor = own_open
            .into_iter()
            .find(|o| o.client_order_id == anchor_cid)
            .expect("anchor order discoverable");
        anchors.push((agent_id.clone(), anchor.order_id));
    }

    let mut tasks = Vec::with_capacity(agents.len());
    for (idx, agent_id) in agents.iter().enumerate() {
        let db_clone = Arc::clone(&db);
        let caller = CallerContext::new(agent_id.clone());
        let owner = agent_id.clone();
        tasks.push(tokio::spawn(async move {
            let mut m = AgentMetrics::default();
            let mut nonce = 10u64;
            for op in 0..180usize {
                let side = if (op + idx) % 2 == 0 {
                    crate::order_book::OrderSide::Bid
                } else {
                    crate::order_book::OrderSide::Ask
                };
                let price = if matches!(side, crate::order_book::OrderSide::Bid) {
                    1_001
                } else {
                    998
                };
                m.primary_attempted += 1;
                let res = db_clone
                    .order_book_new_as(
                        caller.clone(),
                        "p",
                        "app",
                        req(
                            "BTC-USD",
                            &owner,
                            format!("{owner}-p-{op}"),
                            side,
                            crate::order_book::TimeInForce::Ioc,
                            price,
                            1 + (op % 4) as u64,
                            nonce,
                        ),
                    )
                    .await;
                nonce += 1;
                match res {
                    Ok(_) => m.primary_accepted += 1,
                    Err(AedbError::Validation(_)) | Err(AedbError::Conflict(_)) => {
                        m.primary_rejected += 1
                    }
                    Err(other) => return Err(other),
                }

                if op % 30 == 0 {
                    m.lifecycle_attempted += 1;
                    let cid = format!("{owner}-l-{op}");
                    let opened = db_clone
                        .order_book_new_as(
                            caller.clone(),
                            "p",
                            "app",
                            req(
                                "BTC-USD",
                                &owner,
                                cid.clone(),
                                crate::order_book::OrderSide::Bid,
                                crate::order_book::TimeInForce::Gtc,
                                970,
                                1,
                                nonce,
                            ),
                        )
                        .await;
                    nonce += 1;
                    match opened {
                        Ok(_) => m.lifecycle_accepted += 1,
                        Err(AedbError::Validation(_)) | Err(AedbError::Conflict(_)) => {
                            m.lifecycle_rejected += 1
                        }
                        Err(other) => return Err(other),
                    }

                    m.lifecycle_attempted += 1;
                    match db_clone
                        .order_book_cancel_by_client_id_as(
                            caller.clone(),
                            "p",
                            "app",
                            "BTC-USD",
                            &cid,
                            &owner,
                        )
                        .await
                    {
                        Ok(_) => m.lifecycle_accepted += 1,
                        Err(AedbError::Validation(_)) | Err(AedbError::Conflict(_)) => {
                            m.lifecycle_rejected += 1
                        }
                        Err(other) => return Err(other),
                    }
                }

                if op % 40 == 0 {
                    let own = db_clone
                        .open_orders(
                            "p",
                            "app",
                            "BTC-USD",
                            &owner,
                            ConsistencyMode::AtLatest,
                            &caller,
                        )
                        .await
                        .map_err(|e| {
                            AedbError::Validation(format!("own open_orders failed: {e}"))
                        })?;
                    assert!(
                        own.iter().all(|o| o.owner == owner),
                        "open_orders must only return owner rows"
                    );
                    m.own_read_checks += 1;
                }
            }
            Ok::<_, AedbError>(m)
        }));
    }

    let mut metrics = Vec::with_capacity(agents.len());
    for task in tasks {
        metrics.push(task.await.expect("join agent task").expect("agent run"));
    }

    for (idx, agent_id) in agents.iter().enumerate() {
        let caller = CallerContext::new(agent_id.clone());
        let (target_owner, target_order_id) = &anchors[(idx + 1) % anchors.len()];

        let err = db
            .order_status(
                "p",
                "app",
                "BTC-USD",
                *target_order_id,
                ConsistencyMode::AtLatest,
                &caller,
            )
            .await
            .expect_err("cross-owner order_status must be denied");
        assert!(
            matches!(err, QueryError::PermissionDenied { .. }),
            "expected permission denied, got {err:?}"
        );

        let err = db
            .open_orders(
                "p",
                "app",
                "BTC-USD",
                target_owner,
                ConsistencyMode::AtLatest,
                &caller,
            )
            .await
            .expect_err("cross-owner open_orders must be denied");
        assert!(
            matches!(err, QueryError::PermissionDenied { .. }),
            "expected permission denied, got {err:?}"
        );
    }

    for m in &metrics {
        assert_eq!(
            m.primary_accepted + m.primary_rejected,
            m.primary_attempted,
            "primary accounting mismatch"
        );
        assert_eq!(
            m.lifecycle_accepted + m.lifecycle_rejected,
            m.lifecycle_attempted,
            "lifecycle accounting mismatch"
        );
        assert!(
            m.own_read_checks > 0,
            "agent should perform own-read checks"
        );
    }
}

#[tokio::test]
async fn query_sql_read_only_respects_adapter_scan_policy() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    db.create_scope("p", "app").await.expect("scope");
    create_table(
        &db,
        "p",
        "app",
        "items",
        vec![ColumnDef {
            name: "id".into(),
            col_type: ColumnType::Integer,
            nullable: false,
        }],
        vec!["id"],
    )
    .await;
    db.commit(Mutation::Upsert {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "items".into(),
        primary_key: vec![Value::Integer(1)],
        row: Row::from_values(vec![Value::Integer(1)]),
    })
    .await
    .expect("seed");

    let err = db
        .query_sql_read_only(
            &RestrictiveSqlAdapter,
            "p",
            "app",
            "select * from items",
            ConsistencyMode::AtLatest,
        )
        .await
        .expect_err("adapter should not be forced into full scan mode");
    assert!(matches!(err, QueryError::InvalidQuery { .. }));
}

#[tokio::test]
#[ignore = "long-running multi-agent user-perspective profile"]
async fn secure_multi_agent_profile_identifies_core_shortcomings() {
    fn u256_be(v: u64) -> [u8; 32] {
        let mut out = [0u8; 32];
        out[24..].copy_from_slice(&v.to_be_bytes());
        out
    }
    fn decode_u256_u64(bytes: [u8; 32]) -> u64 {
        let mut out = [0u8; 8];
        out.copy_from_slice(&bytes[24..]);
        u64::from_be_bytes(out)
    }

    #[derive(Debug, Default, Clone, Copy)]
    struct WorkerStats {
        attempted: usize,
        accepted: usize,
        rejected: usize,
        unauthorized_attempted: usize,
        unauthorized_denied: usize,
        latency_sum_us: u128,
        latency_max_us: u64,
    }

    #[derive(Debug, Default, Clone, Copy)]
    struct RuntimePeaks {
        queue_depth: usize,
        inflight: usize,
        durable_lag: u64,
        conflict_rate: f64,
        durable_wait_ops: u64,
        durable_wait_avg_us: u64,
        wal_append_ops: u64,
        wal_append_bytes: u64,
        wal_append_avg_us: u64,
        wal_sync_ops: u64,
        wal_sync_avg_us: u64,
        commit_errors: u64,
        permission_rejections: u64,
        validation_rejections: u64,
        queue_full_rejections: u64,
        timeout_rejections: u64,
        read_set_conflicts: u64,
    }

    let agents = std::env::var("AEDB_MULTI_AGENT_PROFILE_AGENTS")
        .ok()
        .and_then(|v| v.parse::<usize>().ok())
        .unwrap_or(12)
        .max(4);
    let ops_per_agent = std::env::var("AEDB_MULTI_AGENT_PROFILE_OPS")
        .ok()
        .and_then(|v| v.parse::<usize>().ok())
        .unwrap_or(800)
        .max(100);

    let dir = tempdir().expect("temp");
    let db = Arc::new(
        AedbInstance::open_secure(AedbConfig::production([3u8; 32]), dir.path())
            .expect("open secure"),
    );
    let system = CallerContext::system_internal();
    db.commit_as(
        system.clone(),
        Mutation::Ddl(DdlOperation::CreateProject {
            owner_id: None,
            if_not_exists: true,
            project_id: "p".into(),
        }),
    )
    .await
    .expect("create project");

    let agent_ids: Vec<String> = (0..agents).map(|i| format!("prof_agent_{i}")).collect();
    for a in &agent_ids {
        db.commit_as(
            system.clone(),
            Mutation::Ddl(DdlOperation::GrantPermission {
                caller_id: a.clone(),
                permission: Permission::KvRead {
                    project_id: "p".into(),
                    scope_id: Some("app".into()),
                    prefix: Some(format!("agent:{a}:").into_bytes()),
                },
                actor_id: Some("system".into()),
                delegable: false,
            }),
        )
        .await
        .expect("grant kv read");
        db.commit_as(
            system.clone(),
            Mutation::Ddl(DdlOperation::GrantPermission {
                caller_id: a.clone(),
                permission: Permission::KvWrite {
                    project_id: "p".into(),
                    scope_id: Some("app".into()),
                    prefix: Some(format!("agent:{a}:").into_bytes()),
                },
                actor_id: Some("system".into()),
                delegable: false,
            }),
        )
        .await
        .expect("grant kv write");
        db.commit_as(
            system.clone(),
            Mutation::KvSet {
                project_id: "p".into(),
                scope_id: "app".into(),
                key: format!("agent:{a}:balance").into_bytes(),
                value: u256_be(0).to_vec(),
            },
        )
        .await
        .expect("seed balance");
    }

    let done = Arc::new(tokio::sync::Notify::new());
    let peaks = Arc::new(std::sync::Mutex::new(RuntimePeaks::default()));
    let monitor_db = Arc::clone(&db);
    let monitor_done = Arc::clone(&done);
    let monitor_peaks = Arc::clone(&peaks);
    let monitor = tokio::spawn(async move {
        loop {
            tokio::select! {
                _ = monitor_done.notified() => break,
                _ = tokio::time::sleep(Duration::from_millis(10)) => {
                    let op = monitor_db.operational_metrics().await;
                    let mut p = monitor_peaks.lock().expect("peaks lock");
                    p.queue_depth = p.queue_depth.max(op.queue_depth);
                    p.inflight = p.inflight.max(op.inflight_commits);
                    p.durable_lag = p.durable_lag.max(op.durable_head_lag);
                    p.conflict_rate = p.conflict_rate.max(op.conflict_rate);
                    p.durable_wait_ops = p.durable_wait_ops.max(op.durable_wait_ops);
                    p.durable_wait_avg_us = p.durable_wait_avg_us.max(op.avg_durable_wait_micros);
                    p.wal_append_ops = p.wal_append_ops.max(op.wal_append_ops);
                    p.wal_append_bytes = p.wal_append_bytes.max(op.wal_append_bytes);
                    p.wal_append_avg_us = p.wal_append_avg_us.max(op.avg_wal_append_micros);
                    p.wal_sync_ops = p.wal_sync_ops.max(op.wal_sync_ops);
                    p.wal_sync_avg_us = p.wal_sync_avg_us.max(op.avg_wal_sync_micros);
                    p.commit_errors = p.commit_errors.max(op.commit_errors);
                    p.permission_rejections = p.permission_rejections.max(op.permission_rejections);
                    p.validation_rejections = p.validation_rejections.max(op.validation_rejections);
                    p.queue_full_rejections = p.queue_full_rejections.max(op.queue_full_rejections);
                    p.timeout_rejections = p.timeout_rejections.max(op.timeout_rejections);
                    p.read_set_conflicts = p.read_set_conflicts.max(op.read_set_conflicts);
                }
            }
        }
    });

    let started = Instant::now();
    let mut tasks = Vec::with_capacity(agent_ids.len());
    for (idx, agent) in agent_ids.iter().enumerate() {
        let db_clone = Arc::clone(&db);
        let caller = CallerContext::new(agent.clone());
        let own_key = format!("agent:{agent}:balance").into_bytes();
        let neighbor = &agent_ids[(idx + 1) % agent_ids.len()];
        let cross_key = format!("agent:{neighbor}:balance").into_bytes();
        tasks.push(tokio::spawn(async move {
            let mut stats = WorkerStats::default();
            for op in 0..ops_per_agent {
                stats.attempted += 1;
                let now = Instant::now();
                let res = db_clone
                    .commit_as(
                        caller.clone(),
                        Mutation::KvIncU256 {
                            project_id: "p".into(),
                            scope_id: "app".into(),
                            key: own_key.clone(),
                            amount_be: u256_be(1),
                        },
                    )
                    .await;
                let us = now.elapsed().as_micros() as u64;
                stats.latency_sum_us += us as u128;
                stats.latency_max_us = stats.latency_max_us.max(us);
                match res {
                    Ok(_) => stats.accepted += 1,
                    Err(AedbError::Validation(_)) | Err(AedbError::Conflict(_)) => {
                        stats.rejected += 1
                    }
                    Err(other) => return Err(other),
                }

                if op % 25 == 0 {
                    stats.unauthorized_attempted += 1;
                    let denied = db_clone
                        .commit_as(
                            caller.clone(),
                            Mutation::KvIncU256 {
                                project_id: "p".into(),
                                scope_id: "app".into(),
                                key: cross_key.clone(),
                                amount_be: u256_be(1),
                            },
                        )
                        .await;
                    if matches!(denied, Err(AedbError::PermissionDenied(_))) {
                        stats.unauthorized_denied += 1;
                    } else {
                        denied?;
                    }
                }
            }
            Ok::<_, AedbError>(stats)
        }));
    }

    let mut merged = WorkerStats::default();
    for t in tasks {
        let s = t.await.expect("join profile task").expect("profile run");
        merged.attempted += s.attempted;
        merged.accepted += s.accepted;
        merged.rejected += s.rejected;
        merged.unauthorized_attempted += s.unauthorized_attempted;
        merged.unauthorized_denied += s.unauthorized_denied;
        merged.latency_sum_us += s.latency_sum_us;
        merged.latency_max_us = merged.latency_max_us.max(s.latency_max_us);
    }

    done.notify_waiters();
    monitor.await.expect("join monitor");
    db.force_fsync().await.expect("force fsync");

    let elapsed = started.elapsed().as_secs_f64().max(0.001);
    let attempted_tps = (merged.attempted as f64 / elapsed) as u64;
    let accepted_tps = (merged.accepted as f64 / elapsed) as u64;
    let avg_lat_us = (merged.latency_sum_us / merged.attempted.max(1) as u128) as u64;
    let peaks = *peaks.lock().expect("peaks lock");
    let heads = db.head_state().await;

    eprintln!(
        "multi_agent_profile: agents={} ops_per_agent={} attempted={} accepted={} rejected={} unauthorized_denied={} attempted_tps={} accepted_tps={} avg_lat_us={} max_lat_us={} peak_queue_depth={} peak_inflight={} peak_durable_lag={} peak_conflict_rate={:.4} peak_durable_wait_ops={} peak_durable_wait_avg_us={} peak_wal_append_ops={} peak_wal_append_bytes={} peak_wal_append_avg_us={} peak_wal_sync_ops={} peak_wal_sync_avg_us={} peak_commit_errors={} peak_permission_rejections={} peak_validation_rejections={} peak_queue_full_rejections={} peak_timeout_rejections={} peak_read_set_conflicts={} heads(v={},d={})",
        agents,
        ops_per_agent,
        merged.attempted,
        merged.accepted,
        merged.rejected,
        merged.unauthorized_denied,
        attempted_tps,
        accepted_tps,
        avg_lat_us,
        merged.latency_max_us,
        peaks.queue_depth,
        peaks.inflight,
        peaks.durable_lag,
        peaks.conflict_rate,
        peaks.durable_wait_ops,
        peaks.durable_wait_avg_us,
        peaks.wal_append_ops,
        peaks.wal_append_bytes,
        peaks.wal_append_avg_us,
        peaks.wal_sync_ops,
        peaks.wal_sync_avg_us,
        peaks.commit_errors,
        peaks.permission_rejections,
        peaks.validation_rejections,
        peaks.queue_full_rejections,
        peaks.timeout_rejections,
        peaks.read_set_conflicts,
        heads.visible_head_seq,
        heads.durable_head_seq
    );

    assert_eq!(
        merged.accepted + merged.rejected,
        merged.attempted,
        "profile run must not drop operations"
    );
    assert_eq!(
        merged.unauthorized_denied, merged.unauthorized_attempted,
        "cross-agent unauthorized writes must always be denied"
    );
    assert!(
        heads.visible_head_seq >= heads.durable_head_seq,
        "visible head should be >= durable head"
    );
    assert!(
        attempted_tps >= 200,
        "profile indicates severe throughput regression: attempted_tps={attempted_tps}"
    );
    assert!(
        peaks.commit_errors <= (merged.rejected + merged.unauthorized_denied) as u64,
        "unexpected commit errors exceed rejected + unauthorized-denied operations"
    );
    assert_eq!(
        peaks.permission_rejections, merged.unauthorized_denied as u64,
        "permission rejection accounting must match denied unauthorized attempts"
    );
    assert_eq!(
        peaks.validation_rejections, merged.rejected as u64,
        "validation rejection accounting must match application-level rejected operations"
    );
    assert_eq!(
        peaks.queue_full_rejections, 0,
        "unexpected queue-full rejections under baseline profile load"
    );
    assert_eq!(
        peaks.timeout_rejections, 0,
        "unexpected timeout rejections under baseline profile load"
    );
    assert_eq!(
        peaks.durable_wait_ops, 0,
        "baseline non-durable profile should not accumulate durable wait operations"
    );
    assert!(
        peaks.wal_sync_ops > 0,
        "full-durability secure profile should execute WAL sync operations"
    );

    for agent in &agent_ids {
        let caller = CallerContext::new(agent.clone());
        let entry = db
            .kv_get(
                "p",
                "app",
                format!("agent:{agent}:balance").as_bytes(),
                ConsistencyMode::AtLatest,
                &caller,
            )
            .await
            .expect("read own balance")
            .expect("balance exists");
        let mut bal = [0u8; 32];
        bal.copy_from_slice(&entry.value);
        assert!(
            decode_u256_u64(bal) as usize <= ops_per_agent,
            "agent balance must not exceed own attempts"
        );
    }
}

#[tokio::test]
async fn project_and_scope_admin_grants_allow_table_writes() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    db.create_scope("p", "app").await.expect("scope");
    create_table(
        &db,
        "p",
        "app",
        "items",
        vec![ColumnDef {
            name: "id".into(),
            col_type: ColumnType::Integer,
            nullable: false,
        }],
        vec!["id"],
    )
    .await;

    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        caller_id: "project-admin".into(),
        permission: Permission::ProjectAdmin {
            project_id: "p".into(),
        },
        actor_id: None,
        delegable: false,
    }))
    .await
    .expect("grant project admin");
    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        caller_id: "scope-admin".into(),
        permission: Permission::ScopeAdmin {
            project_id: "p".into(),
            scope_id: "app".into(),
        },
        actor_id: None,
        delegable: false,
    }))
    .await
    .expect("grant scope admin");

    db.commit_as(
        CallerContext::new("project-admin"),
        Mutation::Insert {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "items".into(),
            primary_key: vec![Value::Integer(1)],
            row: Row::from_values(vec![Value::Integer(1)]),
        },
    )
    .await
    .expect("project admin insert");
    db.commit_as(
        CallerContext::new("scope-admin"),
        Mutation::Insert {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "items".into(),
            primary_key: vec![Value::Integer(2)],
            row: Row::from_values(vec![Value::Integer(2)]),
        },
    )
    .await
    .expect("scope admin insert");
}

#[tokio::test]
async fn delete_where_respects_row_level_read_policy() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    db.create_scope("p", "app").await.expect("scope");
    create_table(
        &db,
        "p",
        "app",
        "items",
        vec![
            ColumnDef {
                name: "id".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "owner".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
        ],
        vec!["id"],
    )
    .await;

    for (id, owner) in [(1_i64, "alice"), (2_i64, "bob")] {
        db.commit(Mutation::Insert {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "items".into(),
            primary_key: vec![Value::Integer(id)],
            row: Row::from_values(vec![Value::Integer(id), Value::Text(owner.into())]),
        })
        .await
        .expect("seed item");
    }

    db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
        caller_id: "alice".into(),
        permission: Permission::TableWrite {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "items".into(),
        },
        actor_id: None,
        delegable: false,
    }))
    .await
    .expect("grant table write");
    db.set_read_policy(
        "p",
        "app",
        "items",
        Expr::Eq("owner".into(), Value::Text("$caller_id".into())),
    )
    .await
    .expect("set policy");

    db.commit_as(
        CallerContext::new("alice"),
        Mutation::DeleteWhere {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "items".into(),
            predicate: Expr::IsNotNull("id".into()),
            limit: None,
        },
    )
    .await
    .expect("delete visible rows only");

    let remaining = db
        .query_no_auth(
            "p",
            "app",
            Query::select(&["id", "owner"]).from("items").limit(10),
            QueryOptions::default(),
        )
        .await
        .expect("query remaining rows");
    assert_eq!(remaining.rows.len(), 1, "one row should remain");
    assert_eq!(remaining.rows[0].values[0], Value::Integer(2));
    assert_eq!(remaining.rows[0].values[1], Value::Text("bob".into()));
}

#[tokio::test]
async fn idempotency_keys_are_scoped_to_caller() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");

    for caller_id in ["alice", "bob"] {
        db.commit(Mutation::Ddl(DdlOperation::GrantPermission {
            caller_id: caller_id.into(),
            permission: Permission::KvWrite {
                project_id: "p".into(),
                scope_id: Some("app".into()),
                prefix: None,
            },
            actor_id: None,
            delegable: false,
        }))
        .await
        .expect("grant kv write");
    }

    let key = IdempotencyKey([7u8; 16]);
    let alice = db
        .commit_envelope(TransactionEnvelope {
            caller: Some(CallerContext::new("alice")),
            idempotency_key: Some(key.clone()),
            write_class: WriteClass::Standard,
            assertions: Vec::new(),
            read_set: Default::default(),
            write_intent: WriteIntent {
                mutations: vec![Mutation::KvSet {
                    project_id: "p".into(),
                    scope_id: "app".into(),
                    key: b"alice".to_vec(),
                    value: b"1".to_vec(),
                }],
            },
            base_seq: 0,
        })
        .await
        .expect("alice commit");
    let bob = db
        .commit_envelope(TransactionEnvelope {
            caller: Some(CallerContext::new("bob")),
            idempotency_key: Some(key),
            write_class: WriteClass::Standard,
            assertions: Vec::new(),
            read_set: Default::default(),
            write_intent: WriteIntent {
                mutations: vec![Mutation::KvSet {
                    project_id: "p".into(),
                    scope_id: "app".into(),
                    key: b"bob".to_vec(),
                    value: b"1".to_vec(),
                }],
            },
            base_seq: 0,
        })
        .await
        .expect("bob commit");

    assert!(matches!(
        alice.idempotency,
        crate::commit::executor::IdempotencyOutcome::Applied
    ));
    assert!(matches!(
        bob.idempotency,
        crate::commit::executor::IdempotencyOutcome::Applied
    ));
    assert_ne!(alice.commit_seq, bob.commit_seq);
}

#[derive(Clone, Copy)]
enum AuthzMatrixCase {
    TableRead,
    TableWrite,
    KvRead,
    KvWrite,
    AsyncIndexRead,
    EventStreamRead,
    ProcessorLagRead,
}

impl AuthzMatrixCase {
    fn name(self) -> &'static str {
        match self {
            Self::TableRead => "table read",
            Self::TableWrite => "table write",
            Self::KvRead => "kv read",
            Self::KvWrite => "kv write",
            Self::AsyncIndexRead => "async index read",
            Self::EventStreamRead => "event stream read",
            Self::ProcessorLagRead => "processor lag read",
        }
    }

    fn direct_permissions(self) -> Vec<Permission> {
        match self {
            Self::TableRead => vec![Permission::TableRead {
                project_id: "p".into(),
                scope_id: "s1".into(),
                table_name: "events".into(),
            }],
            Self::TableWrite => vec![Permission::TableWrite {
                project_id: "p".into(),
                scope_id: "s1".into(),
                table_name: "events".into(),
            }],
            Self::KvRead => vec![Permission::KvRead {
                project_id: "p".into(),
                scope_id: Some("s1".into()),
                prefix: Some(b"acct:".to_vec()),
            }],
            Self::KvWrite => vec![Permission::KvWrite {
                project_id: "p".into(),
                scope_id: Some("s1".into()),
                prefix: Some(b"acct:".to_vec()),
            }],
            Self::AsyncIndexRead => vec![
                Permission::TableRead {
                    project_id: "p".into(),
                    scope_id: "s1".into(),
                    table_name: "events".into(),
                },
                Permission::IndexRead {
                    project_id: "p".into(),
                    scope_id: "s1".into(),
                    table_name: "events".into(),
                    index_name: "events_view".into(),
                },
            ],
            Self::EventStreamRead => vec![Permission::TableRead {
                project_id: crate::catalog::SYSTEM_PROJECT_ID.into(),
                scope_id: crate::SYSTEM_SCOPE_ID.into(),
                table_name: crate::EVENT_OUTBOX_TABLE.into(),
            }],
            Self::ProcessorLagRead => vec![Permission::TableRead {
                project_id: crate::catalog::SYSTEM_PROJECT_ID.into(),
                scope_id: crate::SYSTEM_SCOPE_ID.into(),
                table_name: crate::REACTIVE_PROCESSOR_CHECKPOINTS_TABLE.into(),
            }],
        }
    }
}

async fn setup_authz_matrix_db(db: &AedbInstance) {
    db.commit_as(
        CallerContext::system_internal(),
        Mutation::Ddl(DdlOperation::CreateProject {
            project_id: "p".into(),
            owner_id: Some("project_owner".into()),
            if_not_exists: true,
        }),
    )
    .await
    .expect("project");
    db.commit_as(
        CallerContext::system_internal(),
        Mutation::Ddl(DdlOperation::CreateScope {
            project_id: "p".into(),
            scope_id: "s1".into(),
            owner_id: Some("scope_owner".into()),
            if_not_exists: true,
        }),
    )
    .await
    .expect("scope");
    db.commit_as(
        CallerContext::system_internal(),
        Mutation::Ddl(DdlOperation::CreateTable {
            project_id: "p".into(),
            scope_id: "s1".into(),
            table_name: "events".into(),
            owner_id: Some("table_owner".into()),
            if_not_exists: false,
            columns: vec![
                ColumnDef {
                    name: "id".into(),
                    col_type: ColumnType::Integer,
                    nullable: false,
                },
                ColumnDef {
                    name: "amount".into(),
                    col_type: ColumnType::Integer,
                    nullable: false,
                },
            ],
            primary_key: vec!["id".into()],
        }),
    )
    .await
    .expect("table");
    for mutation in [
        Mutation::Ddl(DdlOperation::CreateAsyncIndex {
            project_id: "p".into(),
            scope_id: "s1".into(),
            table_name: "events".into(),
            index_name: "events_view".into(),
            if_not_exists: false,
            projected_columns: vec!["id".into(), "amount".into()],
        }),
        Mutation::Upsert {
            project_id: "p".into(),
            scope_id: "s1".into(),
            table_name: "events".into(),
            primary_key: vec![Value::Integer(1)],
            row: Row::from_values(vec![Value::Integer(1), Value::Integer(10)]),
        },
        Mutation::KvSet {
            project_id: "p".into(),
            scope_id: "s1".into(),
            key: b"acct:1".to_vec(),
            value: b"10".to_vec(),
        },
        Mutation::EmitEvent {
            project_id: "p".into(),
            scope_id: "s1".into(),
            topic: "payments".into(),
            event_key: "evt-1".into(),
            payload_json: "{\"amount\":10}".into(),
        },
    ] {
        db.commit_as(CallerContext::system_internal(), mutation)
            .await
            .expect("seed matrix mutation");
    }
}

async fn grant_matrix_permissions(
    db: &AedbInstance,
    caller_id: &str,
    permissions: Vec<Permission>,
) {
    for permission in permissions {
        db.commit_as(
            CallerContext::system_internal(),
            Mutation::Ddl(DdlOperation::GrantPermission {
                caller_id: caller_id.into(),
                permission,
                actor_id: Some(SYSTEM_CALLER_ID.into()),
                delegable: false,
            }),
        )
        .await
        .expect("grant matrix permission");
    }
}

async fn revoke_matrix_permissions(
    db: &AedbInstance,
    caller_id: &str,
    permissions: Vec<Permission>,
) {
    for permission in permissions {
        db.commit_as(
            CallerContext::system_internal(),
            Mutation::Ddl(DdlOperation::RevokePermission {
                caller_id: caller_id.into(),
                permission,
                actor_id: Some(SYSTEM_CALLER_ID.into()),
            }),
        )
        .await
        .expect("revoke matrix permission");
    }
}

async fn assert_matrix_access(db: &AedbInstance, case: AuthzMatrixCase, caller_id: &str) {
    let caller = CallerContext::new(caller_id);
    match case {
        AuthzMatrixCase::TableRead => {
            db.query_with_options_as(
                Some(&caller),
                "p",
                "s1",
                Query::select(&["id"]).from("events").limit(1),
                QueryOptions::default(),
            )
            .await
            .unwrap_or_else(|err| panic!("{} should allow {caller_id}: {err:?}", case.name()));
        }
        AuthzMatrixCase::TableWrite => {
            db.commit_as(
                caller,
                Mutation::Upsert {
                    project_id: "p".into(),
                    scope_id: "s1".into(),
                    table_name: "events".into(),
                    primary_key: vec![Value::Integer(20)],
                    row: Row::from_values(vec![Value::Integer(20), Value::Integer(20)]),
                },
            )
            .await
            .unwrap_or_else(|err| panic!("{} should allow {caller_id}: {err:?}", case.name()));
        }
        AuthzMatrixCase::KvRead => {
            db.kv_get("p", "s1", b"acct:1", ConsistencyMode::AtLatest, &caller)
                .await
                .unwrap_or_else(|err| panic!("{} should allow {caller_id}: {err:?}", case.name()));
        }
        AuthzMatrixCase::KvWrite => {
            db.commit_as(
                caller,
                Mutation::KvSet {
                    project_id: "p".into(),
                    scope_id: "s1".into(),
                    key: b"acct:2".to_vec(),
                    value: b"20".to_vec(),
                },
            )
            .await
            .unwrap_or_else(|err| panic!("{} should allow {caller_id}: {err:?}", case.name()));
        }
        AuthzMatrixCase::AsyncIndexRead => {
            db.query_with_options_as(
                Some(&caller),
                "p",
                "s1",
                Query::select(&["id"]).from("events").limit(1),
                QueryOptions {
                    async_index: Some("events_view".into()),
                    allow_full_scan: true,
                    ..QueryOptions::default()
                },
            )
            .await
            .unwrap_or_else(|err| panic!("{} should allow {caller_id}: {err:?}", case.name()));
        }
        AuthzMatrixCase::EventStreamRead => {
            db.read_event_stream_as(&caller, None, 0, 10, ConsistencyMode::AtLatest)
                .await
                .unwrap_or_else(|err| panic!("{} should allow {caller_id}: {err:?}", case.name()));
        }
        AuthzMatrixCase::ProcessorLagRead => {
            db.reactive_processor_lag_as(&caller, "matrix_processor", ConsistencyMode::AtLatest)
                .await
                .unwrap_or_else(|err| panic!("{} should allow {caller_id}: {err:?}", case.name()));
        }
    }
}

async fn assert_matrix_denied(db: &AedbInstance, case: AuthzMatrixCase, caller_id: &str) {
    let caller = CallerContext::new(caller_id);
    let denied = match case {
        AuthzMatrixCase::TableRead => db
            .query_with_options_as(
                Some(&caller),
                "p",
                "s1",
                Query::select(&["id"]).from("events").limit(1),
                QueryOptions::default(),
            )
            .await
            .is_err(),
        AuthzMatrixCase::TableWrite => db
            .commit_as(
                caller,
                Mutation::Upsert {
                    project_id: "p".into(),
                    scope_id: "s1".into(),
                    table_name: "events".into(),
                    primary_key: vec![Value::Integer(99)],
                    row: Row::from_values(vec![Value::Integer(99), Value::Integer(99)]),
                },
            )
            .await
            .is_err(),
        AuthzMatrixCase::KvRead => db
            .kv_get("p", "s1", b"acct:1", ConsistencyMode::AtLatest, &caller)
            .await
            .is_err(),
        AuthzMatrixCase::KvWrite => db
            .commit_as(
                caller,
                Mutation::KvSet {
                    project_id: "p".into(),
                    scope_id: "s1".into(),
                    key: b"acct:denied".to_vec(),
                    value: b"x".to_vec(),
                },
            )
            .await
            .is_err(),
        AuthzMatrixCase::AsyncIndexRead => db
            .query_with_options_as(
                Some(&caller),
                "p",
                "s1",
                Query::select(&["id"]).from("events").limit(1),
                QueryOptions {
                    async_index: Some("events_view".into()),
                    allow_full_scan: true,
                    ..QueryOptions::default()
                },
            )
            .await
            .is_err(),
        AuthzMatrixCase::EventStreamRead => db
            .read_event_stream_as(&caller, None, 0, 10, ConsistencyMode::AtLatest)
            .await
            .is_err(),
        AuthzMatrixCase::ProcessorLagRead => db
            .reactive_processor_lag_as(&caller, "matrix_processor", ConsistencyMode::AtLatest)
            .await
            .is_err(),
    };
    assert!(denied, "{} should deny {caller_id}", case.name());
}

#[tokio::test]
async fn authorization_matrix_helper_covers_core_resource_access() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_production(AedbConfig::production([14u8; 32]), dir.path())
        .expect("open");
    setup_authz_matrix_db(&db).await;

    db.commit_as(
        CallerContext::system_internal(),
        Mutation::Ddl(DdlOperation::GrantPermission {
            caller_id: "global_admin".into(),
            permission: Permission::GlobalAdmin,
            actor_id: Some(SYSTEM_CALLER_ID.into()),
            delegable: false,
        }),
    )
    .await
    .expect("grant global admin");
    grant_matrix_permissions(
        &db,
        "project_admin",
        vec![Permission::ProjectAdmin {
            project_id: "p".into(),
        }],
    )
    .await;
    grant_matrix_permissions(
        &db,
        "scope_admin",
        vec![Permission::ScopeAdmin {
            project_id: "p".into(),
            scope_id: "s1".into(),
        }],
    )
    .await;

    let project_cases = [
        AuthzMatrixCase::TableRead,
        AuthzMatrixCase::TableWrite,
        AuthzMatrixCase::KvRead,
        AuthzMatrixCase::KvWrite,
        AuthzMatrixCase::AsyncIndexRead,
    ];
    for case in project_cases {
        grant_matrix_permissions(&db, "direct", case.direct_permissions()).await;
        assert_matrix_access(&db, case, "direct").await;
        revoke_matrix_permissions(&db, "direct", case.direct_permissions()).await;
        assert_matrix_denied(&db, case, "direct").await;

        assert_matrix_access(&db, case, "project_admin").await;
        assert_matrix_access(&db, case, "scope_admin").await;
        assert_matrix_access(&db, case, "global_admin").await;
    }

    db.commit_as(
        CallerContext::new("table_owner"),
        Mutation::Ddl(DdlOperation::GrantPermission {
            caller_id: "owned_table_reader".into(),
            permission: Permission::TableRead {
                project_id: "p".into(),
                scope_id: "s1".into(),
                table_name: "events".into(),
            },
            actor_id: None,
            delegable: false,
        }),
    )
    .await
    .expect("table owner can grant table read");
    assert_matrix_access(&db, AuthzMatrixCase::TableRead, "owned_table_reader").await;

    for case in [
        AuthzMatrixCase::EventStreamRead,
        AuthzMatrixCase::ProcessorLagRead,
    ] {
        grant_matrix_permissions(&db, "direct_system_reader", case.direct_permissions()).await;
        assert_matrix_access(&db, case, "direct_system_reader").await;
        revoke_matrix_permissions(&db, "direct_system_reader", case.direct_permissions()).await;
        assert_matrix_denied(&db, case, "direct_system_reader").await;
        assert_matrix_access(&db, case, "global_admin").await;
    }

    assert_matrix_denied(&db, AuthzMatrixCase::TableRead, "missing").await;
    assert_matrix_denied(&db, AuthzMatrixCase::KvRead, "missing").await;
}

#[tokio::test]
async fn secure_mode_rejects_all_public_no_auth_read_escape_hatches() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_production(AedbConfig::production([16u8; 32]), dir.path())
        .expect("open secure");

    let cases = [
        db.query_no_auth(
            "secret_project",
            "app",
            Query::select(&["*"]).from("secret_table").limit(1),
            QueryOptions::default(),
        )
        .await
        .err()
        .map(|err| err.to_string())
        .expect("query should deny"),
        db.query_no_auth(
            "secret_project",
            "app",
            Query::select(&["*"]).from("secret_table").limit(1),
            QueryOptions::default(),
        )
        .await
        .err()
        .map(|err| err.to_string())
        .expect("query_with_options should deny"),
        db.query_with_options_capturing_as(
            None,
            "secret_project",
            "app",
            Query::select(&["*"]).from("secret_table").limit(1),
            QueryOptions::default(),
        )
        .await
        .err()
        .map(|err| err.to_string())
        .expect("query_with_read_set should deny"),
        db.query_no_auth(
            "secret_project",
            "app",
            Query::select(&["*"]).from("secret_table").limit(1),
            QueryOptions::default(),
        )
        .await
        .err()
        .map(|err| err.to_string())
        .expect("query_no_auth should deny"),
        db.kv_get_no_auth(
            "secret_project",
            "app",
            b"secret-key",
            ConsistencyMode::AtLatest,
        )
        .await
        .err()
        .map(|err| err.to_string())
        .expect("kv_get_no_auth should deny"),
        db.kv_get_many_no_auth(
            "secret_project",
            "app",
            &[b"secret-key".to_vec()],
            ConsistencyMode::AtLatest,
        )
        .await
        .err()
        .map(|err| err.to_string())
        .expect("kv_get_many_no_auth should deny"),
        db.kv_scan_prefix_no_auth(
            "secret_project",
            "app",
            b"secret-prefix",
            10,
            ConsistencyMode::AtLatest,
        )
        .await
        .err()
        .map(|err| err.to_string())
        .expect("kv_scan_prefix_no_auth should deny"),
    ];

    for rendered in cases {
        assert!(rendered.contains("permission denied"));
        assert!(!rendered.contains("secret_project"));
        assert!(!rendered.contains("secret_table"));
        assert!(!rendered.contains("secret-key"));
        assert!(!rendered.contains("secret-prefix"));
    }

    assert!(matches!(
        db.read_event_stream(None, 0, 1, ConsistencyMode::AtLatest)
            .await
            .expect_err("anonymous event stream should deny"),
        AedbError::PermissionDenied(_)
    ));
    assert!(matches!(
        db.reactive_processor_lag("secret_processor", ConsistencyMode::AtLatest)
            .await
            .expect_err("anonymous processor lag should deny"),
        AedbError::PermissionDenied(_)
    ));
    let backup_dir = tempdir().expect("backup temp");
    assert!(matches!(
        db.backup_full(backup_dir.path())
            .await
            .expect_err("anonymous backup should deny"),
        AedbError::PermissionDenied(_)
    ));
    let incremental_backup_dir = tempdir().expect("incremental backup temp");
    let parent_backup_dir = tempdir().expect("parent backup temp");
    assert!(matches!(
        db.backup_incremental(incremental_backup_dir.path(), parent_backup_dir.path())
            .await
            .expect_err("anonymous incremental backup should deny"),
        AedbError::PermissionDenied(_)
    ));
    let backup_file_dir = tempdir().expect("backup file temp");
    let backup_file = backup_file_dir.path().join("backup.aedbarc");
    assert!(matches!(
        db.backup_full_to_file(&backup_file)
            .await
            .expect_err("anonymous backup file should deny"),
        AedbError::PermissionDenied(_)
    ));
    assert!(matches!(
        db.apply_migration(crate::migration::Migration {
            version: 1,
            name: "secret migration".into(),
            project_id: "secret_project".into(),
            scope_id: "app".into(),
            mutations: Vec::new(),
            down_mutations: None,
        })
        .await
        .expect_err("anonymous migration should deny"),
        AedbError::PermissionDenied(_)
    ));
}

#[test]
fn public_no_auth_and_unchecked_surface_is_intentional() {
    let root = Path::new(env!("CARGO_MANIFEST_DIR"));
    let mut exposed = Vec::new();
    for rel in [
        "src/lib.rs",
        "src/api/kv_api.rs",
        "src/api/query_api.rs",
        "src/sync_bridge.rs",
    ] {
        let source = fs::read_to_string(root.join(rel)).expect("read source");
        for line in source.lines() {
            let trimmed = line.trim();
            if trimmed.starts_with("pub ")
                && (trimmed.contains("_no_auth") || trimmed.contains("unchecked"))
            {
                // Record only up to the opening paren so rustfmt reflowing the
                // argument list (one line vs. several) does not change the
                // recorded surface; the method name — the security-relevant
                // part — always precedes `(`.
                let sig = match trimmed.find('(') {
                    Some(i) => &trimmed[..=i],
                    None => trimmed,
                };
                exposed.push(format!("{rel}:{sig}"));
            }
        }
    }
    exposed.sort();
    assert_eq!(
        exposed,
        vec![
            "src/api/kv_api.rs:pub async fn kv_del_no_auth(".to_string(),
            "src/api/kv_api.rs:pub async fn kv_get_many_no_auth(".to_string(),
            "src/api/kv_api.rs:pub async fn kv_get_no_auth(".to_string(),
            "src/api/kv_api.rs:pub async fn kv_scan_prefix_no_auth(".to_string(),
            "src/api/kv_api.rs:pub async fn kv_set_no_auth(".to_string(),
            "src/api/query_api.rs:pub async fn query_no_auth(".to_string(),
            "src/lib.rs:pub async fn commit_no_auth(".to_string(),
            "src/sync_bridge.rs:pub fn query_no_auth(".to_string(),
        ],
        "public no_auth/unchecked API surface changed; update secure-mode tests and docs intentionally"
    );
}

#[tokio::test]
async fn permission_denied_message_snapshots_do_not_disclose_protected_data() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_production(AedbConfig::production([17u8; 32]), dir.path())
        .expect("open");
    setup_secure_assertion_denial_fixture(&db).await;

    let commit_denied = db
        .commit_as(
            CallerContext::new("mallory"),
            Mutation::Upsert {
                project_id: "p".into(),
                scope_id: "app".into(),
                table_name: "secret_rows".into(),
                primary_key: vec![Value::Integer(777)],
                row: Row::from_values(vec![Value::Integer(777), Value::Integer(42)]),
            },
        )
        .await
        .expect_err("table write should deny")
        .to_string();
    let query_denied = db
        .query_with_options_as(
            Some(&CallerContext::new("mallory")),
            "p",
            "app",
            Query::select(&["amount"])
                .from("secret_rows")
                .where_(Expr::Eq("amount".into(), Value::Integer(42))),
            QueryOptions::default(),
        )
        .await
        .expect_err("query should deny")
        .to_string();
    let kv_denied = db
        .kv_get(
            "p",
            "app",
            b"secret:key",
            ConsistencyMode::AtLatest,
            &CallerContext::new("mallory"),
        )
        .await
        .expect_err("kv should deny")
        .to_string();
    let event_denied = db
        .read_event_stream_as(
            &CallerContext::new("mallory"),
            None,
            0,
            1,
            ConsistencyMode::AtLatest,
        )
        .await
        .expect_err("event stream should deny")
        .to_string();
    let processor_denied = db
        .reactive_processor_lag_as(
            &CallerContext::new("mallory"),
            "secret_processor",
            ConsistencyMode::AtLatest,
        )
        .await
        .expect_err("processor lag should deny")
        .to_string();

    let snapshots = [
        (
            "commit",
            commit_denied.as_str(),
            "permission denied: permission denied",
        ),
        (
            "query",
            query_denied.as_str(),
            "permission denied: permission denied",
        ),
        (
            "kv",
            kv_denied.as_str(),
            "permission denied: permission denied",
        ),
        (
            "event",
            event_denied.as_str(),
            "permission denied: permission denied",
        ),
        (
            "processor",
            processor_denied.as_str(),
            "permission denied: permission denied",
        ),
    ];
    for (label, rendered, expected) in snapshots {
        assert_eq!(rendered, expected, "{label} denial snapshot changed");
        for forbidden in [
            "secret_rows",
            "secret:key",
            "secret_processor",
            "amount",
            "777",
            "42",
            "mallory",
        ] {
            assert!(
                !rendered.contains(forbidden),
                "{label} denial leaked {forbidden}: {rendered}"
            );
        }
    }
}

async fn setup_secure_assertion_denial_fixture(db: &AedbInstance) {
    let system = CallerContext::system_internal();
    db.commit_as(
        system.clone(),
        Mutation::Ddl(DdlOperation::CreateProject {
            owner_id: None,
            if_not_exists: true,
            project_id: "p".into(),
        }),
    )
    .await
    .expect("create project");
    db.commit_as(
        system.clone(),
        Mutation::Ddl(DdlOperation::CreateTable {
            owner_id: None,
            if_not_exists: false,
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "secret_rows".into(),
            columns: vec![
                ColumnDef {
                    name: "id".into(),
                    col_type: ColumnType::Integer,
                    nullable: false,
                },
                ColumnDef {
                    name: "amount".into(),
                    col_type: ColumnType::Integer,
                    nullable: false,
                },
            ],
            primary_key: vec!["id".into()],
        }),
    )
    .await
    .expect("create secret table");
    for mutation in [
        Mutation::KvSet {
            project_id: "p".into(),
            scope_id: "app".into(),
            key: b"secret:key".to_vec(),
            value: b"secret-value".to_vec(),
        },
        Mutation::Upsert {
            project_id: "p".into(),
            scope_id: "app".into(),
            table_name: "secret_rows".into(),
            primary_key: vec![Value::Integer(1)],
            row: Row::from_values(vec![Value::Integer(1), Value::Integer(42)]),
        },
        Mutation::Ddl(DdlOperation::GrantPermission {
            caller_id: "writer".into(),
            permission: Permission::KvWrite {
                project_id: "p".into(),
                scope_id: Some("app".into()),
                prefix: Some(b"public:".to_vec()),
            },
            actor_id: Some(SYSTEM_CALLER_ID.into()),
            delegable: false,
        }),
    ] {
        db.commit_as(system.clone(), mutation)
            .await
            .expect("seed assertion denial fixture");
    }
}

fn assertion_denial_cases() -> Vec<(&'static str, ReadAssertion)> {
    vec![
        (
            "key equals",
            ReadAssertion::KeyEquals {
                project_id: "p".into(),
                scope_id: "app".into(),
                key: b"secret:key".to_vec(),
                expected: b"wrong".to_vec(),
            },
        ),
        (
            "key compare",
            ReadAssertion::KeyCompare {
                project_id: "p".into(),
                scope_id: "app".into(),
                key: b"secret:key".to_vec(),
                op: CompareOp::Eq,
                threshold: b"wrong".to_vec(),
            },
        ),
        (
            "key exists",
            ReadAssertion::KeyExists {
                project_id: "p".into(),
                scope_id: "app".into(),
                key: b"secret:key".to_vec(),
                expected: false,
            },
        ),
        (
            "key version",
            ReadAssertion::KeyVersion {
                project_id: "p".into(),
                scope_id: "app".into(),
                key: b"secret:key".to_vec(),
                expected_seq: 999,
            },
        ),
        (
            "row version",
            ReadAssertion::RowVersion {
                project_id: "p".into(),
                scope_id: "app".into(),
                table_name: "secret_rows".into(),
                primary_key: vec![Value::Integer(1)],
                expected_seq: 999,
            },
        ),
        (
            "row exists",
            ReadAssertion::RowExists {
                project_id: "p".into(),
                scope_id: "app".into(),
                table_name: "secret_rows".into(),
                primary_key: vec![Value::Integer(1)],
                expected: false,
            },
        ),
        (
            "row column compare",
            ReadAssertion::RowColumnCompare {
                project_id: "p".into(),
                scope_id: "app".into(),
                table_name: "secret_rows".into(),
                primary_key: vec![Value::Integer(1)],
                column: "amount".into(),
                op: CompareOp::Eq,
                threshold: Value::Integer(0),
            },
        ),
        (
            "count compare",
            ReadAssertion::CountCompare {
                project_id: "p".into(),
                scope_id: "app".into(),
                table_name: "secret_rows".into(),
                filter: Some(Expr::Eq("amount".into(), Value::Integer(42))),
                op: CompareOp::Eq,
                threshold: 0,
            },
        ),
        (
            "sum compare",
            ReadAssertion::SumCompare {
                project_id: "p".into(),
                scope_id: "app".into(),
                table_name: "secret_rows".into(),
                column: "amount".into(),
                filter: Some(Expr::Eq("id".into(), Value::Integer(1))),
                op: CompareOp::Eq,
                threshold: Value::Integer(0),
            },
        ),
        (
            "nested all",
            ReadAssertion::All(vec![ReadAssertion::KeyExists {
                project_id: "p".into(),
                scope_id: "app".into(),
                key: b"secret:key".to_vec(),
                expected: false,
            }]),
        ),
        (
            "nested any",
            ReadAssertion::Any(vec![ReadAssertion::RowExists {
                project_id: "p".into(),
                scope_id: "app".into(),
                table_name: "secret_rows".into(),
                primary_key: vec![Value::Integer(1)],
                expected: false,
            }]),
        ),
        (
            "nested not",
            ReadAssertion::Not(Box::new(ReadAssertion::KeyEquals {
                project_id: "p".into(),
                scope_id: "app".into(),
                key: b"secret:key".to_vec(),
                expected: b"secret-value".to_vec(),
            })),
        ),
    ]
}

#[tokio::test]
async fn secure_mode_denies_every_read_assertion_variant_before_evaluation() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_production(AedbConfig::production([15u8; 32]), dir.path())
        .expect("open");
    setup_secure_assertion_denial_fixture(&db).await;

    for (label, assertion) in assertion_denial_cases() {
        let result = db
            .commit_envelope(TransactionEnvelope {
                caller: Some(CallerContext::new("writer")),
                idempotency_key: None,
                write_class: WriteClass::Standard,
                assertions: vec![assertion],
                read_set: ReadSet::default(),
                write_intent: WriteIntent {
                    mutations: vec![Mutation::KvSet {
                        project_id: "p".into(),
                        scope_id: "app".into(),
                        key: format!("public:{label}").into_bytes(),
                        value: b"ok".to_vec(),
                    }],
                },
                base_seq: 0,
            })
            .await;
        let err = match result {
            Ok(ok) => panic!(
                "{label} assertion unexpectedly committed at seq {}",
                ok.commit_seq
            ),
            Err(err) => err,
        };
        assert!(
            matches!(err, AedbError::PermissionDenied(_)),
            "{label} should be denied before assertion evaluation, got {err:?}"
        );
        let rendered = err.to_string();
        for forbidden in [
            "secret_rows",
            "secret:key",
            "secret-value",
            "amount",
            "42",
            "public:",
        ] {
            assert!(
                !rendered.contains(forbidden),
                "{label} permission denial leaked {forbidden}: {rendered}"
            );
        }
    }
}

#[tokio::test]
async fn open_default_requires_authenticated_caller_for_writes() {
    let dir = tempdir().expect("temp");
    // `open` is secure-by-default: anonymous writes/reads must be rejected.
    let db = AedbInstance::open(AedbConfig::default(), dir.path()).expect("open");

    let set = db.kv_set("p", "app", b"k".to_vec(), b"v".to_vec()).await;
    assert!(
        matches!(set, Err(AedbError::PermissionDenied(_))),
        "anonymous kv_set must be denied under secure-default open, got {set:?}"
    );

    let commit = db
        .commit(Mutation::KvSet {
            project_id: "p".into(),
            scope_id: "app".into(),
            key: b"k".to_vec(),
            value: b"v".to_vec(),
        })
        .await;
    assert!(
        matches!(commit, Err(AedbError::PermissionDenied(_))),
        "anonymous commit must be denied under secure-default open, got {commit:?}"
    );

    // The explicit anonymous twins are also unavailable in secure mode.
    let no_auth = db
        .kv_set_no_auth("p", "app", b"k".to_vec(), b"v".to_vec())
        .await;
    assert!(
        matches!(no_auth, Err(AedbError::PermissionDenied(_))),
        "kv_set_no_auth must be denied under secure-default open, got {no_auth:?}"
    );
}

#[tokio::test]
async fn open_anonymous_allows_no_auth_write_twins() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");

    db.kv_set_no_auth("p", "app", b"k".to_vec(), b"v".to_vec())
        .await
        .expect("anonymous kv_set_no_auth should succeed");

    let got = db
        .kv_get_no_auth("p", "app", b"k", ConsistencyMode::AtLatest)
        .await
        .expect("kv_get_no_auth")
        .expect("value present");
    assert_eq!(got.value, b"v".to_vec());

    db.kv_del_no_auth("p", "app", b"k".to_vec())
        .await
        .expect("anonymous kv_del_no_auth should succeed");
    assert!(
        db.kv_get_no_auth("p", "app", b"k", ConsistencyMode::AtLatest)
            .await
            .expect("kv_get_no_auth")
            .is_none(),
        "key should be deleted after kv_del_no_auth"
    );
}
