Struct RecallQuery

Source

pub struct RecallQuery {
    pub text: String,
    pub tags: Vec<String>,
    pub kinds: Vec<TraceKind>,
    pub limit: usize,
    pub now_ms: u64,
    pub include_superseded: bool,
}

Fields§

§text: String§tags: Vec<String>§kinds: Vec<TraceKind>§limit: usize§now_ms: u64§include_superseded: bool

Also recall superseded traces (history), each with superseded_by set. Off by default: a replaced belief must not be activated as if it were current.

Implementations§

Source §

impl RecallQuery

Source

pub fn new(text: impl Into<String>) -> Self

Examples found in repository ?

examples/retrieval_bench.rs (line 219)

144fn retrieve(corpus: &Path, queries: &Path, dir: &Path, dataset: &str, k: usize) -> Result<()> {
145    if !(1..=100).contains(&k) || dataset.trim().is_empty() {
146        return Err("k must be 1..100 and dataset must be nonempty".into());
147    }
148    let policy = KernelPolicy {
149        recall_limit: k,
150        ..KernelPolicy::default()
151    };
152    let path = fresh(dir)?;
153    let source = format!("beir:{dataset}");
154    let mut store = MemoryStore::open(&path, policy.clone())?;
155    let mut baseline = Bm25::default();
156    let mut ids = HashSet::new();
157    let mut rejected = Vec::new();
158    let mut batch = Vec::new();
159    let mut baseline_index_ms = 0.0;
160    let ingest = Instant::now();
161    for line in BufReader::new(File::open(corpus)?).lines() {
162        let doc: Document = serde_json::from_str(&line?)?;
163        if doc.id.trim().is_empty() || !ids.insert(doc.id.clone()) {
164            return Err("empty or duplicate document id".into());
165        }
166        let trace = make_trace(&doc.id, &doc.content, &source, TraceKind::Observation);
167        if let Err(reason) = trace.validate() {
168            rejected.push(json!({"id": doc.id, "reason": reason.to_string()}));
169            continue;
170        }
171        let timer = Instant::now();
172        baseline.add(&doc);
173        baseline_index_ms += ms(timer);
174        batch.push(trace);
175        if batch.len() == 1000 {
176            store.remember_many(std::mem::take(&mut batch))?;
177        }
178    }
179    if !batch.is_empty() {
180        store.remember_many(batch)?;
181    }
182    let ingest_ms = ms(ingest);
183    store.verify()?;
184    let head_before = serde_json::to_value(store.head())?;
185    drop(store);
186    let start = Instant::now();
187    let store = MemoryStore::open_existing(&path, policy)?;
188    let reopen_ms = ms(start);
189    store.verify()?;
190    let provenance_retained = store
191        .all()
192        .filter(|entry| {
193            let t = &entry.trace;
194            t.provenance.source == source
195                && !t.provenance.verified
196                && t.created_at_ms == CLOCK
197                && t.provenance.observed_at_ms == CLOCK
198                && t.provenance.locator.as_deref() == Some(format!("{source}:{}", t.id).as_str())
199        })
200        .count();
201    if store.record_count() != baseline.ids.len()
202        || provenance_retained != baseline.ids.len()
203        || serde_json::to_value(store.head())? != head_before
204    {
205        return Err("replay, head or provenance invariant failed".into());
206    }
207    emit(
208        json!({"kind": "loaded", "corpus_total": ids.len(), "admitted": baseline.ids.len(),
209        "rejected": rejected, "provenance_retained": provenance_retained,
210        "ingest_both_ms": ingest_ms, "bm25_index_ms": baseline_index_ms, "reopen_ms": reopen_ms,
211        "policy": store.policy(), "bm25": {"k1": 1.2, "b": 0.75}, "query_clock_ms": CLOCK}),
212    )?;
213    let mut query_ids = HashSet::new();
214    for line in BufReader::new(File::open(queries)?).lines() {
215        let q: Query = serde_json::from_str(&line?)?;
216        if q.id.trim().is_empty() || !query_ids.insert(q.id.clone()) {
217            return Err("invalid query id".into());
218        }
219        let mut query = RecallQuery::new(&q.text);
220        query.now_ms = CLOCK;
221        query.limit = k;
222        let start = Instant::now();
223        let recalled = store.recall(&query);
224        let hikmah_ms = ms(start);
225        let start = Instant::now();
226        let bm25 = baseline.search(&q.text, k);
227        let bm25_ms = ms(start);
228        let hikmah: Vec<_> = recalled.into_iter().map(|r| json!({
229            "id": r.trace.id, "score": r.score, "duplicates_folded": r.duplicates,
230            "provenance_source": r.trace.provenance.source, "verified": r.trace.provenance.verified,
231        })).collect();
232        emit(
233            json!({"kind": "query", "id": q.id, "hikmah": hikmah, "bm25": bm25,
234                    "hikmah_ms": hikmah_ms, "bm25_ms": bm25_ms}),
235        )?;
236    }
237    emit(json!({"kind": "finished", "queries": query_ids.len()}))
238}
239
240fn corrections(dir: &Path, cases: usize) -> Result<()> {
241    if !(1..=10000).contains(&cases) {
242        return Err("cases must be 1..10000".into());
243    }
244    let path = fresh(dir)?;
245    let mut store = MemoryStore::open(&path, KernelPolicy::default())?;
246    let mut stale_hits = 0;
247    let mut latest_hits = 0;
248    let mut model_rejections = 0;
249    let mut link_hits = 0;
250    let mut expected = Vec::new();
251    for i in 0..cases {
252        let subject = format!("service{i}topic");
253        let key = format!("{subject}.region");
254        let mut predecessor = None;
255        for (version, value) in ["oldregion", "middlezone", "newlocation"]
256            .iter()
257            .enumerate()
258        {
259            let id = format!("{i}-v{version}");
260            let mut trace = make_trace(
261                &id,
262                &format!("{subject} {value}"),
263                "synthetic:correction",
264                if version == 0 {
265                    TraceKind::Belief
266                } else {
267                    TraceKind::Correction
268                },
269            );
270            trace.claim_key = Some(key.clone());
271            trace.claim_value = Some((*value).into());
272            trace.supersedes = predecessor;
273            store.remember(trace)?;
274            predecessor = Some(id);
275        }
276        let mut malicious = make_trace(
277            &format!("{i}-model"),
278            "invented overwrite",
279            "model:fixture",
280            TraceKind::Correction,
281        );
282        malicious.supersedes = predecessor;
283        model_rejections += usize::from(store.remember(malicious).is_err());
284        let results = store.recall(&RecallQuery::new(&subject));
285        stale_hits += results
286            .iter()
287            .filter(|r| r.trace.id == format!("{i}-v0") || r.trace.id == format!("{i}-v1"))
288            .count();
289        latest_hits += usize::from(results.iter().any(|r| r.trace.id == format!("{i}-v2")));
290        link_hits +=
291            usize::from(results.iter().any(|r| {
292                r.trace.id == format!("{i}-v2") && r.supersedes == Some(format!("{i}-v1"))
293            }));
294        expected.push((subject, format!("{i}-v2")));
295    }
296    let old_only_results = store
297        .recall(&RecallQuery::new("oldregion middlezone"))
298        .len();
299    drop(store);
300    let store = MemoryStore::open_existing(&path, KernelPolicy::default())?;
301    store.verify()?;
302    let restart_hits = expected
303        .iter()
304        .filter(|(q, id)| {
305            store
306                .recall(&RecallQuery::new(q))
307                .iter()
308                .any(|r| &r.trace.id == id)
309        })
310        .count();
311    let provenance = store
312        .all()
313        .filter(|e| {
314            e.trace.provenance.source == "synthetic:correction" && !e.trace.provenance.verified
315        })
316        .count();
317    let ok = stale_hits == 0
318        && latest_hits == cases
319        && model_rejections == cases
320        && link_hits == cases
321        && old_only_results == 0
322        && restart_hits == cases
323        && provenance == 3 * cases;
324    emit(
325        json!({"kind": "synthetic_corrections", "cases": cases, "ok": ok,
326        "stale_hits": stale_hits, "latest_hits": latest_hits, "supersession_link_hits": link_hits,
327        "model_overwrite_rejections": model_rejections, "old_only_query_results": old_only_results,
328        "latest_hits_after_restart": restart_hits, "provenance_retained": provenance,
329        "limits": ["Generated structured histories; not extracted real incident corrections.",
330          "Subject variants of one generated template, not independent real incidents or a proof across all histories.",
331          "A query matching only retired terms abstains; no semantic alias expansion is implied."]}),
332    )?;
333    if !ok {
334        return Err("correction invariant failure".into());
335    }
336    Ok(())
337}

