1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
//! Bitemporal recall — `recall_as_of(query, t)`: what did this database
//! believe at time `t`?
//!
//! The engine never silently loses state: `correct()` mutates in place
//! but archives the prior text/metadata/importance/valence in
//! `record_revisions` (with `applied_at`), and supersession is an edge
//! in `record_links` (with `created_at`). Those two ledgers are enough
//! to answer an as-of query WITHOUT a new storage model:
//!
//! 1. a record that did not exist at `t` is excluded
//! (`memories.created_at > t`);
//! 2. a record superseded at `t` is excluded — but only by edges that
//! already existed at `t` (`record_links.created_at <= t`); a later
//! supersession does not rewrite what was believed then;
//! 3. a record corrected after `t` is rolled back to its `t`-state by
//! walking `record_revisions` for the earliest revision applied
//! after `t` — its `prior_*` columns ARE the state at `t`.
//!
//! HONEST LIMITS, documented rather than hidden:
//!
//! - **Ranking is present-day.** Similarity is computed against today's
//! vectors (a corrected record's embedding cannot be un-embedded;
//! `prior_embedding_hash` lets history explain the drift, not undo
//! it), and decay/importance components are today's. `recall_as_of`
//! reconstructs CONTENT and CURRENCY at `t`, not the exact ranking a
//! query would have produced at `t`.
//! - **Forgotten records stay forgotten.** A tombstoned record is
//! physically out of the recall path (cache and index), so an as-of
//! query cannot resurface something forgotten since `t`. The oplog
//! holds that history for explicit forensics; recall does not.
//! - **Access stats are untouched.** The pool query runs with
//! `skip_reinforce` — archaeology must not masquerade as usage.
//!
//! Results keep their present-day `current_status`: a hit can honestly
//! read "this was current at `t`; today it is superseded", which is
//! exactly what an audit wants to see.
use rusqlite::{params, OptionalExtension};
use crate::error::Result;
use crate::types::RecallResult;
use super::YantrikDB;
/// Over-fetch multiplier for the candidate pool: as-of filtering can
/// only shrink the pool, so the underlying recall casts a wider net.
const AS_OF_POOL_FACTOR: usize = 4;
/// Upper bound on the pool a single as-of query may hydrate.
const AS_OF_POOL_CAP: usize = 100;
impl YantrikDB {
/// Recall against the state of belief at `as_of` (epoch seconds).
///
/// See the module docs for exact semantics and honest limits.
pub fn recall_as_of(
&self,
query_embedding: &[f32],
top_k: usize,
as_of: f64,
namespace: Option<&str>,
memory_type: Option<&str>,
) -> Result<Vec<RecallResult>> {
let fetch_k = top_k
.saturating_mul(AS_OF_POOL_FACTOR)
.clamp(top_k, AS_OF_POOL_CAP.max(top_k));
let mut pool = self.recall(
query_embedding,
fetch_k,
None, // time_window — as-of does its own temporal filtering
memory_type,
false, // include_consolidated
// Aligned 2026-08-15: graph expansion measured −0.24 MRR and is
// off by default on every other surface; this pre-decision
// hardcode was never revisited (no comment defended it), and
// as_of passes query_text=None — the weakest seeding mode.
false, // expand_entities
None, // query_text
true, // skip_reinforce: archaeology is not usage
namespace,
None, // domain
None, // source
None, // certainty_min
None, // order
true, // include_superseded: the as-of filter decides, not today's
None, // event_after (#149)
None, // event_before (#149)
)?;
// 1. Nothing that did not exist yet.
pool.retain(|r| r.created_at <= as_of);
// 2. Nothing already superseded at `as_of`. Edges created after
// `as_of` do not suppress what was still current then.
if !pool.is_empty() {
let rids: Vec<&str> = pool.iter().map(|r| r.rid.as_str()).collect();
let superseded = self.superseded_rids_as_of(&rids, as_of)?;
if !superseded.is_empty() {
pool.retain(|r| !superseded.contains(&r.rid));
}
}
// 3. Roll corrected records back to their state at `as_of`.
for result in pool.iter_mut() {
self.rollback_result_to(result, as_of)?;
}
pool.truncate(top_k);
Ok(pool)
}
/// Which of `rids` were superseded by an edge that already existed
/// at `as_of`. Same shape as `superseded_rids_among`, with the
/// edge-creation clock applied.
fn superseded_rids_as_of(
&self,
rids: &[&str],
as_of: f64,
) -> Result<std::collections::HashSet<String>> {
if rids.is_empty() {
return Ok(std::collections::HashSet::new());
}
let placeholders: String = (0..rids.len())
.map(|i| format!("?{}", i + 1))
.collect::<Vec<_>>()
.join(",");
let sql = format!(
"SELECT DISTINCT target_rid FROM record_links \
WHERE link_type = 'supersedes' \
AND status = 'active' AND selection_state = 'selected' \
AND created_at <= ?{} \
AND target_rid IN ({placeholders})",
rids.len() + 1
);
let mut param_values: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
for r in rids {
param_values.push(Box::new(r.to_string()));
}
param_values.push(Box::new(as_of));
let params_ref: Vec<&dyn rusqlite::types::ToSql> =
param_values.iter().map(|p| p.as_ref()).collect();
let conn = self.read_conn();
let mut stmt = conn.prepare(&sql)?;
let rows = stmt
.query_map(params_ref.as_slice(), |row| row.get::<_, String>(0))?
.collect::<std::result::Result<std::collections::HashSet<_>, _>>()?;
Ok(rows)
}
/// If `result`'s record was corrected after `as_of`, swap in the
/// state it had at `as_of`. The earliest revision applied AFTER
/// `as_of` archived exactly the pre-correction state we want.
fn rollback_result_to(&self, result: &mut RecallResult, as_of: f64) -> Result<()> {
let conn = self.read_conn();
let prior = conn
.query_row(
"SELECT prior_text, prior_metadata, prior_importance, \
prior_valence \
FROM record_revisions \
WHERE rid = ?1 AND applied_at > ?2 \
ORDER BY revision_num ASC LIMIT 1",
params![result.rid, as_of],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, f64>(2)?,
row.get::<_, f64>(3)?,
))
},
)
.optional()?;
if let Some((text, metadata, importance, valence)) = prior {
// Revisions are archived in STORED (encrypted-or-plain) form by
// design (see the correct() revision insert). Every other read
// path decrypts on hydration; this one assigned the stored form
// straight into the result — on encrypted DBs the caller got
// ciphertext text and Null metadata (serde on ciphertext),
// silently. 2026-08-15 surface audit.
result.text = self.decrypt_text(&text)?;
let metadata_plain = self.decrypt_text(&metadata)?;
result.metadata =
serde_json::from_str(&metadata_plain).unwrap_or(serde_json::Value::Null);
result.importance = importance;
result.valence = valence;
// #181: recall stamped the bounds from the live row, and
// correct() re-derives event time — so without this the result
// pairs yesterday's metadata with today's dates. The revision's
// own JSON is the only record of the bounds as of `as_of`.
let (event_time_min, event_time_max) =
crate::base::datetext::event_time_bounds(&result.metadata);
result.event_time_min = event_time_min;
result.event_time_max = event_time_max;
result
.why_retrieved
.push("as_of: rolled back to pre-correction state".to_string());
}
Ok(())
}
}