anda_brain 0.13.3

use super::*;
use anda_cognitive_nexus::attention::{AttentionConfig, RuntimePin};
use object_store::{PutMode, memory::InMemory};
use std::sync::Arc;

fn config(id: &str) -> AttentionConfig {
    AttentionConfig {
        scope: RuntimeScope {
            space_id: "kip:space:default".into(),
            space_instance: id.into(),
        },
        pins: RuntimePins {
            policy: RuntimePin {
                id: "test".into(),
                digest: format!("sha256:{}", "a".repeat(64)),
            },
            evaluator: None,
            binding: None,
        },
    }
}

#[tokio::test]
async fn directory_cas_does_not_clear_new_dirty_work_and_reads_all_pages() {
    let store = Arc::new(InMemory::new());
    let d = Directory::new(store.clone(), 7);
    let mut old = d.register("first", &config("first"), true).await.unwrap();
    let fresh = d.register("first", &config("first"), true).await.unwrap();
    old.last_report.scan_complete = true;
    assert!(!d.finish(old).await.unwrap());
    assert_eq!(
        d.entry("first")
            .await
            .unwrap()
            .unwrap()
            .value
            .registration
            .reconciled_generation,
        0
    );
    assert_eq!(fresh.registration.dirty_generation, 2);
    for id in ["second", "third", "fourth", "fifth"] {
        d.register(id, &config(id), true).await.unwrap();
    }
    let d = Directory::new(store, 7);
    let mut seen = std::collections::BTreeSet::new();
    for _ in 0..3 {
        let page = d.page(2).await.unwrap();
        assert!(page.len() <= 2);
        for slot in page {
            seen.insert(
                d.read_slot(slot)
                    .await
                    .unwrap()
                    .unwrap()
                    .registration
                    .scope
                    .space_id,
            );
            d.advance_cursor(slot).await.unwrap();
        }
    }
    assert_eq!(seen.len(), 5);
    assert_eq!(d.page(2).await.unwrap(), vec![1, 2]);
    assert!(
        d.register("first", &config("replacement-instance"), true)
            .await
            .is_err()
    );
}

#[tokio::test]
async fn directory_reconciles_lost_ack_and_rejects_wrong_scope() {
    use anda_object_store::fault::{FaultKind, FaultOp, FaultRule, FaultStore};
    let (store, faults) = FaultStore::wrap(InMemory::new());
    let d = Directory::new(Arc::new(store), 3);
    faults.push_rule(FaultRule {
        kind: FaultKind::ErrorAfter,
        ..FaultRule::fail_once(FaultOp::Put, "/spaces/")
    });
    let entry = d.register("safe", &config("safe"), true).await.unwrap();
    let mut bad = entry.clone();
    bad.registration.shard = 999;
    let key = format!(
        "spaces/{}",
        &anda_cognitive_nexus::content_digest(&serde_json::json!("safe")).unwrap()[7..]
    );
    d.put(&key, &bad, PutMode::Overwrite).await.unwrap();
    assert!(d.read_slot(entry.slot).await.is_err());
}

#[tokio::test]
async fn local_directory_cas_survives_metastore_restart() {
    use anda_object_store::MetaStoreBuilder;
    let dir = std::env::temp_dir().join(format!("brain-r2-{}", rand::random::<u64>()));
    std::fs::create_dir_all(&dir).unwrap();
    let make = || {
        Arc::new(
            MetaStoreBuilder::new(
                object_store::local::LocalFileSystem::new_with_prefix(&dir).unwrap(),
                100,
            )
            .build(),
        )
    };
    let d = Directory::new(make(), 0);
    let mut entry = d.register("local", &config("local"), true).await.unwrap();
    entry.last_report.scan_complete = true;
    assert!(d.finish(entry).await.unwrap());
    drop(d);
    let reopened = Directory::new(make(), 0);
    let entry = reopened.read_slot(1).await.unwrap().unwrap();
    assert_eq!(
        entry.registration.dirty_generation,
        entry.registration.reconciled_generation
    );
    drop(reopened);
    std::fs::remove_dir_all(dir).unwrap();
}