kmp-adapter-embedded 0.18.3

Embedded SQLite storage adapters for every KMP persistence port
Documentation
use super::*;
use crate::EmbeddedKernelStore;
use kmp_domain::{
    GraphNeighborhoodReader, NodeRelationProjection, ProjectionMutation, ProjectionWriter,
    RelationDirection, RelationExplanation, RelationSemanticClass, TraceSearchLimits,
    TraceSearchRequest, TraceSearchStop, bounded_trace_search,
};

fn node(id: &str, about: &str) -> ProjectionMutation {
    ProjectionMutation::UpsertNode(NodeProjection {
        node_id: id.into(),
        node_kind: "observation".into(),
        title: id.into(),
        summary: id.into(),
        status: "ACTIVE".into(),
        labels: vec!["entry".into()],
        properties: [("memory_about".into(), about.into())].into(),
        provenance: None,
    })
}

fn edge(from: &str, to: &str, source: &str) -> ProjectionMutation {
    ProjectionMutation::UpsertNodeRelation(Box::new(NodeRelationProjection {
        source_node_id: from.into(),
        target_node_id: to.into(),
        relation_type: "depends_on".into(),
        explanation: RelationExplanation::new(RelationSemanticClass::Causal)
            .with_optional_rationale(Some("The register declares a dependency.".into()))
            .with_optional_evidence(Some(source.into())),
    }))
}

#[tokio::test]
async fn ownership_and_all_route_explanations_share_one_snapshot_during_independent_commit() {
    let dir = tempfile::tempdir().expect("temporary store");
    let reader = EmbeddedKernelStore::open(dir.path()).expect("open fixture store");
    let writer = EmbeddedKernelStore::open(dir.path()).expect("open fixture store");
    writer
        .apply_mutations(vec![
            node("a", "project:test"),
            node("b", "project:test"),
            node("c", "project:test"),
            edge("a", "b", "old AB"),
            edge("b", "c", "old BC"),
        ])
        .await
        .expect("fixture write or bounded read");
    let tx = reader.begin_read().expect("read snapshot");
    let snapshot = TraceSnapshot(tx.as_ref());
    assert!(
        snapshot
            .node("a")
            .expect("fixture write or bounded read")
            .is_some()
    );
    writer
        .apply_mutations(vec![edge("b", "c", "new BC"), node("b", "project:foreign")])
        .await
        .expect("fixture write or bounded read");
    let request = TraceSearchRequest {
        proof: false,
        about: "project:test".into(),
        from: "a".into(),
        targets: ["c".into()].into(),
        direction: RelationDirection::Outgoing,
        dimensions: Default::default(),
        select: None,
        follow: vec![],
        paths_per_target: 1,
        relations: Default::default(),
        limits: TraceSearchLimits::default(),
        temporal: Default::default(),
        body: Default::default(),
    };
    let result = bounded_trace_search(&snapshot, &request).expect("fixture write or bounded read");
    assert_eq!(result.stop, TraceSearchStop::TargetsReached);
    assert_eq!(result.relations[1].explanation.evidence(), Some("old BC"));
    drop(tx);
    let fresh = reader
        .load_bounded_trace(&request)
        .await
        .expect("fixture write or bounded read");
    assert_eq!(fresh.stop, TraceSearchStop::FrontierExhausted);
    assert!(
        fresh.relations.is_empty(),
        "new ownership excludes the entire foreign path"
    );
}