More examples

Hide additional examples

examples/bench_kernel.rs (line 144)

114fn scale(dir: &Path, records: usize, queries: usize) -> Result<Value> {
115    if records == 0 || queries == 0 {
116        return Err("records and queries must be positive".into());
117    }
118    let path = fresh(dir)?;
119    let mut store = MemoryStore::open(&path, KernelPolicy::default())?;
120    let single_count = records.min(50);
121    let batched = records - single_count;
122    let mut batches = Vec::new();
123    let seed_started = Instant::now();
124    for begin in (0..batched).step_by(1000) {
125        let traces = (begin..(begin + 1000).min(batched)).map(trace).collect();
126        let start = Instant::now();
127        store.remember_many(traces)?;
128        batches.push(ms(start));
129    }
130    let mut appends = Vec::new();
131    for i in batched..records {
132        let t = trace(i);
133        let start = Instant::now();
134        store.remember(t)?;
135        appends.push(ms(start));
136    }
137    let seed_ms = ms(seed_started);
138    let memory_after_seed = rss();
139    let mut recall_ms = Vec::new();
140    let mut hits = 0;
141    let mut returned = 0;
142    let mut result_ids = Vec::new();
143    // Warm up with an exact cue; quality here is a synthetic exact-cue check, not field recall.
144    std::hint::black_box(store.recall(&RecallQuery::new(token(0))));
145    for i in 0..queries {
146        let selected = i * records / queries;
147        let mut query = RecallQuery::new(token(selected));
148        query.now_ms = CLOCK;
149        query.limit = 1;
150        let start = Instant::now();
151        let found = store.recall(&query);
152        recall_ms.push(ms(start));
153        returned += found.len();
154        hits += usize::from(
155            found
156                .first()
157                .is_some_and(|r| r.trace.id == format!("sample-{selected}")),
158        );
159        result_ids.push(found.iter().map(|r| r.trace.id.clone()).collect::<Vec<_>>());
160    }
161    let start = Instant::now();
162    store.verify()?;
163    let verify_ms = ms(start);
164    let before = store.head();
165    drop(store);
166    let start = Instant::now();
167    let store = MemoryStore::open_existing(&path, KernelPolicy::default())?;
168    let replay_ms = ms(start);
169    store.verify()?;
170    let retained = store
171        .all()
172        .filter(|e| {
173            e.trace.provenance.source == "synthetic-kernel-benchmark"
174                && e.trace.provenance.observed_at_ms == CLOCK
175                && !e.trace.provenance.verified
176        })
177        .count();
178    let mut replay_matches = 0;
179    for (i, expected) in result_ids.iter().enumerate() {
180        let mut query = RecallQuery::new(token(i * records / queries));
181        query.now_ms = CLOCK;
182        query.limit = 1;
183        let ids: Vec<_> = store
184            .recall(&query)
185            .into_iter()
186            .map(|r| r.trace.id)
187            .collect();
188        replay_matches += usize::from(ids == *expected);
189    }
190    let head_same = serde_json::to_value(before)? == serde_json::to_value(store.head())?;
191    let ok = store.record_count() == records
192        && retained == records
193        && hits == queries
194        && replay_matches == queries
195        && head_same;
196    Ok(json!({
197        "kind": "synthetic_scale", "ok": ok, "records": records, "queries": queries,
198        "workload": "unclaimed observation traces, unique exact lexical cues, fixed clock",
199        "seed_ms": seed_ms, "batch_size": 1000, "batched_records": batched,
200        "batch_append_ms": quantiles(batches), "single_append_ms": quantiles(appends),
201        "recall_ms": quantiles(recall_ms), "verify_ms": verify_ms, "replay_ms": replay_ms,
202        "exact_cue_hits": hits, "returned": returned, "recall_at_1": hits as f64 / queries as f64,
203        "provenance_retained": retained, "replayed_records": store.record_count(),
204        "recall_replay_matches": replay_matches, "head_unchanged": head_same,
205        "ledger_bytes": fs::metadata(path)?.len(), "memory_after_seed": memory_after_seed,
206        "memory_after_replay": rss(), "policy": store.policy(),
207        "limits": ["Batched seeding does not measure per-record durable append latency.",
208          "Warm filesystem cache; reopening in the same process is not a cold-start measurement.",
209          "No structured claims in this workload; conflict-heavy stores need a separate workload.",
210          "Synthetic exact-cue quality is not semantic or real-world retrieval accuracy.",
211          "RSS is process-wide Linux /proc data; unavailable platforms report null."]
212    }))
213}