kmp-adapter-embedded 0.1.5

Embedded edition storage adapters: every KMP kernel port on a single local store file (redb by default, SQLite opt-in)
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113

//! Crash-test workload driver: appends context events and applies
//! projections in a tight loop, printing each committed revision to stdout.
//! The crash-recovery test SIGKILLs this process mid-loop and verifies the
//! store's durability contract; the scale test runs it to corpus size.

use std::io::Write;
use std::time::SystemTime;

use kmp_adapter_embedded::EmbeddedKernelStore;
use kmp_domain::{
    ContextEventChange, ContextEventStore, ContextUpdatedEvent, NodeDetailProjection,
    NodeProjection, ProjectionMutation, ProjectionWriter,
};

const DEFAULT_ABOUT: &str = "crash:test";
const ROLE: &str = "memory";
const SCOPE: &str = "about:crash:test:dimension:conversation:s1";

fn entry_id(revision: u64) -> String {
    format!("claim:{revision:06}")
}

fn event_for(about: &str, revision: u64) -> ContextUpdatedEvent {
    let entry = entry_id(revision);
    let payload = serde_json::json!({
        "id": entry,
        "kind": "claim",
        "text": format!("Crash-test decision {revision}."),
        "coordinates": [{
            "dimension": "conversation",
            "scope_id": SCOPE,
            "occurred_at": format!("2026-07-21T{:02}:{:02}:00Z", revision / 60 % 24, revision % 60),
            "sequence": revision.max(1),
        }],
    })
    .to_string();

    ContextUpdatedEvent {
        root_node_id: about.to_string(),
        role: ROLE.to_string(),
        revision,
        content_hash: format!("hash-{revision}"),
        changes: vec![ContextEventChange {
            operation: "UPSERT".to_string(),
            entity_kind: "memory_entry".to_string(),
            entity_id: entry,
            payload_json: payload,
            reason: Some("crash-test ingest".to_string()),
            scopes: vec![SCOPE.to_string()],
        }],
        idempotency_key: Some(format!("ingest:crash-{revision}")),
        logical_digest: None,
        requested_by: None,
        occurred_at: SystemTime::now(),
    }
}

fn projection_for(revision: u64) -> Vec<ProjectionMutation> {
    let entry = entry_id(revision);
    vec![
        ProjectionMutation::UpsertNode(NodeProjection {
            node_id: entry.clone(),
            node_kind: "claim".to_string(),
            title: format!("Crash-test decision {revision}"),
            summary: format!("Crash-test decision {revision}."),
            status: "ACTIVE".to_string(),
            labels: vec!["memory".to_string(), "entry".to_string()],
            properties: Default::default(),
            provenance: None,
        }),
        ProjectionMutation::UpsertNodeDetail(NodeDetailProjection {
            node_id: entry,
            detail: format!("Crash-test decision {revision}."),
            content_hash: format!("detail-hash-{revision}"),
            revision,
        }),
    ]
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let mut args = std::env::args().skip(1);
    let data_dir = args
        .next()
        .expect("usage: embedded_crash_writer <data-dir> <count>");
    let count: u64 = args
        .next()
        .expect("usage: embedded_crash_writer <data-dir> <count> [about]")
        .parse()
        .expect("count must be a positive integer");
    // A third argument picks the aggregate. Two writers on one store need
    // two aggregates: the revision check is per aggregate, and that is what
    // lets both of them commit instead of one conflicting the other out.
    let about = args.next().unwrap_or_else(|| DEFAULT_ABOUT.to_string());

    let store = EmbeddedKernelStore::open(std::path::Path::new(&data_dir))
        .expect("embedded store should open");
    let stdout = std::io::stdout();

    for revision in 1..=count {
        let committed = store
            .append(event_for(&about, revision), revision - 1)
            .await
            .expect("append should succeed");
        store
            .apply_mutations(projection_for(committed))
            .await
            .expect("projection should apply");
        let mut lock = stdout.lock();
        writeln!(lock, "{committed}").expect("stdout write");
        lock.flush().expect("stdout flush");
    }
}