#[tokio::test]
async fn coordinate_admission_and_proof_share_one_snapshot_during_retiming() {
    use kmp_domain::{MemoryDimensionIdentity, TemporalAxis, TemporalCursor, TemporalSelection};
    let dir = tempfile::tempdir().expect("directory");
    let reader = EmbeddedKernelStore::open(dir.path()).expect("reader");
    let writer = EmbeddedKernelStore::open(dir.path()).expect("writer");
    let lane = MemoryDimensionIdentity::new("project:test", "work", "one")
        .expect("label")
        .node_id();
    let coordinate = |id: &str, at: &str| {
        ProjectionMutation::UpsertNodeRelation(Box::new(NodeRelationProjection {
            source_node_id: lane.clone(),
            target_node_id: id.into(),
            relation_type: "contains_entry".into(),
            explanation: RelationExplanation::new(RelationSemanticClass::Structural)
                .with_dimension("work")
                .with_scope_id(lane.clone())
                .with_observed_at(at),
        }))
    };
    writer
        .apply_mutations(vec![
            node("a", "project:test"),
            node("b", "project:test"),
            coordinate("a", "2026-09-11T09:00:00Z"),
            coordinate("b", "2026-09-11T09:00:00Z"),
            edge("a", "b", "old proof"),
        ])
        .await
        .expect("initial");
    let tx = reader.begin_read().expect("snapshot");
    let snapshot = TraceSnapshot(tx.as_ref());
    assert!(snapshot.node("a").expect("fix snapshot").is_some());
    writer
        .apply_mutations(vec![
            coordinate("b", "2026-09-11T11:00:00Z"),
            edge("a", "b", "new proof"),
        ])
        .await
        .expect("later commit");
    let request = TraceSearchRequest {
        proof: false,
        about: "project:test".into(),
        from: "a".into(),
        targets: ["b".into()].into(),
        direction: RelationDirection::Outgoing,
        dimensions: Default::default(),
        select: None,
        follow: vec![],
        paths_per_target: 1,
        relations: Default::default(),
        limits: Default::default(),
        temporal: TemporalSelection::as_of(
            TemporalCursor::time("2026-09-11T10:00:00Z").expect("cut"),
            TemporalAxis::Observed,
        )
        .expect("selection"),
        body: Default::default(),
    };
    let result = bounded_trace_search(&snapshot, &request).expect("old snapshot");
    assert_eq!(result.stop, TraceSearchStop::TargetsReached);
    assert_eq!(
        result.relations[0].explanation.evidence(),
        Some("old proof")
    );
    drop(tx);
    assert!(
        reader
            .load_bounded_trace(&request)
            .await
            .expect("fresh snapshot")
            .routes
            .is_empty()
    );
}

#[tokio::test]
async fn focused_resumed_pages_keep_ownership_and_evidence_in_the_original_snapshot() {
    use kmp_domain::{DimensionSelection, LabelSelector, LabelSelectorOperator};
    let dir = tempfile::tempdir().expect("directory");
    let reader = EmbeddedKernelStore::open(dir.path()).expect("reader");
    let writer = EmbeddedKernelStore::open(dir.path()).expect("writer");
    let mut mutations = vec![
        node("a", "project:test"),
        node("z", "project:test"),
        node("t", "project:test"),
        edge("a", "z", "old AZ"),
        edge("z", "t", "old ZT"),
    ];
    for i in 0..12 {
        let child = format!("n{i:02}");
        mutations.extend([
            node(&child, "project:test"),
            edge("a", &child, "declared distractor"),
        ]);
    }
    writer
        .apply_mutations(mutations)
        .await
        .expect("initial graph");
    let tx = reader.begin_read().expect("snapshot");
    let snapshot = TraceSnapshot(tx.as_ref());
    assert!(snapshot.node("a").expect("fix snapshot").is_some());
    writer
        .apply_mutations(vec![node("z", "project:foreign"), edge("z", "t", "new ZT")])
        .await
        .expect("independent commit");
    let request = TraceSearchRequest {
        proof: false,
        about: "project:test".into(),
        from: "a".into(),
        targets: ["t".into()].into(),
        direction: RelationDirection::Outgoing,
        relations: Default::default(),
        follow: vec![],
        paths_per_target: 1,
        select: None,
        limits: Default::default(),
        temporal: Default::default(),
        dimensions: kmp_domain::TraceDimensionPolicy {
            required: None,
            preferred: Some(
                DimensionSelection::all().with_selectors([LabelSelector::new(
                    "env",
                    LabelSelectorOperator::Exists,
                    Vec::<String>::new(),
                )
                .expect("selector")]),
            ),
        },
        body: Default::default(),
    };
    let old = bounded_trace_search(&snapshot, &request).expect("resumed old snapshot");
    assert_eq!(old.stop, TraceSearchStop::TargetsReached);
    assert_eq!(old.relations[1].explanation.evidence(), Some("old ZT"));
    assert!(old.routing.expect("routing").resumed_states >= 3);
    drop(tx);
    let fresh = reader
        .load_bounded_trace(&request)
        .await
        .expect("new snapshot");
    assert_eq!(fresh.stop, TraceSearchStop::FrontierExhausted);
    assert!(fresh.routes.is_empty());
}