Struct MemoryStore 

pub struct MemoryStore { /* private fields */ }

Implementations

impl MemoryStore

pub fn calibration(&self, family: Option<&str>) -> CalibrationReport

Calibration of every family, or of one family.

pub fn calibration_report(&self, filter: FamilyFilter<'_>) -> CalibrationReport

Calibration of the families filter selects; a prefix filter adds pooled rows.

impl MemoryStore

pub fn active_conflicts(&self) -> Vec<ActiveConflict>

Every claim key whose active traces carry more than one normalized value. Conflicts are derived from current state on every call, so a supersession or purge resolves them without any extra event. Sensitive traces are left out unless the policy allows them.

impl MemoryStore

pub fn consolidation_proposals(&self) -> Vec<ConsolidationProposal>

Quiet Replay: generate evidence-preserving consolidation proposals. This never writes a new belief automatically.

impl MemoryStore

pub fn focus(&self, query: &RecallQuery) -> FocusCapsule

Start a working set from one recall, bounded by the policy’s working_set_limit.

impl MemoryStore

pub fn open(path: impl AsRef<Path>, policy: KernelPolicy) -> Result<Self>

Open a store, creating the file (and parent directories) when it does not exist. Use this for commands that write. Read-only commands should use Self::open_existing.

Examples found in repository

examples/retrieval_bench.rs (line 154)

fn retrieve(corpus: &Path, queries: &Path, dir: &Path, dataset: &str, k: usize) -> Result<()> {
    if !(1..=100).contains(&k) || dataset.trim().is_empty() {
        return Err("k must be 1..100 and dataset must be nonempty".into());
    }
    let policy = KernelPolicy {
        recall_limit: k,
        ..KernelPolicy::default()
    };
    let path = fresh(dir)?;
    let source = format!("beir:{dataset}");
    let mut store = MemoryStore::open(&path, policy.clone())?;
    let mut baseline = Bm25::default();
    let mut ids = HashSet::new();
    let mut rejected = Vec::new();
    let mut batch = Vec::new();
    let mut baseline_index_ms = 0.0;
    let ingest = Instant::now();
    for line in BufReader::new(File::open(corpus)?).lines() {
        let doc: Document = serde_json::from_str(&line?)?;
        if doc.id.trim().is_empty() || !ids.insert(doc.id.clone()) {
            return Err("empty or duplicate document id".into());
        }
        let trace = make_trace(&doc.id, &doc.content, &source, TraceKind::Observation);
        if let Err(reason) = trace.validate() {
            rejected.push(json!({"id": doc.id, "reason": reason.to_string()}));
            continue;
        }
        let timer = Instant::now();
        baseline.add(&doc);
        baseline_index_ms += ms(timer);
        batch.push(trace);
        if batch.len() == 1000 {
            store.remember_many(std::mem::take(&mut batch))?;
        }
    }
    if !batch.is_empty() {
        store.remember_many(batch)?;
    }
    let ingest_ms = ms(ingest);
    store.verify()?;
    let head_before = serde_json::to_value(store.head())?;
    drop(store);
    let start = Instant::now();
    let store = MemoryStore::open_existing(&path, policy)?;
    let reopen_ms = ms(start);
    store.verify()?;
    let provenance_retained = store
        .all()
        .filter(|entry| {
            let t = &entry.trace;
            t.provenance.source == source
                && !t.provenance.verified
                && t.created_at_ms == CLOCK
                && t.provenance.observed_at_ms == CLOCK
                && t.provenance.locator.as_deref() == Some(format!("{source}:{}", t.id).as_str())
        })
        .count();
    if store.record_count() != baseline.ids.len()
        || provenance_retained != baseline.ids.len()
        || serde_json::to_value(store.head())? != head_before
    {
        return Err("replay, head or provenance invariant failed".into());
    }
    emit(
        json!({"kind": "loaded", "corpus_total": ids.len(), "admitted": baseline.ids.len(),
        "rejected": rejected, "provenance_retained": provenance_retained,
        "ingest_both_ms": ingest_ms, "bm25_index_ms": baseline_index_ms, "reopen_ms": reopen_ms,
        "policy": store.policy(), "bm25": {"k1": 1.2, "b": 0.75}, "query_clock_ms": CLOCK}),
    )?;
    let mut query_ids = HashSet::new();
    for line in BufReader::new(File::open(queries)?).lines() {
        let q: Query = serde_json::from_str(&line?)?;
        if q.id.trim().is_empty() || !query_ids.insert(q.id.clone()) {
            return Err("invalid query id".into());
        }
        let mut query = RecallQuery::new(&q.text);
        query.now_ms = CLOCK;
        query.limit = k;
        let start = Instant::now();
        let recalled = store.recall(&query);
        let hikmah_ms = ms(start);
        let start = Instant::now();
        let bm25 = baseline.search(&q.text, k);
        let bm25_ms = ms(start);
        let hikmah: Vec<_> = recalled.into_iter().map(|r| json!({
            "id": r.trace.id, "score": r.score, "duplicates_folded": r.duplicates,
            "provenance_source": r.trace.provenance.source, "verified": r.trace.provenance.verified,
        })).collect();
        emit(
            json!({"kind": "query", "id": q.id, "hikmah": hikmah, "bm25": bm25,
                    "hikmah_ms": hikmah_ms, "bm25_ms": bm25_ms}),
        )?;
    }
    emit(json!({"kind": "finished", "queries": query_ids.len()}))
}

fn corrections(dir: &Path, cases: usize) -> Result<()> {
    if !(1..=10000).contains(&cases) {
        return Err("cases must be 1..10000".into());
    }
    let path = fresh(dir)?;
    let mut store = MemoryStore::open(&path, KernelPolicy::default())?;
    let mut stale_hits = 0;
    let mut latest_hits = 0;
    let mut model_rejections = 0;
    let mut link_hits = 0;
    let mut expected = Vec::new();
    for i in 0..cases {
        let subject = format!("service{i}topic");
        let key = format!("{subject}.region");
        let mut predecessor = None;
        for (version, value) in ["oldregion", "middlezone", "newlocation"]
            .iter()
            .enumerate()
        {
            let id = format!("{i}-v{version}");
            let mut trace = make_trace(
                &id,
                &format!("{subject} {value}"),
                "synthetic:correction",
                if version == 0 {
                    TraceKind::Belief
                } else {
                    TraceKind::Correction
                },
            );
            trace.claim_key = Some(key.clone());
            trace.claim_value = Some((*value).into());
            trace.supersedes = predecessor;
            store.remember(trace)?;
            predecessor = Some(id);
        }
        let mut malicious = make_trace(
            &format!("{i}-model"),
            "invented overwrite",
            "model:fixture",
            TraceKind::Correction,
        );
        malicious.supersedes = predecessor;
        model_rejections += usize::from(store.remember(malicious).is_err());
        let results = store.recall(&RecallQuery::new(&subject));
        stale_hits += results
            .iter()
            .filter(|r| r.trace.id == format!("{i}-v0") || r.trace.id == format!("{i}-v1"))
            .count();
        latest_hits += usize::from(results.iter().any(|r| r.trace.id == format!("{i}-v2")));
        link_hits +=
            usize::from(results.iter().any(|r| {
                r.trace.id == format!("{i}-v2") && r.supersedes == Some(format!("{i}-v1"))
            }));
        expected.push((subject, format!("{i}-v2")));
    }
    let old_only_results = store
        .recall(&RecallQuery::new("oldregion middlezone"))
        .len();
    drop(store);
    let store = MemoryStore::open_existing(&path, KernelPolicy::default())?;
    store.verify()?;
    let restart_hits = expected
        .iter()
        .filter(|(q, id)| {
            store
                .recall(&RecallQuery::new(q))
                .iter()
                .any(|r| &r.trace.id == id)
        })
        .count();
    let provenance = store
        .all()
        .filter(|e| {
            e.trace.provenance.source == "synthetic:correction" && !e.trace.provenance.verified
        })
        .count();
    let ok = stale_hits == 0
        && latest_hits == cases
        && model_rejections == cases
        && link_hits == cases
        && old_only_results == 0
        && restart_hits == cases
        && provenance == 3 * cases;
    emit(
        json!({"kind": "synthetic_corrections", "cases": cases, "ok": ok,
        "stale_hits": stale_hits, "latest_hits": latest_hits, "supersession_link_hits": link_hits,
        "model_overwrite_rejections": model_rejections, "old_only_query_results": old_only_results,
        "latest_hits_after_restart": restart_hits, "provenance_retained": provenance,
        "limits": ["Generated structured histories; not extracted real incident corrections.",
          "Subject variants of one generated template, not independent real incidents or a proof across all histories.",
          "A query matching only retired terms abstains; no semantic alias expansion is implied."]}),
    )?;
    if !ok {
        return Err("correction invariant failure".into());
    }
    Ok(())
}

examples/bench_kernel.rs (line 119)

fn scale(dir: &Path, records: usize, queries: usize) -> Result<Value> {
    if records == 0 || queries == 0 {
        return Err("records and queries must be positive".into());
    }
    let path = fresh(dir)?;
    let mut store = MemoryStore::open(&path, KernelPolicy::default())?;
    let single_count = records.min(50);
    let batched = records - single_count;
    let mut batches = Vec::new();
    let seed_started = Instant::now();
    for begin in (0..batched).step_by(1000) {
        let traces = (begin..(begin + 1000).min(batched)).map(trace).collect();
        let start = Instant::now();
        store.remember_many(traces)?;
        batches.push(ms(start));
    }
    let mut appends = Vec::new();
    for i in batched..records {
        let t = trace(i);
        let start = Instant::now();
        store.remember(t)?;
        appends.push(ms(start));
    }
    let seed_ms = ms(seed_started);
    let memory_after_seed = rss();
    let mut recall_ms = Vec::new();
    let mut hits = 0;
    let mut returned = 0;
    let mut result_ids = Vec::new();
    // Warm up with an exact cue; quality here is a synthetic exact-cue check, not field recall.
    std::hint::black_box(store.recall(&RecallQuery::new(token(0))));
    for i in 0..queries {
        let selected = i * records / queries;
        let mut query = RecallQuery::new(token(selected));
        query.now_ms = CLOCK;
        query.limit = 1;
        let start = Instant::now();
        let found = store.recall(&query);
        recall_ms.push(ms(start));
        returned += found.len();
        hits += usize::from(
            found
                .first()
                .is_some_and(|r| r.trace.id == format!("sample-{selected}")),
        );
        result_ids.push(found.iter().map(|r| r.trace.id.clone()).collect::<Vec<_>>());
    }
    let start = Instant::now();
    store.verify()?;
    let verify_ms = ms(start);
    let before = store.head();
    drop(store);
    let start = Instant::now();
    let store = MemoryStore::open_existing(&path, KernelPolicy::default())?;
    let replay_ms = ms(start);
    store.verify()?;
    let retained = store
        .all()
        .filter(|e| {
            e.trace.provenance.source == "synthetic-kernel-benchmark"
                && e.trace.provenance.observed_at_ms == CLOCK
                && !e.trace.provenance.verified
        })
        .count();
    let mut replay_matches = 0;
    for (i, expected) in result_ids.iter().enumerate() {
        let mut query = RecallQuery::new(token(i * records / queries));
        query.now_ms = CLOCK;
        query.limit = 1;
        let ids: Vec<_> = store
            .recall(&query)
            .into_iter()
            .map(|r| r.trace.id)
            .collect();
        replay_matches += usize::from(ids == *expected);
    }
    let head_same = serde_json::to_value(before)? == serde_json::to_value(store.head())?;
    let ok = store.record_count() == records
        && retained == records
        && hits == queries
        && replay_matches == queries
        && head_same;
    Ok(json!({
        "kind": "synthetic_scale", "ok": ok, "records": records, "queries": queries,
        "workload": "unclaimed observation traces, unique exact lexical cues, fixed clock",
        "seed_ms": seed_ms, "batch_size": 1000, "batched_records": batched,
        "batch_append_ms": quantiles(batches), "single_append_ms": quantiles(appends),
        "recall_ms": quantiles(recall_ms), "verify_ms": verify_ms, "replay_ms": replay_ms,
        "exact_cue_hits": hits, "returned": returned, "recall_at_1": hits as f64 / queries as f64,
        "provenance_retained": retained, "replayed_records": store.record_count(),
        "recall_replay_matches": replay_matches, "head_unchanged": head_same,
        "ledger_bytes": fs::metadata(path)?.len(), "memory_after_seed": memory_after_seed,
        "memory_after_replay": rss(), "policy": store.policy(),
        "limits": ["Batched seeding does not measure per-record durable append latency.",
          "Warm filesystem cache; reopening in the same process is not a cold-start measurement.",
          "No structured claims in this workload; conflict-heavy stores need a separate workload.",
          "Synthetic exact-cue quality is not semantic or real-world retrieval accuracy.",
          "RSS is process-wide Linux /proc data; unavailable platforms report null."]
    }))
}

// Specified, platform-independent SplitMix64; never a cryptographic RNG.
struct Rng(u64);
impl Rng {
    fn next(&mut self) -> u64 {
        self.0 = self.0.wrapping_add(0x9e3779b97f4a7c15);
        let mut x = self.0;
        x = (x ^ (x >> 30)).wrapping_mul(0xbf58476d1ce4e5b9);
        x = (x ^ (x >> 27)).wrapping_mul(0x94d049bb133111eb);
        x ^ (x >> 31)
    }
    fn unit(&mut self) -> f64 {
        (self.next() % 10001) as f64 / 10000.0
    }
}

fn option(name: &str, a: f64, b: f64) -> DecisionOption {
    DecisionOption {
        name: name.into(),
        description: None,
        scores: BTreeMap::from([("a".into(), a), ("b".into(), b)]),
        model_scores: BTreeMap::new(),
        evidence_confidence: 1.0,
        hard_blocks: vec![],
        reversible: false,
    }
}

fn decisions(cases: usize, seed: u64) -> Result<Value> {
    if cases < 6 {
        return Err("at least 6 cases required to exercise every family".into());
    }
    let mut rng = Rng(seed);
    let mut counts: BTreeMap<&str, usize> = BTreeMap::new();
    let mut failures = Vec::new();
    let mut violations = 0;
    let start = Instant::now();
    for i in 0..cases {
        let mut a = option("a", rng.unit(), rng.unit());
        let mut b = option("b", rng.unit(), rng.unit());
        let family;
        let mut expected: Option<Option<String>> = None;
        match i % 6 {
            0 => {
                family = "high_score_hard_block";
                a.scores.values_mut().for_each(|s| *s = 1.0);
                a.hard_blocks.push("no authorization".into());
                expected = Some(Some("b".into()));
            }
            1 => {
                family = "all_blocked";
                a.hard_blocks.push("privacy".into());
                b.hard_blocks.push("consent".into());
                expected = Some(None);
            }
            2 => {
                family = "complete_score_oracle";
                let sum_a: f64 = a.scores.values().sum();
                let sum_b: f64 = b.scores.values().sum();
                expected = Some(Some(if sum_a >= sum_b { "a" } else { "b" }.into()));
            }
            3 => {
                family = "unknowns_and_model_estimates";
                a.scores.clear();
                a.model_scores.insert("a".into(), rng.unit());
            }
            4 => {
                family = "reversibility_band";
                let top = 0.5 + rng.unit() * 0.4;
                let gap = if rng.next().is_multiple_of(2) {
                    0.01
                } else {
                    0.03
                };
                a.scores.values_mut().for_each(|s| *s = top);
                a.evidence_confidence = 0.1;
                b.scores.values_mut().for_each(|s| *s = top - gap);
                b.reversible = true;
                expected = Some(Some(if gap < 0.02 { "b" } else { "a" }.into()));
            }
            _ => {
                family = "invalid_numeric_input";
                a.scores.insert("a".into(), 1.01 + rng.unit());
            }
        }
        *counts.entry(family).or_default() += 1;
        let frame = DecisionFrame {
            question: "synthetic decision".into(),
            criteria: ["a", "b"]
                .into_iter()
                .map(|id| Criterion {
                    id: id.into(),
                    weight: 1.0,
                    description: id.into(),
                })
                .collect(),
            options: vec![a, b],
        };
        let actual = evaluate(&frame);
        let mut reasons = Vec::new();
        if family == "invalid_numeric_input" {
            if actual.is_ok() {
                reasons.push("invalid input admitted");
            }
        } else if let Ok(answer) = &actual {
            if let Some(expected) = expected {
                if answer.recommended != expected {
                    reasons.push("wrong recommendation");
                }
            }
            if let Some(name) = &answer.recommended {
                if frame
                    .options
                    .iter()
                    .any(|o| &o.name == name && !o.hard_blocks.is_empty())
                {
                    reasons.push("hard-blocked option selected");
                }
            }
            if family == "unknowns_and_model_estimates" {
                let ranked = answer.ranking.iter().find(|o| o.name == "a").unwrap();
                if ranked.coverage != 0.0
                    || ranked.evidence_interval != [0.0, 1.0]
                    || ranked.missing_criteria != ["b"]
                    || answer.decisive
                {
                    reasons.push("unknown evidence hidden or model guess counted as evidence");
                }
            }
            let mut permuted = frame.clone();
            permuted.options.reverse();
            permuted.question = "irrelevant wording changed".into();
            let other = evaluate(&permuted)?;
            if answer.recommended != other.recommended
                || serde_json::to_value(&answer.ranking)? != serde_json::to_value(&other.ranking)?
            {
                reasons.push("order or irrelevant text changed result");
            }
        } else {
            reasons.push("valid frame rejected");
        }
        if !reasons.is_empty() {
            violations += 1;
            if failures.len() < 20 {
                failures.push(
                    json!({"case_index": i, "family": family, "reasons": reasons,
                    "frame": frame, "actual": actual.as_ref().ok()}),
                );
            }
        }
    }
    Ok(
        json!({"kind": "synthetic_decision_invariants", "seed": seed, "rng": "splitmix64",
        "cases": cases, "families": counts, "violations": violations,
        "violation_rate": violations as f64 / cases as f64, "ok": violations == 0,
        "elapsed_ms": ms(start), "counterexamples": failures,
        "limits": ["Generated families are not real decision-quality labels or a proof over all inputs.",
          "Metamorphic checks keep criteria fixed; weight perturbations require separate expectations."]}),
    )
}

fn committed_child(dir: &Path) -> Result<()> {
    let path = fresh(dir)?;
    let mut store = MemoryStore::open(path, KernelPolicy::default())?;
    store.remember_many((0..50).map(trace).collect())?;
    println!("committed:50");
    io::stdout().flush()?;
    // Parent kills us after the acknowledged commit; this is not an in-flight crash claim.
    loop {
        std::thread::park();
    }
}

fn recovery(dir: &Path) -> Result<Value> {
    fs::create_dir(dir)?;
    let child_dir = dir.join("committed");
    let mut child = Command::new(std::env::current_exe()?)
        .args(["committed-child", "--dir"])
        .arg(&child_dir)
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .spawn()?;
    let mut ack = String::new();
    let read_result = io::BufReader::new(child.stdout.take().unwrap()).read_line(&mut ack);
    let kill_result = child.kill();
    let wait_result = child.wait();
    read_result?;
    kill_result?;
    wait_result?;
    if ack.trim() != "committed:50" {
        return Err("child did not acknowledge commit".into());
    }
    let path = child_dir.join("memory.jsonl");
    let start = Instant::now();
    let store = MemoryStore::open_existing(&path, KernelPolicy::default())?;
    store.verify()?;
    let restart_ms = ms(start);
    let acknowledged_survived =
        store.record_count() == 50 && (0..50).all(|i| store.get(&format!("sample-{i}")).is_some());
    let head_path = store.head_path();
    drop(store);
    let original = fs::read(&path)?;
    let original_head = fs::read(head_path)?;
    let copied_store = |name: &str, bytes: &[u8]| -> Result<PathBuf> {
        let target = fresh(&dir.join(name))?;
        let store = MemoryStore::open(&target, KernelPolicy::default())?;
        let target_head = store.head_path();
        drop(store);
        fs::write(&target, bytes)?;
        fs::write(target_head, &original_head)?;
        Ok(target)
    };
    let torn = copied_store("torn", &original)?;
    OpenOptions::new()
        .append(true)
        .open(&torn)?
        .write_all(b"{\"seq\":")?;
    let mut store = MemoryStore::open_existing(&torn, KernelPolicy::default())?;
    store.remember(trace(50))?;
    store.verify()?;
    let torn_repaired = store.record_count() == 51;
    let mut edited = String::from_utf8(original.clone())?;
    edited = edited.replacen("deployment observation", "deployment fabrication", 1);
    let tampered = copied_store("edited", edited.as_bytes())?;
    let rejected = |path: &Path| match MemoryStore::open_existing(path, KernelPolicy::default()) {
        Err(_) => true,
        Ok(store) => store.verify().is_err(),
    };
    let edit_detected = rejected(&tampered);
    let last = original[..original.len() - 1]
        .iter()
        .rposition(|b| *b == b'\n')
        .unwrap()
        + 1;
    let truncated = copied_store("truncated", &original[..last])?;
    let truncation_detected = rejected(&truncated);
    Ok(
        json!({"kind": "synthetic_recovery", "acknowledged_records": 50,
        "acknowledged_survived": acknowledged_survived, "restart_ms": restart_ms,
        "torn_tail_repaired": torn_repaired, "edit_detected": edit_detected,
        "truncation_detected": truncation_detected,
        "ok": acknowledged_survived && torn_repaired && edit_detected && truncation_detected,
        "limits": ["Process killed after an acknowledged commit, not during fsync or head replacement.",
          "Torn tail, edit and truncation are injected into separate copies.",
          "This does not simulate power loss or an attacker rewriting both ledger and head."]}),
    )
}

pub fn open_existing( path: impl AsRef<Path>, policy: KernelPolicy, ) -> Result<Self>

Open a store that must already exist. A mistyped path is an error, not an empty store.

Examples found in repository

examples/retrieval_bench.rs (line 187)

fn retrieve(corpus: &Path, queries: &Path, dir: &Path, dataset: &str, k: usize) -> Result<()> {
    if !(1..=100).contains(&k) || dataset.trim().is_empty() {
        return Err("k must be 1..100 and dataset must be nonempty".into());
    }
    let policy = KernelPolicy {
        recall_limit: k,
        ..KernelPolicy::default()
    };
    let path = fresh(dir)?;
    let source = format!("beir:{dataset}");
    let mut store = MemoryStore::open(&path, policy.clone())?;
    let mut baseline = Bm25::default();
    let mut ids = HashSet::new();
    let mut rejected = Vec::new();
    let mut batch = Vec::new();
    let mut baseline_index_ms = 0.0;
    let ingest = Instant::now();
    for line in BufReader::new(File::open(corpus)?).lines() {
        let doc: Document = serde_json::from_str(&line?)?;
        if doc.id.trim().is_empty() || !ids.insert(doc.id.clone()) {
            return Err("empty or duplicate document id".into());
        }
        let trace = make_trace(&doc.id, &doc.content, &source, TraceKind::Observation);
        if let Err(reason) = trace.validate() {
            rejected.push(json!({"id": doc.id, "reason": reason.to_string()}));
            continue;
        }
        let timer = Instant::now();
        baseline.add(&doc);
        baseline_index_ms += ms(timer);
        batch.push(trace);
        if batch.len() == 1000 {
            store.remember_many(std::mem::take(&mut batch))?;
        }
    }
    if !batch.is_empty() {
        store.remember_many(batch)?;
    }
    let ingest_ms = ms(ingest);
    store.verify()?;
    let head_before = serde_json::to_value(store.head())?;
    drop(store);
    let start = Instant::now();
    let store = MemoryStore::open_existing(&path, policy)?;
    let reopen_ms = ms(start);
    store.verify()?;
    let provenance_retained = store
        .all()
        .filter(|entry| {
            let t = &entry.trace;
            t.provenance.source == source
                && !t.provenance.verified
                && t.created_at_ms == CLOCK
                && t.provenance.observed_at_ms == CLOCK
                && t.provenance.locator.as_deref() == Some(format!("{source}:{}", t.id).as_str())
        })
        .count();
    if store.record_count() != baseline.ids.len()
        || provenance_retained != baseline.ids.len()
        || serde_json::to_value(store.head())? != head_before
    {
        return Err("replay, head or provenance invariant failed".into());
    }
    emit(
        json!({"kind": "loaded", "corpus_total": ids.len(), "admitted": baseline.ids.len(),
        "rejected": rejected, "provenance_retained": provenance_retained,
        "ingest_both_ms": ingest_ms, "bm25_index_ms": baseline_index_ms, "reopen_ms": reopen_ms,
        "policy": store.policy(), "bm25": {"k1": 1.2, "b": 0.75}, "query_clock_ms": CLOCK}),
    )?;
    let mut query_ids = HashSet::new();
    for line in BufReader::new(File::open(queries)?).lines() {
        let q: Query = serde_json::from_str(&line?)?;
        if q.id.trim().is_empty() || !query_ids.insert(q.id.clone()) {
            return Err("invalid query id".into());
        }
        let mut query = RecallQuery::new(&q.text);
        query.now_ms = CLOCK;
        query.limit = k;
        let start = Instant::now();
        let recalled = store.recall(&query);
        let hikmah_ms = ms(start);
        let start = Instant::now();
        let bm25 = baseline.search(&q.text, k);
        let bm25_ms = ms(start);
        let hikmah: Vec<_> = recalled.into_iter().map(|r| json!({
            "id": r.trace.id, "score": r.score, "duplicates_folded": r.duplicates,
            "provenance_source": r.trace.provenance.source, "verified": r.trace.provenance.verified,
        })).collect();
        emit(
            json!({"kind": "query", "id": q.id, "hikmah": hikmah, "bm25": bm25,
                    "hikmah_ms": hikmah_ms, "bm25_ms": bm25_ms}),
        )?;
    }
    emit(json!({"kind": "finished", "queries": query_ids.len()}))
}

fn corrections(dir: &Path, cases: usize) -> Result<()> {
    if !(1..=10000).contains(&cases) {
        return Err("cases must be 1..10000".into());
    }
    let path = fresh(dir)?;
    let mut store = MemoryStore::open(&path, KernelPolicy::default())?;
    let mut stale_hits = 0;
    let mut latest_hits = 0;
    let mut model_rejections = 0;
    let mut link_hits = 0;
    let mut expected = Vec::new();
    for i in 0..cases {
        let subject = format!("service{i}topic");
        let key = format!("{subject}.region");
        let mut predecessor = None;
        for (version, value) in ["oldregion", "middlezone", "newlocation"]
            .iter()
            .enumerate()
        {
            let id = format!("{i}-v{version}");
            let mut trace = make_trace(
                &id,
                &format!("{subject} {value}"),
                "synthetic:correction",
                if version == 0 {
                    TraceKind::Belief
                } else {
                    TraceKind::Correction
                },
            );
            trace.claim_key = Some(key.clone());
            trace.claim_value = Some((*value).into());
            trace.supersedes = predecessor;
            store.remember(trace)?;
            predecessor = Some(id);
        }
        let mut malicious = make_trace(
            &format!("{i}-model"),
            "invented overwrite",
            "model:fixture",
            TraceKind::Correction,
        );
        malicious.supersedes = predecessor;
        model_rejections += usize::from(store.remember(malicious).is_err());
        let results = store.recall(&RecallQuery::new(&subject));
        stale_hits += results
            .iter()
            .filter(|r| r.trace.id == format!("{i}-v0") || r.trace.id == format!("{i}-v1"))
            .count();
        latest_hits += usize::from(results.iter().any(|r| r.trace.id == format!("{i}-v2")));
        link_hits +=
            usize::from(results.iter().any(|r| {
                r.trace.id == format!("{i}-v2") && r.supersedes == Some(format!("{i}-v1"))
            }));
        expected.push((subject, format!("{i}-v2")));
    }
    let old_only_results = store
        .recall(&RecallQuery::new("oldregion middlezone"))
        .len();
    drop(store);
    let store = MemoryStore::open_existing(&path, KernelPolicy::default())?;
    store.verify()?;
    let restart_hits = expected
        .iter()
        .filter(|(q, id)| {
            store
                .recall(&RecallQuery::new(q))
                .iter()
                .any(|r| &r.trace.id == id)
        })
        .count();
    let provenance = store
        .all()
        .filter(|e| {
            e.trace.provenance.source == "synthetic:correction" && !e.trace.provenance.verified
        })
        .count();
    let ok = stale_hits == 0
        && latest_hits == cases
        && model_rejections == cases
        && link_hits == cases
        && old_only_results == 0
        && restart_hits == cases
        && provenance == 3 * cases;
    emit(
        json!({"kind": "synthetic_corrections", "cases": cases, "ok": ok,
        "stale_hits": stale_hits, "latest_hits": latest_hits, "supersession_link_hits": link_hits,
        "model_overwrite_rejections": model_rejections, "old_only_query_results": old_only_results,
        "latest_hits_after_restart": restart_hits, "provenance_retained": provenance,
        "limits": ["Generated structured histories; not extracted real incident corrections.",
          "Subject variants of one generated template, not independent real incidents or a proof across all histories.",
          "A query matching only retired terms abstains; no semantic alias expansion is implied."]}),
    )?;
    if !ok {
        return Err("correction invariant failure".into());
    }
    Ok(())
}

examples/bench_kernel.rs (line 167)

fn scale(dir: &Path, records: usize, queries: usize) -> Result<Value> {
    if records == 0 || queries == 0 {
        return Err("records and queries must be positive".into());
    }
    let path = fresh(dir)?;
    let mut store = MemoryStore::open(&path, KernelPolicy::default())?;
    let single_count = records.min(50);
    let batched = records - single_count;
    let mut batches = Vec::new();
    let seed_started = Instant::now();
    for begin in (0..batched).step_by(1000) {
        let traces = (begin..(begin + 1000).min(batched)).map(trace).collect();
        let start = Instant::now();
        store.remember_many(traces)?;
        batches.push(ms(start));
    }
    let mut appends = Vec::new();
    for i in batched..records {
        let t = trace(i);
        let start = Instant::now();
        store.remember(t)?;
        appends.push(ms(start));
    }
    let seed_ms = ms(seed_started);
    let memory_after_seed = rss();
    let mut recall_ms = Vec::new();
    let mut hits = 0;
    let mut returned = 0;
    let mut result_ids = Vec::new();
    // Warm up with an exact cue; quality here is a synthetic exact-cue check, not field recall.
    std::hint::black_box(store.recall(&RecallQuery::new(token(0))));
    for i in 0..queries {
        let selected = i * records / queries;
        let mut query = RecallQuery::new(token(selected));
        query.now_ms = CLOCK;
        query.limit = 1;
        let start = Instant::now();
        let found = store.recall(&query);
        recall_ms.push(ms(start));
        returned += found.len();
        hits += usize::from(
            found
                .first()
                .is_some_and(|r| r.trace.id == format!("sample-{selected}")),
        );
        result_ids.push(found.iter().map(|r| r.trace.id.clone()).collect::<Vec<_>>());
    }
    let start = Instant::now();
    store.verify()?;
    let verify_ms = ms(start);
    let before = store.head();
    drop(store);
    let start = Instant::now();
    let store = MemoryStore::open_existing(&path, KernelPolicy::default())?;
    let replay_ms = ms(start);
    store.verify()?;
    let retained = store
        .all()
        .filter(|e| {
            e.trace.provenance.source == "synthetic-kernel-benchmark"
                && e.trace.provenance.observed_at_ms == CLOCK
                && !e.trace.provenance.verified
        })
        .count();
    let mut replay_matches = 0;
    for (i, expected) in result_ids.iter().enumerate() {
        let mut query = RecallQuery::new(token(i * records / queries));
        query.now_ms = CLOCK;
        query.limit = 1;
        let ids: Vec<_> = store
            .recall(&query)
            .into_iter()
            .map(|r| r.trace.id)
            .collect();
        replay_matches += usize::from(ids == *expected);
    }
    let head_same = serde_json::to_value(before)? == serde_json::to_value(store.head())?;
    let ok = store.record_count() == records
        && retained == records
        && hits == queries
        && replay_matches == queries
        && head_same;
    Ok(json!({
        "kind": "synthetic_scale", "ok": ok, "records": records, "queries": queries,
        "workload": "unclaimed observation traces, unique exact lexical cues, fixed clock",
        "seed_ms": seed_ms, "batch_size": 1000, "batched_records": batched,
        "batch_append_ms": quantiles(batches), "single_append_ms": quantiles(appends),
        "recall_ms": quantiles(recall_ms), "verify_ms": verify_ms, "replay_ms": replay_ms,
        "exact_cue_hits": hits, "returned": returned, "recall_at_1": hits as f64 / queries as f64,
        "provenance_retained": retained, "replayed_records": store.record_count(),
        "recall_replay_matches": replay_matches, "head_unchanged": head_same,
        "ledger_bytes": fs::metadata(path)?.len(), "memory_after_seed": memory_after_seed,
        "memory_after_replay": rss(), "policy": store.policy(),
        "limits": ["Batched seeding does not measure per-record durable append latency.",
          "Warm filesystem cache; reopening in the same process is not a cold-start measurement.",
          "No structured claims in this workload; conflict-heavy stores need a separate workload.",
          "Synthetic exact-cue quality is not semantic or real-world retrieval accuracy.",
          "RSS is process-wide Linux /proc data; unavailable platforms report null."]
    }))
}

// Specified, platform-independent SplitMix64; never a cryptographic RNG.
struct Rng(u64);
impl Rng {
    fn next(&mut self) -> u64 {
        self.0 = self.0.wrapping_add(0x9e3779b97f4a7c15);
        let mut x = self.0;
        x = (x ^ (x >> 30)).wrapping_mul(0xbf58476d1ce4e5b9);
        x = (x ^ (x >> 27)).wrapping_mul(0x94d049bb133111eb);
        x ^ (x >> 31)
    }
    fn unit(&mut self) -> f64 {
        (self.next() % 10001) as f64 / 10000.0
    }
}

fn option(name: &str, a: f64, b: f64) -> DecisionOption {
    DecisionOption {
        name: name.into(),
        description: None,
        scores: BTreeMap::from([("a".into(), a), ("b".into(), b)]),
        model_scores: BTreeMap::new(),
        evidence_confidence: 1.0,
        hard_blocks: vec![],
        reversible: false,
    }
}

fn decisions(cases: usize, seed: u64) -> Result<Value> {
    if cases < 6 {
        return Err("at least 6 cases required to exercise every family".into());
    }
    let mut rng = Rng(seed);
    let mut counts: BTreeMap<&str, usize> = BTreeMap::new();
    let mut failures = Vec::new();
    let mut violations = 0;
    let start = Instant::now();
    for i in 0..cases {
        let mut a = option("a", rng.unit(), rng.unit());
        let mut b = option("b", rng.unit(), rng.unit());
        let family;
        let mut expected: Option<Option<String>> = None;
        match i % 6 {
            0 => {
                family = "high_score_hard_block";
                a.scores.values_mut().for_each(|s| *s = 1.0);
                a.hard_blocks.push("no authorization".into());
                expected = Some(Some("b".into()));
            }
            1 => {
                family = "all_blocked";
                a.hard_blocks.push("privacy".into());
                b.hard_blocks.push("consent".into());
                expected = Some(None);
            }
            2 => {
                family = "complete_score_oracle";
                let sum_a: f64 = a.scores.values().sum();
                let sum_b: f64 = b.scores.values().sum();
                expected = Some(Some(if sum_a >= sum_b { "a" } else { "b" }.into()));
            }
            3 => {
                family = "unknowns_and_model_estimates";
                a.scores.clear();
                a.model_scores.insert("a".into(), rng.unit());
            }
            4 => {
                family = "reversibility_band";
                let top = 0.5 + rng.unit() * 0.4;
                let gap = if rng.next().is_multiple_of(2) {
                    0.01
                } else {
                    0.03
                };
                a.scores.values_mut().for_each(|s| *s = top);
                a.evidence_confidence = 0.1;
                b.scores.values_mut().for_each(|s| *s = top - gap);
                b.reversible = true;
                expected = Some(Some(if gap < 0.02 { "b" } else { "a" }.into()));
            }
            _ => {
                family = "invalid_numeric_input";
                a.scores.insert("a".into(), 1.01 + rng.unit());
            }
        }
        *counts.entry(family).or_default() += 1;
        let frame = DecisionFrame {
            question: "synthetic decision".into(),
            criteria: ["a", "b"]
                .into_iter()
                .map(|id| Criterion {
                    id: id.into(),
                    weight: 1.0,
                    description: id.into(),
                })
                .collect(),
            options: vec![a, b],
        };
        let actual = evaluate(&frame);
        let mut reasons = Vec::new();
        if family == "invalid_numeric_input" {
            if actual.is_ok() {
                reasons.push("invalid input admitted");
            }
        } else if let Ok(answer) = &actual {
            if let Some(expected) = expected {
                if answer.recommended != expected {
                    reasons.push("wrong recommendation");
                }
            }
            if let Some(name) = &answer.recommended {
                if frame
                    .options
                    .iter()
                    .any(|o| &o.name == name && !o.hard_blocks.is_empty())
                {
                    reasons.push("hard-blocked option selected");
                }
            }
            if family == "unknowns_and_model_estimates" {
                let ranked = answer.ranking.iter().find(|o| o.name == "a").unwrap();
                if ranked.coverage != 0.0
                    || ranked.evidence_interval != [0.0, 1.0]
                    || ranked.missing_criteria != ["b"]
                    || answer.decisive
                {
                    reasons.push("unknown evidence hidden or model guess counted as evidence");
                }
            }
            let mut permuted = frame.clone();
            permuted.options.reverse();
            permuted.question = "irrelevant wording changed".into();
            let other = evaluate(&permuted)?;
            if answer.recommended != other.recommended
                || serde_json::to_value(&answer.ranking)? != serde_json::to_value(&other.ranking)?
            {
                reasons.push("order or irrelevant text changed result");
            }
        } else {
            reasons.push("valid frame rejected");
        }
        if !reasons.is_empty() {
            violations += 1;
            if failures.len() < 20 {
                failures.push(
                    json!({"case_index": i, "family": family, "reasons": reasons,
                    "frame": frame, "actual": actual.as_ref().ok()}),
                );
            }
        }
    }
    Ok(
        json!({"kind": "synthetic_decision_invariants", "seed": seed, "rng": "splitmix64",
        "cases": cases, "families": counts, "violations": violations,
        "violation_rate": violations as f64 / cases as f64, "ok": violations == 0,
        "elapsed_ms": ms(start), "counterexamples": failures,
        "limits": ["Generated families are not real decision-quality labels or a proof over all inputs.",
          "Metamorphic checks keep criteria fixed; weight perturbations require separate expectations."]}),
    )
}

fn committed_child(dir: &Path) -> Result<()> {
    let path = fresh(dir)?;
    let mut store = MemoryStore::open(path, KernelPolicy::default())?;
    store.remember_many((0..50).map(trace).collect())?;
    println!("committed:50");
    io::stdout().flush()?;
    // Parent kills us after the acknowledged commit; this is not an in-flight crash claim.
    loop {
        std::thread::park();
    }
}

fn recovery(dir: &Path) -> Result<Value> {
    fs::create_dir(dir)?;
    let child_dir = dir.join("committed");
    let mut child = Command::new(std::env::current_exe()?)
        .args(["committed-child", "--dir"])
        .arg(&child_dir)
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .spawn()?;
    let mut ack = String::new();
    let read_result = io::BufReader::new(child.stdout.take().unwrap()).read_line(&mut ack);
    let kill_result = child.kill();
    let wait_result = child.wait();
    read_result?;
    kill_result?;
    wait_result?;
    if ack.trim() != "committed:50" {
        return Err("child did not acknowledge commit".into());
    }
    let path = child_dir.join("memory.jsonl");
    let start = Instant::now();
    let store = MemoryStore::open_existing(&path, KernelPolicy::default())?;
    store.verify()?;
    let restart_ms = ms(start);
    let acknowledged_survived =
        store.record_count() == 50 && (0..50).all(|i| store.get(&format!("sample-{i}")).is_some());
    let head_path = store.head_path();
    drop(store);
    let original = fs::read(&path)?;
    let original_head = fs::read(head_path)?;
    let copied_store = |name: &str, bytes: &[u8]| -> Result<PathBuf> {
        let target = fresh(&dir.join(name))?;
        let store = MemoryStore::open(&target, KernelPolicy::default())?;
        let target_head = store.head_path();
        drop(store);
        fs::write(&target, bytes)?;
        fs::write(target_head, &original_head)?;
        Ok(target)
    };
    let torn = copied_store("torn", &original)?;
    OpenOptions::new()
        .append(true)
        .open(&torn)?
        .write_all(b"{\"seq\":")?;
    let mut store = MemoryStore::open_existing(&torn, KernelPolicy::default())?;
    store.remember(trace(50))?;
    store.verify()?;
    let torn_repaired = store.record_count() == 51;
    let mut edited = String::from_utf8(original.clone())?;
    edited = edited.replacen("deployment observation", "deployment fabrication", 1);
    let tampered = copied_store("edited", edited.as_bytes())?;
    let rejected = |path: &Path| match MemoryStore::open_existing(path, KernelPolicy::default()) {
        Err(_) => true,
        Ok(store) => store.verify().is_err(),
    };
    let edit_detected = rejected(&tampered);
    let last = original[..original.len() - 1]
        .iter()
        .rposition(|b| *b == b'\n')
        .unwrap()
        + 1;
    let truncated = copied_store("truncated", &original[..last])?;
    let truncation_detected = rejected(&truncated);
    Ok(
        json!({"kind": "synthetic_recovery", "acknowledged_records": 50,
        "acknowledged_survived": acknowledged_survived, "restart_ms": restart_ms,
        "torn_tail_repaired": torn_repaired, "edit_detected": edit_detected,
        "truncation_detected": truncation_detected,
        "ok": acknowledged_survived && torn_repaired && edit_detected && truncation_detected,
        "limits": ["Process killed after an acknowledged commit, not during fsync or head replacement.",
          "Torn tail, edit and truncation are injected into separate copies.",
          "This does not simulate power loss or an attacker rewriting both ledger and head."]}),
    )
}

pub fn remember(&mut self, trace: Trace) -> Result<(Trace, Vec<ClaimConflict>)>

Examples found in repository

examples/retrieval_bench.rs (line 273)

fn corrections(dir: &Path, cases: usize) -> Result<()> {
    if !(1..=10000).contains(&cases) {
        return Err("cases must be 1..10000".into());
    }
    let path = fresh(dir)?;
    let mut store = MemoryStore::open(&path, KernelPolicy::default())?;
    let mut stale_hits = 0;
    let mut latest_hits = 0;
    let mut model_rejections = 0;
    let mut link_hits = 0;
    let mut expected = Vec::new();
    for i in 0..cases {
        let subject = format!("service{i}topic");
        let key = format!("{subject}.region");
        let mut predecessor = None;
        for (version, value) in ["oldregion", "middlezone", "newlocation"]
            .iter()
            .enumerate()
        {
            let id = format!("{i}-v{version}");
            let mut trace = make_trace(
                &id,
                &format!("{subject} {value}"),
                "synthetic:correction",
                if version == 0 {
                    TraceKind::Belief
                } else {
                    TraceKind::Correction
                },
            );
            trace.claim_key = Some(key.clone());
            trace.claim_value = Some((*value).into());
            trace.supersedes = predecessor;
            store.remember(trace)?;
            predecessor = Some(id);
        }
        let mut malicious = make_trace(
            &format!("{i}-model"),
            "invented overwrite",
            "model:fixture",
            TraceKind::Correction,
        );
        malicious.supersedes = predecessor;
        model_rejections += usize::from(store.remember(malicious).is_err());
        let results = store.recall(&RecallQuery::new(&subject));
        stale_hits += results
            .iter()
            .filter(|r| r.trace.id == format!("{i}-v0") || r.trace.id == format!("{i}-v1"))
            .count();
        latest_hits += usize::from(results.iter().any(|r| r.trace.id == format!("{i}-v2")));
        link_hits +=
            usize::from(results.iter().any(|r| {
                r.trace.id == format!("{i}-v2") && r.supersedes == Some(format!("{i}-v1"))
            }));
        expected.push((subject, format!("{i}-v2")));
    }
    let old_only_results = store
        .recall(&RecallQuery::new("oldregion middlezone"))
        .len();
    drop(store);
    let store = MemoryStore::open_existing(&path, KernelPolicy::default())?;
    store.verify()?;
    let restart_hits = expected
        .iter()
        .filter(|(q, id)| {
            store
                .recall(&RecallQuery::new(q))
                .iter()
                .any(|r| &r.trace.id == id)
        })
        .count();
    let provenance = store
        .all()
        .filter(|e| {
            e.trace.provenance.source == "synthetic:correction" && !e.trace.provenance.verified
        })
        .count();
    let ok = stale_hits == 0
        && latest_hits == cases
        && model_rejections == cases
        && link_hits == cases
        && old_only_results == 0
        && restart_hits == cases
        && provenance == 3 * cases;
    emit(
        json!({"kind": "synthetic_corrections", "cases": cases, "ok": ok,
        "stale_hits": stale_hits, "latest_hits": latest_hits, "supersession_link_hits": link_hits,
        "model_overwrite_rejections": model_rejections, "old_only_query_results": old_only_results,
        "latest_hits_after_restart": restart_hits, "provenance_retained": provenance,
        "limits": ["Generated structured histories; not extracted real incident corrections.",
          "Subject variants of one generated template, not independent real incidents or a proof across all histories.",
          "A query matching only retired terms abstains; no semantic alias expansion is implied."]}),
    )?;
    if !ok {
        return Err("correction invariant failure".into());
    }
    Ok(())
}

examples/bench_kernel.rs (line 134)

fn scale(dir: &Path, records: usize, queries: usize) -> Result<Value> {
    if records == 0 || queries == 0 {
        return Err("records and queries must be positive".into());
    }
    let path = fresh(dir)?;
    let mut store = MemoryStore::open(&path, KernelPolicy::default())?;
    let single_count = records.min(50);
    let batched = records - single_count;
    let mut batches = Vec::new();
    let seed_started = Instant::now();
    for begin in (0..batched).step_by(1000) {
        let traces = (begin..(begin + 1000).min(batched)).map(trace).collect();
        let start = Instant::now();
        store.remember_many(traces)?;
        batches.push(ms(start));
    }
    let mut appends = Vec::new();
    for i in batched..records {
        let t = trace(i);
        let start = Instant::now();
        store.remember(t)?;
        appends.push(ms(start));
    }
    let seed_ms = ms(seed_started);
    let memory_after_seed = rss();
    let mut recall_ms = Vec::new();
    let mut hits = 0;
    let mut returned = 0;
    let mut result_ids = Vec::new();
    // Warm up with an exact cue; quality here is a synthetic exact-cue check, not field recall.
    std::hint::black_box(store.recall(&RecallQuery::new(token(0))));
    for i in 0..queries {
        let selected = i * records / queries;
        let mut query = RecallQuery::new(token(selected));
        query.now_ms = CLOCK;
        query.limit = 1;
        let start = Instant::now();
        let found = store.recall(&query);
        recall_ms.push(ms(start));
        returned += found.len();
        hits += usize::from(
            found
                .first()
                .is_some_and(|r| r.trace.id == format!("sample-{selected}")),
        );
        result_ids.push(found.iter().map(|r| r.trace.id.clone()).collect::<Vec<_>>());
    }
    let start = Instant::now();
    store.verify()?;
    let verify_ms = ms(start);
    let before = store.head();
    drop(store);
    let start = Instant::now();
    let store = MemoryStore::open_existing(&path, KernelPolicy::default())?;
    let replay_ms = ms(start);
    store.verify()?;
    let retained = store
        .all()
        .filter(|e| {
            e.trace.provenance.source == "synthetic-kernel-benchmark"
                && e.trace.provenance.observed_at_ms == CLOCK
                && !e.trace.provenance.verified
        })
        .count();
    let mut replay_matches = 0;
    for (i, expected) in result_ids.iter().enumerate() {
        let mut query = RecallQuery::new(token(i * records / queries));
        query.now_ms = CLOCK;
        query.limit = 1;
        let ids: Vec<_> = store
            .recall(&query)
            .into_iter()
            .map(|r| r.trace.id)
            .collect();
        replay_matches += usize::from(ids == *expected);
    }
    let head_same = serde_json::to_value(before)? == serde_json::to_value(store.head())?;
    let ok = store.record_count() == records
        && retained == records
        && hits == queries
        && replay_matches == queries
        && head_same;
    Ok(json!({
        "kind": "synthetic_scale", "ok": ok, "records": records, "queries": queries,
        "workload": "unclaimed observation traces, unique exact lexical cues, fixed clock",
        "seed_ms": seed_ms, "batch_size": 1000, "batched_records": batched,
        "batch_append_ms": quantiles(batches), "single_append_ms": quantiles(appends),
        "recall_ms": quantiles(recall_ms), "verify_ms": verify_ms, "replay_ms": replay_ms,
        "exact_cue_hits": hits, "returned": returned, "recall_at_1": hits as f64 / queries as f64,
        "provenance_retained": retained, "replayed_records": store.record_count(),
        "recall_replay_matches": replay_matches, "head_unchanged": head_same,
        "ledger_bytes": fs::metadata(path)?.len(), "memory_after_seed": memory_after_seed,
        "memory_after_replay": rss(), "policy": store.policy(),
        "limits": ["Batched seeding does not measure per-record durable append latency.",
          "Warm filesystem cache; reopening in the same process is not a cold-start measurement.",
          "No structured claims in this workload; conflict-heavy stores need a separate workload.",
          "Synthetic exact-cue quality is not semantic or real-world retrieval accuracy.",
          "RSS is process-wide Linux /proc data; unavailable platforms report null."]
    }))
}

// Specified, platform-independent SplitMix64; never a cryptographic RNG.
struct Rng(u64);
impl Rng {
    fn next(&mut self) -> u64 {
        self.0 = self.0.wrapping_add(0x9e3779b97f4a7c15);
        let mut x = self.0;
        x = (x ^ (x >> 30)).wrapping_mul(0xbf58476d1ce4e5b9);
        x = (x ^ (x >> 27)).wrapping_mul(0x94d049bb133111eb);
        x ^ (x >> 31)
    }
    fn unit(&mut self) -> f64 {
        (self.next() % 10001) as f64 / 10000.0
    }
}

fn option(name: &str, a: f64, b: f64) -> DecisionOption {
    DecisionOption {
        name: name.into(),
        description: None,
        scores: BTreeMap::from([("a".into(), a), ("b".into(), b)]),
        model_scores: BTreeMap::new(),
        evidence_confidence: 1.0,
        hard_blocks: vec![],
        reversible: false,
    }
}

fn decisions(cases: usize, seed: u64) -> Result<Value> {
    if cases < 6 {
        return Err("at least 6 cases required to exercise every family".into());
    }
    let mut rng = Rng(seed);
    let mut counts: BTreeMap<&str, usize> = BTreeMap::new();
    let mut failures = Vec::new();
    let mut violations = 0;
    let start = Instant::now();
    for i in 0..cases {
        let mut a = option("a", rng.unit(), rng.unit());
        let mut b = option("b", rng.unit(), rng.unit());
        let family;
        let mut expected: Option<Option<String>> = None;
        match i % 6 {
            0 => {
                family = "high_score_hard_block";
                a.scores.values_mut().for_each(|s| *s = 1.0);
                a.hard_blocks.push("no authorization".into());
                expected = Some(Some("b".into()));
            }
            1 => {
                family = "all_blocked";
                a.hard_blocks.push("privacy".into());
                b.hard_blocks.push("consent".into());
                expected = Some(None);
            }
            2 => {
                family = "complete_score_oracle";
                let sum_a: f64 = a.scores.values().sum();
                let sum_b: f64 = b.scores.values().sum();
                expected = Some(Some(if sum_a >= sum_b { "a" } else { "b" }.into()));
            }
            3 => {
                family = "unknowns_and_model_estimates";
                a.scores.clear();
                a.model_scores.insert("a".into(), rng.unit());
            }
            4 => {
                family = "reversibility_band";
                let top = 0.5 + rng.unit() * 0.4;
                let gap = if rng.next().is_multiple_of(2) {
                    0.01
                } else {
                    0.03
                };
                a.scores.values_mut().for_each(|s| *s = top);
                a.evidence_confidence = 0.1;
                b.scores.values_mut().for_each(|s| *s = top - gap);
                b.reversible = true;
                expected = Some(Some(if gap < 0.02 { "b" } else { "a" }.into()));
            }
            _ => {
                family = "invalid_numeric_input";
                a.scores.insert("a".into(), 1.01 + rng.unit());
            }
        }
        *counts.entry(family).or_default() += 1;
        let frame = DecisionFrame {
            question: "synthetic decision".into(),
            criteria: ["a", "b"]
                .into_iter()
                .map(|id| Criterion {
                    id: id.into(),
                    weight: 1.0,
                    description: id.into(),
                })
                .collect(),
            options: vec![a, b],
        };
        let actual = evaluate(&frame);
        let mut reasons = Vec::new();
        if family == "invalid_numeric_input" {
            if actual.is_ok() {
                reasons.push("invalid input admitted");
            }
        } else if let Ok(answer) = &actual {
            if let Some(expected) = expected {
                if answer.recommended != expected {
                    reasons.push("wrong recommendation");
                }
            }
            if let Some(name) = &answer.recommended {
                if frame
                    .options
                    .iter()
                    .any(|o| &o.name == name && !o.hard_blocks.is_empty())
                {
                    reasons.push("hard-blocked option selected");
                }
            }
            if family == "unknowns_and_model_estimates" {
                let ranked = answer.ranking.iter().find(|o| o.name == "a").unwrap();
                if ranked.coverage != 0.0
                    || ranked.evidence_interval != [0.0, 1.0]
                    || ranked.missing_criteria != ["b"]
                    || answer.decisive
                {
                    reasons.push("unknown evidence hidden or model guess counted as evidence");
                }
            }
            let mut permuted = frame.clone();
            permuted.options.reverse();
            permuted.question = "irrelevant wording changed".into();
            let other = evaluate(&permuted)?;
            if answer.recommended != other.recommended
                || serde_json::to_value(&answer.ranking)? != serde_json::to_value(&other.ranking)?
            {
                reasons.push("order or irrelevant text changed result");
            }
        } else {
            reasons.push("valid frame rejected");
        }
        if !reasons.is_empty() {
            violations += 1;
            if failures.len() < 20 {
                failures.push(
                    json!({"case_index": i, "family": family, "reasons": reasons,
                    "frame": frame, "actual": actual.as_ref().ok()}),
                );
            }
        }
    }
    Ok(
        json!({"kind": "synthetic_decision_invariants", "seed": seed, "rng": "splitmix64",
        "cases": cases, "families": counts, "violations": violations,
        "violation_rate": violations as f64 / cases as f64, "ok": violations == 0,
        "elapsed_ms": ms(start), "counterexamples": failures,
        "limits": ["Generated families are not real decision-quality labels or a proof over all inputs.",
          "Metamorphic checks keep criteria fixed; weight perturbations require separate expectations."]}),
    )
}

fn committed_child(dir: &Path) -> Result<()> {
    let path = fresh(dir)?;
    let mut store = MemoryStore::open(path, KernelPolicy::default())?;
    store.remember_many((0..50).map(trace).collect())?;
    println!("committed:50");
    io::stdout().flush()?;
    // Parent kills us after the acknowledged commit; this is not an in-flight crash claim.
    loop {
        std::thread::park();
    }
}

fn recovery(dir: &Path) -> Result<Value> {
    fs::create_dir(dir)?;
    let child_dir = dir.join("committed");
    let mut child = Command::new(std::env::current_exe()?)
        .args(["committed-child", "--dir"])
        .arg(&child_dir)
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .spawn()?;
    let mut ack = String::new();
    let read_result = io::BufReader::new(child.stdout.take().unwrap()).read_line(&mut ack);
    let kill_result = child.kill();
    let wait_result = child.wait();
    read_result?;
    kill_result?;
    wait_result?;
    if ack.trim() != "committed:50" {
        return Err("child did not acknowledge commit".into());
    }
    let path = child_dir.join("memory.jsonl");
    let start = Instant::now();
    let store = MemoryStore::open_existing(&path, KernelPolicy::default())?;
    store.verify()?;
    let restart_ms = ms(start);
    let acknowledged_survived =
        store.record_count() == 50 && (0..50).all(|i| store.get(&format!("sample-{i}")).is_some());
    let head_path = store.head_path();
    drop(store);
    let original = fs::read(&path)?;
    let original_head = fs::read(head_path)?;
    let copied_store = |name: &str, bytes: &[u8]| -> Result<PathBuf> {
        let target = fresh(&dir.join(name))?;
        let store = MemoryStore::open(&target, KernelPolicy::default())?;
        let target_head = store.head_path();
        drop(store);
        fs::write(&target, bytes)?;
        fs::write(target_head, &original_head)?;
        Ok(target)
    };
    let torn = copied_store("torn", &original)?;
    OpenOptions::new()
        .append(true)
        .open(&torn)?
        .write_all(b"{\"seq\":")?;
    let mut store = MemoryStore::open_existing(&torn, KernelPolicy::default())?;
    store.remember(trace(50))?;
    store.verify()?;
    let torn_repaired = store.record_count() == 51;
    let mut edited = String::from_utf8(original.clone())?;
    edited = edited.replacen("deployment observation", "deployment fabrication", 1);
    let tampered = copied_store("edited", edited.as_bytes())?;
    let rejected = |path: &Path| match MemoryStore::open_existing(path, KernelPolicy::default()) {
        Err(_) => true,
        Ok(store) => store.verify().is_err(),
    };
    let edit_detected = rejected(&tampered);
    let last = original[..original.len() - 1]
        .iter()
        .rposition(|b| *b == b'\n')
        .unwrap()
        + 1;
    let truncated = copied_store("truncated", &original[..last])?;
    let truncation_detected = rejected(&truncated);
    Ok(
        json!({"kind": "synthetic_recovery", "acknowledged_records": 50,
        "acknowledged_survived": acknowledged_survived, "restart_ms": restart_ms,
        "torn_tail_repaired": torn_repaired, "edit_detected": edit_detected,
        "truncation_detected": truncation_detected,
        "ok": acknowledged_survived && torn_repaired && edit_detected && truncation_detected,
        "limits": ["Process killed after an acknowledged commit, not during fsync or head replacement.",
          "Torn tail, edit and truncation are injected into separate copies.",
          "This does not simulate power loss or an attacker rewriting both ledger and head."]}),
    )
}

pub fn remember_many( &mut self, traces: Vec<Trace>, ) -> Result<Vec<(Trace, Vec<ClaimConflict>)>>

Append several traces as one batch: one lock, one validation pass, one write, so either every trace is stored or none is. Each trace gets the same checks as Self::remember against the stored state; an outcome must resolve a prediction that is already stored. Conflicts are reported against stored traces, not between traces of the same batch.

Examples found in repository

examples/retrieval_bench.rs (line 176)

fn retrieve(corpus: &Path, queries: &Path, dir: &Path, dataset: &str, k: usize) -> Result<()> {
    if !(1..=100).contains(&k) || dataset.trim().is_empty() {
        return Err("k must be 1..100 and dataset must be nonempty".into());
    }
    let policy = KernelPolicy {
        recall_limit: k,
        ..KernelPolicy::default()
    };
    let path = fresh(dir)?;
    let source = format!("beir:{dataset}");
    let mut store = MemoryStore::open(&path, policy.clone())?;
    let mut baseline = Bm25::default();
    let mut ids = HashSet::new();
    let mut rejected = Vec::new();
    let mut batch = Vec::new();
    let mut baseline_index_ms = 0.0;
    let ingest = Instant::now();
    for line in BufReader::new(File::open(corpus)?).lines() {
        let doc: Document = serde_json::from_str(&line?)?;
        if doc.id.trim().is_empty() || !ids.insert(doc.id.clone()) {
            return Err("empty or duplicate document id".into());
        }
        let trace = make_trace(&doc.id, &doc.content, &source, TraceKind::Observation);
        if let Err(reason) = trace.validate() {
            rejected.push(json!({"id": doc.id, "reason": reason.to_string()}));
            continue;
        }
        let timer = Instant::now();
        baseline.add(&doc);
        baseline_index_ms += ms(timer);
        batch.push(trace);
        if batch.len() == 1000 {
            store.remember_many(std::mem::take(&mut batch))?;
        }
    }
    if !batch.is_empty() {
        store.remember_many(batch)?;
    }
    let ingest_ms = ms(ingest);
    store.verify()?;
    let head_before = serde_json::to_value(store.head())?;
    drop(store);
    let start = Instant::now();
    let store = MemoryStore::open_existing(&path, policy)?;
    let reopen_ms = ms(start);
    store.verify()?;
    let provenance_retained = store
        .all()
        .filter(|entry| {
            let t = &entry.trace;
            t.provenance.source == source
                && !t.provenance.verified
                && t.created_at_ms == CLOCK
                && t.provenance.observed_at_ms == CLOCK
                && t.provenance.locator.as_deref() == Some(format!("{source}:{}", t.id).as_str())
        })
        .count();
    if store.record_count() != baseline.ids.len()
        || provenance_retained != baseline.ids.len()
        || serde_json::to_value(store.head())? != head_before
    {
        return Err("replay, head or provenance invariant failed".into());
    }
    emit(
        json!({"kind": "loaded", "corpus_total": ids.len(), "admitted": baseline.ids.len(),
        "rejected": rejected, "provenance_retained": provenance_retained,
        "ingest_both_ms": ingest_ms, "bm25_index_ms": baseline_index_ms, "reopen_ms": reopen_ms,
        "policy": store.policy(), "bm25": {"k1": 1.2, "b": 0.75}, "query_clock_ms": CLOCK}),
    )?;
    let mut query_ids = HashSet::new();
    for line in BufReader::new(File::open(queries)?).lines() {
        let q: Query = serde_json::from_str(&line?)?;
        if q.id.trim().is_empty() || !query_ids.insert(q.id.clone()) {
            return Err("invalid query id".into());
        }
        let mut query = RecallQuery::new(&q.text);
        query.now_ms = CLOCK;
        query.limit = k;
        let start = Instant::now();
        let recalled = store.recall(&query);
        let hikmah_ms = ms(start);
        let start = Instant::now();
        let bm25 = baseline.search(&q.text, k);
        let bm25_ms = ms(start);
        let hikmah: Vec<_> = recalled.into_iter().map(|r| json!({
            "id": r.trace.id, "score": r.score, "duplicates_folded": r.duplicates,
            "provenance_source": r.trace.provenance.source, "verified": r.trace.provenance.verified,
        })).collect();
        emit(
            json!({"kind": "query", "id": q.id, "hikmah": hikmah, "bm25": bm25,
                    "hikmah_ms": hikmah_ms, "bm25_ms": bm25_ms}),
        )?;
    }
    emit(json!({"kind": "finished", "queries": query_ids.len()}))
}

examples/bench_kernel.rs (line 127)

fn scale(dir: &Path, records: usize, queries: usize) -> Result<Value> {
    if records == 0 || queries == 0 {
        return Err("records and queries must be positive".into());
    }
    let path = fresh(dir)?;
    let mut store = MemoryStore::open(&path, KernelPolicy::default())?;
    let single_count = records.min(50);
    let batched = records - single_count;
    let mut batches = Vec::new();
    let seed_started = Instant::now();
    for begin in (0..batched).step_by(1000) {
        let traces = (begin..(begin + 1000).min(batched)).map(trace).collect();
        let start = Instant::now();
        store.remember_many(traces)?;
        batches.push(ms(start));
    }
    let mut appends = Vec::new();
    for i in batched..records {
        let t = trace(i);
        let start = Instant::now();
        store.remember(t)?;
        appends.push(ms(start));
    }
    let seed_ms = ms(seed_started);
    let memory_after_seed = rss();
    let mut recall_ms = Vec::new();
    let mut hits = 0;
    let mut returned = 0;
    let mut result_ids = Vec::new();
    // Warm up with an exact cue; quality here is a synthetic exact-cue check, not field recall.
    std::hint::black_box(store.recall(&RecallQuery::new(token(0))));
    for i in 0..queries {
        let selected = i * records / queries;
        let mut query = RecallQuery::new(token(selected));
        query.now_ms = CLOCK;
        query.limit = 1;
        let start = Instant::now();
        let found = store.recall(&query);
        recall_ms.push(ms(start));
        returned += found.len();
        hits += usize::from(
            found
                .first()
                .is_some_and(|r| r.trace.id == format!("sample-{selected}")),
        );
        result_ids.push(found.iter().map(|r| r.trace.id.clone()).collect::<Vec<_>>());
    }
    let start = Instant::now();
    store.verify()?;
    let verify_ms = ms(start);
    let before = store.head();
    drop(store);
    let start = Instant::now();
    let store = MemoryStore::open_existing(&path, KernelPolicy::default())?;
    let replay_ms = ms(start);
    store.verify()?;
    let retained = store
        .all()
        .filter(|e| {
            e.trace.provenance.source == "synthetic-kernel-benchmark"
                && e.trace.provenance.observed_at_ms == CLOCK
                && !e.trace.provenance.verified
        })
        .count();
    let mut replay_matches = 0;
    for (i, expected) in result_ids.iter().enumerate() {
        let mut query = RecallQuery::new(token(i * records / queries));
        query.now_ms = CLOCK;
        query.limit = 1;
        let ids: Vec<_> = store
            .recall(&query)
            .into_iter()
            .map(|r| r.trace.id)
            .collect();
        replay_matches += usize::from(ids == *expected);
    }
    let head_same = serde_json::to_value(before)? == serde_json::to_value(store.head())?;
    let ok = store.record_count() == records
        && retained == records
        && hits == queries
        && replay_matches == queries
        && head_same;
    Ok(json!({
        "kind": "synthetic_scale", "ok": ok, "records": records, "queries": queries,
        "workload": "unclaimed observation traces, unique exact lexical cues, fixed clock",
        "seed_ms": seed_ms, "batch_size": 1000, "batched_records": batched,
        "batch_append_ms": quantiles(batches), "single_append_ms": quantiles(appends),
        "recall_ms": quantiles(recall_ms), "verify_ms": verify_ms, "replay_ms": replay_ms,
        "exact_cue_hits": hits, "returned": returned, "recall_at_1": hits as f64 / queries as f64,
        "provenance_retained": retained, "replayed_records": store.record_count(),
        "recall_replay_matches": replay_matches, "head_unchanged": head_same,
        "ledger_bytes": fs::metadata(path)?.len(), "memory_after_seed": memory_after_seed,
        "memory_after_replay": rss(), "policy": store.policy(),
        "limits": ["Batched seeding does not measure per-record durable append latency.",
          "Warm filesystem cache; reopening in the same process is not a cold-start measurement.",
          "No structured claims in this workload; conflict-heavy stores need a separate workload.",
          "Synthetic exact-cue quality is not semantic or real-world retrieval accuracy.",
          "RSS is process-wide Linux /proc data; unavailable platforms report null."]
    }))
}

// Specified, platform-independent SplitMix64; never a cryptographic RNG.
struct Rng(u64);
impl Rng {
    fn next(&mut self) -> u64 {
        self.0 = self.0.wrapping_add(0x9e3779b97f4a7c15);
        let mut x = self.0;
        x = (x ^ (x >> 30)).wrapping_mul(0xbf58476d1ce4e5b9);
        x = (x ^ (x >> 27)).wrapping_mul(0x94d049bb133111eb);
        x ^ (x >> 31)
    }
    fn unit(&mut self) -> f64 {
        (self.next() % 10001) as f64 / 10000.0
    }
}

fn option(name: &str, a: f64, b: f64) -> DecisionOption {
    DecisionOption {
        name: name.into(),
        description: None,
        scores: BTreeMap::from([("a".into(), a), ("b".into(), b)]),
        model_scores: BTreeMap::new(),
        evidence_confidence: 1.0,
        hard_blocks: vec![],
        reversible: false,
    }
}

fn decisions(cases: usize, seed: u64) -> Result<Value> {
    if cases < 6 {
        return Err("at least 6 cases required to exercise every family".into());
    }
    let mut rng = Rng(seed);
    let mut counts: BTreeMap<&str, usize> = BTreeMap::new();
    let mut failures = Vec::new();
    let mut violations = 0;
    let start = Instant::now();
    for i in 0..cases {
        let mut a = option("a", rng.unit(), rng.unit());
        let mut b = option("b", rng.unit(), rng.unit());
        let family;
        let mut expected: Option<Option<String>> = None;
        match i % 6 {
            0 => {
                family = "high_score_hard_block";
                a.scores.values_mut().for_each(|s| *s = 1.0);
                a.hard_blocks.push("no authorization".into());
                expected = Some(Some("b".into()));
            }
            1 => {
                family = "all_blocked";
                a.hard_blocks.push("privacy".into());
                b.hard_blocks.push("consent".into());
                expected = Some(None);
            }
            2 => {
                family = "complete_score_oracle";
                let sum_a: f64 = a.scores.values().sum();
                let sum_b: f64 = b.scores.values().sum();
                expected = Some(Some(if sum_a >= sum_b { "a" } else { "b" }.into()));
            }
            3 => {
                family = "unknowns_and_model_estimates";
                a.scores.clear();
                a.model_scores.insert("a".into(), rng.unit());
            }
            4 => {
                family = "reversibility_band";
                let top = 0.5 + rng.unit() * 0.4;
                let gap = if rng.next().is_multiple_of(2) {
                    0.01
                } else {
                    0.03
                };
                a.scores.values_mut().for_each(|s| *s = top);
                a.evidence_confidence = 0.1;
                b.scores.values_mut().for_each(|s| *s = top - gap);
                b.reversible = true;
                expected = Some(Some(if gap < 0.02 { "b" } else { "a" }.into()));
            }
            _ => {
                family = "invalid_numeric_input";
                a.scores.insert("a".into(), 1.01 + rng.unit());
            }
        }
        *counts.entry(family).or_default() += 1;
        let frame = DecisionFrame {
            question: "synthetic decision".into(),
            criteria: ["a", "b"]
                .into_iter()
                .map(|id| Criterion {
                    id: id.into(),
                    weight: 1.0,
                    description: id.into(),
                })
                .collect(),
            options: vec![a, b],
        };
        let actual = evaluate(&frame);
        let mut reasons = Vec::new();
        if family == "invalid_numeric_input" {
            if actual.is_ok() {
                reasons.push("invalid input admitted");
            }
        } else if let Ok(answer) = &actual {
            if let Some(expected) = expected {
                if answer.recommended != expected {
                    reasons.push("wrong recommendation");
                }
            }
            if let Some(name) = &answer.recommended {
                if frame
                    .options
                    .iter()
                    .any(|o| &o.name == name && !o.hard_blocks.is_empty())
                {
                    reasons.push("hard-blocked option selected");
                }
            }
            if family == "unknowns_and_model_estimates" {
                let ranked = answer.ranking.iter().find(|o| o.name == "a").unwrap();
                if ranked.coverage != 0.0
                    || ranked.evidence_interval != [0.0, 1.0]
                    || ranked.missing_criteria != ["b"]
                    || answer.decisive
                {
                    reasons.push("unknown evidence hidden or model guess counted as evidence");
                }
            }
            let mut permuted = frame.clone();
            permuted.options.reverse();
            permuted.question = "irrelevant wording changed".into();
            let other = evaluate(&permuted)?;
            if answer.recommended != other.recommended
                || serde_json::to_value(&answer.ranking)? != serde_json::to_value(&other.ranking)?
            {
                reasons.push("order or irrelevant text changed result");
            }
        } else {
            reasons.push("valid frame rejected");
        }
        if !reasons.is_empty() {
            violations += 1;
            if failures.len() < 20 {
                failures.push(
                    json!({"case_index": i, "family": family, "reasons": reasons,
                    "frame": frame, "actual": actual.as_ref().ok()}),
                );
            }
        }
    }
    Ok(
        json!({"kind": "synthetic_decision_invariants", "seed": seed, "rng": "splitmix64",
        "cases": cases, "families": counts, "violations": violations,
        "violation_rate": violations as f64 / cases as f64, "ok": violations == 0,
        "elapsed_ms": ms(start), "counterexamples": failures,
        "limits": ["Generated families are not real decision-quality labels or a proof over all inputs.",
          "Metamorphic checks keep criteria fixed; weight perturbations require separate expectations."]}),
    )
}

fn committed_child(dir: &Path) -> Result<()> {
    let path = fresh(dir)?;
    let mut store = MemoryStore::open(path, KernelPolicy::default())?;
    store.remember_many((0..50).map(trace).collect())?;
    println!("committed:50");
    io::stdout().flush()?;
    // Parent kills us after the acknowledged commit; this is not an in-flight crash claim.
    loop {
        std::thread::park();
    }
}

pub fn fulfill(&mut self, id: impl Into<String>) -> Result<()>

pub fn purge( &mut self, id: impl Into<String>, reason: impl Into<String>, ) -> Result<()>

Tombstone a trace: it leaves recall, consolidation, and commitments. The content stays in the append-only ledger; this is not erasure.

pub fn get(&self, id: &str) -> Option<&TraceEntry>

Examples found in repository

examples/bench_kernel.rs (line 412)

fn recovery(dir: &Path) -> Result<Value> {
    fs::create_dir(dir)?;
    let child_dir = dir.join("committed");
    let mut child = Command::new(std::env::current_exe()?)
        .args(["committed-child", "--dir"])
        .arg(&child_dir)
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .spawn()?;
    let mut ack = String::new();
    let read_result = io::BufReader::new(child.stdout.take().unwrap()).read_line(&mut ack);
    let kill_result = child.kill();
    let wait_result = child.wait();
    read_result?;
    kill_result?;
    wait_result?;
    if ack.trim() != "committed:50" {
        return Err("child did not acknowledge commit".into());
    }
    let path = child_dir.join("memory.jsonl");
    let start = Instant::now();
    let store = MemoryStore::open_existing(&path, KernelPolicy::default())?;
    store.verify()?;
    let restart_ms = ms(start);
    let acknowledged_survived =
        store.record_count() == 50 && (0..50).all(|i| store.get(&format!("sample-{i}")).is_some());
    let head_path = store.head_path();
    drop(store);
    let original = fs::read(&path)?;
    let original_head = fs::read(head_path)?;
    let copied_store = |name: &str, bytes: &[u8]| -> Result<PathBuf> {
        let target = fresh(&dir.join(name))?;
        let store = MemoryStore::open(&target, KernelPolicy::default())?;
        let target_head = store.head_path();
        drop(store);
        fs::write(&target, bytes)?;
        fs::write(target_head, &original_head)?;
        Ok(target)
    };
    let torn = copied_store("torn", &original)?;
    OpenOptions::new()
        .append(true)
        .open(&torn)?
        .write_all(b"{\"seq\":")?;
    let mut store = MemoryStore::open_existing(&torn, KernelPolicy::default())?;
    store.remember(trace(50))?;
    store.verify()?;
    let torn_repaired = store.record_count() == 51;
    let mut edited = String::from_utf8(original.clone())?;
    edited = edited.replacen("deployment observation", "deployment fabrication", 1);
    let tampered = copied_store("edited", edited.as_bytes())?;
    let rejected = |path: &Path| match MemoryStore::open_existing(path, KernelPolicy::default()) {
        Err(_) => true,
        Ok(store) => store.verify().is_err(),
    };
    let edit_detected = rejected(&tampered);
    let last = original[..original.len() - 1]
        .iter()
        .rposition(|b| *b == b'\n')
        .unwrap()
        + 1;
    let truncated = copied_store("truncated", &original[..last])?;
    let truncation_detected = rejected(&truncated);
    Ok(
        json!({"kind": "synthetic_recovery", "acknowledged_records": 50,
        "acknowledged_survived": acknowledged_survived, "restart_ms": restart_ms,
        "torn_tail_repaired": torn_repaired, "edit_detected": edit_detected,
        "truncation_detected": truncation_detected,
        "ok": acknowledged_survived && torn_repaired && edit_detected && truncation_detected,
        "limits": ["Process killed after an acknowledged commit, not during fsync or head replacement.",
          "Torn tail, edit and truncation are injected into separate copies.",
          "This does not simulate power loss or an attacker rewriting both ledger and head."]}),
    )
}

pub fn all(&self) -> impl Iterator<Item = &TraceEntry>

Examples found in repository

examples/retrieval_bench.rs (line 191)

fn retrieve(corpus: &Path, queries: &Path, dir: &Path, dataset: &str, k: usize) -> Result<()> {
    if !(1..=100).contains(&k) || dataset.trim().is_empty() {
        return Err("k must be 1..100 and dataset must be nonempty".into());
    }
    let policy = KernelPolicy {
        recall_limit: k,
        ..KernelPolicy::default()
    };
    let path = fresh(dir)?;
    let source = format!("beir:{dataset}");
    let mut store = MemoryStore::open(&path, policy.clone())?;
    let mut baseline = Bm25::default();
    let mut ids = HashSet::new();
    let mut rejected = Vec::new();
    let mut batch = Vec::new();
    let mut baseline_index_ms = 0.0;
    let ingest = Instant::now();
    for line in BufReader::new(File::open(corpus)?).lines() {
        let doc: Document = serde_json::from_str(&line?)?;
        if doc.id.trim().is_empty() || !ids.insert(doc.id.clone()) {
            return Err("empty or duplicate document id".into());
        }
        let trace = make_trace(&doc.id, &doc.content, &source, TraceKind::Observation);
        if let Err(reason) = trace.validate() {
            rejected.push(json!({"id": doc.id, "reason": reason.to_string()}));
            continue;
        }
        let timer = Instant::now();
        baseline.add(&doc);
        baseline_index_ms += ms(timer);
        batch.push(trace);
        if batch.len() == 1000 {
            store.remember_many(std::mem::take(&mut batch))?;
        }
    }
    if !batch.is_empty() {
        store.remember_many(batch)?;
    }
    let ingest_ms = ms(ingest);
    store.verify()?;
    let head_before = serde_json::to_value(store.head())?;
    drop(store);
    let start = Instant::now();
    let store = MemoryStore::open_existing(&path, policy)?;
    let reopen_ms = ms(start);
    store.verify()?;
    let provenance_retained = store
        .all()
        .filter(|entry| {
            let t = &entry.trace;
            t.provenance.source == source
                && !t.provenance.verified
                && t.created_at_ms == CLOCK
                && t.provenance.observed_at_ms == CLOCK
                && t.provenance.locator.as_deref() == Some(format!("{source}:{}", t.id).as_str())
        })
        .count();
    if store.record_count() != baseline.ids.len()
        || provenance_retained != baseline.ids.len()
        || serde_json::to_value(store.head())? != head_before
    {
        return Err("replay, head or provenance invariant failed".into());
    }
    emit(
        json!({"kind": "loaded", "corpus_total": ids.len(), "admitted": baseline.ids.len(),
        "rejected": rejected, "provenance_retained": provenance_retained,
        "ingest_both_ms": ingest_ms, "bm25_index_ms": baseline_index_ms, "reopen_ms": reopen_ms,
        "policy": store.policy(), "bm25": {"k1": 1.2, "b": 0.75}, "query_clock_ms": CLOCK}),
    )?;
    let mut query_ids = HashSet::new();
    for line in BufReader::new(File::open(queries)?).lines() {
        let q: Query = serde_json::from_str(&line?)?;
        if q.id.trim().is_empty() || !query_ids.insert(q.id.clone()) {
            return Err("invalid query id".into());
        }
        let mut query = RecallQuery::new(&q.text);
        query.now_ms = CLOCK;
        query.limit = k;
        let start = Instant::now();
        let recalled = store.recall(&query);
        let hikmah_ms = ms(start);
        let start = Instant::now();
        let bm25 = baseline.search(&q.text, k);
        let bm25_ms = ms(start);
        let hikmah: Vec<_> = recalled.into_iter().map(|r| json!({
            "id": r.trace.id, "score": r.score, "duplicates_folded": r.duplicates,
            "provenance_source": r.trace.provenance.source, "verified": r.trace.provenance.verified,
        })).collect();
        emit(
            json!({"kind": "query", "id": q.id, "hikmah": hikmah, "bm25": bm25,
                    "hikmah_ms": hikmah_ms, "bm25_ms": bm25_ms}),
        )?;
    }
    emit(json!({"kind": "finished", "queries": query_ids.len()}))
}

fn corrections(dir: &Path, cases: usize) -> Result<()> {
    if !(1..=10000).contains(&cases) {
        return Err("cases must be 1..10000".into());
    }
    let path = fresh(dir)?;
    let mut store = MemoryStore::open(&path, KernelPolicy::default())?;
    let mut stale_hits = 0;
    let mut latest_hits = 0;
    let mut model_rejections = 0;
    let mut link_hits = 0;
    let mut expected = Vec::new();
    for i in 0..cases {
        let subject = format!("service{i}topic");
        let key = format!("{subject}.region");
        let mut predecessor = None;
        for (version, value) in ["oldregion", "middlezone", "newlocation"]
            .iter()
            .enumerate()
        {
            let id = format!("{i}-v{version}");
            let mut trace = make_trace(
                &id,
                &format!("{subject} {value}"),
                "synthetic:correction",
                if version == 0 {
                    TraceKind::Belief
                } else {
                    TraceKind::Correction
                },
            );
            trace.claim_key = Some(key.clone());
            trace.claim_value = Some((*value).into());
            trace.supersedes = predecessor;
            store.remember(trace)?;
            predecessor = Some(id);
        }
        let mut malicious = make_trace(
            &format!("{i}-model"),
            "invented overwrite",
            "model:fixture",
            TraceKind::Correction,
        );
        malicious.supersedes = predecessor;
        model_rejections += usize::from(store.remember(malicious).is_err());
        let results = store.recall(&RecallQuery::new(&subject));
        stale_hits += results
            .iter()
            .filter(|r| r.trace.id == format!("{i}-v0") || r.trace.id == format!("{i}-v1"))
            .count();
        latest_hits += usize::from(results.iter().any(|r| r.trace.id == format!("{i}-v2")));
        link_hits +=
            usize::from(results.iter().any(|r| {
                r.trace.id == format!("{i}-v2") && r.supersedes == Some(format!("{i}-v1"))
            }));
        expected.push((subject, format!("{i}-v2")));
    }
    let old_only_results = store
        .recall(&RecallQuery::new("oldregion middlezone"))
        .len();
    drop(store);
    let store = MemoryStore::open_existing(&path, KernelPolicy::default())?;
    store.verify()?;
    let restart_hits = expected
        .iter()
        .filter(|(q, id)| {
            store
                .recall(&RecallQuery::new(q))
                .iter()
                .any(|r| &r.trace.id == id)
        })
        .count();
    let provenance = store
        .all()
        .filter(|e| {
            e.trace.provenance.source == "synthetic:correction" && !e.trace.provenance.verified
        })
        .count();
    let ok = stale_hits == 0
        && latest_hits == cases
        && model_rejections == cases
        && link_hits == cases
        && old_only_results == 0
        && restart_hits == cases
        && provenance == 3 * cases;
    emit(
        json!({"kind": "synthetic_corrections", "cases": cases, "ok": ok,
        "stale_hits": stale_hits, "latest_hits": latest_hits, "supersession_link_hits": link_hits,
        "model_overwrite_rejections": model_rejections, "old_only_query_results": old_only_results,
        "latest_hits_after_restart": restart_hits, "provenance_retained": provenance,
        "limits": ["Generated structured histories; not extracted real incident corrections.",
          "Subject variants of one generated template, not independent real incidents or a proof across all histories.",
          "A query matching only retired terms abstains; no semantic alias expansion is implied."]}),
    )?;
    if !ok {
        return Err("correction invariant failure".into());
    }
    Ok(())
}

examples/bench_kernel.rs (line 171)

fn scale(dir: &Path, records: usize, queries: usize) -> Result<Value> {
    if records == 0 || queries == 0 {
        return Err("records and queries must be positive".into());
    }
    let path = fresh(dir)?;
    let mut store = MemoryStore::open(&path, KernelPolicy::default())?;
    let single_count = records.min(50);
    let batched = records - single_count;
    let mut batches = Vec::new();
    let seed_started = Instant::now();
    for begin in (0..batched).step_by(1000) {
        let traces = (begin..(begin + 1000).min(batched)).map(trace).collect();
        let start = Instant::now();
        store.remember_many(traces)?;
        batches.push(ms(start));
    }
    let mut appends = Vec::new();
    for i in batched..records {
        let t = trace(i);
        let start = Instant::now();
        store.remember(t)?;
        appends.push(ms(start));
    }
    let seed_ms = ms(seed_started);
    let memory_after_seed = rss();
    let mut recall_ms = Vec::new();
    let mut hits = 0;
    let mut returned = 0;
    let mut result_ids = Vec::new();
    // Warm up with an exact cue; quality here is a synthetic exact-cue check, not field recall.
    std::hint::black_box(store.recall(&RecallQuery::new(token(0))));
    for i in 0..queries {
        let selected = i * records / queries;
        let mut query = RecallQuery::new(token(selected));
        query.now_ms = CLOCK;
        query.limit = 1;
        let start = Instant::now();
        let found = store.recall(&query);
        recall_ms.push(ms(start));
        returned += found.len();
        hits += usize::from(
            found
                .first()
                .is_some_and(|r| r.trace.id == format!("sample-{selected}")),
        );
        result_ids.push(found.iter().map(|r| r.trace.id.clone()).collect::<Vec<_>>());
    }
    let start = Instant::now();
    store.verify()?;
    let verify_ms = ms(start);
    let before = store.head();
    drop(store);
    let start = Instant::now();
    let store = MemoryStore::open_existing(&path, KernelPolicy::default())?;
    let replay_ms = ms(start);
    store.verify()?;
    let retained = store
        .all()
        .filter(|e| {
            e.trace.provenance.source == "synthetic-kernel-benchmark"
                && e.trace.provenance.observed_at_ms == CLOCK
                && !e.trace.provenance.verified
        })
        .count();
    let mut replay_matches = 0;
    for (i, expected) in result_ids.iter().enumerate() {
        let mut query = RecallQuery::new(token(i * records / queries));
        query.now_ms = CLOCK;
        query.limit = 1;
        let ids: Vec<_> = store
            .recall(&query)
            .into_iter()
            .map(|r| r.trace.id)
            .collect();
        replay_matches += usize::from(ids == *expected);
    }
    let head_same = serde_json::to_value(before)? == serde_json::to_value(store.head())?;
    let ok = store.record_count() == records
        && retained == records
        && hits == queries
        && replay_matches == queries
        && head_same;
    Ok(json!({
        "kind": "synthetic_scale", "ok": ok, "records": records, "queries": queries,
        "workload": "unclaimed observation traces, unique exact lexical cues, fixed clock",
        "seed_ms": seed_ms, "batch_size": 1000, "batched_records": batched,
        "batch_append_ms": quantiles(batches), "single_append_ms": quantiles(appends),
        "recall_ms": quantiles(recall_ms), "verify_ms": verify_ms, "replay_ms": replay_ms,
        "exact_cue_hits": hits, "returned": returned, "recall_at_1": hits as f64 / queries as f64,
        "provenance_retained": retained, "replayed_records": store.record_count(),
        "recall_replay_matches": replay_matches, "head_unchanged": head_same,
        "ledger_bytes": fs::metadata(path)?.len(), "memory_after_seed": memory_after_seed,
        "memory_after_replay": rss(), "policy": store.policy(),
        "limits": ["Batched seeding does not measure per-record durable append latency.",
          "Warm filesystem cache; reopening in the same process is not a cold-start measurement.",
          "No structured claims in this workload; conflict-heavy stores need a separate workload.",
          "Synthetic exact-cue quality is not semantic or real-world retrieval accuracy.",
          "RSS is process-wide Linux /proc data; unavailable platforms report null."]
    }))
}

pub fn active_traces(&self) -> impl Iterator<Item = &Trace>

pub fn policy(&self) -> &KernelPolicy

Examples found in repository

examples/retrieval_bench.rs (line 211)

fn retrieve(corpus: &Path, queries: &Path, dir: &Path, dataset: &str, k: usize) -> Result<()> {
    if !(1..=100).contains(&k) || dataset.trim().is_empty() {
        return Err("k must be 1..100 and dataset must be nonempty".into());
    }
    let policy = KernelPolicy {
        recall_limit: k,
        ..KernelPolicy::default()
    };
    let path = fresh(dir)?;
    let source = format!("beir:{dataset}");
    let mut store = MemoryStore::open(&path, policy.clone())?;
    let mut baseline = Bm25::default();
    let mut ids = HashSet::new();
    let mut rejected = Vec::new();
    let mut batch = Vec::new();
    let mut baseline_index_ms = 0.0;
    let ingest = Instant::now();
    for line in BufReader::new(File::open(corpus)?).lines() {
        let doc: Document = serde_json::from_str(&line?)?;
        if doc.id.trim().is_empty() || !ids.insert(doc.id.clone()) {
            return Err("empty or duplicate document id".into());
        }
        let trace = make_trace(&doc.id, &doc.content, &source, TraceKind::Observation);
        if let Err(reason) = trace.validate() {
            rejected.push(json!({"id": doc.id, "reason": reason.to_string()}));
            continue;
        }
        let timer = Instant::now();
        baseline.add(&doc);
        baseline_index_ms += ms(timer);
        batch.push(trace);
        if batch.len() == 1000 {
            store.remember_many(std::mem::take(&mut batch))?;
        }
    }
    if !batch.is_empty() {
        store.remember_many(batch)?;
    }
    let ingest_ms = ms(ingest);
    store.verify()?;
    let head_before = serde_json::to_value(store.head())?;
    drop(store);
    let start = Instant::now();
    let store = MemoryStore::open_existing(&path, policy)?;
    let reopen_ms = ms(start);
    store.verify()?;
    let provenance_retained = store
        .all()
        .filter(|entry| {
            let t = &entry.trace;
            t.provenance.source == source
                && !t.provenance.verified
                && t.created_at_ms == CLOCK
                && t.provenance.observed_at_ms == CLOCK
                && t.provenance.locator.as_deref() == Some(format!("{source}:{}", t.id).as_str())
        })
        .count();
    if store.record_count() != baseline.ids.len()
        || provenance_retained != baseline.ids.len()
        || serde_json::to_value(store.head())? != head_before
    {
        return Err("replay, head or provenance invariant failed".into());
    }
    emit(
        json!({"kind": "loaded", "corpus_total": ids.len(), "admitted": baseline.ids.len(),
        "rejected": rejected, "provenance_retained": provenance_retained,
        "ingest_both_ms": ingest_ms, "bm25_index_ms": baseline_index_ms, "reopen_ms": reopen_ms,
        "policy": store.policy(), "bm25": {"k1": 1.2, "b": 0.75}, "query_clock_ms": CLOCK}),
    )?;
    let mut query_ids = HashSet::new();
    for line in BufReader::new(File::open(queries)?).lines() {
        let q: Query = serde_json::from_str(&line?)?;
        if q.id.trim().is_empty() || !query_ids.insert(q.id.clone()) {
            return Err("invalid query id".into());
        }
        let mut query = RecallQuery::new(&q.text);
        query.now_ms = CLOCK;
        query.limit = k;
        let start = Instant::now();
        let recalled = store.recall(&query);
        let hikmah_ms = ms(start);
        let start = Instant::now();
        let bm25 = baseline.search(&q.text, k);
        let bm25_ms = ms(start);
        let hikmah: Vec<_> = recalled.into_iter().map(|r| json!({
            "id": r.trace.id, "score": r.score, "duplicates_folded": r.duplicates,
            "provenance_source": r.trace.provenance.source, "verified": r.trace.provenance.verified,
        })).collect();
        emit(
            json!({"kind": "query", "id": q.id, "hikmah": hikmah, "bm25": bm25,
                    "hikmah_ms": hikmah_ms, "bm25_ms": bm25_ms}),
        )?;
    }
    emit(json!({"kind": "finished", "queries": query_ids.len()}))
}

examples/bench_kernel.rs (line 206)

fn scale(dir: &Path, records: usize, queries: usize) -> Result<Value> {
    if records == 0 || queries == 0 {
        return Err("records and queries must be positive".into());
    }
    let path = fresh(dir)?;
    let mut store = MemoryStore::open(&path, KernelPolicy::default())?;
    let single_count = records.min(50);
    let batched = records - single_count;
    let mut batches = Vec::new();
    let seed_started = Instant::now();
    for begin in (0..batched).step_by(1000) {
        let traces = (begin..(begin + 1000).min(batched)).map(trace).collect();
        let start = Instant::now();
        store.remember_many(traces)?;
        batches.push(ms(start));
    }
    let mut appends = Vec::new();
    for i in batched..records {
        let t = trace(i);
        let start = Instant::now();
        store.remember(t)?;
        appends.push(ms(start));
    }
    let seed_ms = ms(seed_started);
    let memory_after_seed = rss();
    let mut recall_ms = Vec::new();
    let mut hits = 0;
    let mut returned = 0;
    let mut result_ids = Vec::new();
    // Warm up with an exact cue; quality here is a synthetic exact-cue check, not field recall.
    std::hint::black_box(store.recall(&RecallQuery::new(token(0))));
    for i in 0..queries {
        let selected = i * records / queries;
        let mut query = RecallQuery::new(token(selected));
        query.now_ms = CLOCK;
        query.limit = 1;
        let start = Instant::now();
        let found = store.recall(&query);
        recall_ms.push(ms(start));
        returned += found.len();
        hits += usize::from(
            found
                .first()
                .is_some_and(|r| r.trace.id == format!("sample-{selected}")),
        );
        result_ids.push(found.iter().map(|r| r.trace.id.clone()).collect::<Vec<_>>());
    }
    let start = Instant::now();
    store.verify()?;
    let verify_ms = ms(start);
    let before = store.head();
    drop(store);
    let start = Instant::now();
    let store = MemoryStore::open_existing(&path, KernelPolicy::default())?;
    let replay_ms = ms(start);
    store.verify()?;
    let retained = store
        .all()
        .filter(|e| {
            e.trace.provenance.source == "synthetic-kernel-benchmark"
                && e.trace.provenance.observed_at_ms == CLOCK
                && !e.trace.provenance.verified
        })
        .count();
    let mut replay_matches = 0;
    for (i, expected) in result_ids.iter().enumerate() {
        let mut query = RecallQuery::new(token(i * records / queries));
        query.now_ms = CLOCK;
        query.limit = 1;
        let ids: Vec<_> = store
            .recall(&query)
            .into_iter()
            .map(|r| r.trace.id)
            .collect();
        replay_matches += usize::from(ids == *expected);
    }
    let head_same = serde_json::to_value(before)? == serde_json::to_value(store.head())?;
    let ok = store.record_count() == records
        && retained == records
        && hits == queries
        && replay_matches == queries
        && head_same;
    Ok(json!({
        "kind": "synthetic_scale", "ok": ok, "records": records, "queries": queries,
        "workload": "unclaimed observation traces, unique exact lexical cues, fixed clock",
        "seed_ms": seed_ms, "batch_size": 1000, "batched_records": batched,
        "batch_append_ms": quantiles(batches), "single_append_ms": quantiles(appends),
        "recall_ms": quantiles(recall_ms), "verify_ms": verify_ms, "replay_ms": replay_ms,
        "exact_cue_hits": hits, "returned": returned, "recall_at_1": hits as f64 / queries as f64,
        "provenance_retained": retained, "replayed_records": store.record_count(),
        "recall_replay_matches": replay_matches, "head_unchanged": head_same,
        "ledger_bytes": fs::metadata(path)?.len(), "memory_after_seed": memory_after_seed,
        "memory_after_replay": rss(), "policy": store.policy(),
        "limits": ["Batched seeding does not measure per-record durable append latency.",
          "Warm filesystem cache; reopening in the same process is not a cold-start measurement.",
          "No structured claims in this workload; conflict-heavy stores need a separate workload.",
          "Synthetic exact-cue quality is not semantic or real-world retrieval accuracy.",
          "RSS is process-wide Linux /proc data; unavailable platforms report null."]
    }))
}

pub fn record_count(&self) -> usize

Examples found in repository

examples/retrieval_bench.rs (line 201)

fn retrieve(corpus: &Path, queries: &Path, dir: &Path, dataset: &str, k: usize) -> Result<()> {
    if !(1..=100).contains(&k) || dataset.trim().is_empty() {
        return Err("k must be 1..100 and dataset must be nonempty".into());
    }
    let policy = KernelPolicy {
        recall_limit: k,
        ..KernelPolicy::default()
    };
    let path = fresh(dir)?;
    let source = format!("beir:{dataset}");
    let mut store = MemoryStore::open(&path, policy.clone())?;
    let mut baseline = Bm25::default();
    let mut ids = HashSet::new();
    let mut rejected = Vec::new();
    let mut batch = Vec::new();
    let mut baseline_index_ms = 0.0;
    let ingest = Instant::now();
    for line in BufReader::new(File::open(corpus)?).lines() {
        let doc: Document = serde_json::from_str(&line?)?;
        if doc.id.trim().is_empty() || !ids.insert(doc.id.clone()) {
            return Err("empty or duplicate document id".into());
        }
        let trace = make_trace(&doc.id, &doc.content, &source, TraceKind::Observation);
        if let Err(reason) = trace.validate() {
            rejected.push(json!({"id": doc.id, "reason": reason.to_string()}));
            continue;
        }
        let timer = Instant::now();
        baseline.add(&doc);
        baseline_index_ms += ms(timer);
        batch.push(trace);
        if batch.len() == 1000 {
            store.remember_many(std::mem::take(&mut batch))?;
        }
    }
    if !batch.is_empty() {
        store.remember_many(batch)?;
    }
    let ingest_ms = ms(ingest);
    store.verify()?;
    let head_before = serde_json::to_value(store.head())?;
    drop(store);
    let start = Instant::now();
    let store = MemoryStore::open_existing(&path, policy)?;
    let reopen_ms = ms(start);
    store.verify()?;
    let provenance_retained = store
        .all()
        .filter(|entry| {
            let t = &entry.trace;
            t.provenance.source == source
                && !t.provenance.verified
                && t.created_at_ms == CLOCK
                && t.provenance.observed_at_ms == CLOCK
                && t.provenance.locator.as_deref() == Some(format!("{source}:{}", t.id).as_str())
        })
        .count();
    if store.record_count() != baseline.ids.len()
        || provenance_retained != baseline.ids.len()
        || serde_json::to_value(store.head())? != head_before
    {
        return Err("replay, head or provenance invariant failed".into());
    }
    emit(
        json!({"kind": "loaded", "corpus_total": ids.len(), "admitted": baseline.ids.len(),
        "rejected": rejected, "provenance_retained": provenance_retained,
        "ingest_both_ms": ingest_ms, "bm25_index_ms": baseline_index_ms, "reopen_ms": reopen_ms,
        "policy": store.policy(), "bm25": {"k1": 1.2, "b": 0.75}, "query_clock_ms": CLOCK}),
    )?;
    let mut query_ids = HashSet::new();
    for line in BufReader::new(File::open(queries)?).lines() {
        let q: Query = serde_json::from_str(&line?)?;
        if q.id.trim().is_empty() || !query_ids.insert(q.id.clone()) {
            return Err("invalid query id".into());
        }
        let mut query = RecallQuery::new(&q.text);
        query.now_ms = CLOCK;
        query.limit = k;
        let start = Instant::now();
        let recalled = store.recall(&query);
        let hikmah_ms = ms(start);
        let start = Instant::now();
        let bm25 = baseline.search(&q.text, k);
        let bm25_ms = ms(start);
        let hikmah: Vec<_> = recalled.into_iter().map(|r| json!({
            "id": r.trace.id, "score": r.score, "duplicates_folded": r.duplicates,
            "provenance_source": r.trace.provenance.source, "verified": r.trace.provenance.verified,
        })).collect();
        emit(
            json!({"kind": "query", "id": q.id, "hikmah": hikmah, "bm25": bm25,
                    "hikmah_ms": hikmah_ms, "bm25_ms": bm25_ms}),
        )?;
    }
    emit(json!({"kind": "finished", "queries": query_ids.len()}))
}

examples/bench_kernel.rs (line 191)

fn scale(dir: &Path, records: usize, queries: usize) -> Result<Value> {
    if records == 0 || queries == 0 {
        return Err("records and queries must be positive".into());
    }
    let path = fresh(dir)?;
    let mut store = MemoryStore::open(&path, KernelPolicy::default())?;
    let single_count = records.min(50);
    let batched = records - single_count;
    let mut batches = Vec::new();
    let seed_started = Instant::now();
    for begin in (0..batched).step_by(1000) {
        let traces = (begin..(begin + 1000).min(batched)).map(trace).collect();
        let start = Instant::now();
        store.remember_many(traces)?;
        batches.push(ms(start));
    }
    let mut appends = Vec::new();
    for i in batched..records {
        let t = trace(i);
        let start = Instant::now();
        store.remember(t)?;
        appends.push(ms(start));
    }
    let seed_ms = ms(seed_started);
    let memory_after_seed = rss();
    let mut recall_ms = Vec::new();
    let mut hits = 0;
    let mut returned = 0;
    let mut result_ids = Vec::new();
    // Warm up with an exact cue; quality here is a synthetic exact-cue check, not field recall.
    std::hint::black_box(store.recall(&RecallQuery::new(token(0))));
    for i in 0..queries {
        let selected = i * records / queries;
        let mut query = RecallQuery::new(token(selected));
        query.now_ms = CLOCK;
        query.limit = 1;
        let start = Instant::now();
        let found = store.recall(&query);
        recall_ms.push(ms(start));
        returned += found.len();
        hits += usize::from(
            found
                .first()
                .is_some_and(|r| r.trace.id == format!("sample-{selected}")),
        );
        result_ids.push(found.iter().map(|r| r.trace.id.clone()).collect::<Vec<_>>());
    }
    let start = Instant::now();
    store.verify()?;
    let verify_ms = ms(start);
    let before = store.head();
    drop(store);
    let start = Instant::now();
    let store = MemoryStore::open_existing(&path, KernelPolicy::default())?;
    let replay_ms = ms(start);
    store.verify()?;
    let retained = store
        .all()
        .filter(|e| {
            e.trace.provenance.source == "synthetic-kernel-benchmark"
                && e.trace.provenance.observed_at_ms == CLOCK
                && !e.trace.provenance.verified
        })
        .count();
    let mut replay_matches = 0;
    for (i, expected) in result_ids.iter().enumerate() {
        let mut query = RecallQuery::new(token(i * records / queries));
        query.now_ms = CLOCK;
        query.limit = 1;
        let ids: Vec<_> = store
            .recall(&query)
            .into_iter()
            .map(|r| r.trace.id)
            .collect();
        replay_matches += usize::from(ids == *expected);
    }
    let head_same = serde_json::to_value(before)? == serde_json::to_value(store.head())?;
    let ok = store.record_count() == records
        && retained == records
        && hits == queries
        && replay_matches == queries
        && head_same;
    Ok(json!({
        "kind": "synthetic_scale", "ok": ok, "records": records, "queries": queries,
        "workload": "unclaimed observation traces, unique exact lexical cues, fixed clock",
        "seed_ms": seed_ms, "batch_size": 1000, "batched_records": batched,
        "batch_append_ms": quantiles(batches), "single_append_ms": quantiles(appends),
        "recall_ms": quantiles(recall_ms), "verify_ms": verify_ms, "replay_ms": replay_ms,
        "exact_cue_hits": hits, "returned": returned, "recall_at_1": hits as f64 / queries as f64,
        "provenance_retained": retained, "replayed_records": store.record_count(),
        "recall_replay_matches": replay_matches, "head_unchanged": head_same,
        "ledger_bytes": fs::metadata(path)?.len(), "memory_after_seed": memory_after_seed,
        "memory_after_replay": rss(), "policy": store.policy(),
        "limits": ["Batched seeding does not measure per-record durable append latency.",
          "Warm filesystem cache; reopening in the same process is not a cold-start measurement.",
          "No structured claims in this workload; conflict-heavy stores need a separate workload.",
          "Synthetic exact-cue quality is not semantic or real-world retrieval accuracy.",
          "RSS is process-wide Linux /proc data; unavailable platforms report null."]
    }))
}

// Specified, platform-independent SplitMix64; never a cryptographic RNG.
struct Rng(u64);
impl Rng {
    fn next(&mut self) -> u64 {
        self.0 = self.0.wrapping_add(0x9e3779b97f4a7c15);
        let mut x = self.0;
        x = (x ^ (x >> 30)).wrapping_mul(0xbf58476d1ce4e5b9);
        x = (x ^ (x >> 27)).wrapping_mul(0x94d049bb133111eb);
        x ^ (x >> 31)
    }
    fn unit(&mut self) -> f64 {
        (self.next() % 10001) as f64 / 10000.0
    }
}

fn option(name: &str, a: f64, b: f64) -> DecisionOption {
    DecisionOption {
        name: name.into(),
        description: None,
        scores: BTreeMap::from([("a".into(), a), ("b".into(), b)]),
        model_scores: BTreeMap::new(),
        evidence_confidence: 1.0,
        hard_blocks: vec![],
        reversible: false,
    }
}

fn decisions(cases: usize, seed: u64) -> Result<Value> {
    if cases < 6 {
        return Err("at least 6 cases required to exercise every family".into());
    }
    let mut rng = Rng(seed);
    let mut counts: BTreeMap<&str, usize> = BTreeMap::new();
    let mut failures = Vec::new();
    let mut violations = 0;
    let start = Instant::now();
    for i in 0..cases {
        let mut a = option("a", rng.unit(), rng.unit());
        let mut b = option("b", rng.unit(), rng.unit());
        let family;
        let mut expected: Option<Option<String>> = None;
        match i % 6 {
            0 => {
                family = "high_score_hard_block";
                a.scores.values_mut().for_each(|s| *s = 1.0);
                a.hard_blocks.push("no authorization".into());
                expected = Some(Some("b".into()));
            }
            1 => {
                family = "all_blocked";
                a.hard_blocks.push("privacy".into());
                b.hard_blocks.push("consent".into());
                expected = Some(None);
            }
            2 => {
                family = "complete_score_oracle";
                let sum_a: f64 = a.scores.values().sum();
                let sum_b: f64 = b.scores.values().sum();
                expected = Some(Some(if sum_a >= sum_b { "a" } else { "b" }.into()));
            }
            3 => {
                family = "unknowns_and_model_estimates";
                a.scores.clear();
                a.model_scores.insert("a".into(), rng.unit());
            }
            4 => {
                family = "reversibility_band";
                let top = 0.5 + rng.unit() * 0.4;
                let gap = if rng.next().is_multiple_of(2) {
                    0.01
                } else {
                    0.03
                };
                a.scores.values_mut().for_each(|s| *s = top);
                a.evidence_confidence = 0.1;
                b.scores.values_mut().for_each(|s| *s = top - gap);
                b.reversible = true;
                expected = Some(Some(if gap < 0.02 { "b" } else { "a" }.into()));
            }
            _ => {
                family = "invalid_numeric_input";
                a.scores.insert("a".into(), 1.01 + rng.unit());
            }
        }
        *counts.entry(family).or_default() += 1;
        let frame = DecisionFrame {
            question: "synthetic decision".into(),
            criteria: ["a", "b"]
                .into_iter()
                .map(|id| Criterion {
                    id: id.into(),
                    weight: 1.0,
                    description: id.into(),
                })
                .collect(),
            options: vec![a, b],
        };
        let actual = evaluate(&frame);
        let mut reasons = Vec::new();
        if family == "invalid_numeric_input" {
            if actual.is_ok() {
                reasons.push("invalid input admitted");
            }
        } else if let Ok(answer) = &actual {
            if let Some(expected) = expected {
                if answer.recommended != expected {
                    reasons.push("wrong recommendation");
                }
            }
            if let Some(name) = &answer.recommended {
                if frame
                    .options
                    .iter()
                    .any(|o| &o.name == name && !o.hard_blocks.is_empty())
                {
                    reasons.push("hard-blocked option selected");
                }
            }
            if family == "unknowns_and_model_estimates" {
                let ranked = answer.ranking.iter().find(|o| o.name == "a").unwrap();
                if ranked.coverage != 0.0
                    || ranked.evidence_interval != [0.0, 1.0]
                    || ranked.missing_criteria != ["b"]
                    || answer.decisive
                {
                    reasons.push("unknown evidence hidden or model guess counted as evidence");
                }
            }
            let mut permuted = frame.clone();
            permuted.options.reverse();
            permuted.question = "irrelevant wording changed".into();
            let other = evaluate(&permuted)?;
            if answer.recommended != other.recommended
                || serde_json::to_value(&answer.ranking)? != serde_json::to_value(&other.ranking)?
            {
                reasons.push("order or irrelevant text changed result");
            }
        } else {
            reasons.push("valid frame rejected");
        }
        if !reasons.is_empty() {
            violations += 1;
            if failures.len() < 20 {
                failures.push(
                    json!({"case_index": i, "family": family, "reasons": reasons,
                    "frame": frame, "actual": actual.as_ref().ok()}),
                );
            }
        }
    }
    Ok(
        json!({"kind": "synthetic_decision_invariants", "seed": seed, "rng": "splitmix64",
        "cases": cases, "families": counts, "violations": violations,
        "violation_rate": violations as f64 / cases as f64, "ok": violations == 0,
        "elapsed_ms": ms(start), "counterexamples": failures,
        "limits": ["Generated families are not real decision-quality labels or a proof over all inputs.",
          "Metamorphic checks keep criteria fixed; weight perturbations require separate expectations."]}),
    )
}

fn committed_child(dir: &Path) -> Result<()> {
    let path = fresh(dir)?;
    let mut store = MemoryStore::open(path, KernelPolicy::default())?;
    store.remember_many((0..50).map(trace).collect())?;
    println!("committed:50");
    io::stdout().flush()?;
    // Parent kills us after the acknowledged commit; this is not an in-flight crash claim.
    loop {
        std::thread::park();
    }
}

fn recovery(dir: &Path) -> Result<Value> {
    fs::create_dir(dir)?;
    let child_dir = dir.join("committed");
    let mut child = Command::new(std::env::current_exe()?)
        .args(["committed-child", "--dir"])
        .arg(&child_dir)
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .spawn()?;
    let mut ack = String::new();
    let read_result = io::BufReader::new(child.stdout.take().unwrap()).read_line(&mut ack);
    let kill_result = child.kill();
    let wait_result = child.wait();
    read_result?;
    kill_result?;
    wait_result?;
    if ack.trim() != "committed:50" {
        return Err("child did not acknowledge commit".into());
    }
    let path = child_dir.join("memory.jsonl");
    let start = Instant::now();
    let store = MemoryStore::open_existing(&path, KernelPolicy::default())?;
    store.verify()?;
    let restart_ms = ms(start);
    let acknowledged_survived =
        store.record_count() == 50 && (0..50).all(|i| store.get(&format!("sample-{i}")).is_some());
    let head_path = store.head_path();
    drop(store);
    let original = fs::read(&path)?;
    let original_head = fs::read(head_path)?;
    let copied_store = |name: &str, bytes: &[u8]| -> Result<PathBuf> {
        let target = fresh(&dir.join(name))?;
        let store = MemoryStore::open(&target, KernelPolicy::default())?;
        let target_head = store.head_path();
        drop(store);
        fs::write(&target, bytes)?;
        fs::write(target_head, &original_head)?;
        Ok(target)
    };
    let torn = copied_store("torn", &original)?;
    OpenOptions::new()
        .append(true)
        .open(&torn)?
        .write_all(b"{\"seq\":")?;
    let mut store = MemoryStore::open_existing(&torn, KernelPolicy::default())?;
    store.remember(trace(50))?;
    store.verify()?;
    let torn_repaired = store.record_count() == 51;
    let mut edited = String::from_utf8(original.clone())?;
    edited = edited.replacen("deployment observation", "deployment fabrication", 1);
    let tampered = copied_store("edited", edited.as_bytes())?;
    let rejected = |path: &Path| match MemoryStore::open_existing(path, KernelPolicy::default()) {
        Err(_) => true,
        Ok(store) => store.verify().is_err(),
    };
    let edit_detected = rejected(&tampered);
    let last = original[..original.len() - 1]
        .iter()
        .rposition(|b| *b == b'\n')
        .unwrap()
        + 1;
    let truncated = copied_store("truncated", &original[..last])?;
    let truncation_detected = rejected(&truncated);
    Ok(
        json!({"kind": "synthetic_recovery", "acknowledged_records": 50,
        "acknowledged_survived": acknowledged_survived, "restart_ms": restart_ms,
        "torn_tail_repaired": torn_repaired, "edit_detected": edit_detected,
        "truncation_detected": truncation_detected,
        "ok": acknowledged_survived && torn_repaired && edit_detected && truncation_detected,
        "limits": ["Process killed after an acknowledged commit, not during fsync or head replacement.",
          "Torn tail, edit and truncation are injected into separate copies.",
          "This does not simulate power loss or an attacker rewriting both ledger and head."]}),
    )
}

pub fn records(&self) -> &[LedgerRecord]

pub fn head(&self) -> Option<LedgerHead>

Examples found in repository

examples/retrieval_bench.rs (line 184)

fn retrieve(corpus: &Path, queries: &Path, dir: &Path, dataset: &str, k: usize) -> Result<()> {
    if !(1..=100).contains(&k) || dataset.trim().is_empty() {
        return Err("k must be 1..100 and dataset must be nonempty".into());
    }
    let policy = KernelPolicy {
        recall_limit: k,
        ..KernelPolicy::default()
    };
    let path = fresh(dir)?;
    let source = format!("beir:{dataset}");
    let mut store = MemoryStore::open(&path, policy.clone())?;
    let mut baseline = Bm25::default();
    let mut ids = HashSet::new();
    let mut rejected = Vec::new();
    let mut batch = Vec::new();
    let mut baseline_index_ms = 0.0;
    let ingest = Instant::now();
    for line in BufReader::new(File::open(corpus)?).lines() {
        let doc: Document = serde_json::from_str(&line?)?;
        if doc.id.trim().is_empty() || !ids.insert(doc.id.clone()) {
            return Err("empty or duplicate document id".into());
        }
        let trace = make_trace(&doc.id, &doc.content, &source, TraceKind::Observation);
        if let Err(reason) = trace.validate() {
            rejected.push(json!({"id": doc.id, "reason": reason.to_string()}));
            continue;
        }
        let timer = Instant::now();
        baseline.add(&doc);
        baseline_index_ms += ms(timer);
        batch.push(trace);
        if batch.len() == 1000 {
            store.remember_many(std::mem::take(&mut batch))?;
        }
    }
    if !batch.is_empty() {
        store.remember_many(batch)?;
    }
    let ingest_ms = ms(ingest);
    store.verify()?;
    let head_before = serde_json::to_value(store.head())?;
    drop(store);
    let start = Instant::now();
    let store = MemoryStore::open_existing(&path, policy)?;
    let reopen_ms = ms(start);
    store.verify()?;
    let provenance_retained = store
        .all()
        .filter(|entry| {
            let t = &entry.trace;
            t.provenance.source == source
                && !t.provenance.verified
                && t.created_at_ms == CLOCK
                && t.provenance.observed_at_ms == CLOCK
                && t.provenance.locator.as_deref() == Some(format!("{source}:{}", t.id).as_str())
        })
        .count();
    if store.record_count() != baseline.ids.len()
        || provenance_retained != baseline.ids.len()
        || serde_json::to_value(store.head())? != head_before
    {
        return Err("replay, head or provenance invariant failed".into());
    }
    emit(
        json!({"kind": "loaded", "corpus_total": ids.len(), "admitted": baseline.ids.len(),
        "rejected": rejected, "provenance_retained": provenance_retained,
        "ingest_both_ms": ingest_ms, "bm25_index_ms": baseline_index_ms, "reopen_ms": reopen_ms,
        "policy": store.policy(), "bm25": {"k1": 1.2, "b": 0.75}, "query_clock_ms": CLOCK}),
    )?;
    let mut query_ids = HashSet::new();
    for line in BufReader::new(File::open(queries)?).lines() {
        let q: Query = serde_json::from_str(&line?)?;
        if q.id.trim().is_empty() || !query_ids.insert(q.id.clone()) {
            return Err("invalid query id".into());
        }
        let mut query = RecallQuery::new(&q.text);
        query.now_ms = CLOCK;
        query.limit = k;
        let start = Instant::now();
        let recalled = store.recall(&query);
        let hikmah_ms = ms(start);
        let start = Instant::now();
        let bm25 = baseline.search(&q.text, k);
        let bm25_ms = ms(start);
        let hikmah: Vec<_> = recalled.into_iter().map(|r| json!({
            "id": r.trace.id, "score": r.score, "duplicates_folded": r.duplicates,
            "provenance_source": r.trace.provenance.source, "verified": r.trace.provenance.verified,
        })).collect();
        emit(
            json!({"kind": "query", "id": q.id, "hikmah": hikmah, "bm25": bm25,
                    "hikmah_ms": hikmah_ms, "bm25_ms": bm25_ms}),
        )?;
    }
    emit(json!({"kind": "finished", "queries": query_ids.len()}))
}

examples/bench_kernel.rs (line 164)

fn scale(dir: &Path, records: usize, queries: usize) -> Result<Value> {
    if records == 0 || queries == 0 {
        return Err("records and queries must be positive".into());
    }
    let path = fresh(dir)?;
    let mut store = MemoryStore::open(&path, KernelPolicy::default())?;
    let single_count = records.min(50);
    let batched = records - single_count;
    let mut batches = Vec::new();
    let seed_started = Instant::now();
    for begin in (0..batched).step_by(1000) {
        let traces = (begin..(begin + 1000).min(batched)).map(trace).collect();
        let start = Instant::now();
        store.remember_many(traces)?;
        batches.push(ms(start));
    }
    let mut appends = Vec::new();
    for i in batched..records {
        let t = trace(i);
        let start = Instant::now();
        store.remember(t)?;
        appends.push(ms(start));
    }
    let seed_ms = ms(seed_started);
    let memory_after_seed = rss();
    let mut recall_ms = Vec::new();
    let mut hits = 0;
    let mut returned = 0;
    let mut result_ids = Vec::new();
    // Warm up with an exact cue; quality here is a synthetic exact-cue check, not field recall.
    std::hint::black_box(store.recall(&RecallQuery::new(token(0))));
    for i in 0..queries {
        let selected = i * records / queries;
        let mut query = RecallQuery::new(token(selected));
        query.now_ms = CLOCK;
        query.limit = 1;
        let start = Instant::now();
        let found = store.recall(&query);
        recall_ms.push(ms(start));
        returned += found.len();
        hits += usize::from(
            found
                .first()
                .is_some_and(|r| r.trace.id == format!("sample-{selected}")),
        );
        result_ids.push(found.iter().map(|r| r.trace.id.clone()).collect::<Vec<_>>());
    }
    let start = Instant::now();
    store.verify()?;
    let verify_ms = ms(start);
    let before = store.head();
    drop(store);
    let start = Instant::now();
    let store = MemoryStore::open_existing(&path, KernelPolicy::default())?;
    let replay_ms = ms(start);
    store.verify()?;
    let retained = store
        .all()
        .filter(|e| {
            e.trace.provenance.source == "synthetic-kernel-benchmark"
                && e.trace.provenance.observed_at_ms == CLOCK
                && !e.trace.provenance.verified
        })
        .count();
    let mut replay_matches = 0;
    for (i, expected) in result_ids.iter().enumerate() {
        let mut query = RecallQuery::new(token(i * records / queries));
        query.now_ms = CLOCK;
        query.limit = 1;
        let ids: Vec<_> = store
            .recall(&query)
            .into_iter()
            .map(|r| r.trace.id)
            .collect();
        replay_matches += usize::from(ids == *expected);
    }
    let head_same = serde_json::to_value(before)? == serde_json::to_value(store.head())?;
    let ok = store.record_count() == records
        && retained == records
        && hits == queries
        && replay_matches == queries
        && head_same;
    Ok(json!({
        "kind": "synthetic_scale", "ok": ok, "records": records, "queries": queries,
        "workload": "unclaimed observation traces, unique exact lexical cues, fixed clock",
        "seed_ms": seed_ms, "batch_size": 1000, "batched_records": batched,
        "batch_append_ms": quantiles(batches), "single_append_ms": quantiles(appends),
        "recall_ms": quantiles(recall_ms), "verify_ms": verify_ms, "replay_ms": replay_ms,
        "exact_cue_hits": hits, "returned": returned, "recall_at_1": hits as f64 / queries as f64,
        "provenance_retained": retained, "replayed_records": store.record_count(),
        "recall_replay_matches": replay_matches, "head_unchanged": head_same,
        "ledger_bytes": fs::metadata(path)?.len(), "memory_after_seed": memory_after_seed,
        "memory_after_replay": rss(), "policy": store.policy(),
        "limits": ["Batched seeding does not measure per-record durable append latency.",
          "Warm filesystem cache; reopening in the same process is not a cold-start measurement.",
          "No structured claims in this workload; conflict-heavy stores need a separate workload.",
          "Synthetic exact-cue quality is not semantic or real-world retrieval accuracy.",
          "RSS is process-wide Linux /proc data; unavailable platforms report null."]
    }))
}

pub fn head_path(&self) -> PathBuf

Examples found in repository

examples/bench_kernel.rs (line 413)

fn recovery(dir: &Path) -> Result<Value> {
    fs::create_dir(dir)?;
    let child_dir = dir.join("committed");
    let mut child = Command::new(std::env::current_exe()?)
        .args(["committed-child", "--dir"])
        .arg(&child_dir)
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .spawn()?;
    let mut ack = String::new();
    let read_result = io::BufReader::new(child.stdout.take().unwrap()).read_line(&mut ack);
    let kill_result = child.kill();
    let wait_result = child.wait();
    read_result?;
    kill_result?;
    wait_result?;
    if ack.trim() != "committed:50" {
        return Err("child did not acknowledge commit".into());
    }
    let path = child_dir.join("memory.jsonl");
    let start = Instant::now();
    let store = MemoryStore::open_existing(&path, KernelPolicy::default())?;
    store.verify()?;
    let restart_ms = ms(start);
    let acknowledged_survived =
        store.record_count() == 50 && (0..50).all(|i| store.get(&format!("sample-{i}")).is_some());
    let head_path = store.head_path();
    drop(store);
    let original = fs::read(&path)?;
    let original_head = fs::read(head_path)?;
    let copied_store = |name: &str, bytes: &[u8]| -> Result<PathBuf> {
        let target = fresh(&dir.join(name))?;
        let store = MemoryStore::open(&target, KernelPolicy::default())?;
        let target_head = store.head_path();
        drop(store);
        fs::write(&target, bytes)?;
        fs::write(target_head, &original_head)?;
        Ok(target)
    };
    let torn = copied_store("torn", &original)?;
    OpenOptions::new()
        .append(true)
        .open(&torn)?
        .write_all(b"{\"seq\":")?;
    let mut store = MemoryStore::open_existing(&torn, KernelPolicy::default())?;
    store.remember(trace(50))?;
    store.verify()?;
    let torn_repaired = store.record_count() == 51;
    let mut edited = String::from_utf8(original.clone())?;
    edited = edited.replacen("deployment observation", "deployment fabrication", 1);
    let tampered = copied_store("edited", edited.as_bytes())?;
    let rejected = |path: &Path| match MemoryStore::open_existing(path, KernelPolicy::default()) {
        Err(_) => true,
        Ok(store) => store.verify().is_err(),
    };
    let edit_detected = rejected(&tampered);
    let last = original[..original.len() - 1]
        .iter()
        .rposition(|b| *b == b'\n')
        .unwrap()
        + 1;
    let truncated = copied_store("truncated", &original[..last])?;
    let truncation_detected = rejected(&truncated);
    Ok(
        json!({"kind": "synthetic_recovery", "acknowledged_records": 50,
        "acknowledged_survived": acknowledged_survived, "restart_ms": restart_ms,
        "torn_tail_repaired": torn_repaired, "edit_detected": edit_detected,
        "truncation_detected": truncation_detected,
        "ok": acknowledged_survived && torn_repaired && edit_detected && truncation_detected,
        "limits": ["Process killed after an acknowledged commit, not during fsync or head replacement.",
          "Torn tail, edit and truncation are injected into separate copies.",
          "This does not simulate power loss or an attacker rewriting both ledger and head."]}),
    )
}

pub fn verify(&self) -> Result<()>

Integrity check. Errors on any chain break; returns Ok(()) only when the report is ok.

Examples found in repository

examples/retrieval_bench.rs (line 183)

fn retrieve(corpus: &Path, queries: &Path, dir: &Path, dataset: &str, k: usize) -> Result<()> {
    if !(1..=100).contains(&k) || dataset.trim().is_empty() {
        return Err("k must be 1..100 and dataset must be nonempty".into());
    }
    let policy = KernelPolicy {
        recall_limit: k,
        ..KernelPolicy::default()
    };
    let path = fresh(dir)?;
    let source = format!("beir:{dataset}");
    let mut store = MemoryStore::open(&path, policy.clone())?;
    let mut baseline = Bm25::default();
    let mut ids = HashSet::new();
    let mut rejected = Vec::new();
    let mut batch = Vec::new();
    let mut baseline_index_ms = 0.0;
    let ingest = Instant::now();
    for line in BufReader::new(File::open(corpus)?).lines() {
        let doc: Document = serde_json::from_str(&line?)?;
        if doc.id.trim().is_empty() || !ids.insert(doc.id.clone()) {
            return Err("empty or duplicate document id".into());
        }
        let trace = make_trace(&doc.id, &doc.content, &source, TraceKind::Observation);
        if let Err(reason) = trace.validate() {
            rejected.push(json!({"id": doc.id, "reason": reason.to_string()}));
            continue;
        }
        let timer = Instant::now();
        baseline.add(&doc);
        baseline_index_ms += ms(timer);
        batch.push(trace);
        if batch.len() == 1000 {
            store.remember_many(std::mem::take(&mut batch))?;
        }
    }
    if !batch.is_empty() {
        store.remember_many(batch)?;
    }
    let ingest_ms = ms(ingest);
    store.verify()?;
    let head_before = serde_json::to_value(store.head())?;
    drop(store);
    let start = Instant::now();
    let store = MemoryStore::open_existing(&path, policy)?;
    let reopen_ms = ms(start);
    store.verify()?;
    let provenance_retained = store
        .all()
        .filter(|entry| {
            let t = &entry.trace;
            t.provenance.source == source
                && !t.provenance.verified
                && t.created_at_ms == CLOCK
                && t.provenance.observed_at_ms == CLOCK
                && t.provenance.locator.as_deref() == Some(format!("{source}:{}", t.id).as_str())
        })
        .count();
    if store.record_count() != baseline.ids.len()
        || provenance_retained != baseline.ids.len()
        || serde_json::to_value(store.head())? != head_before
    {
        return Err("replay, head or provenance invariant failed".into());
    }
    emit(
        json!({"kind": "loaded", "corpus_total": ids.len(), "admitted": baseline.ids.len(),
        "rejected": rejected, "provenance_retained": provenance_retained,
        "ingest_both_ms": ingest_ms, "bm25_index_ms": baseline_index_ms, "reopen_ms": reopen_ms,
        "policy": store.policy(), "bm25": {"k1": 1.2, "b": 0.75}, "query_clock_ms": CLOCK}),
    )?;
    let mut query_ids = HashSet::new();
    for line in BufReader::new(File::open(queries)?).lines() {
        let q: Query = serde_json::from_str(&line?)?;
        if q.id.trim().is_empty() || !query_ids.insert(q.id.clone()) {
            return Err("invalid query id".into());
        }
        let mut query = RecallQuery::new(&q.text);
        query.now_ms = CLOCK;
        query.limit = k;
        let start = Instant::now();
        let recalled = store.recall(&query);
        let hikmah_ms = ms(start);
        let start = Instant::now();
        let bm25 = baseline.search(&q.text, k);
        let bm25_ms = ms(start);
        let hikmah: Vec<_> = recalled.into_iter().map(|r| json!({
            "id": r.trace.id, "score": r.score, "duplicates_folded": r.duplicates,
            "provenance_source": r.trace.provenance.source, "verified": r.trace.provenance.verified,
        })).collect();
        emit(
            json!({"kind": "query", "id": q.id, "hikmah": hikmah, "bm25": bm25,
                    "hikmah_ms": hikmah_ms, "bm25_ms": bm25_ms}),
        )?;
    }
    emit(json!({"kind": "finished", "queries": query_ids.len()}))
}

fn corrections(dir: &Path, cases: usize) -> Result<()> {
    if !(1..=10000).contains(&cases) {
        return Err("cases must be 1..10000".into());
    }
    let path = fresh(dir)?;
    let mut store = MemoryStore::open(&path, KernelPolicy::default())?;
    let mut stale_hits = 0;
    let mut latest_hits = 0;
    let mut model_rejections = 0;
    let mut link_hits = 0;
    let mut expected = Vec::new();
    for i in 0..cases {
        let subject = format!("service{i}topic");
        let key = format!("{subject}.region");
        let mut predecessor = None;
        for (version, value) in ["oldregion", "middlezone", "newlocation"]
            .iter()
            .enumerate()
        {
            let id = format!("{i}-v{version}");
            let mut trace = make_trace(
                &id,
                &format!("{subject} {value}"),
                "synthetic:correction",
                if version == 0 {
                    TraceKind::Belief
                } else {
                    TraceKind::Correction
                },
            );
            trace.claim_key = Some(key.clone());
            trace.claim_value = Some((*value).into());
            trace.supersedes = predecessor;
            store.remember(trace)?;
            predecessor = Some(id);
        }
        let mut malicious = make_trace(
            &format!("{i}-model"),
            "invented overwrite",
            "model:fixture",
            TraceKind::Correction,
        );
        malicious.supersedes = predecessor;
        model_rejections += usize::from(store.remember(malicious).is_err());
        let results = store.recall(&RecallQuery::new(&subject));
        stale_hits += results
            .iter()
            .filter(|r| r.trace.id == format!("{i}-v0") || r.trace.id == format!("{i}-v1"))
            .count();
        latest_hits += usize::from(results.iter().any(|r| r.trace.id == format!("{i}-v2")));
        link_hits +=
            usize::from(results.iter().any(|r| {
                r.trace.id == format!("{i}-v2") && r.supersedes == Some(format!("{i}-v1"))
            }));
        expected.push((subject, format!("{i}-v2")));
    }
    let old_only_results = store
        .recall(&RecallQuery::new("oldregion middlezone"))
        .len();
    drop(store);
    let store = MemoryStore::open_existing(&path, KernelPolicy::default())?;
    store.verify()?;
    let restart_hits = expected
        .iter()
        .filter(|(q, id)| {
            store
                .recall(&RecallQuery::new(q))
                .iter()
                .any(|r| &r.trace.id == id)
        })
        .count();
    let provenance = store
        .all()
        .filter(|e| {
            e.trace.provenance.source == "synthetic:correction" && !e.trace.provenance.verified
        })
        .count();
    let ok = stale_hits == 0
        && latest_hits == cases
        && model_rejections == cases
        && link_hits == cases
        && old_only_results == 0
        && restart_hits == cases
        && provenance == 3 * cases;
    emit(
        json!({"kind": "synthetic_corrections", "cases": cases, "ok": ok,
        "stale_hits": stale_hits, "latest_hits": latest_hits, "supersession_link_hits": link_hits,
        "model_overwrite_rejections": model_rejections, "old_only_query_results": old_only_results,
        "latest_hits_after_restart": restart_hits, "provenance_retained": provenance,
        "limits": ["Generated structured histories; not extracted real incident corrections.",
          "Subject variants of one generated template, not independent real incidents or a proof across all histories.",
          "A query matching only retired terms abstains; no semantic alias expansion is implied."]}),
    )?;
    if !ok {
        return Err("correction invariant failure".into());
    }
    Ok(())
}

examples/bench_kernel.rs (line 162)

fn scale(dir: &Path, records: usize, queries: usize) -> Result<Value> {
    if records == 0 || queries == 0 {
        return Err("records and queries must be positive".into());
    }
    let path = fresh(dir)?;
    let mut store = MemoryStore::open(&path, KernelPolicy::default())?;
    let single_count = records.min(50);
    let batched = records - single_count;
    let mut batches = Vec::new();
    let seed_started = Instant::now();
    for begin in (0..batched).step_by(1000) {
        let traces = (begin..(begin + 1000).min(batched)).map(trace).collect();
        let start = Instant::now();
        store.remember_many(traces)?;
        batches.push(ms(start));
    }
    let mut appends = Vec::new();
    for i in batched..records {
        let t = trace(i);
        let start = Instant::now();
        store.remember(t)?;
        appends.push(ms(start));
    }
    let seed_ms = ms(seed_started);
    let memory_after_seed = rss();
    let mut recall_ms = Vec::new();
    let mut hits = 0;
    let mut returned = 0;
    let mut result_ids = Vec::new();
    // Warm up with an exact cue; quality here is a synthetic exact-cue check, not field recall.
    std::hint::black_box(store.recall(&RecallQuery::new(token(0))));
    for i in 0..queries {
        let selected = i * records / queries;
        let mut query = RecallQuery::new(token(selected));
        query.now_ms = CLOCK;
        query.limit = 1;
        let start = Instant::now();
        let found = store.recall(&query);
        recall_ms.push(ms(start));
        returned += found.len();
        hits += usize::from(
            found
                .first()
                .is_some_and(|r| r.trace.id == format!("sample-{selected}")),
        );
        result_ids.push(found.iter().map(|r| r.trace.id.clone()).collect::<Vec<_>>());
    }
    let start = Instant::now();
    store.verify()?;
    let verify_ms = ms(start);
    let before = store.head();
    drop(store);
    let start = Instant::now();
    let store = MemoryStore::open_existing(&path, KernelPolicy::default())?;
    let replay_ms = ms(start);
    store.verify()?;
    let retained = store
        .all()
        .filter(|e| {
            e.trace.provenance.source == "synthetic-kernel-benchmark"
                && e.trace.provenance.observed_at_ms == CLOCK
                && !e.trace.provenance.verified
        })
        .count();
    let mut replay_matches = 0;
    for (i, expected) in result_ids.iter().enumerate() {
        let mut query = RecallQuery::new(token(i * records / queries));
        query.now_ms = CLOCK;
        query.limit = 1;
        let ids: Vec<_> = store
            .recall(&query)
            .into_iter()
            .map(|r| r.trace.id)
            .collect();
        replay_matches += usize::from(ids == *expected);
    }
    let head_same = serde_json::to_value(before)? == serde_json::to_value(store.head())?;
    let ok = store.record_count() == records
        && retained == records
        && hits == queries
        && replay_matches == queries
        && head_same;
    Ok(json!({
        "kind": "synthetic_scale", "ok": ok, "records": records, "queries": queries,
        "workload": "unclaimed observation traces, unique exact lexical cues, fixed clock",
        "seed_ms": seed_ms, "batch_size": 1000, "batched_records": batched,
        "batch_append_ms": quantiles(batches), "single_append_ms": quantiles(appends),
        "recall_ms": quantiles(recall_ms), "verify_ms": verify_ms, "replay_ms": replay_ms,
        "exact_cue_hits": hits, "returned": returned, "recall_at_1": hits as f64 / queries as f64,
        "provenance_retained": retained, "replayed_records": store.record_count(),
        "recall_replay_matches": replay_matches, "head_unchanged": head_same,
        "ledger_bytes": fs::metadata(path)?.len(), "memory_after_seed": memory_after_seed,
        "memory_after_replay": rss(), "policy": store.policy(),
        "limits": ["Batched seeding does not measure per-record durable append latency.",
          "Warm filesystem cache; reopening in the same process is not a cold-start measurement.",
          "No structured claims in this workload; conflict-heavy stores need a separate workload.",
          "Synthetic exact-cue quality is not semantic or real-world retrieval accuracy.",
          "RSS is process-wide Linux /proc data; unavailable platforms report null."]
    }))
}

// Specified, platform-independent SplitMix64; never a cryptographic RNG.
struct Rng(u64);
impl Rng {
    fn next(&mut self) -> u64 {
        self.0 = self.0.wrapping_add(0x9e3779b97f4a7c15);
        let mut x = self.0;
        x = (x ^ (x >> 30)).wrapping_mul(0xbf58476d1ce4e5b9);
        x = (x ^ (x >> 27)).wrapping_mul(0x94d049bb133111eb);
        x ^ (x >> 31)
    }
    fn unit(&mut self) -> f64 {
        (self.next() % 10001) as f64 / 10000.0
    }
}

fn option(name: &str, a: f64, b: f64) -> DecisionOption {
    DecisionOption {
        name: name.into(),
        description: None,
        scores: BTreeMap::from([("a".into(), a), ("b".into(), b)]),
        model_scores: BTreeMap::new(),
        evidence_confidence: 1.0,
        hard_blocks: vec![],
        reversible: false,
    }
}

fn decisions(cases: usize, seed: u64) -> Result<Value> {
    if cases < 6 {
        return Err("at least 6 cases required to exercise every family".into());
    }
    let mut rng = Rng(seed);
    let mut counts: BTreeMap<&str, usize> = BTreeMap::new();
    let mut failures = Vec::new();
    let mut violations = 0;
    let start = Instant::now();
    for i in 0..cases {
        let mut a = option("a", rng.unit(), rng.unit());
        let mut b = option("b", rng.unit(), rng.unit());
        let family;
        let mut expected: Option<Option<String>> = None;
        match i % 6 {
            0 => {
                family = "high_score_hard_block";
                a.scores.values_mut().for_each(|s| *s = 1.0);
                a.hard_blocks.push("no authorization".into());
                expected = Some(Some("b".into()));
            }
            1 => {
                family = "all_blocked";
                a.hard_blocks.push("privacy".into());
                b.hard_blocks.push("consent".into());
                expected = Some(None);
            }
            2 => {
                family = "complete_score_oracle";
                let sum_a: f64 = a.scores.values().sum();
                let sum_b: f64 = b.scores.values().sum();
                expected = Some(Some(if sum_a >= sum_b { "a" } else { "b" }.into()));
            }
            3 => {
                family = "unknowns_and_model_estimates";
                a.scores.clear();
                a.model_scores.insert("a".into(), rng.unit());
            }
            4 => {
                family = "reversibility_band";
                let top = 0.5 + rng.unit() * 0.4;
                let gap = if rng.next().is_multiple_of(2) {
                    0.01
                } else {
                    0.03
                };
                a.scores.values_mut().for_each(|s| *s = top);
                a.evidence_confidence = 0.1;
                b.scores.values_mut().for_each(|s| *s = top - gap);
                b.reversible = true;
                expected = Some(Some(if gap < 0.02 { "b" } else { "a" }.into()));
            }
            _ => {
                family = "invalid_numeric_input";
                a.scores.insert("a".into(), 1.01 + rng.unit());
            }
        }
        *counts.entry(family).or_default() += 1;
        let frame = DecisionFrame {
            question: "synthetic decision".into(),
            criteria: ["a", "b"]
                .into_iter()
                .map(|id| Criterion {
                    id: id.into(),
                    weight: 1.0,
                    description: id.into(),
                })
                .collect(),
            options: vec![a, b],
        };
        let actual = evaluate(&frame);
        let mut reasons = Vec::new();
        if family == "invalid_numeric_input" {
            if actual.is_ok() {
                reasons.push("invalid input admitted");
            }
        } else if let Ok(answer) = &actual {
            if let Some(expected) = expected {
                if answer.recommended != expected {
                    reasons.push("wrong recommendation");
                }
            }
            if let Some(name) = &answer.recommended {
                if frame
                    .options
                    .iter()
                    .any(|o| &o.name == name && !o.hard_blocks.is_empty())
                {
                    reasons.push("hard-blocked option selected");
                }
            }
            if family == "unknowns_and_model_estimates" {
                let ranked = answer.ranking.iter().find(|o| o.name == "a").unwrap();
                if ranked.coverage != 0.0
                    || ranked.evidence_interval != [0.0, 1.0]
                    || ranked.missing_criteria != ["b"]
                    || answer.decisive
                {
                    reasons.push("unknown evidence hidden or model guess counted as evidence");
                }
            }
            let mut permuted = frame.clone();
            permuted.options.reverse();
            permuted.question = "irrelevant wording changed".into();
            let other = evaluate(&permuted)?;
            if answer.recommended != other.recommended
                || serde_json::to_value(&answer.ranking)? != serde_json::to_value(&other.ranking)?
            {
                reasons.push("order or irrelevant text changed result");
            }
        } else {
            reasons.push("valid frame rejected");
        }
        if !reasons.is_empty() {
            violations += 1;
            if failures.len() < 20 {
                failures.push(
                    json!({"case_index": i, "family": family, "reasons": reasons,
                    "frame": frame, "actual": actual.as_ref().ok()}),
                );
            }
        }
    }
    Ok(
        json!({"kind": "synthetic_decision_invariants", "seed": seed, "rng": "splitmix64",
        "cases": cases, "families": counts, "violations": violations,
        "violation_rate": violations as f64 / cases as f64, "ok": violations == 0,
        "elapsed_ms": ms(start), "counterexamples": failures,
        "limits": ["Generated families are not real decision-quality labels or a proof over all inputs.",
          "Metamorphic checks keep criteria fixed; weight perturbations require separate expectations."]}),
    )
}

fn committed_child(dir: &Path) -> Result<()> {
    let path = fresh(dir)?;
    let mut store = MemoryStore::open(path, KernelPolicy::default())?;
    store.remember_many((0..50).map(trace).collect())?;
    println!("committed:50");
    io::stdout().flush()?;
    // Parent kills us after the acknowledged commit; this is not an in-flight crash claim.
    loop {
        std::thread::park();
    }
}

fn recovery(dir: &Path) -> Result<Value> {
    fs::create_dir(dir)?;
    let child_dir = dir.join("committed");
    let mut child = Command::new(std::env::current_exe()?)
        .args(["committed-child", "--dir"])
        .arg(&child_dir)
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .spawn()?;
    let mut ack = String::new();
    let read_result = io::BufReader::new(child.stdout.take().unwrap()).read_line(&mut ack);
    let kill_result = child.kill();
    let wait_result = child.wait();
    read_result?;
    kill_result?;
    wait_result?;
    if ack.trim() != "committed:50" {
        return Err("child did not acknowledge commit".into());
    }
    let path = child_dir.join("memory.jsonl");
    let start = Instant::now();
    let store = MemoryStore::open_existing(&path, KernelPolicy::default())?;
    store.verify()?;
    let restart_ms = ms(start);
    let acknowledged_survived =
        store.record_count() == 50 && (0..50).all(|i| store.get(&format!("sample-{i}")).is_some());
    let head_path = store.head_path();
    drop(store);
    let original = fs::read(&path)?;
    let original_head = fs::read(head_path)?;
    let copied_store = |name: &str, bytes: &[u8]| -> Result<PathBuf> {
        let target = fresh(&dir.join(name))?;
        let store = MemoryStore::open(&target, KernelPolicy::default())?;
        let target_head = store.head_path();
        drop(store);
        fs::write(&target, bytes)?;
        fs::write(target_head, &original_head)?;
        Ok(target)
    };
    let torn = copied_store("torn", &original)?;
    OpenOptions::new()
        .append(true)
        .open(&torn)?
        .write_all(b"{\"seq\":")?;
    let mut store = MemoryStore::open_existing(&torn, KernelPolicy::default())?;
    store.remember(trace(50))?;
    store.verify()?;
    let torn_repaired = store.record_count() == 51;
    let mut edited = String::from_utf8(original.clone())?;
    edited = edited.replacen("deployment observation", "deployment fabrication", 1);
    let tampered = copied_store("edited", edited.as_bytes())?;
    let rejected = |path: &Path| match MemoryStore::open_existing(path, KernelPolicy::default()) {
        Err(_) => true,
        Ok(store) => store.verify().is_err(),
    };
    let edit_detected = rejected(&tampered);
    let last = original[..original.len() - 1]
        .iter()
        .rposition(|b| *b == b'\n')
        .unwrap()
        + 1;
    let truncated = copied_store("truncated", &original[..last])?;
    let truncation_detected = rejected(&truncated);
    Ok(
        json!({"kind": "synthetic_recovery", "acknowledged_records": 50,
        "acknowledged_survived": acknowledged_survived, "restart_ms": restart_ms,
        "torn_tail_repaired": torn_repaired, "edit_detected": edit_detected,
        "truncation_detected": truncation_detected,
        "ok": acknowledged_survived && torn_repaired && edit_detected && truncation_detected,
        "limits": ["Process killed after an acknowledged commit, not during fsync or head replacement.",
          "Torn tail, edit and truncation are injected into separate copies.",
          "This does not simulate power loss or an attacker rewriting both ledger and head."]}),
    )
}

pub fn verify_report(&self, expected_head: Option<&str>) -> Result<VerifyReport>

Full verification report. expected_head pins the latest hash (for example a value kept in CI or in git) so a wholesale re-chain by someone with write access is detected. When the loaded ledger holds records past its head snapshot, this waits for in-flight writes (a shared lock on the <store>.lock sidecar) before reporting them as unacknowledged.

pub fn accept_tail(&mut self) -> Result<TailAcceptance>

Acknowledge records appended after the head file (after inspecting them): the head moves to the end of the ledger and writes are allowed again. Unlike Self::reset_head this accepts only appended records. It refuses when records were removed or rewritten, when the head file is unreadable, and when there is no head file to extend.

pub fn reset_head(&mut self) -> Result<Option<LedgerHead>>

Explicitly accept the current ledger as the new head (after a deliberate repair).

pub fn set_nonblocking_writes(&mut self, nonblocking: bool)

Make writes fail immediately when another process holds the writer lock, instead of waiting. For callers such as the Stop hook that must never stall on a busy store.

pub fn lock_path(&self) -> PathBuf

impl MemoryStore

pub fn commitments_due(&self, now_ms: u64, within_ms: u64) -> Vec<CommitmentDue>

impl MemoryStore

pub fn recall(&self, query: &RecallQuery) -> Vec<RecallResult>

Examples found in repository

examples/retrieval_bench.rs (line 223)

fn retrieve(corpus: &Path, queries: &Path, dir: &Path, dataset: &str, k: usize) -> Result<()> {
    if !(1..=100).contains(&k) || dataset.trim().is_empty() {
        return Err("k must be 1..100 and dataset must be nonempty".into());
    }
    let policy = KernelPolicy {
        recall_limit: k,
        ..KernelPolicy::default()
    };
    let path = fresh(dir)?;
    let source = format!("beir:{dataset}");
    let mut store = MemoryStore::open(&path, policy.clone())?;
    let mut baseline = Bm25::default();
    let mut ids = HashSet::new();
    let mut rejected = Vec::new();
    let mut batch = Vec::new();
    let mut baseline_index_ms = 0.0;
    let ingest = Instant::now();
    for line in BufReader::new(File::open(corpus)?).lines() {
        let doc: Document = serde_json::from_str(&line?)?;
        if doc.id.trim().is_empty() || !ids.insert(doc.id.clone()) {
            return Err("empty or duplicate document id".into());
        }
        let trace = make_trace(&doc.id, &doc.content, &source, TraceKind::Observation);
        if let Err(reason) = trace.validate() {
            rejected.push(json!({"id": doc.id, "reason": reason.to_string()}));
            continue;
        }
        let timer = Instant::now();
        baseline.add(&doc);
        baseline_index_ms += ms(timer);
        batch.push(trace);
        if batch.len() == 1000 {
            store.remember_many(std::mem::take(&mut batch))?;
        }
    }
    if !batch.is_empty() {
        store.remember_many(batch)?;
    }
    let ingest_ms = ms(ingest);
    store.verify()?;
    let head_before = serde_json::to_value(store.head())?;
    drop(store);
    let start = Instant::now();
    let store = MemoryStore::open_existing(&path, policy)?;
    let reopen_ms = ms(start);
    store.verify()?;
    let provenance_retained = store
        .all()
        .filter(|entry| {
            let t = &entry.trace;
            t.provenance.source == source
                && !t.provenance.verified
                && t.created_at_ms == CLOCK
                && t.provenance.observed_at_ms == CLOCK
                && t.provenance.locator.as_deref() == Some(format!("{source}:{}", t.id).as_str())
        })
        .count();
    if store.record_count() != baseline.ids.len()
        || provenance_retained != baseline.ids.len()
        || serde_json::to_value(store.head())? != head_before
    {
        return Err("replay, head or provenance invariant failed".into());
    }
    emit(
        json!({"kind": "loaded", "corpus_total": ids.len(), "admitted": baseline.ids.len(),
        "rejected": rejected, "provenance_retained": provenance_retained,
        "ingest_both_ms": ingest_ms, "bm25_index_ms": baseline_index_ms, "reopen_ms": reopen_ms,
        "policy": store.policy(), "bm25": {"k1": 1.2, "b": 0.75}, "query_clock_ms": CLOCK}),
    )?;
    let mut query_ids = HashSet::new();
    for line in BufReader::new(File::open(queries)?).lines() {
        let q: Query = serde_json::from_str(&line?)?;
        if q.id.trim().is_empty() || !query_ids.insert(q.id.clone()) {
            return Err("invalid query id".into());
        }
        let mut query = RecallQuery::new(&q.text);
        query.now_ms = CLOCK;
        query.limit = k;
        let start = Instant::now();
        let recalled = store.recall(&query);
        let hikmah_ms = ms(start);
        let start = Instant::now();
        let bm25 = baseline.search(&q.text, k);
        let bm25_ms = ms(start);
        let hikmah: Vec<_> = recalled.into_iter().map(|r| json!({
            "id": r.trace.id, "score": r.score, "duplicates_folded": r.duplicates,
            "provenance_source": r.trace.provenance.source, "verified": r.trace.provenance.verified,
        })).collect();
        emit(
            json!({"kind": "query", "id": q.id, "hikmah": hikmah, "bm25": bm25,
                    "hikmah_ms": hikmah_ms, "bm25_ms": bm25_ms}),
        )?;
    }
    emit(json!({"kind": "finished", "queries": query_ids.len()}))
}

fn corrections(dir: &Path, cases: usize) -> Result<()> {
    if !(1..=10000).contains(&cases) {
        return Err("cases must be 1..10000".into());
    }
    let path = fresh(dir)?;
    let mut store = MemoryStore::open(&path, KernelPolicy::default())?;
    let mut stale_hits = 0;
    let mut latest_hits = 0;
    let mut model_rejections = 0;
    let mut link_hits = 0;
    let mut expected = Vec::new();
    for i in 0..cases {
        let subject = format!("service{i}topic");
        let key = format!("{subject}.region");
        let mut predecessor = None;
        for (version, value) in ["oldregion", "middlezone", "newlocation"]
            .iter()
            .enumerate()
        {
            let id = format!("{i}-v{version}");
            let mut trace = make_trace(
                &id,
                &format!("{subject} {value}"),
                "synthetic:correction",
                if version == 0 {
                    TraceKind::Belief
                } else {
                    TraceKind::Correction
                },
            );
            trace.claim_key = Some(key.clone());
            trace.claim_value = Some((*value).into());
            trace.supersedes = predecessor;
            store.remember(trace)?;
            predecessor = Some(id);
        }
        let mut malicious = make_trace(
            &format!("{i}-model"),
            "invented overwrite",
            "model:fixture",
            TraceKind::Correction,
        );
        malicious.supersedes = predecessor;
        model_rejections += usize::from(store.remember(malicious).is_err());
        let results = store.recall(&RecallQuery::new(&subject));
        stale_hits += results
            .iter()
            .filter(|r| r.trace.id == format!("{i}-v0") || r.trace.id == format!("{i}-v1"))
            .count();
        latest_hits += usize::from(results.iter().any(|r| r.trace.id == format!("{i}-v2")));
        link_hits +=
            usize::from(results.iter().any(|r| {
                r.trace.id == format!("{i}-v2") && r.supersedes == Some(format!("{i}-v1"))
            }));
        expected.push((subject, format!("{i}-v2")));
    }
    let old_only_results = store
        .recall(&RecallQuery::new("oldregion middlezone"))
        .len();
    drop(store);
    let store = MemoryStore::open_existing(&path, KernelPolicy::default())?;
    store.verify()?;
    let restart_hits = expected
        .iter()
        .filter(|(q, id)| {
            store
                .recall(&RecallQuery::new(q))
                .iter()
                .any(|r| &r.trace.id == id)
        })
        .count();
    let provenance = store
        .all()
        .filter(|e| {
            e.trace.provenance.source == "synthetic:correction" && !e.trace.provenance.verified
        })
        .count();
    let ok = stale_hits == 0
        && latest_hits == cases
        && model_rejections == cases
        && link_hits == cases
        && old_only_results == 0
        && restart_hits == cases
        && provenance == 3 * cases;
    emit(
        json!({"kind": "synthetic_corrections", "cases": cases, "ok": ok,
        "stale_hits": stale_hits, "latest_hits": latest_hits, "supersession_link_hits": link_hits,
        "model_overwrite_rejections": model_rejections, "old_only_query_results": old_only_results,
        "latest_hits_after_restart": restart_hits, "provenance_retained": provenance,
        "limits": ["Generated structured histories; not extracted real incident corrections.",
          "Subject variants of one generated template, not independent real incidents or a proof across all histories.",
          "A query matching only retired terms abstains; no semantic alias expansion is implied."]}),
    )?;
    if !ok {
        return Err("correction invariant failure".into());
    }
    Ok(())
}

examples/bench_kernel.rs (line 144)

fn scale(dir: &Path, records: usize, queries: usize) -> Result<Value> {
    if records == 0 || queries == 0 {
        return Err("records and queries must be positive".into());
    }
    let path = fresh(dir)?;
    let mut store = MemoryStore::open(&path, KernelPolicy::default())?;
    let single_count = records.min(50);
    let batched = records - single_count;
    let mut batches = Vec::new();
    let seed_started = Instant::now();
    for begin in (0..batched).step_by(1000) {
        let traces = (begin..(begin + 1000).min(batched)).map(trace).collect();
        let start = Instant::now();
        store.remember_many(traces)?;
        batches.push(ms(start));
    }
    let mut appends = Vec::new();
    for i in batched..records {
        let t = trace(i);
        let start = Instant::now();
        store.remember(t)?;
        appends.push(ms(start));
    }
    let seed_ms = ms(seed_started);
    let memory_after_seed = rss();
    let mut recall_ms = Vec::new();
    let mut hits = 0;
    let mut returned = 0;
    let mut result_ids = Vec::new();
    // Warm up with an exact cue; quality here is a synthetic exact-cue check, not field recall.
    std::hint::black_box(store.recall(&RecallQuery::new(token(0))));
    for i in 0..queries {
        let selected = i * records / queries;
        let mut query = RecallQuery::new(token(selected));
        query.now_ms = CLOCK;
        query.limit = 1;
        let start = Instant::now();
        let found = store.recall(&query);
        recall_ms.push(ms(start));
        returned += found.len();
        hits += usize::from(
            found
                .first()
                .is_some_and(|r| r.trace.id == format!("sample-{selected}")),
        );
        result_ids.push(found.iter().map(|r| r.trace.id.clone()).collect::<Vec<_>>());
    }
    let start = Instant::now();
    store.verify()?;
    let verify_ms = ms(start);
    let before = store.head();
    drop(store);
    let start = Instant::now();
    let store = MemoryStore::open_existing(&path, KernelPolicy::default())?;
    let replay_ms = ms(start);
    store.verify()?;
    let retained = store
        .all()
        .filter(|e| {
            e.trace.provenance.source == "synthetic-kernel-benchmark"
                && e.trace.provenance.observed_at_ms == CLOCK
                && !e.trace.provenance.verified
        })
        .count();
    let mut replay_matches = 0;
    for (i, expected) in result_ids.iter().enumerate() {
        let mut query = RecallQuery::new(token(i * records / queries));
        query.now_ms = CLOCK;
        query.limit = 1;
        let ids: Vec<_> = store
            .recall(&query)
            .into_iter()
            .map(|r| r.trace.id)
            .collect();
        replay_matches += usize::from(ids == *expected);
    }
    let head_same = serde_json::to_value(before)? == serde_json::to_value(store.head())?;
    let ok = store.record_count() == records
        && retained == records
        && hits == queries
        && replay_matches == queries
        && head_same;
    Ok(json!({
        "kind": "synthetic_scale", "ok": ok, "records": records, "queries": queries,
        "workload": "unclaimed observation traces, unique exact lexical cues, fixed clock",
        "seed_ms": seed_ms, "batch_size": 1000, "batched_records": batched,
        "batch_append_ms": quantiles(batches), "single_append_ms": quantiles(appends),
        "recall_ms": quantiles(recall_ms), "verify_ms": verify_ms, "replay_ms": replay_ms,
        "exact_cue_hits": hits, "returned": returned, "recall_at_1": hits as f64 / queries as f64,
        "provenance_retained": retained, "replayed_records": store.record_count(),
        "recall_replay_matches": replay_matches, "head_unchanged": head_same,
        "ledger_bytes": fs::metadata(path)?.len(), "memory_after_seed": memory_after_seed,
        "memory_after_replay": rss(), "policy": store.policy(),
        "limits": ["Batched seeding does not measure per-record durable append latency.",
          "Warm filesystem cache; reopening in the same process is not a cold-start measurement.",
          "No structured claims in this workload; conflict-heavy stores need a separate workload.",
          "Synthetic exact-cue quality is not semantic or real-world retrieval accuracy.",
          "RSS is process-wide Linux /proc data; unavailable platforms report null."]
    }))
}

impl MemoryStore

pub fn gate_threshold( &self, max_false_block: f64, ) -> Result<GateThresholdReport>

Pair recorded Truth Gate engine predictions (model: sources only) with their outcomes and choose the engine threshold. Refuses with fewer than MIN_OUTCOMES resolved predictions, or without both classes.

impl MemoryStore

pub fn provenance_chain(&self, id: &str) -> Option<TraceLineage>

The full provenance chain of one trace: itself, its supersession lineage, and its unresolved conflicts. None when no trace carries id.

Trait Implementations

impl Debug for MemoryStore

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.