Skip to main content

grit_core/
query.rs

1//! Read path: point lookups, validity-filtered traversal, bi-temporal history,
2//! and merge-candidate scoring.
3
4use rusqlite::params;
5use serde::{Deserialize, Serialize};
6use uuid::Uuid;
7
8use crate::Grit;
9use crate::clock::TimestampMs;
10use crate::error::Result;
11use crate::model::{
12    self, EDGE_COLS, EPISODE_COLS, Edge, Episode, NODE_COLS, Node, Subgraph, collect,
13};
14
15/// Parameters for [`Grit::traverse`]. `Default` gives depth 3 (the AGENTS.md
16/// default bound), both timelines evaluated "now", no group filter.
17#[derive(Debug, Clone)]
18pub struct Traversal {
19    /// Maximum hops from the seed set (inclusive bound; default 3).
20    pub depth: u32,
21    /// Event-time instant: only walk edges whose `[valid_at, invalid_at)`
22    /// interval contains this. `None` = clock now.
23    pub as_of: Option<TimestampMs>,
24    /// System-time instant ("what did I believe at t?"): only walk edges whose
25    /// `[created_at, expired_at)` interval contains this. `None` = clock now.
26    pub as_at: Option<TimestampMs>,
27    /// Restrict the walk to one namespace.
28    pub group_id: Option<String>,
29    /// Node budget: the walk halts (breadth-first, nearest wins) once this
30    /// many nodes are reached. Keeps traversal latency independent of
31    /// neighborhood size — Layer 3 asks for "context that fits", never the
32    /// 7,000-node ball around a hub. Default 256.
33    pub max_nodes: usize,
34}
35
36impl Default for Traversal {
37    fn default() -> Self {
38        Self {
39            depth: 3,
40            as_of: None,
41            as_at: None,
42            group_id: None,
43            max_nodes: 256,
44        }
45    }
46}
47
48impl Traversal {
49    /// Set the hop bound.
50    pub fn depth(mut self, depth: u32) -> Self {
51        self.depth = depth;
52        self
53    }
54
55    /// Set the event-time instant.
56    pub fn as_of(mut self, t: TimestampMs) -> Self {
57        self.as_of = Some(t);
58        self
59    }
60
61    /// Set the system-time (belief) instant — this is the time-travel knob.
62    pub fn as_at(mut self, t: TimestampMs) -> Self {
63        self.as_at = Some(t);
64        self
65    }
66
67    /// Restrict to a namespace.
68    pub fn group(mut self, group_id: impl Into<String>) -> Self {
69        self.group_id = Some(group_id.into());
70        self
71    }
72
73    /// Set the node budget (`usize::MAX` for an unbounded walk).
74    pub fn max_nodes(mut self, n: usize) -> Self {
75        self.max_nodes = n;
76        self
77    }
78}
79
80/// Row counts, as reported by [`Grit::stats`].
81#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
82pub struct Stats {
83    /// Total node rows (including expired/merged-away ones).
84    pub nodes: usize,
85    /// Total edge rows.
86    pub edges: usize,
87    /// Total episodes.
88    pub episodes: usize,
89    /// Total mention links.
90    pub mentions: usize,
91    /// Total oplog entries.
92    pub oplog: usize,
93    /// Purged (tombstoned) ids.
94    pub purged: usize,
95}
96
97/// Bi-temporal audit trail for one node: every incident edge row ever
98/// believed, including invalidated and expired ones, oldest belief first.
99#[derive(Debug, Clone, Serialize, Deserialize)]
100pub struct NodeHistory {
101    /// The node row (even if expired/merged away).
102    pub node: Node,
103    /// All incident edges, unfiltered, ordered by `created_at`.
104    pub edges: Vec<Edge>,
105}
106
107/// A scored suggestion from [`Grit::find_merge_candidates`]. Grit never merges
108/// on its own — deciding is Layer 2's LLM-judgment call; grit only scores.
109#[derive(Debug, Clone, Serialize, Deserialize)]
110pub struct MergeCandidate {
111    /// The candidate duplicate.
112    pub node: Node,
113    /// Jaro-Winkler similarity of names, in `[0, 1]`.
114    pub name_score: f64,
115    /// Cosine similarity of embeddings, in `[-1, 1]`, if both are embedded.
116    pub vector_score: Option<f64>,
117    /// max(name_score, vector_score) — the ranking key.
118    pub score: f64,
119}
120
121/// One BFS level: expand every frontier node (bound as a JSON array — no SQL
122/// built from user input) across valid edges to the DISTINCT set of neighbor
123/// ids. ?1 frontier, ?2 group, ?3 as_at, ?4 as_of.
124///
125/// This is the current-belief variant: `as_at` is now, so the denormalized
126/// `edges.invalid_at` (maintained as MIN over all invalidation records) IS
127/// the belief-current value and the per-edge subquery is unnecessary.
128#[doc(hidden)]
129pub const EXPAND_SQL_CURRENT: &str = "
130SELECT DISTINCT CASE WHEN e.src = f.value THEN e.dst ELSE e.src END
131FROM json_each(?1) AS f
132JOIN edges AS e ON e.src = f.value OR e.dst = f.value
133WHERE (?2 IS NULL OR e.group_id = ?2)
134  AND e.created_at <= ?3 AND (e.expired_at IS NULL OR e.expired_at > ?3)
135  AND (e.valid_at IS NULL OR e.valid_at <= ?4)
136  AND (e.invalid_at IS NULL OR e.invalid_at > ?4)";
137
138/// Time-travel variant of [`EXPAND_SQL_CURRENT`]: a historical `as_at` must
139/// ignore invalidations recorded after it, so validity consults the
140/// belief-versioned `edge_invalidations` records instead of the folded
141/// `edges.invalid_at`.
142#[doc(hidden)]
143pub const EXPAND_SQL_AS_AT: &str = "
144SELECT DISTINCT CASE WHEN e.src = f.value THEN e.dst ELSE e.src END
145FROM json_each(?1) AS f
146JOIN edges AS e ON e.src = f.value OR e.dst = f.value
147WHERE (?2 IS NULL OR e.group_id = ?2)
148  AND e.created_at <= ?3 AND (e.expired_at IS NULL OR e.expired_at > ?3)
149  AND (e.valid_at IS NULL OR e.valid_at <= ?4)
150  AND NOT EXISTS (SELECT 1 FROM edge_invalidations AS iv
151                  WHERE iv.edge_id = e.id
152                    AND iv.recorded_at <= ?3 AND iv.invalid_at <= ?4)";
153
154/// Keep only ids that are believed nodes at ?2 (purged and merged-away nodes
155/// fail this — the walk never passes through ghosts). ORDER BY id makes the
156/// budget cut deterministic.
157const BELIEVED_NODES_SQL: &str = "
158SELECT id FROM nodes
159WHERE id IN (SELECT value FROM json_each(?1))
160  AND created_at <= ?2 AND (expired_at IS NULL OR expired_at > ?2)
161ORDER BY id";
162
163impl Grit {
164    /// Row counts across the whole file, from one consistent snapshot.
165    pub fn stats(&self) -> Result<Stats> {
166        let conn = self.read();
167        let tx = conn.unchecked_transaction()?;
168        let count = |table: &str| -> Result<usize> {
169            // Table names are compile-time constants below, never user input.
170            let n: i64 =
171                tx.query_row(&format!("SELECT COUNT(*) FROM {table}"), [], |r| r.get(0))?;
172            Ok(n as usize)
173        };
174        Ok(Stats {
175            nodes: count("nodes")?,
176            edges: count("edges")?,
177            episodes: count("episodes")?,
178            mentions: count("mentions")?,
179            oplog: count("oplog")?,
180            purged: count("purged")?,
181        })
182    }
183
184    /// Fetch one node by id (regardless of validity — point lookups see
185    /// everything; filtering is for traversal/search).
186    pub fn node(&self, id: Uuid) -> Result<Option<Node>> {
187        node_on(&self.read(), id)
188    }
189
190    /// Fetch one edge by id.
191    pub fn edge(&self, id: Uuid) -> Result<Option<Edge>> {
192        let conn = self.read();
193        let mut stmt =
194            conn.prepare_cached(&format!("SELECT {EDGE_COLS} FROM edges WHERE id = ?1"))?;
195        let mut rows = collect(stmt.query_map(params![id.to_string()], Edge::from_row)?)?;
196        Ok(rows.pop())
197    }
198
199    /// Fetch one episode by id.
200    pub fn episode(&self, id: Uuid) -> Result<Option<Episode>> {
201        let conn = self.read();
202        let mut stmt = conn.prepare_cached(&format!(
203            "SELECT {EPISODE_COLS} FROM episodes WHERE id = ?1"
204        ))?;
205        let mut rows = collect(stmt.query_map(params![id.to_string()], Episode::from_row)?)?;
206        Ok(rows.pop())
207    }
208
209    /// Every node in a group, `id`-ordered (deterministic). Includes expired
210    /// and merged-away rows — callers filter on [`Node::expired_at`] /
211    /// [`Node::merged_into`] when they want only live entities.
212    ///
213    /// # Example
214    /// ```no_run
215    /// # use grit_core::{Grit, Options};
216    /// # let g = Grit::open("memory.db", Options::new("laptop"))?;
217    /// let live: Vec<_> = g
218    ///     .nodes_in_group("algebra")?
219    ///     .into_iter()
220    ///     .filter(|n| n.expired_at.is_none())
221    ///     .collect();
222    /// # Ok::<(), grit_core::Error>(())
223    /// ```
224    pub fn nodes_in_group(&self, group_id: &str) -> Result<Vec<Node>> {
225        let conn = self.read();
226        let mut stmt = conn.prepare_cached(&format!(
227            "SELECT {NODE_COLS} FROM nodes WHERE group_id = ?1 ORDER BY id"
228        ))?;
229        collect(stmt.query_map(params![group_id], Node::from_row)?)
230    }
231
232    /// Every edge in a group, `id`-ordered (deterministic). Includes
233    /// invalidated and expired rows — the full bi-temporal record, same view
234    /// as [`Grit::export_jsonl`].
235    ///
236    /// # Example
237    /// ```no_run
238    /// # use grit_core::{Grit, Options};
239    /// # let g = Grit::open("memory.db", Options::new("laptop"))?;
240    /// let facts = g.edges_in_group("algebra")?.len();
241    /// # Ok::<(), grit_core::Error>(())
242    /// ```
243    pub fn edges_in_group(&self, group_id: &str) -> Result<Vec<Edge>> {
244        let conn = self.read();
245        let mut stmt = conn.prepare_cached(&format!(
246            "SELECT {EDGE_COLS} FROM edges WHERE group_id = ?1 ORDER BY id"
247        ))?;
248        collect(stmt.query_map(params![group_id], Edge::from_row)?)
249    }
250
251    /// Every episode in a group, ordered by event time then id
252    /// (deterministic, chronological — the natural order for building
253    /// previous-episode context windows).
254    ///
255    /// # Example
256    /// ```no_run
257    /// # use grit_core::{Grit, Options};
258    /// # let g = Grit::open("memory.db", Options::new("laptop"))?;
259    /// let history = g.episodes_in_group("chat")?;
260    /// let latest = history.last();
261    /// # Ok::<(), grit_core::Error>(())
262    /// ```
263    pub fn episodes_in_group(&self, group_id: &str) -> Result<Vec<Episode>> {
264        let conn = self.read();
265        let mut stmt = conn.prepare_cached(&format!(
266            "SELECT {EPISODE_COLS} FROM episodes WHERE group_id = ?1
267             ORDER BY occurred_at, id"
268        ))?;
269        collect(stmt.query_map(params![group_id], Episode::from_row)?)
270    }
271
272    /// The embedding stored for a node via [`Grit::set_node_embedding`], or
273    /// `None` when the node has no vector (or no model is registered).
274    ///
275    /// # Example
276    /// ```no_run
277    /// # use grit_core::{Grit, Options};
278    /// # let g = Grit::open("memory.db", Options::new("laptop"))?;
279    /// # let id = g.new_id();
280    /// if g.get_node_embedding(id)?.is_none() {
281    ///     // embed and g.set_node_embedding(id, vector)?
282    /// }
283    /// # Ok::<(), grit_core::Error>(())
284    /// ```
285    pub fn get_node_embedding(&self, id: Uuid) -> Result<Option<Vec<f32>>> {
286        self.get_embedding("vec_nodes", id)
287    }
288
289    /// The embedding stored for an edge via [`Grit::set_edge_embedding`], or
290    /// `None` when the edge has no vector (or no model is registered).
291    pub fn get_edge_embedding(&self, id: Uuid) -> Result<Option<Vec<f32>>> {
292        self.get_embedding("vec_edges", id)
293    }
294
295    fn get_embedding(&self, table: &str, id: Uuid) -> Result<Option<Vec<f32>>> {
296        let conn = self.read();
297        // The vec tables are created lazily by register_embedding_model.
298        let has_table: i64 = conn.query_row(
299            "SELECT COUNT(*) FROM sqlite_master WHERE name = ?1",
300            params![table],
301            |r| r.get(0),
302        )?;
303        if has_table == 0 {
304            return Ok(None);
305        }
306        // `table` is one of two crate-internal constants — never user input.
307        let mut stmt =
308            conn.prepare_cached(&format!("SELECT embedding FROM {table} WHERE id = ?1"))?;
309        let mut rows =
310            collect(stmt.query_map(params![id.to_string()], |r| r.get::<_, Vec<u8>>(0))?)?;
311        Ok(rows.pop().map(|bytes| crate::vecext::bytes_as_f32s(&bytes)))
312    }
313
314    /// Episode ids that mention the given node or edge (provenance lookup).
315    /// Mention rows keep their original target ids, so this resolves `target`
316    /// to its canonical node and gathers mentions across all merged aliases.
317    pub fn mentions_of(&self, target: Uuid) -> Result<Vec<Uuid>> {
318        let conn = self.read();
319        let tx = conn.unchecked_transaction()?;
320        episodes_mentioning(&tx, &target.to_string())
321    }
322
323    /// Expand from `from` along edges valid at the requested instant, up to
324    /// `spec.depth` hops, and return the induced subgraph. Time-travel is the
325    /// public API here (Design Invariant 4): pass `as_at` for "what did I
326    /// believe at t?", `as_of` for "what was true at t?".
327    ///
328    /// # Example
329    /// ```no_run
330    /// # use grit_core::{Grit, Options, Traversal};
331    /// # let g = Grit::open("memory.db", Options::new("laptop"))?;
332    /// # let (node_id, march) = (g.new_id(), 0);
333    /// let believed_in_march = g.traverse(&[node_id], &Traversal::default().as_at(march))?;
334    /// # Ok::<(), grit_core::Error>(())
335    /// ```
336    pub fn traverse(&self, from: &[Uuid], spec: &Traversal) -> Result<Subgraph> {
337        use std::collections::HashSet;
338
339        let now = self.now_ms();
340        let as_of = spec.as_of.unwrap_or(now);
341        let as_at = spec.as_at.unwrap_or(now);
342        // A historical as_at must ignore invalidations recorded after it;
343        // otherwise the folded edges.invalid_at is exact and much cheaper.
344        let expand_sql = if spec.as_at.is_none_or(|t| t >= now) {
345            EXPAND_SQL_CURRENT
346        } else {
347            EXPAND_SQL_AS_AT
348        };
349
350        let conn = self.read();
351        // One read transaction for the whole walk: every BFS level and the
352        // final fetches see the same committed snapshot — a write landing
353        // mid-traversal cannot produce a subgraph no database state ever held.
354        let tx = conn.unchecked_transaction()?;
355
356        // Stale handles are fine: seeds follow merged_into to their canonical
357        // node before the walk starts, then must themselves be believed at
358        // as_at (purged / merged-away nodes are dropped, so the walk never
359        // starts inside a ghost).
360        let mut in_visited: HashSet<String> = HashSet::new();
361        let mut seeds: Vec<String> = Vec::with_capacity(from.len());
362        for id in from {
363            let canon = resolve_read(&tx, &id.to_string())?;
364            if in_visited.insert(canon.clone()) {
365                seeds.push(canon);
366            }
367        }
368        let believed: HashSet<String> = believed_nodes(&tx, &seeds, as_at)?.into_iter().collect();
369        let mut visited: Vec<String> = seeds
370            .into_iter()
371            .filter(|s| believed.contains(s))
372            .take(spec.max_nodes)
373            .collect();
374        in_visited = visited.iter().cloned().collect();
375
376        // Level-synchronous BFS: one batched expansion per depth level, each
377        // node expanded exactly once, believed-ness checked once per NEW node
378        // (not once per edge). The budget cut is deterministic — nearest
379        // level first, ordered by id within a level.
380        let mut frontier = visited.clone();
381        for _ in 0..spec.depth {
382            if frontier.is_empty() || visited.len() >= spec.max_nodes {
383                break;
384            }
385            let neighbors: Vec<String> = {
386                let mut stmt = tx.prepare_cached(expand_sql)?;
387                let rows = stmt.query_map(
388                    params![
389                        serde_json::to_string(&frontier)?,
390                        spec.group_id,
391                        as_at,
392                        as_of
393                    ],
394                    |r| r.get::<_, String>(0),
395                )?;
396                collect(rows)?
397            };
398            let fresh: Vec<String> = neighbors
399                .into_iter()
400                .filter(|id| !in_visited.contains(id))
401                .collect();
402            let fresh: Vec<String> = believed_nodes(&tx, &fresh, as_at)?
403                .into_iter()
404                .filter(|id| !in_visited.contains(id))
405                .take(spec.max_nodes - visited.len())
406                .collect();
407            for id in &fresh {
408                in_visited.insert(id.clone());
409            }
410            visited.extend(fresh.iter().cloned());
411            frontier = fresh;
412        }
413        let visited_json = serde_json::to_string(&visited)?;
414
415        let nodes = {
416            let mut stmt = tx.prepare_cached(&format!(
417                "SELECT {NODE_COLS} FROM nodes
418                 WHERE id IN (SELECT value FROM json_each(?1))
419                 ORDER BY id"
420            ))?;
421            collect(stmt.query_map(params![visited_json], Node::from_row)?)?
422        };
423        // Induced subgraph: both endpoints in the visited set, which contains
424        // only believed nodes by construction — ghost edges (an endpoint
425        // purged or merged away) can never qualify.
426        let invalid_clause = if spec.as_at.is_none_or(|t| t >= now) {
427            "(e.invalid_at IS NULL OR e.invalid_at > ?4)"
428        } else {
429            "NOT EXISTS (SELECT 1 FROM edge_invalidations AS iv
430                         WHERE iv.edge_id = e.id
431                           AND iv.recorded_at <= ?3 AND iv.invalid_at <= ?4)"
432        };
433        let edges = {
434            let mut stmt = tx.prepare_cached(&format!(
435                "SELECT {EDGE_COLS} FROM edges AS e
436                 WHERE e.src IN (SELECT value FROM json_each(?1))
437                   AND e.dst IN (SELECT value FROM json_each(?1))
438                   AND (?2 IS NULL OR e.group_id = ?2)
439                   AND e.created_at <= ?3 AND (e.expired_at IS NULL OR e.expired_at > ?3)
440                   AND (e.valid_at IS NULL OR e.valid_at <= ?4)
441                   AND {invalid_clause}
442                 ORDER BY e.id"
443            ))?;
444            collect(stmt.query_map(
445                params![visited_json, spec.group_id, as_at, as_of],
446                Edge::from_row,
447            )?)?
448        };
449        Ok(Subgraph { nodes, edges })
450    }
451
452    /// Full bi-temporal audit for a node: the node row plus every incident
453    /// edge ever believed, invalidated and expired included — one snapshot.
454    pub fn node_history(&self, id: Uuid) -> Result<NodeHistory> {
455        let conn = self.read();
456        let tx = conn.unchecked_transaction()?;
457        let node = node_on(&tx, id)?.ok_or(crate::Error::NotFound(id))?;
458        let mut stmt = tx.prepare_cached(&format!(
459            "SELECT {EDGE_COLS} FROM edges WHERE src = ?1 OR dst = ?1
460             ORDER BY created_at, id"
461        ))?;
462        let edges = collect(stmt.query_map(params![id.to_string()], Edge::from_row)?)?;
463        Ok(NodeHistory { node, edges })
464    }
465
466    /// Score possible duplicates of `id` within its group: Jaro-Winkler on
467    /// names, cosine over embeddings when present. Returns candidates with
468    /// `score >= min_score`, best first. Grit never acts on these — executing
469    /// a [`crate::GraphOp::MergeNodes`] is Layer 2's decision.
470    pub fn find_merge_candidates(&self, id: Uuid, min_score: f64) -> Result<Vec<MergeCandidate>> {
471        let conn = self.read();
472        let tx = conn.unchecked_transaction()?;
473        let target = node_on(&tx, id)?.ok_or(crate::Error::NotFound(id))?;
474        let vector_scores = node_vector_neighbors(&tx, id)?;
475
476        let others = {
477            let mut stmt = tx.prepare_cached(&format!(
478                "SELECT {NODE_COLS} FROM nodes
479                 WHERE group_id = ?1 AND id != ?2 AND expired_at IS NULL"
480            ))?;
481            collect(stmt.query_map(params![target.group_id, id.to_string()], Node::from_row)?)?
482        };
483        drop(tx);
484        drop(conn);
485
486        let target_name = target.name.to_lowercase();
487        let mut out: Vec<MergeCandidate> = others
488            .into_iter()
489            .map(|node| {
490                let name_score = strsim::jaro_winkler(&target_name, &node.name.to_lowercase());
491                let vector_score = vector_scores.get(&node.id).copied();
492                let score = vector_score.map_or(name_score, |v| name_score.max(v));
493                MergeCandidate {
494                    node,
495                    name_score,
496                    vector_score,
497                    score,
498                }
499            })
500            .filter(|c| c.score >= min_score)
501            .collect();
502        out.sort_by(|a, b| b.score.total_cmp(&a.score));
503        Ok(out)
504    }
505}
506
507/// Cosine similarity of every embedded node against `id`'s embedding,
508/// empty if `id` has no embedding (or no model is registered).
509fn node_vector_neighbors(
510    conn: &rusqlite::Connection,
511    id: Uuid,
512) -> Result<std::collections::HashMap<Uuid, f64>> {
513    let has_vec: i64 = conn.query_row(
514        "SELECT COUNT(*) FROM sqlite_master WHERE name = 'vec_nodes'",
515        [],
516        |r| r.get(0),
517    )?;
518    let mut out = std::collections::HashMap::new();
519    if has_vec == 0 {
520        return Ok(out);
521    }
522    let embedding: Option<Vec<u8>> = {
523        let mut stmt = conn.prepare_cached("SELECT embedding FROM vec_nodes WHERE id = ?1")?;
524        let mut rows =
525            collect(stmt.query_map(params![id.to_string()], |r| r.get::<_, Vec<u8>>(0))?)?;
526        rows.pop()
527    };
528    let Some(embedding) = embedding else {
529        return Ok(out);
530    };
531    // distance_metric=cosine ⇒ distance = 1 - cos_sim.
532    // KNN queries only allow the MATCH + k constraints; the self-hit is
533    // filtered out here instead.
534    let mut stmt = conn
535        .prepare_cached("SELECT id, distance FROM vec_nodes WHERE embedding MATCH ?1 AND k = 64")?;
536    let rows = stmt.query_map(params![embedding], |r| {
537        Ok((r.get::<_, String>(0)?, r.get::<_, f64>(1)?))
538    })?;
539    for row in rows {
540        let (nid, distance) = row?;
541        if let Ok(nid) = Uuid::parse_str(&nid)
542            && nid != id
543        {
544            out.insert(nid, 1.0 - distance);
545        }
546    }
547    Ok(out)
548}
549
550/// Point lookup on an already-held connection/transaction.
551fn node_on(conn: &rusqlite::Connection, id: Uuid) -> Result<Option<Node>> {
552    let mut stmt = conn.prepare_cached(&format!("SELECT {NODE_COLS} FROM nodes WHERE id = ?1"))?;
553    let mut rows = collect(stmt.query_map(params![id.to_string()], Node::from_row)?)?;
554    Ok(rows.pop())
555}
556
557pub(crate) fn ids_json<'a>(ids: impl Iterator<Item = &'a Uuid>) -> String {
558    serde_json::to_string(&ids.map(Uuid::to_string).collect::<Vec<_>>())
559        .expect("Vec<String> serialization cannot fail")
560}
561
562/// Filter `ids` down to believed nodes at `as_at`, sorted by id (see
563/// [`BELIEVED_NODES_SQL`]).
564fn believed_nodes(
565    conn: &rusqlite::Connection,
566    ids: &[String],
567    as_at: TimestampMs,
568) -> Result<Vec<String>> {
569    if ids.is_empty() {
570        return Ok(Vec::new());
571    }
572    let mut stmt = conn.prepare_cached(BELIEVED_NODES_SQL)?;
573    let rows = stmt.query_map(params![serde_json::to_string(ids)?, as_at], |r| {
574        r.get::<_, String>(0)
575    })?;
576    collect(rows)
577}
578
579/// Provenance across merges: resolve `id` to its canonical node, then collect
580/// mentions of every alias that folded into it (mention rows keep original
581/// target ids — the writer never re-points them).
582pub(crate) fn episodes_mentioning(
583    conn: &rusqlite::Connection,
584    id: &str,
585) -> Result<Vec<uuid::Uuid>> {
586    let canon = resolve_read(conn, id)?;
587    let mut stmt = conn.prepare_cached(
588        "WITH RECURSIVE aliases (id) AS (
589             SELECT ?1
590             UNION
591             SELECT n.id FROM nodes AS n JOIN aliases AS a ON n.merged_into = a.id
592             UNION
593             SELECT p.id FROM purged AS p JOIN aliases AS a ON p.merged_into = a.id
594         )
595         SELECT DISTINCT episode_id FROM mentions
596         WHERE target_id IN (SELECT id FROM aliases)
597         ORDER BY episode_id",
598    )?;
599    let rows = stmt.query_map(params![canon], |r| model::uuid_col(r, 0))?;
600    collect(rows)
601}
602
603/// Read-side twin of the writer's canonical resolution: follow `merged_into`
604/// to the surviving node, cycle-safe (min id wins, same rule as apply).
605fn resolve_read(conn: &rusqlite::Connection, id: &str) -> Result<String> {
606    use rusqlite::OptionalExtension;
607    let mut chain: Vec<String> = vec![id.to_owned()];
608    let mut seen: std::collections::HashSet<String> = chain.iter().cloned().collect();
609    loop {
610        let current = chain.last().expect("chain never empty");
611        let mut stmt = conn.prepare_cached(
612            "SELECT merged_into FROM nodes WHERE id = ?1
613             UNION ALL
614             SELECT merged_into FROM purged WHERE id = ?1
615             LIMIT 1",
616        )?;
617        let next: Option<Option<String>> =
618            stmt.query_row(params![current], |r| r.get(0)).optional()?;
619        match next {
620            None | Some(None) => return Ok(chain.pop().expect("chain never empty")),
621            Some(Some(next)) => {
622                if seen.contains(&next) {
623                    let pos = chain
624                        .iter()
625                        .position(|x| x == &next)
626                        .expect("seen implies present");
627                    return Ok(chain[pos..]
628                        .iter()
629                        .min()
630                        .expect("cycle slice non-empty")
631                        .clone());
632                }
633                seen.insert(next.clone());
634                chain.push(next);
635            }
636        }
637    }
638}