rto_graph/store.rs
1//! SQLite-backed graph store.
2
3use std::path::Path;
4
5use rusqlite::{Connection, OptionalExtension, params};
6
7use crate::findings::{self, AnalysisRun, Finding, FindingsApplied, FindingsLayer};
8use crate::media::{self, MediaFilter, MediaKind, MediaRecord, MediaWrite, ProducerSummary};
9use crate::memory::{
10 self, CacheEntry, CacheStats, CacheSweep, CacheWrite, MemoryError, MemoryFilter,
11 MemoryForgotten, MemoryListing, MemoryRecord, MemoryWrite, Recall, RecallOptions,
12};
13use crate::migrations;
14use crate::model::{Direction, Edge, EdgeKind, FactSet, Node, NodeKind, Span};
15use crate::provenance::Provenance;
16
17/// Errors raised by the store.
18#[derive(Debug, thiserror::Error)]
19pub enum StoreError {
20 /// Underlying `SQLite` failure.
21 #[error("sqlite error: {0}")]
22 Sqlite(#[from] rusqlite::Error),
23 /// A node's `meta` could not be (de)serialized as JSON.
24 #[error("json error: {0}")]
25 Json(#[from] serde_json::Error),
26 /// An edge referenced a node key that does not exist in the store.
27 #[error("unknown node key: {0}")]
28 UnknownNode(String),
29 /// An edge violated the provenance/confidence invariant.
30 #[error("invalid edge: {0}")]
31 InvalidEdge(String),
32 /// A stored value could not be interpreted (database corruption).
33 #[error("corrupt store: {0}")]
34 Corrupt(String),
35}
36
37/// A store written by a build **newer** than this one: it records migrations
38/// this binary has never heard of. Produced by [`Store::schema_ahead`].
39///
40/// # Why this is its own type, and not a `StoreError` variant
41///
42/// Opening such a store is not an error and reading it is provably sound —
43/// migrations are additive in effect, so every column an older build selects is
44/// still there (the one `DROP TABLE` in [`crate::migrations`] is a table rebuild
45/// that re-selects every prior column). What is *not* sound is **rewriting** the
46/// graph: this build would re-extract every file with its older extractor and
47/// replace the newer build's content with worse content, silently. So the
48/// refusal belongs on the write paths, and the carrier of that refusal has no
49/// business in the type every `Store` call returns.
50///
51/// The second reason is semver, and it is the binding one. `StoreError` is a
52/// public enum on a published 1.x crate and is not `#[non_exhaustive]`, so
53/// adding a variant would stop a downstream exhaustive `match` from compiling —
54/// a breaking change, which in this workspace means a major bump of all seven
55/// crates. A new type is purely additive. Its fields are private and read
56/// through accessors for the same reason: adding one later stays non-breaking.
57#[derive(Debug, Clone, PartialEq, Eq)]
58pub struct SchemaAhead {
59 store: u32,
60 build: u32,
61 unknown: Vec<u32>,
62}
63
64impl SchemaAhead {
65 /// The highest schema version this store records — necessarily one this
66 /// build does not know.
67 #[must_use]
68 pub fn store_version(&self) -> u32 {
69 self.store
70 }
71
72 /// The highest schema version this build knows how to apply.
73 #[must_use]
74 pub fn build_version(&self) -> u32 {
75 self.build
76 }
77
78 /// Every recorded version this build has never heard of, ascending. Never
79 /// empty. More than one means the store ran several unknown migrations, not
80 /// that anything is inconsistent.
81 #[must_use]
82 pub fn unknown_versions(&self) -> &[u32] {
83 &self.unknown
84 }
85}
86
87impl std::fmt::Display for SchemaAhead {
88 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
89 let unknown = self
90 .unknown
91 .iter()
92 .map(u32::to_string)
93 .collect::<Vec<_>>()
94 .join(", ");
95 // Both versions, and what to do about it. A bare "schema mismatch" would
96 // leave the reader with nothing to act on: the useful facts are *which*
97 // side is behind (this binary) and that the fix is upgrading it, not
98 // deleting the store.
99 write!(
100 f,
101 "this graph store was written by a newer Roteiro: it is at schema \
102 version {store}, and this build knows only up to {build} \
103 (unrecognised migration(s): {unknown}). Refusing to rewrite the \
104 graph — this build would re-extract every file with its older \
105 extractor and replace the newer build's graph with worse content, \
106 silently. Upgrade the `roteiro` binary to one that knows schema \
107 version {store} or later, then retry. Reading this store is \
108 unaffected and needs no upgrade.",
109 store = self.store,
110 build = self.build,
111 )
112 }
113}
114
115impl std::error::Error for SchemaAhead {}
116
117/// A summary of applying/re-applying import layers (see
118/// [`Store::apply_import_layer`] and [`Store::reapply_imports`]).
119#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize)]
120pub struct ImportApplied {
121 /// Number of import layers processed.
122 pub layers: usize,
123 /// Import nodes upserted (across all layers).
124 pub nodes: usize,
125 /// Import edges applied — both endpoints resolved. A duplicate of an
126 /// already-present edge is a harmless no-op but still counted as applied.
127 pub edges_applied: usize,
128 /// Import edges **pruned**: an endpoint was absent (a cross-reference to code
129 /// that no longer exists), so the edge was dropped from the persisted layer
130 /// rather than kept as stale data.
131 pub edges_pruned: usize,
132}
133
134/// Qualified node columns for `SELECT`s that alias the `nodes` table as `n`.
135const NODE_COLS: &str = "n.key, n.kind, n.name, n.path, n.lang, n.blob_hash, n.span_start, n.span_end, n.provenance, n.meta";
136
137/// `SELECT` prefix that yields an [`Edge`] row (endpoints resolved back to keys).
138const EDGE_SELECT: &str = "SELECT ns.key AS src, nd.key AS dst, e.kind, e.provenance, \
139 e.confidence, e.src_ref \
140 FROM edges e JOIN nodes ns ON ns.id = e.src JOIN nodes nd ON nd.id = e.dst";
141
142/// A Roteiro graph store backed by a single `SQLite` database.
143pub struct Store {
144 conn: Connection,
145}
146
147impl Store {
148 /// Open (creating if absent) a store at `path` and apply pending migrations.
149 ///
150 /// # Errors
151 /// Returns [`StoreError::Sqlite`] if the database cannot be opened or a
152 /// migration fails.
153 pub fn open(path: &Path) -> Result<Self, StoreError> {
154 let conn = Connection::open(path)?;
155 Self::from_conn(conn)
156 }
157
158 /// Open an in-memory store (tests, previews).
159 ///
160 /// # Errors
161 /// Returns [`StoreError::Sqlite`] if a migration fails.
162 pub fn open_in_memory() -> Result<Self, StoreError> {
163 let conn = Connection::open_in_memory()?;
164 Self::from_conn(conn)
165 }
166
167 fn from_conn(mut conn: Connection) -> Result<Self, StoreError> {
168 conn.execute_batch("PRAGMA foreign_keys = ON;")?;
169 // Wait briefly for a concurrent writer instead of failing a read with
170 // `database is locked`. Matters for workspace `serve` (ADR-0008), where a
171 // long-lived server reads a project's graph while that repo's own
172 // `roteiro sync` commits an update to the same file. Syncs are
173 // sub-second, so this only ever costs a short wait, never a lost query.
174 conn.busy_timeout(std::time::Duration::from_secs(5))?;
175 migrations::apply(&mut conn)?;
176 Ok(Self { conn })
177 }
178
179 /// The schema version this store has been migrated to: the highest `v` for
180 /// which every migration `1..=v` is recorded as applied.
181 ///
182 /// **Not the maximum recorded version.** Migrations are selected by set
183 /// membership rather than `> MAX(version)` (see [`crate::migrations`]), so a
184 /// store *can* hold a gap — one written by a build that knew a higher
185 /// migration but not a lower one. For a store recorded as `1..11, 13`, the
186 /// maximum is 13 while none of migration 12's schema is present; reporting 13
187 /// would be a higher number than the truth. The contiguous prefix reports 11,
188 /// which is the version the store can actually be relied on to provide.
189 ///
190 /// The gap is transient in practice — the next [`Store::open`] repairs it —
191 /// but this must not answer wrongly in the window where it exists, which is
192 /// exactly the window in which someone is debugging.
193 ///
194 /// # Errors
195 /// Returns [`StoreError::Sqlite`] on query failure.
196 pub fn schema_version(&self) -> Result<u32, StoreError> {
197 Ok(migrations::store_version(&self.conn)?)
198 }
199
200 /// The highest schema version **this build** knows how to apply — the
201 /// binary's half of the comparison [`Store::schema_ahead`] makes.
202 #[must_use]
203 pub fn build_schema_version() -> u32 {
204 migrations::latest_version()
205 }
206
207 /// `Some` when this store was written by a **newer** build: it records
208 /// migrations this binary has never heard of. `None` — the ordinary case —
209 /// when the store is level with this build or behind it.
210 ///
211 /// Call this before **rewriting** a graph (sync, reconcile, rebuild). Do not
212 /// call it to gate reads: an older binary's reads are sound, which is the
213 /// whole reason this is not checked in [`Store::open`] (issue #342).
214 ///
215 /// # Which version this compares, and why it is not [`Store::schema_version`]
216 ///
217 /// It compares the **set** of recorded versions against the migrations this
218 /// build carries, so it answers *has anything newer than me written here?*
219 /// [`Store::schema_version`] answers a different question — the highest
220 /// **gap-free** version, a floor describing what a reader may rely on — and
221 /// using it here would let the bug through: a store recorded `1..=13, 15`,
222 /// read by a build that knows 13, has a gap-free version of 13, so
223 /// `schema_version() > build` is false while migration 15's schema sits in
224 /// the file and a newer build plainly assembled the graph.
225 ///
226 /// Conflating the two situations would also misreport both, so it does not.
227 /// A **gap** — a store missing a lower migration this build knows — is not
228 /// a store from the future, and [`Store::open`] has already repaired it by
229 /// applying the missing migration before this can be called. What is left is
230 /// only ever a version no build of this vintage could have written.
231 ///
232 /// # Errors
233 /// Returns [`StoreError::Sqlite`] on query failure.
234 pub fn schema_ahead(&self) -> Result<Option<SchemaAhead>, StoreError> {
235 let unknown = migrations::versions_ahead_of_build(&self.conn)?;
236 let Some(&store) = unknown.last() else {
237 return Ok(None);
238 };
239 Ok(Some(SchemaAhead {
240 store,
241 build: Self::build_schema_version(),
242 unknown,
243 }))
244 }
245
246 /// Number of nodes currently in the store.
247 ///
248 /// # Errors
249 /// Returns [`StoreError::Sqlite`] on query failure.
250 pub fn node_count(&self) -> Result<u64, StoreError> {
251 let n: i64 = self
252 .conn
253 .query_row("SELECT COUNT(*) FROM nodes", [], |r| r.get(0))?;
254 Ok(u64::try_from(n).unwrap_or(0))
255 }
256
257 /// Number of edges currently in the store.
258 ///
259 /// # Errors
260 /// Returns [`StoreError::Sqlite`] on query failure.
261 pub fn edge_count(&self) -> Result<u64, StoreError> {
262 let n: i64 = self
263 .conn
264 .query_row("SELECT COUNT(*) FROM edges", [], |r| r.get(0))?;
265 Ok(u64::try_from(n).unwrap_or(0))
266 }
267
268 /// Insert or update a node, keyed by its natural [`Node::key`].
269 ///
270 /// # Errors
271 /// Returns [`StoreError::Json`] if `meta` cannot be serialized, or
272 /// [`StoreError::Sqlite`] on write failure.
273 pub fn upsert_node(&self, node: &Node) -> Result<(), StoreError> {
274 upsert_node(&self.conn, node)
275 }
276
277 /// Insert an edge. Both endpoints must already resolve to nodes.
278 ///
279 /// # Errors
280 /// Returns [`StoreError::InvalidEdge`] if the provenance/confidence
281 /// invariant is violated, [`StoreError::UnknownNode`] if an endpoint key is
282 /// absent, or [`StoreError::Sqlite`] on write failure.
283 pub fn insert_edge(&self, edge: &Edge) -> Result<(), StoreError> {
284 insert_edge(&self.conn, edge)
285 }
286
287 /// Apply a fact set atomically: all nodes are upserted, then all edges are
288 /// inserted, in a single transaction. On any error nothing is committed.
289 ///
290 /// # Errors
291 /// Returns the first error encountered (see [`Store::upsert_node`] and
292 /// [`Store::insert_edge`]); the transaction is rolled back.
293 pub fn apply_factset(&mut self, facts: &FactSet) -> Result<(), StoreError> {
294 let tx = self.conn.transaction()?;
295 for node in &facts.nodes {
296 upsert_node(&tx, node)?;
297 }
298 for edge in &facts.edges {
299 insert_edge(&tx, edge)?;
300 }
301 tx.commit()?;
302 Ok(())
303 }
304
305 /// The `HEAD` tree id recorded at the last successful [`Store::rebuild`], if
306 /// any. Used by the sync engine to detect an unchanged tree.
307 ///
308 /// # Errors
309 /// Returns [`StoreError::Sqlite`] on query failure.
310 pub fn sync_state(&self) -> Result<Option<String>, StoreError> {
311 Ok(self
312 .conn
313 .query_row("SELECT tree FROM sync_state WHERE id = 0", [], |r| r.get(0))
314 .optional()?)
315 }
316
317 /// The extractor environment recorded with the last committed [`sync`],
318 /// `None` if unset (a legacy row, or the last sync was a worktree/index
319 /// preview). The incremental committed `sync` compares this to the current
320 /// env and falls back to a full re-extraction when they differ.
321 ///
322 /// # Errors
323 /// Returns [`StoreError::Sqlite`] on query failure.
324 pub fn sync_env(&self) -> Result<Option<String>, StoreError> {
325 Ok(self
326 .conn
327 .query_row("SELECT env FROM sync_state WHERE id = 0", [], |r| r.get(0))
328 .optional()?
329 .flatten())
330 }
331
332 /// Record the extractor environment for the current synced tree. Called by a
333 /// committed `sync` right after it writes the tree, so a later sync can decide
334 /// whether the incremental fast path is sound. A no-op if no tree is recorded.
335 ///
336 /// # Errors
337 /// Returns [`StoreError::Sqlite`] on write failure.
338 pub fn set_sync_env(&self, env: &str) -> Result<(), StoreError> {
339 self.conn
340 .execute("UPDATE sync_state SET env = ?1 WHERE id = 0", [env])?;
341 Ok(())
342 }
343
344 /// Which working tree this graph was assembled from, or `None` when the store
345 /// has never been synced or predates the column (issue #330).
346 ///
347 /// `graph.db` is an assembled view of **one** tree, so a store that came to
348 /// describe a different tree — restored from a backup, copied along with a
349 /// `.git` directory, or reached after a layout change — would otherwise
350 /// answer confidently about the wrong one: `sync` reporting "up to date"
351 /// against a state id belonging to someone else's tree, `check` validating a
352 /// tree nobody is looking at. `None` reads as "unknown" and is *adopted*
353 /// rather than treated as a mismatch, so an existing store is never rebuilt
354 /// merely for predating the stamp.
355 ///
356 /// # Errors
357 /// Returns [`StoreError::Sqlite`] on query failure.
358 pub fn synced_worktree(&self) -> Result<Option<String>, StoreError> {
359 Ok(self
360 .conn
361 .query_row("SELECT worktree FROM sync_state WHERE id = 0", [], |r| {
362 r.get(0)
363 })
364 .optional()?
365 .flatten())
366 }
367
368 /// Stamp this graph as assembled from `worktree`. Called by every sync entry
369 /// point right after it writes the tree state, so the two always agree about
370 /// whose tree the state describes. A no-op if no tree is recorded.
371 ///
372 /// Unlike [`Store::set_sync_env`]'s value, this survives a tree write: it
373 /// identifies the *store*, not the tree, and a new commit does not move the
374 /// graph to a different working tree.
375 ///
376 /// # Errors
377 /// Returns [`StoreError::Sqlite`] on write failure.
378 pub fn set_synced_worktree(&self, worktree: &str) -> Result<(), StoreError> {
379 self.conn.execute(
380 "UPDATE sync_state SET worktree = ?1 WHERE id = 0",
381 [worktree],
382 )?;
383 Ok(())
384 }
385
386 /// Atomically replace the entire graph with `facts`, recording `tree` as the
387 /// synced state (or clearing it when `tree` is `None`). All existing nodes
388 /// and edges are deleted first, so the store reflects exactly the given fact
389 /// set.
390 ///
391 /// Passing `None` records *no* synced tree — distinct from an empty string —
392 /// so [`Store::sync_state`] returns `None` and a later `sync` will not
393 /// spuriously short-circuit.
394 ///
395 /// # Errors
396 /// Returns the first error encountered (see [`Store::apply_factset`]); on any
397 /// error nothing is committed.
398 pub fn rebuild(&mut self, facts: &FactSet, tree: Option<&str>) -> Result<(), StoreError> {
399 let tx = self.conn.transaction()?;
400 tx.execute("DELETE FROM edges", [])?;
401 tx.execute("DELETE FROM nodes", [])?;
402 for node in &facts.nodes {
403 upsert_node(&tx, node)?;
404 }
405 for edge in &facts.edges {
406 insert_edge(&tx, edge)?;
407 }
408 write_sync_state(&tx, tree)?;
409 tx.commit()?;
410 Ok(())
411 }
412
413 /// Bring the store to exactly `facts` (as [`Store::rebuild`] does) but writing
414 /// only what **differs** instead of wiping and reinserting the whole graph —
415 /// the git-style "write only the delta". Unchanged node rows (which carry the
416 /// heavy JSON `meta`) and unchanged edge rows are left untouched; only removed
417 /// rows are deleted and new/changed rows written. The final state — nodes,
418 /// edges, and `sync_state` — is identical to `rebuild(facts, tree)`.
419 ///
420 /// Leaving unchanged edges in place means their row ids do not match a cold
421 /// rebuild's — which is safe *because* every edge query is content-ordered
422 /// (`(src, dst, kind, provenance)`, see [`Store::all_edges`]), never by row
423 /// id. So an incrementally reconciled store and a fresh rebuild return every
424 /// query identically; the delta is invisible above the storage layer.
425 ///
426 /// # Errors
427 /// Returns [`StoreError`] on a query failure; the transaction is rolled back.
428 pub fn reconcile(&mut self, facts: &FactSet, tree: Option<&str>) -> Result<(), StoreError> {
429 let current_nodes = self.all_nodes()?;
430 let current_edges = self.all_edges()?;
431 let cur_by_key: std::collections::HashMap<&str, &Node> =
432 current_nodes.iter().map(|n| (n.key.as_str(), n)).collect();
433 let new_keys: std::collections::HashSet<&str> =
434 facts.nodes.iter().map(|n| n.key.as_str()).collect();
435
436 // Edge identity is the full tuple, so a changed `confidence`/`src_ref`
437 // counts as remove-old + add-new — keeping the result identical to a
438 // wholesale rebuild, not the store's insert-time `DO NOTHING` semantics.
439 let new_edge_ids: std::collections::HashSet<EdgeId> =
440 facts.edges.iter().map(edge_identity).collect();
441 let cur_edge_ids: std::collections::HashSet<EdgeId> =
442 current_edges.iter().map(edge_identity).collect();
443
444 let tx = self.conn.transaction()?;
445 // 1. Delete removed edges first, so any node they reference can then be
446 // dropped (edges are FK-constrained on node ids, with no cascade). A
447 // removed node's edges are all removals, so they are gone before step 2.
448 for edge in ¤t_edges {
449 if !new_edge_ids.contains(&edge_identity(edge)) {
450 delete_edge(&tx, edge)?;
451 }
452 }
453 // 2. Drop nodes that no longer exist.
454 for old in ¤t_nodes {
455 if !new_keys.contains(old.key.as_str()) {
456 tx.execute("DELETE FROM nodes WHERE key = ?1", [&old.key])?;
457 }
458 }
459 // 3. Upsert only the nodes that are new or whose content changed (an upsert
460 // keeps the row id, so unchanged edges stay valid).
461 for node in &facts.nodes {
462 if cur_by_key
463 .get(node.key.as_str())
464 .is_none_or(|cur| *cur != node)
465 {
466 upsert_node(&tx, node)?;
467 }
468 }
469 // 4. Insert only the added edges (their endpoints now all exist).
470 for edge in &facts.edges {
471 if !cur_edge_ids.contains(&edge_identity(edge)) {
472 insert_edge(&tx, edge)?;
473 }
474 }
475 write_sync_state(&tx, tree)?;
476 tx.commit()?;
477 Ok(())
478 }
479
480 /// Fetch a node by its natural key.
481 ///
482 /// # Errors
483 /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
484 /// [`StoreError::Corrupt`] if a stored value cannot be decoded.
485 pub fn get_node(&self, key: &str) -> Result<Option<Node>, StoreError> {
486 let sql = format!("SELECT {NODE_COLS} FROM nodes n WHERE n.key = ?1");
487 let mut stmt = self.conn.prepare(&sql)?;
488 let mut rows = stmt.query([key])?;
489 match rows.next()? {
490 Some(row) => Ok(Some(row_to_node(row)?)),
491 None => Ok(None),
492 }
493 }
494
495 /// Every node key in the store, ordered. Useful for whole-graph exports.
496 ///
497 /// # Errors
498 /// Returns [`StoreError::Sqlite`] on query failure.
499 pub fn all_keys(&self) -> Result<Vec<String>, StoreError> {
500 let mut stmt = self.conn.prepare("SELECT key FROM nodes ORDER BY key")?;
501 let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
502 let mut out = Vec::new();
503 for row in rows {
504 out.push(row?);
505 }
506 Ok(out)
507 }
508
509 /// Dump the entire graph as a single [`FactSet`], with nodes and edges in a
510 /// deterministic order — suitable for a portable, content-stable artifact.
511 ///
512 /// # Errors
513 /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
514 /// [`StoreError::Corrupt`] on decode failure.
515 pub fn export_factset(&self) -> Result<FactSet, StoreError> {
516 let node_sql = format!("SELECT {NODE_COLS} FROM nodes n ORDER BY n.key");
517 let mut node_stmt = self.conn.prepare(&node_sql)?;
518 let mut node_rows = node_stmt.query([])?;
519 let nodes = collect_nodes(&mut node_rows)?;
520
521 // Order edges by their resolved endpoint keys (not row id) so the dump is
522 // stable regardless of insertion order.
523 let edge_sql = format!("{EDGE_SELECT} ORDER BY ns.key, nd.key, e.kind, e.provenance");
524 let mut edge_stmt = self.conn.prepare(&edge_sql)?;
525 let mut edge_rows = edge_stmt.query([])?;
526 let edges = collect_edges(&mut edge_rows)?;
527
528 Ok(FactSet { nodes, edges })
529 }
530
531 /// All nodes of a given kind.
532 ///
533 /// # Errors
534 /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
535 /// [`StoreError::Corrupt`] on decode failure.
536 pub fn nodes_by_kind(&self, kind: &NodeKind) -> Result<Vec<Node>, StoreError> {
537 let sql = format!("SELECT {NODE_COLS} FROM nodes n WHERE n.kind = ?1 ORDER BY n.key");
538 let mut stmt = self.conn.prepare(&sql)?;
539 let mut rows = stmt.query([kind.as_str()])?;
540 collect_nodes(&mut rows)
541 }
542
543 /// Nodes of a given kind whose `name` equals `name_lower` **case-insensitively**,
544 /// ordered by key. Narrows a lookup at the SQL layer — using the `kind` index and
545 /// filtering `name` in-query — so only matching rows are decoded, never every
546 /// node of that kind. Used by the cross-repo follow bridge to fetch just the
547 /// candidate struct(s) for a config section rather than scanning all structs.
548 ///
549 /// # Errors
550 /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
551 /// [`StoreError::Corrupt`] on decode failure.
552 pub fn nodes_by_kind_named(
553 &self,
554 kind: &NodeKind,
555 name_lower: &str,
556 ) -> Result<Vec<Node>, StoreError> {
557 let sql = format!(
558 "SELECT {NODE_COLS} FROM nodes n \
559 WHERE n.kind = ?1 AND lower(n.name) = ?2 ORDER BY n.key"
560 );
561 let mut stmt = self.conn.prepare(&sql)?;
562 let mut rows = stmt.query([kind.as_str(), name_lower])?;
563 collect_nodes(&mut rows)
564 }
565
566 /// Every `config_key` node's flattened setting (ADR-0009), read back out of
567 /// the graph as [`crate::ConfigKey`]s — the graph-native source the cross-repo
568 /// link matcher (`roteiro links --infer`) consumes, so it never re-parses
569 /// config files. Ordered by node key (deterministic). A node missing the
570 /// `key`/`path` a well-formed `config_key` carries is skipped defensively.
571 ///
572 /// # Errors
573 /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
574 /// [`StoreError::Corrupt`] on decode failure.
575 pub fn config_keys(&self) -> Result<Vec<crate::ConfigKey>, StoreError> {
576 let nodes = self.nodes_by_kind(&NodeKind::Other(crate::config_keys::KIND.to_owned()))?;
577 let mut out = Vec::with_capacity(nodes.len());
578 for n in &nodes {
579 let key = n.meta.get("key").and_then(serde_json::Value::as_str);
580 // A `config_key` node carries a `value` in `meta` when it has a real
581 // setting (file-derived keys always do, even an empty string). A
582 // struct-derived key (`meta.source = "struct"`) omits it — its value is
583 // *unknown*, not empty — so record that absence explicitly rather than
584 // defaulting it to `""`, which would false-match in value agreement.
585 let value = n.meta.get("value").and_then(serde_json::Value::as_str);
586 if let (Some(key), Some(file)) = (key, n.path.as_deref()) {
587 out.push(crate::ConfigKey {
588 file: file.to_owned(),
589 key: key.to_owned(),
590 value: value.unwrap_or_default().to_owned(),
591 value_known: value.is_some(),
592 });
593 }
594 }
595 Ok(out)
596 }
597
598 /// Every node whose source `path` is `path`, ordered by key — the file node
599 /// plus the symbols and markers defined in it. Used to scope a change to the
600 /// graph (e.g. `roteiro review`).
601 ///
602 /// # Errors
603 /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
604 /// [`StoreError::Corrupt`] on decode failure.
605 pub fn nodes_by_path(&self, path: &str) -> Result<Vec<Node>, StoreError> {
606 let sql = format!("SELECT {NODE_COLS} FROM nodes n WHERE n.path = ?1 ORDER BY n.key");
607 let mut stmt = self.conn.prepare(&sql)?;
608 let mut rows = stmt.query([path])?;
609 collect_nodes(&mut rows)
610 }
611
612 /// Every node produced by a given layer, ordered by key. The incremental
613 /// `sync` loads the `Derived` layer to reconstruct the extraction graph
614 /// without re-reading every blob.
615 ///
616 /// # Errors
617 /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
618 /// [`StoreError::Corrupt`] on decode failure.
619 pub fn nodes_by_provenance(&self, provenance: Provenance) -> Result<Vec<Node>, StoreError> {
620 let sql = format!("SELECT {NODE_COLS} FROM nodes n WHERE n.provenance = ?1 ORDER BY n.key");
621 let mut stmt = self.conn.prepare(&sql)?;
622 let mut rows = stmt.query([provenance.as_str()])?;
623 collect_nodes(&mut rows)
624 }
625
626 /// Every node in the store, ordered by key. Unlike [`Store::export_factset`]
627 /// this decodes no edges, so it is cheap for node-only scans (e.g. search).
628 ///
629 /// # Errors
630 /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
631 /// [`StoreError::Corrupt`] on decode failure.
632 pub fn all_nodes(&self) -> Result<Vec<Node>, StoreError> {
633 let sql = format!("SELECT {NODE_COLS} FROM nodes n ORDER BY n.key");
634 let mut stmt = self.conn.prepare(&sql)?;
635 let mut rows = stmt.query([])?;
636 collect_nodes(&mut rows)
637 }
638
639 /// Every edge in the store, with endpoints resolved to their node keys. Used
640 /// by [`Store::reconcile`] to diff the edge set.
641 ///
642 /// Ordered by the edge's **content** — `(src key, dst key, kind, provenance)`,
643 /// the table's unique tuple — not by row id. This makes the order a function
644 /// of the *graph*, not of insertion history, so an incrementally
645 /// [`reconcile`](Store::reconcile)d store and a cold [`rebuild`](Store::rebuild)
646 /// return edges identically. (The same reason node scans order by `key`.)
647 ///
648 /// # Errors
649 /// Returns [`StoreError::Sqlite`] or [`StoreError::Corrupt`] on failure.
650 pub fn all_edges(&self) -> Result<Vec<Edge>, StoreError> {
651 let sql = format!("{EDGE_SELECT} ORDER BY ns.key, nd.key, e.kind, e.provenance");
652 let mut stmt = self.conn.prepare(&sql)?;
653 let mut rows = stmt.query([])?;
654 collect_edges(&mut rows)
655 }
656
657 /// Edges whose source is the node with the given key, in content order
658 /// (`(dst key, kind, provenance)` — `src` is fixed). Content-ordered rather
659 /// than by row id so the result is history-independent; see [`Store::all_edges`].
660 ///
661 /// # Errors
662 /// Returns [`StoreError::Sqlite`] or [`StoreError::Corrupt`] on failure.
663 pub fn edges_from(&self, key: &str) -> Result<Vec<Edge>, StoreError> {
664 let sql = format!("{EDGE_SELECT} WHERE ns.key = ?1 ORDER BY nd.key, e.kind, e.provenance");
665 let mut stmt = self.conn.prepare(&sql)?;
666 let mut rows = stmt.query([key])?;
667 collect_edges(&mut rows)
668 }
669
670 /// Edges whose destination is the node with the given key, in content order
671 /// (`(src key, kind, provenance)` — `dst` is fixed). See [`Store::all_edges`].
672 ///
673 /// # Errors
674 /// Returns [`StoreError::Sqlite`] or [`StoreError::Corrupt`] on failure.
675 pub fn edges_to(&self, key: &str) -> Result<Vec<Edge>, StoreError> {
676 let sql = format!("{EDGE_SELECT} WHERE nd.key = ?1 ORDER BY ns.key, e.kind, e.provenance");
677 let mut stmt = self.conn.prepare(&sql)?;
678 let mut rows = stmt.query([key])?;
679 collect_edges(&mut rows)
680 }
681
682 /// All edges with the given provenance, in content order
683 /// (`(src key, dst key, kind)` — `provenance` is fixed). See [`Store::all_edges`].
684 ///
685 /// # Errors
686 /// Returns [`StoreError::Sqlite`] or [`StoreError::Corrupt`] on failure.
687 pub fn edges_by_provenance(&self, provenance: Provenance) -> Result<Vec<Edge>, StoreError> {
688 let sql = format!("{EDGE_SELECT} WHERE e.provenance = ?1 ORDER BY ns.key, nd.key, e.kind");
689 let mut stmt = self.conn.prepare(&sql)?;
690 let mut rows = stmt.query([provenance.as_str()])?;
691 collect_edges(&mut rows)
692 }
693
694 /// Delete all edges with the given provenance, returning how many were
695 /// removed. Used to re-derive a whole provenance class authoritatively (e.g.
696 /// `inferred` edges when re-running inference with different parameters).
697 ///
698 /// # Errors
699 /// Returns [`StoreError::Sqlite`] on write failure.
700 pub fn delete_edges_by_provenance(&self, provenance: Provenance) -> Result<u64, StoreError> {
701 let n = self.conn.execute(
702 "DELETE FROM edges WHERE provenance = ?1",
703 [provenance.as_str()],
704 )?;
705 Ok(u64::try_from(n).unwrap_or(0))
706 }
707
708 /// Delete all edges carrying the given `src_ref`, returning how many were
709 /// removed. Lets one producer of `inferred` edges (e.g. the embedding layer,
710 /// or a Graphify import) re-derive its own edges authoritatively without
711 /// touching edges another producer contributed.
712 ///
713 /// # Errors
714 /// Returns [`StoreError::Sqlite`] on write failure.
715 pub fn delete_edges_by_src_ref(&self, src_ref: &str) -> Result<u64, StoreError> {
716 let n = self
717 .conn
718 .execute("DELETE FROM edges WHERE src_ref = ?1", [src_ref])?;
719 Ok(u64::try_from(n).unwrap_or(0))
720 }
721
722 /// Apply an import layer to the live graph **and** persist it durably under
723 /// `src_ref`, validating as it goes: this ref's prior edges are cleared
724 /// (an authoritative re-import), the layer's nodes are upserted, and each
725 /// edge is applied only if both endpoints resolve. Dangling edges — cross-
726 /// references to code that is not present — are dropped, and only the
727 /// validated (trimmed) layer is persisted, so stale data is never stored.
728 ///
729 /// This is the "validate on import" half; [`Store::reapply_imports`] is the
730 /// "validate on sync" half, re-checking layers against the rebuilt graph.
731 ///
732 /// # Errors
733 /// Returns [`StoreError::Json`] if `facts` cannot be (de)serialized,
734 /// [`StoreError::InvalidEdge`] on a malformed edge, or [`StoreError::Sqlite`]
735 /// on write failure.
736 pub fn apply_import_layer(
737 &mut self,
738 src_ref: &str,
739 facts: &FactSet,
740 ) -> Result<ImportApplied, StoreError> {
741 let tx = self.conn.transaction()?;
742 // Authoritative re-import: drop this ref's prior edges from the live graph.
743 tx.execute("DELETE FROM edges WHERE src_ref = ?1", [src_ref])?;
744 for node in &facts.nodes {
745 upsert_node(&tx, node)?;
746 }
747 let (kept, applied) = apply_edges_pruning(&tx, &facts.edges)?;
748 let trimmed = FactSet {
749 nodes: facts.nodes.clone(),
750 edges: kept,
751 };
752 put_import_row(&tx, src_ref, &trimmed)?;
753 tx.commit()?;
754 Ok(ImportApplied {
755 layers: 1,
756 nodes: facts.nodes.len(),
757 ..applied
758 })
759 }
760
761 /// Remove the persisted import layer for `src_ref`, returning whether one
762 /// existed. Does not remove edges already in the live graph (use
763 /// [`Store::delete_edges_by_src_ref`] for that).
764 ///
765 /// # Errors
766 /// Returns [`StoreError::Sqlite`] on write failure.
767 pub fn delete_import(&self, src_ref: &str) -> Result<bool, StoreError> {
768 let n = self
769 .conn
770 .execute("DELETE FROM imports WHERE src_ref = ?1", [src_ref])?;
771 Ok(n > 0)
772 }
773
774 /// The `src_ref`s of all persisted import layers, ordered.
775 ///
776 /// # Errors
777 /// Returns [`StoreError::Sqlite`] on query failure.
778 pub fn import_refs(&self) -> Result<Vec<String>, StoreError> {
779 let mut stmt = self
780 .conn
781 .prepare("SELECT src_ref FROM imports ORDER BY src_ref")?;
782 let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
783 let mut out = Vec::new();
784 for row in rows {
785 out.push(row?);
786 }
787 Ok(out)
788 }
789
790 /// Re-apply every persisted import layer on top of the current graph and
791 /// **re-validate** it: all import nodes are upserted first (so cross-layer
792 /// and self references resolve), then each edge is applied; an edge whose
793 /// endpoint is now absent — a cross-reference to code a sync removed — is
794 /// pruned from the persisted layer, not merely skipped. So the durable store
795 /// keeps only still-correct data. Idempotent; safe to run after each rebuild.
796 ///
797 /// # Errors
798 /// Returns [`StoreError::Json`] if a stored layer cannot be (de)serialized,
799 /// or [`StoreError::Sqlite`] on write failure.
800 pub fn reapply_imports(&mut self) -> Result<ImportApplied, StoreError> {
801 let layers = self.load_import_layers()?;
802 let tx = self.conn.transaction()?;
803 // Pass 1: upsert every layer's nodes so intra-import edges resolve
804 // regardless of which layer defines the endpoint.
805 for (_, facts) in &layers {
806 for node in &facts.nodes {
807 upsert_node(&tx, node)?;
808 }
809 }
810 // Pass 2: apply edges, pruning (and rewriting) any that dangle.
811 let mut applied = ImportApplied {
812 layers: layers.len(),
813 ..ImportApplied::default()
814 };
815 for (src_ref, facts) in &layers {
816 applied.nodes += facts.nodes.len();
817 let (kept, counts) = apply_edges_pruning(&tx, &facts.edges)?;
818 applied.edges_applied += counts.edges_applied;
819 applied.edges_pruned += counts.edges_pruned;
820 if kept.len() != facts.edges.len() {
821 let trimmed = FactSet {
822 nodes: facts.nodes.clone(),
823 edges: kept,
824 };
825 put_import_row(&tx, src_ref, &trimmed)?;
826 }
827 }
828 tx.commit()?;
829 Ok(applied)
830 }
831
832 /// Load and decode every persisted import layer as `(src_ref, FactSet)`, in
833 /// `src_ref` order.
834 fn load_import_layers(&self) -> Result<Vec<(String, FactSet)>, StoreError> {
835 let mut stmt = self
836 .conn
837 .prepare("SELECT src_ref, facts FROM imports ORDER BY src_ref")?;
838 let rows = stmt.query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)))?;
839 let mut out = Vec::new();
840 for row in rows {
841 let (src_ref, json) = row?;
842 let mut facts: FactSet = serde_json::from_str(&json)?;
843 // An import-layer node is *never* derived (derivation is `sync`'s job),
844 // so a `Derived` tag here is always wrong. It arises two ways, both
845 // repaired the same: a legacy layer persisted before nodes carried
846 // provenance (the field is absent → serde defaults `Derived`), or —
847 // anomalously — a layer that stored an explicit `"provenance":"derived"`
848 // (a producer/data bug). We deliberately repair *both* rather than only
849 // the absent case: leaving an explicit-`derived` import node in place
850 // would let a layer-scoped `sync` treat it as derived and delete it —
851 // the exact corruption this guards against — so repair is the safe
852 // recovery, not silent masking. Idempotent, runs on every reapply
853 // (old stores self-heal), and a no-op for correctly-tagged fresh imports.
854 for node in &mut facts.nodes {
855 if node.provenance == Provenance::Derived {
856 node.provenance = import_node_provenance(&node.key);
857 }
858 }
859 out.push((src_ref, facts));
860 }
861 Ok(out)
862 }
863
864 /// Neighbouring nodes reachable from `key` in the given direction. Returns
865 /// an empty vector if the node does not exist.
866 ///
867 /// # Errors
868 /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
869 /// [`StoreError::Corrupt`] on failure.
870 pub fn neighbors(&self, key: &str, dir: Direction) -> Result<Vec<Node>, StoreError> {
871 let out = format!(
872 "SELECT {NODE_COLS} FROM nodes n JOIN edges e ON n.id = e.dst \
873 JOIN nodes s ON s.id = e.src WHERE s.key = ?1"
874 );
875 let inc = format!(
876 "SELECT {NODE_COLS} FROM nodes n JOIN edges e ON n.id = e.src \
877 JOIN nodes d ON d.id = e.dst WHERE d.key = ?1"
878 );
879 // Order by output column 1 (the node key) so results are deterministic
880 // across SQLite versions/plans. Positional ordering avoids both the
881 // ambiguity of a bare `key` (present in every joined table) and the fact
882 // that a table-qualified name cannot be used after the `Both` UNION.
883 let sql = match dir {
884 Direction::Outgoing => format!("{out} ORDER BY 1"),
885 Direction::Incoming => format!("{inc} ORDER BY 1"),
886 Direction::Both => format!("{out} UNION {inc} ORDER BY 1"),
887 };
888 let mut stmt = self.conn.prepare(&sql)?;
889 let mut rows = stmt.query([key])?;
890 collect_nodes(&mut rows)
891 }
892
893 /// Fetch the cached context bundle for `key` as `(fingerprint, json)`, if
894 /// present. The caller compares the fingerprint to the node's current one to
895 /// decide whether the entry is fresh (see [`crate::context`]).
896 ///
897 /// # Errors
898 /// Returns [`StoreError::Sqlite`] on query failure.
899 pub fn context_cache_get(&self, key: &str) -> Result<Option<(String, String)>, StoreError> {
900 let row = self
901 .conn
902 .query_row(
903 "SELECT fingerprint, json FROM node_context WHERE key = ?1",
904 [key],
905 |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)),
906 )
907 .optional()?;
908 Ok(row)
909 }
910
911 /// Fetch just the cached fingerprint for `key`, without reading the (larger)
912 /// JSON payload — for a cheap freshness check.
913 ///
914 /// # Errors
915 /// Returns [`StoreError::Sqlite`] on query failure.
916 pub fn context_cache_fingerprint(&self, key: &str) -> Result<Option<String>, StoreError> {
917 let fp = self
918 .conn
919 .query_row(
920 "SELECT fingerprint FROM node_context WHERE key = ?1",
921 [key],
922 |r| r.get::<_, String>(0),
923 )
924 .optional()?;
925 Ok(fp)
926 }
927
928 /// Store (or replace) the cached context bundle for `key`.
929 ///
930 /// # Errors
931 /// Returns [`StoreError::Sqlite`] on write failure.
932 pub fn context_cache_put(
933 &self,
934 key: &str,
935 fingerprint: &str,
936 json: &str,
937 ) -> Result<(), StoreError> {
938 self.conn.execute(
939 "INSERT INTO node_context (key, fingerprint, json) VALUES (?1, ?2, ?3)
940 ON CONFLICT(key) DO UPDATE SET
941 fingerprint = excluded.fingerprint, json = excluded.json",
942 [key, fingerprint, json],
943 )?;
944 Ok(())
945 }
946
947 /// Delete the cached context entry for `key`, returning whether one existed.
948 ///
949 /// # Errors
950 /// Returns [`StoreError::Sqlite`] on write failure.
951 pub fn context_cache_delete(&self, key: &str) -> Result<bool, StoreError> {
952 let n = self
953 .conn
954 .execute("DELETE FROM node_context WHERE key = ?1", [key])?;
955 Ok(n > 0)
956 }
957
958 /// Every key with a cached context entry, ordered. Used to prune entries for
959 /// nodes that no longer exist.
960 ///
961 /// # Errors
962 /// Returns [`StoreError::Sqlite`] on query failure.
963 pub fn context_cache_keys(&self) -> Result<Vec<String>, StoreError> {
964 let mut stmt = self
965 .conn
966 .prepare("SELECT key FROM node_context ORDER BY key")?;
967 let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
968 let mut out = Vec::new();
969 for row in rows {
970 out.push(row?);
971 }
972 Ok(out)
973 }
974
975 // --- Analyzer findings (ADR-0012). A separate artifact store: these methods
976 // touch `analysis_runs`/`findings` only, never `nodes`/`edges`, so
977 // `export_factset` — and the published `GraphArtifact` — stays a pure
978 // function of the tree no matter what an analyzer reports. ---
979
980 /// Replace the findings layer `run.layer` **wholesale**, atomically: the
981 /// previous run for that layer and every finding row it owned are deleted,
982 /// then this run and its findings are written. A finding that has since been
983 /// fixed therefore *disappears* instead of lingering, and re-ingesting an
984 /// unchanged report is idempotent — the store ends up with the same rows and
985 /// no growth.
986 ///
987 /// The owned-record cleanup is explicit rather than inherited. The import
988 /// path ([`Store::apply_import_layer`]) deletes a layer's edges but leaves its
989 /// obsolete *nodes* behind; copying that shape here would silently orphan
990 /// findings, so the previous run's rows are deleted by hand and counted in
991 /// [`FindingsApplied::removed`]. The schema's `ON DELETE CASCADE` is kept as
992 /// defence in depth, not as the mechanism.
993 ///
994 /// `findings` must carry distinct [`FindingKey`](crate::FindingKey)s: a
995 /// duplicate identity is a producer bug and is rejected by the unique index
996 /// *inside* the transaction, so nothing is committed. Callers that parse
997 /// untrusted reports should reject duplicates earlier, with a better message.
998 ///
999 /// # Errors
1000 /// Returns [`StoreError::Json`] if the run's command policy or a finding's
1001 /// `meta` cannot be serialized, or [`StoreError::Sqlite`] on write failure. On
1002 /// any error the transaction is rolled back and the previous layer survives
1003 /// intact.
1004 pub fn replace_findings_layer(
1005 &mut self,
1006 run: &AnalysisRun,
1007 findings: &[Finding],
1008 ) -> Result<FindingsApplied, StoreError> {
1009 let tx = self.conn.transaction()?;
1010 let applied = findings::replace_layer(&tx, run, findings)?;
1011 tx.commit()?;
1012 Ok(applied)
1013 }
1014
1015 /// Delete a findings layer and every finding row it owns, returning how many
1016 /// findings went with it, or `None` if the layer was not live.
1017 ///
1018 /// # Errors
1019 /// Returns [`StoreError::Sqlite`] on write failure; nothing is committed on
1020 /// error.
1021 pub fn delete_findings_layer(&mut self, layer: &str) -> Result<Option<usize>, StoreError> {
1022 let tx = self.conn.transaction()?;
1023 let removed = findings::delete_layer(&tx, layer)?;
1024 tx.commit()?;
1025 Ok(removed)
1026 }
1027
1028 /// Every live findings layer with its findings, ordered by layer key and, in
1029 /// each layer, by finding key. Pass `analyzer` to narrow to one analyzer.
1030 ///
1031 /// # Errors
1032 /// Returns [`StoreError::Sqlite`] on query failure, [`StoreError::Json`] if a
1033 /// stored policy or `meta` cannot be decoded, or [`StoreError::Corrupt`] on an
1034 /// unrecognised stored token.
1035 pub fn findings_layers(
1036 &self,
1037 analyzer: Option<&str>,
1038 ) -> Result<Vec<FindingsLayer>, StoreError> {
1039 findings::layers(&self.conn, analyzer)
1040 }
1041
1042 /// Number of findings currently stored, across every layer.
1043 ///
1044 /// # Errors
1045 /// Returns [`StoreError::Sqlite`] on query failure.
1046 pub fn finding_count(&self) -> Result<u64, StoreError> {
1047 findings::count_findings(&self.conn)
1048 }
1049
1050 /// Number of live analysis runs — one per findings layer.
1051 ///
1052 /// # Errors
1053 /// Returns [`StoreError::Sqlite`] on query failure.
1054 pub fn analysis_run_count(&self) -> Result<u64, StoreError> {
1055 findings::count_runs(&self.conn)
1056 }
1057
1058 // --- Generated media content (ADR-0015). A separate artifact store, on the
1059 // same terms as findings above: these methods touch `media_content` only,
1060 // never `nodes`/`edges`, so `export_factset` stays a pure function of the
1061 // tree across a `media build`, and generated text can never reach the
1062 // `authored` relevance boost in `search`. ---
1063
1064 /// Write one generated-content record, returning `true` if a row was written.
1065 ///
1066 /// Keyed by `(blob_id, producer)`. A record for that exact pair already
1067 /// present is **left alone** and `false` is returned — which is what makes
1068 /// `media build` incremental and a second run free. A *different* producer is
1069 /// a new row beside it, never an overwrite: that is the point of keying on the
1070 /// producer identity, so a better model's description can be compared with the
1071 /// one it replaces and a distrusted producer can be dropped wholesale.
1072 /// [`MediaWrite::replace`] (only `media build --force`) is the one path that
1073 /// overwrites, and only for the identical producer.
1074 ///
1075 /// # Errors
1076 /// Returns [`StoreError::Sqlite`] on a write failure; the transaction is
1077 /// rolled back.
1078 pub fn record_media_content(&mut self, write: &MediaWrite<'_>) -> Result<bool, StoreError> {
1079 let tx = self.conn.transaction()?;
1080 let written = media::record(&tx, write)?;
1081 tx.commit()?;
1082 Ok(written)
1083 }
1084
1085 /// Whether a record already exists for exactly this `(blob, producer)`.
1086 ///
1087 /// # Errors
1088 /// Returns [`StoreError::Sqlite`] on query failure.
1089 pub fn has_media_record(&self, blob_id: &str, producer: &str) -> Result<bool, StoreError> {
1090 media::exists(&self.conn, blob_id, producer)
1091 }
1092
1093 /// Stored records matching `filter`, ordered by `(producer, blob id)`.
1094 ///
1095 /// # Errors
1096 /// Returns [`StoreError::Sqlite`] on query failure, or
1097 /// [`StoreError::Corrupt`] if a row carries an unknown modality token.
1098 pub fn media_records(&self, filter: &MediaFilter<'_>) -> Result<Vec<MediaRecord>, StoreError> {
1099 media::records(&self.conn, filter)
1100 }
1101
1102 /// Discard records — all of them, or only those written by `producer`.
1103 /// Returns how many rows went.
1104 ///
1105 /// Nothing in the graph is touched: dropping a model you no longer trust must
1106 /// not cost you a re-sync.
1107 ///
1108 /// # Errors
1109 /// Returns [`StoreError::Sqlite`] on a write failure.
1110 pub fn clear_media_content(&mut self, producer: Option<&str>) -> Result<usize, StoreError> {
1111 let tx = self.conn.transaction()?;
1112 let removed = media::delete(&tx, producer)?;
1113 tx.commit()?;
1114 Ok(removed)
1115 }
1116
1117 /// Number of generated-content records currently stored.
1118 ///
1119 /// # Errors
1120 /// Returns [`StoreError::Sqlite`] on query failure.
1121 pub fn media_content_count(&self) -> Result<u64, StoreError> {
1122 media::count(&self.conn)
1123 }
1124
1125 /// One summary per producer that owns records, ordered by producer id.
1126 ///
1127 /// # Errors
1128 /// Returns [`StoreError::Sqlite`] on query failure, or
1129 /// [`StoreError::Corrupt`] if a row carries an unknown modality token.
1130 pub fn media_producer_summaries(&self) -> Result<Vec<ProducerSummary>, StoreError> {
1131 media::producer_summaries(&self.conn)
1132 }
1133
1134 /// The blob ids that have at least one record carrying **generated text**
1135 /// for `kind`. A blob the pre-generation gate refused is not described, so it
1136 /// is not here — see [`Store::gated_media_blobs`].
1137 ///
1138 /// # Errors
1139 /// Returns [`StoreError::Sqlite`] on query failure.
1140 pub fn described_media_blobs(
1141 &self,
1142 kind: MediaKind,
1143 ) -> Result<std::collections::BTreeSet<String>, StoreError> {
1144 media::described_blobs(&self.conn, kind)
1145 }
1146
1147 /// The blob ids the [pre-generation gate](crate::media::gate) refused for
1148 /// `kind` — silent clips, blank images — each of which has a record naming
1149 /// the value that refused it.
1150 ///
1151 /// # Errors
1152 /// Returns [`StoreError::Sqlite`] on query failure.
1153 pub fn gated_media_blobs(
1154 &self,
1155 kind: MediaKind,
1156 ) -> Result<std::collections::BTreeSet<String>, StoreError> {
1157 media::gated_blobs(&self.conn, kind)
1158 }
1159
1160 /// Records whose blob is no longer anywhere in `present`. Exposed so a caller
1161 /// can see what a tree change has orphaned; nothing deletes them implicitly,
1162 /// because a record is expensive to reproduce and a blob can return.
1163 ///
1164 /// # Errors
1165 /// Returns [`StoreError::Sqlite`] on query failure.
1166 pub fn orphan_media_records(
1167 &self,
1168 present: &std::collections::BTreeSet<String>,
1169 ) -> Result<Vec<MediaRecord>, StoreError> {
1170 Ok(self
1171 .media_records(&MediaFilter::default())?
1172 .into_iter()
1173 .filter(|r| !present.contains(&r.blob_id))
1174 .collect())
1175 }
1176
1177 // --- Episodic agent memory (ADR-0013). A separate artifact store, on the
1178 // same terms as findings and media above: these methods write `agent_memory`
1179 // and nothing else — they *read* `nodes` to capture and check an anchor, and
1180 // never write one. So `export_factset` stays a pure function of the tree
1181 // across every memory write, and nothing an agent remembers can reach the
1182 // `authored` relevance boost in `search`.
1183 //
1184 // These rows are not touched by `rebuild`, following the `imports` precedent:
1185 // what has no generating function must not be destroyed by a re-derivation.
1186 // ---
1187
1188 /// Record one memory, returning its id — the monotonic generation it was
1189 /// written at.
1190 ///
1191 /// The anchor's blob and path, and the `sync_state` tree witness, are
1192 /// captured **here, from the graph**, so a caller cannot record evidence the
1193 /// node never carried. An anchor key naming no node is accepted and reads
1194 /// back as [`crate::AnchorState::Vanished`].
1195 ///
1196 /// [`MemoryWrite::supersedes`] records the supersession explicitly, in the
1197 /// same transaction as the successor: either both land or neither does.
1198 ///
1199 /// # Errors
1200 /// Returns [`MemoryError::InvalidScope`] / [`MemoryError::InvalidBody`] /
1201 /// [`MemoryError::InvalidConfidence`] for a record that could not be recalled,
1202 /// [`MemoryError::NotFound`] or [`MemoryError::AlreadySuperseded`] for a bad
1203 /// supersession target, or [`MemoryError::Store`] on a write failure — in
1204 /// every case with nothing committed.
1205 pub fn record_memory(&mut self, write: &MemoryWrite<'_>) -> Result<i64, MemoryError> {
1206 let tx = self.conn.transaction().map_err(StoreError::from)?;
1207 let id = memory::record(&tx, write)?;
1208 tx.commit().map_err(StoreError::from)?;
1209 Ok(id)
1210 }
1211
1212 /// Memory records matching `filter`, newest generation first.
1213 ///
1214 /// Live records only unless [`MemoryFilter::include_superseded`] is set: a
1215 /// superseded record drops out immediately and regardless of age, because the
1216 /// test is a recorded pointer and not a clock.
1217 ///
1218 /// Each record's [`crate::AnchorState`] is computed against the **current**
1219 /// graph on every call and is never stored.
1220 ///
1221 /// # Errors
1222 /// Returns [`StoreError::Sqlite`] on query failure, or [`StoreError::Corrupt`]
1223 /// if a row carries an unknown kind token.
1224 pub fn memory_records(
1225 &self,
1226 filter: &MemoryFilter<'_>,
1227 ) -> Result<Vec<MemoryRecord>, StoreError> {
1228 memory::records(&self.conn, filter)
1229 }
1230
1231 /// One memory record by id, or `None` if there is no such record.
1232 ///
1233 /// # Errors
1234 /// Returns [`StoreError::Sqlite`] on query failure, or [`StoreError::Corrupt`]
1235 /// if the row carries an unknown kind token.
1236 pub fn memory_record(&self, id: i64) -> Result<Option<MemoryRecord>, StoreError> {
1237 memory::get(&self.conn, id)
1238 }
1239
1240 /// A [`MemoryListing`]: the records matching `filter`, plus the whole store's
1241 /// live and superseded counts, so an empty result is legible as *nothing
1242 /// matched* rather than *nothing is stored*.
1243 ///
1244 /// # Errors
1245 /// Returns [`StoreError::Sqlite`] on query failure, or [`StoreError::Corrupt`]
1246 /// if a row carries an unknown kind token.
1247 pub fn memory_listing(&self, filter: &MemoryFilter<'_>) -> Result<MemoryListing, StoreError> {
1248 let (live, superseded) = memory::counts(&self.conn)?;
1249 Ok(MemoryListing {
1250 schema: memory::MEMORY_SCHEMA,
1251 records: self.memory_records(filter)?,
1252 live,
1253 superseded,
1254 })
1255 }
1256
1257 /// **The only way a memory record is ever removed.** Deletes it, and returns
1258 /// `None` if there was no such record.
1259 ///
1260 /// Episodic memory is unbounded and never auto-evicted — no sweep, no TTL, no
1261 /// capacity bound reaches this table — so an explicit call is the whole
1262 /// reclamation story. It is also the privacy story: memory has no redaction
1263 /// chokepoint, so a record that captured a token or a customer name is
1264 /// removed by asking.
1265 ///
1266 /// Anything the deleted record had superseded becomes **live again** and is
1267 /// named in [`MemoryForgotten::restored`]: leaving it superseded would hide it
1268 /// on the authority of a record that no longer exists.
1269 ///
1270 /// # Errors
1271 /// Returns [`StoreError::Sqlite`] on a write failure; the transaction is
1272 /// rolled back.
1273 pub fn forget_memory(&mut self, id: i64) -> Result<Option<MemoryForgotten>, StoreError> {
1274 let tx = self.conn.transaction()?;
1275 let forgotten = memory::forget(&tx, id)?;
1276 tx.commit()?;
1277 Ok(forgotten)
1278 }
1279
1280 /// How many memory records are stored, as `(live, superseded)`.
1281 ///
1282 /// # Errors
1283 /// Returns [`StoreError::Sqlite`] on query failure.
1284 pub fn memory_counts(&self) -> Result<(u64, u64), StoreError> {
1285 memory::counts(&self.conn)
1286 }
1287
1288 /// **Ranked recall**: the live records that match `opts`, scored
1289 /// `base_confidence × anchor_penalty × decay(age)` and ordered best first.
1290 ///
1291 /// Every term is computed **here, at retrieval time, and written to no
1292 /// column** (ADR-0013). A stored score that decayed would rewrite the store on
1293 /// every read and would be wrong in between, making recall depend on when you
1294 /// last looked. Three consequences follow, and each is a promise:
1295 ///
1296 /// - **This call mutates nothing.** Recall over an unchanged store and an
1297 /// unchanged tree is idempotent, which is what makes
1298 /// [`crate::Decay::None`] byte-identical across runs.
1299 /// - **A superseded record is never returned**, immediately and regardless of
1300 /// age: the test is a recorded pointer, not a clock.
1301 /// - **A record whose anchor no longer resolves is still returned**, demoted
1302 /// and labelled. Drift ranks it down; nothing deletes it.
1303 ///
1304 /// # Errors
1305 /// Returns [`StoreError::Sqlite`] on query failure, or [`StoreError::Corrupt`]
1306 /// if a row carries an unknown kind token.
1307 pub fn recall_memory(&self, opts: &RecallOptions<'_>) -> Result<Recall, StoreError> {
1308 let (live, superseded) = memory::counts(&self.conn)?;
1309 Ok(Recall {
1310 schema: memory::RECALL_SCHEMA,
1311 generation: memory::generation(&self.conn)?,
1312 decay: opts.decay,
1313 reproducible: opts.decay.is_reproducible(),
1314 results: memory::recall(&self.conn, opts)?,
1315 live,
1316 superseded,
1317 })
1318 }
1319
1320 // --- The bounded cache tier (ADR-0013, Tier 2). The *opposite* rules to the
1321 // episodic tier above, because it holds the opposite kind of knowledge:
1322 // everything here is re-derivable, so eviction costs cycles and never
1323 // information. Nothing in this section can reach `agent_memory` — it has no
1324 // `bytes`, no `last_used` and no `hits` for a capacity policy to grip. ---
1325
1326 /// Write (or replace) one cache entry.
1327 ///
1328 /// The payload size is computed here and the anchor's blob is captured from
1329 /// the graph here, so the sweep can order and total the tier without reading
1330 /// it, and a caller cannot record evidence a node never carried.
1331 ///
1332 /// # Errors
1333 /// Returns [`StoreError::Sqlite`] on write failure.
1334 pub fn agent_cache_put(&self, write: &CacheWrite<'_>) -> Result<(), StoreError> {
1335 memory::cache_put(&self.conn, write)
1336 }
1337
1338 /// Read one cache entry back, **recording the access** — `hits` increments and
1339 /// `last_used` advances.
1340 ///
1341 /// This is the one read in the memory store that writes, and what it writes is
1342 /// the cache's own bookkeeping: those two columns exist to be moved by exactly
1343 /// this, and a hit counter nothing increments is a column that lies. It
1344 /// touches nothing outside `agent_cache`, so [`Store::recall_memory`] — the
1345 /// read whose reproducibility is promised — stays free of it.
1346 ///
1347 /// # Errors
1348 /// Returns [`StoreError::Sqlite`] on query failure.
1349 pub fn agent_cache_get(&self, key: &str) -> Result<Option<CacheEntry>, StoreError> {
1350 memory::cache_get(&self.conn, key)
1351 }
1352
1353 /// Every cache entry, ordered by key, **without** recording an access:
1354 /// inspecting a cache is not using it.
1355 ///
1356 /// # Errors
1357 /// Returns [`StoreError::Sqlite`] on query failure.
1358 pub fn agent_cache_entries(&self) -> Result<Vec<CacheEntry>, StoreError> {
1359 memory::cache_entries(&self.conn)
1360 }
1361
1362 /// Delete one cache entry, returning whether there was one.
1363 ///
1364 /// # Errors
1365 /// Returns [`StoreError::Sqlite`] on write failure.
1366 pub fn agent_cache_forget(&self, key: &str) -> Result<bool, StoreError> {
1367 memory::cache_forget(&self.conn, key)
1368 }
1369
1370 /// What the cache tier holds, against `budget_bytes`.
1371 ///
1372 /// # Errors
1373 /// Returns [`StoreError::Sqlite`] on query failure.
1374 pub fn agent_cache_stats(&self, budget_bytes: u64) -> Result<CacheStats, StoreError> {
1375 memory::cache_stats(&self.conn, budget_bytes)
1376 }
1377
1378 /// **Sweep the cache tier down to `budget_bytes`**, evicting oldest-first on
1379 /// `(anchor_valid ASC, last_used ASC)`, and advance the generation.
1380 ///
1381 /// Called at the maintenance seam (beside `refresh_contexts`) and **never on
1382 /// the read path**, so an ordinary query never mutates the store. Three things
1383 /// are never evicted: anything episodic — structurally, there is no column to
1384 /// grip it by; an entry written in the current generation whose anchor still
1385 /// applies, which is the session's own work; and the most-recently-used entry,
1386 /// always, even if it alone exceeds the budget.
1387 ///
1388 /// That last pair means a sweep can legitimately finish still over budget.
1389 /// [`CacheSweep::over_budget`] reports it rather than leaving a bound that
1390 /// silently failed to bind.
1391 ///
1392 /// # Errors
1393 /// Returns [`StoreError::Sqlite`] on failure; the transaction is rolled back.
1394 pub fn sweep_agent_cache(&mut self, budget_bytes: u64) -> Result<CacheSweep, StoreError> {
1395 let tx = self.conn.transaction()?;
1396 let swept = memory::cache_sweep(&tx, budget_bytes)?;
1397 tx.commit()?;
1398 Ok(swept)
1399 }
1400
1401 /// Findings whose owning run no longer exists. Always `0` in a healthy store;
1402 /// exposed so layer replacement can be asserted to clean up its own records
1403 /// rather than orphaning them.
1404 ///
1405 /// # Errors
1406 /// Returns [`StoreError::Sqlite`] on query failure.
1407 pub fn orphan_finding_count(&self) -> Result<u64, StoreError> {
1408 findings::count_orphan_findings(&self.conn)
1409 }
1410}
1411
1412// --- Free helpers operating on a `Connection` (a `Transaction` derefs to one) ---
1413
1414fn node_row_id(conn: &Connection, key: &str) -> rusqlite::Result<Option<i64>> {
1415 conn.query_row("SELECT id FROM nodes WHERE key = ?1", [key], |r| r.get(0))
1416 .optional()
1417}
1418
1419/// Record (or clear) the last-synced `HEAD` tree id. Shared by `rebuild` and
1420/// `reconcile` so both leave identical `sync_state`.
1421fn write_sync_state(conn: &Connection, tree: Option<&str>) -> Result<(), StoreError> {
1422 match tree {
1423 // Clear `env` on every tree write: it is only valid for the tree a
1424 // committed `sync` set it against, and that sync re-records it (via
1425 // `set_sync_env`) immediately after. So a worktree/index sync, or any
1426 // path that does not re-set it, leaves `env` NULL — reading as "unknown"
1427 // and forcing the safe full re-extraction next time.
1428 Some(tree) => conn.execute(
1429 "INSERT INTO sync_state (id, tree) VALUES (0, ?1)
1430 ON CONFLICT(id) DO UPDATE SET tree = excluded.tree, env = NULL",
1431 [tree],
1432 )?,
1433 None => conn.execute("DELETE FROM sync_state WHERE id = 0", [])?,
1434 };
1435 Ok(())
1436}
1437
1438fn upsert_node(conn: &Connection, node: &Node) -> Result<(), StoreError> {
1439 let meta = serde_json::to_string(&node.meta)?;
1440 let (span_start, span_end) = match node.span {
1441 Some(s) => (Some(i64::from(s.start)), Some(i64::from(s.end))),
1442 None => (None, None),
1443 };
1444 conn.execute(
1445 "INSERT INTO nodes (key, kind, name, path, lang, blob_hash, span_start, span_end, provenance, meta)
1446 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)
1447 ON CONFLICT(key) DO UPDATE SET
1448 kind = excluded.kind, name = excluded.name, path = excluded.path,
1449 lang = excluded.lang, blob_hash = excluded.blob_hash,
1450 span_start = excluded.span_start, span_end = excluded.span_end,
1451 provenance = excluded.provenance, meta = excluded.meta",
1452 params![
1453 node.key,
1454 node.kind.as_str(),
1455 node.name,
1456 node.path,
1457 node.lang,
1458 node.blob_hash,
1459 span_start,
1460 span_end,
1461 node.provenance.as_str(),
1462 meta,
1463 ],
1464 )?;
1465 Ok(())
1466}
1467
1468fn insert_edge(conn: &Connection, edge: &Edge) -> Result<(), StoreError> {
1469 validate_edge(edge)?;
1470 let src_id =
1471 node_row_id(conn, &edge.src)?.ok_or_else(|| StoreError::UnknownNode(edge.src.clone()))?;
1472 let dst_id =
1473 node_row_id(conn, &edge.dst)?.ok_or_else(|| StoreError::UnknownNode(edge.dst.clone()))?;
1474 insert_edge_row(conn, edge, src_id, dst_id)
1475}
1476
1477/// Apply `edge` only if both endpoints already resolve to nodes, returning
1478/// whether it was **applied** (both endpoints resolved; a duplicate of an
1479/// existing edge is a harmless no-op via `ON CONFLICT DO NOTHING` but still
1480/// reports `true`). A missing endpoint returns `false` rather than erroring —
1481/// the caller prunes such dangling cross-references from the import layer.
1482fn insert_edge_if_present(conn: &Connection, edge: &Edge) -> Result<bool, StoreError> {
1483 validate_edge(edge)?;
1484 let (Some(src_id), Some(dst_id)) =
1485 (node_row_id(conn, &edge.src)?, node_row_id(conn, &edge.dst)?)
1486 else {
1487 return Ok(false);
1488 };
1489 insert_edge_row(conn, edge, src_id, dst_id)?;
1490 Ok(true)
1491}
1492
1493/// Apply `edges`, keeping those whose endpoints resolve and pruning the rest.
1494/// Returns the kept edges plus the applied/pruned counts (in an [`ImportApplied`]
1495/// whose `layers`/`nodes` are left zero for the caller to fill).
1496fn apply_edges_pruning(
1497 conn: &Connection,
1498 edges: &[Edge],
1499) -> Result<(Vec<Edge>, ImportApplied), StoreError> {
1500 let mut kept = Vec::with_capacity(edges.len());
1501 let mut counts = ImportApplied::default();
1502 for edge in edges {
1503 if insert_edge_if_present(conn, edge)? {
1504 kept.push(edge.clone());
1505 counts.edges_applied += 1;
1506 } else {
1507 counts.edges_pruned += 1;
1508 }
1509 }
1510 Ok((kept, counts))
1511}
1512
1513/// Upsert a persisted import layer row. Free helper so it can run inside the same
1514/// transaction as an apply/prune pass.
1515fn put_import_row(conn: &Connection, src_ref: &str, facts: &FactSet) -> Result<(), StoreError> {
1516 let json = serde_json::to_string(facts)?;
1517 conn.execute(
1518 "INSERT INTO imports (src_ref, facts) VALUES (?1, ?2)
1519 ON CONFLICT(src_ref) DO UPDATE SET facts = excluded.facts, imported_at = datetime('now')",
1520 params![src_ref, json],
1521 )?;
1522 Ok(())
1523}
1524
1525/// The provenance/confidence invariant guard shared by the strict and tolerant
1526/// edge inserts.
1527fn validate_edge(edge: &Edge) -> Result<(), StoreError> {
1528 if edge.is_valid() {
1529 Ok(())
1530 } else {
1531 Err(StoreError::InvalidEdge(format!(
1532 "confidence must be present iff provenance is inferred (src={}, dst={})",
1533 edge.src, edge.dst
1534 )))
1535 }
1536}
1537
1538/// Insert an edge row given already-resolved endpoint ids. Edges are a set: a
1539/// duplicate `(src, dst, kind, provenance)` is a no-op via `ON CONFLICT … DO
1540/// NOTHING`, so re-applying a fact set never accumulates duplicates. Other
1541/// constraint violations (guarded in Rust above) still surface.
1542fn insert_edge_row(
1543 conn: &Connection,
1544 edge: &Edge,
1545 src_id: i64,
1546 dst_id: i64,
1547) -> Result<(), StoreError> {
1548 conn.execute(
1549 "INSERT INTO edges (src, dst, kind, provenance, confidence, src_ref)
1550 VALUES (?1, ?2, ?3, ?4, ?5, ?6)
1551 ON CONFLICT(src, dst, kind, provenance) DO NOTHING",
1552 params![
1553 src_id,
1554 dst_id,
1555 edge.kind.as_str(),
1556 edge.provenance.as_str(),
1557 edge.confidence,
1558 edge.src_ref,
1559 ],
1560 )?;
1561 Ok(())
1562}
1563
1564/// A hashable identity for an edge over **all** its fields — used by
1565/// [`Store::reconcile`] to diff the edge set. A tuple (not a delimiter-joined
1566/// string) so no field value can be confused with a separator: node keys embed
1567/// git paths, which may legally contain any byte (including control characters),
1568/// so a joined string could collapse distinct edges to one identity and drop an
1569/// edge. Confidence is compared by its exact bit pattern (`f64::to_bits`, wrapped
1570/// in `Option` so `None` and `Some(_)` stay distinct), the only non-`Eq` field.
1571fn edge_identity(edge: &Edge) -> EdgeId {
1572 (
1573 edge.src.clone(),
1574 edge.dst.clone(),
1575 edge.kind.as_str().to_owned(),
1576 edge.provenance.as_str().to_owned(),
1577 edge.confidence.map(f64::to_bits),
1578 edge.src_ref.clone(),
1579 )
1580}
1581
1582/// The tuple form of an edge's full-field identity (see [`edge_identity`]):
1583/// `(src, dst, kind, provenance, confidence-bits, src_ref)`.
1584type EdgeId = (String, String, String, String, Option<u64>, Option<String>);
1585
1586/// Delete the edge row identified by `(src, dst, kind, provenance)` — the table's
1587/// unique key — resolving the endpoint node keys to ids. A no-op if absent.
1588fn delete_edge(conn: &Connection, edge: &Edge) -> Result<(), StoreError> {
1589 conn.execute(
1590 "DELETE FROM edges
1591 WHERE src = (SELECT id FROM nodes WHERE key = ?1)
1592 AND dst = (SELECT id FROM nodes WHERE key = ?2)
1593 AND kind = ?3 AND provenance = ?4",
1594 params![
1595 edge.src,
1596 edge.dst,
1597 edge.kind.as_str(),
1598 edge.provenance.as_str()
1599 ],
1600 )?;
1601 Ok(())
1602}
1603
1604fn collect_nodes(rows: &mut rusqlite::Rows) -> Result<Vec<Node>, StoreError> {
1605 let mut out = Vec::new();
1606 while let Some(row) = rows.next()? {
1607 out.push(row_to_node(row)?);
1608 }
1609 Ok(out)
1610}
1611
1612fn collect_edges(rows: &mut rusqlite::Rows) -> Result<Vec<Edge>, StoreError> {
1613 let mut out = Vec::new();
1614 while let Some(row) = rows.next()? {
1615 out.push(row_to_edge(row)?);
1616 }
1617 Ok(out)
1618}
1619
1620fn row_to_node(row: &rusqlite::Row) -> Result<Node, StoreError> {
1621 let kind: String = row.get("kind")?;
1622 let span_start: Option<i64> = row.get("span_start")?;
1623 let span_end: Option<i64> = row.get("span_end")?;
1624 let span = match (span_start, span_end) {
1625 (Some(s), Some(e)) => Some(Span::new(to_u32(s)?, to_u32(e)?)),
1626 _ => None,
1627 };
1628 let meta: String = row.get("meta")?;
1629 let provenance: String = row.get("provenance")?;
1630 let provenance = Provenance::from_token(&provenance)
1631 .ok_or_else(|| StoreError::Corrupt(format!("unknown node provenance: {provenance}")))?;
1632 Ok(Node {
1633 key: row.get("key")?,
1634 kind: NodeKind::from_token(&kind),
1635 name: row.get("name")?,
1636 path: row.get("path")?,
1637 lang: row.get("lang")?,
1638 blob_hash: row.get("blob_hash")?,
1639 span,
1640 provenance,
1641 meta: serde_json::from_str(&meta)?,
1642 })
1643}
1644
1645fn row_to_edge(row: &rusqlite::Row) -> Result<Edge, StoreError> {
1646 let kind: String = row.get("kind")?;
1647 let provenance: String = row.get("provenance")?;
1648 let provenance = Provenance::from_token(&provenance)
1649 .ok_or_else(|| StoreError::Corrupt(format!("unknown provenance: {provenance}")))?;
1650 Ok(Edge {
1651 src: row.get("src")?,
1652 dst: row.get("dst")?,
1653 kind: EdgeKind::from_token(&kind),
1654 provenance,
1655 confidence: row.get("confidence")?,
1656 src_ref: row.get("src_ref")?,
1657 })
1658}
1659
1660fn to_u32(v: i64) -> Result<u32, StoreError> {
1661 u32::try_from(v).map_err(|_| StoreError::Corrupt(format!("span offset out of range: {v}")))
1662}
1663
1664/// The true provenance of an import-layer node, from its key namespace: Graphify
1665/// nodes (`graphify:`) are [`Provenance::Inferred`]; every other import node (lat,
1666/// …) is [`Provenance::Authored`]. Import-layer nodes are never derived, so this
1667/// is used to repair a legacy `Derived` tag on load (see `load_import_layers`).
1668fn import_node_provenance(key: &str) -> Provenance {
1669 if key.starts_with("graphify:") {
1670 Provenance::Inferred
1671 } else {
1672 Provenance::Authored
1673 }
1674}
1675
1676#[cfg(test)]
1677mod tests {
1678 use super::Store;
1679 use crate::model::{Direction, Edge, EdgeKind, FactSet, Node, NodeKind, Span};
1680 use crate::provenance::Provenance;
1681
1682 fn sample_node(key: &str) -> Node {
1683 Node {
1684 key: key.to_owned(),
1685 kind: NodeKind::Fn,
1686 name: "sample".to_owned(),
1687 path: Some("src/lib.rs".to_owned()),
1688 lang: Some("rust".to_owned()),
1689 blob_hash: Some("deadbeef".to_owned()),
1690 span: Some(Span::new(10, 42)),
1691 provenance: Provenance::Derived,
1692 meta: serde_json::json!({"vis": "pub"}),
1693 }
1694 }
1695
1696 #[test]
1697 fn reconcile_matches_a_full_rebuild() {
1698 // reconcile must leave the store identical to a fresh rebuild, across an
1699 // add, a remove, a content change, and edge churn.
1700 let node = |k: &str, name: &str| {
1701 let mut n = sample_node(k);
1702 n.name = name.to_owned();
1703 n
1704 };
1705 let edge =
1706 |src: &str, dst: &str| Edge::derived(src.to_owned(), dst.to_owned(), EdgeKind::Calls);
1707 // An inferred edge carries confidence and a src_ref — exercise both so the
1708 // equivalence claim covers every edge field, not just derived calls.
1709 let inferred = |src: &str, dst: &str, conf: f64| {
1710 let mut e = Edge::inferred(src.to_owned(), dst.to_owned(), EdgeKind::Related, conf);
1711 e.src_ref = Some("import:demo".to_owned());
1712 e
1713 };
1714
1715 let facts1 = FactSet {
1716 nodes: vec![node("a", "A"), node("b", "B"), node("c", "C")],
1717 edges: vec![edge("a", "b"), edge("b", "c"), inferred("a", "c", 0.7)],
1718 };
1719 // b changes (name), c is removed, d is added; edge b->c drops, a->d added,
1720 // and the inferred edge's confidence changes.
1721 let facts2 = FactSet {
1722 nodes: vec![node("a", "A"), node("b", "B2"), node("d", "D")],
1723 edges: vec![edge("a", "b"), edge("a", "d"), inferred("a", "d", 0.9)],
1724 };
1725
1726 // Path 1: rebuild facts1, then reconcile to facts2.
1727 let mut reconciled = Store::open_in_memory().expect("open");
1728 reconciled.rebuild(&facts1, Some("t1")).expect("rebuild");
1729 reconciled
1730 .reconcile(&facts2, Some("t2"))
1731 .expect("reconcile");
1732
1733 // Path 2: a fresh full rebuild of facts2.
1734 let mut rebuilt = Store::open_in_memory().expect("open");
1735 rebuilt.rebuild(&facts2, Some("t2")).expect("rebuild");
1736
1737 let canon = |fs: FactSet| {
1738 let mut nodes = fs.nodes;
1739 nodes.sort_by(|a, b| a.key.cmp(&b.key));
1740 let mut edges: Vec<String> = fs
1741 .edges
1742 .iter()
1743 .map(|e| {
1744 format!(
1745 "{}\0{}\0{}\0{}\0{:?}\0{:?}",
1746 e.kind.as_str(),
1747 e.src,
1748 e.dst,
1749 e.provenance.as_str(),
1750 e.confidence,
1751 e.src_ref
1752 )
1753 })
1754 .collect();
1755 edges.sort();
1756 (nodes, edges)
1757 };
1758 assert_eq!(
1759 canon(reconciled.export_factset().expect("export")),
1760 canon(rebuilt.export_factset().expect("export")),
1761 "reconcile must match a full rebuild",
1762 );
1763 assert_eq!(
1764 reconciled.sync_state().expect("state").as_deref(),
1765 Some("t2")
1766 );
1767 }
1768
1769 /// The worktree stamp (issue #330) identifies the *store*, not the tree: an
1770 /// unstamped store reads as "unknown" and is adopted rather than rebuilt, and
1771 /// a new tree state must not clear the stamp the way it clears `env`.
1772 #[test]
1773 fn the_worktree_stamp_survives_tree_writes_and_starts_unknown() {
1774 let mut store = Store::open_in_memory().expect("open");
1775 let facts = FactSet::new().with_node(sample_node("sym:a"));
1776
1777 // Never synced ⇒ no stamp. A legacy store reads the same way, which is
1778 // why "unknown" must mean adopt, not mismatch.
1779 assert_eq!(store.synced_worktree().expect("stamp"), None);
1780
1781 store.reconcile(&facts, Some("t1")).expect("reconcile");
1782 assert_eq!(
1783 store.synced_worktree().expect("stamp"),
1784 None,
1785 "writing a tree does not invent a stamp; the sync engine records it"
1786 );
1787
1788 store.set_synced_worktree("/w/one").expect("stamp");
1789 store.set_sync_env("env-1").expect("env");
1790 assert_eq!(
1791 store.synced_worktree().expect("stamp").as_deref(),
1792 Some("/w/one")
1793 );
1794
1795 // A later tree write clears `env` (it is only valid for one tree) but must
1796 // KEEP the stamp: a new commit does not move the graph to another tree.
1797 store.reconcile(&facts, Some("t2")).expect("reconcile");
1798 assert_eq!(store.sync_env().expect("env"), None, "env is tree-scoped");
1799 assert_eq!(
1800 store.synced_worktree().expect("stamp").as_deref(),
1801 Some("/w/one"),
1802 "the stamp identifies the store, not the tree, so it must survive"
1803 );
1804
1805 // Re-stamping replaces it (the store was adopted by another tree).
1806 store.set_synced_worktree("/w/two").expect("restamp");
1807 assert_eq!(
1808 store.synced_worktree().expect("stamp").as_deref(),
1809 Some("/w/two")
1810 );
1811
1812 // Clearing the synced state clears the stamp with it: a store with no
1813 // recorded tree describes no working tree either.
1814 store.rebuild(&facts, None).expect("rebuild unstated");
1815 assert_eq!(store.sync_state().expect("state"), None);
1816 assert_eq!(store.synced_worktree().expect("stamp"), None);
1817 }
1818
1819 #[test]
1820 fn reconcile_writes_only_the_edge_delta() {
1821 // An unchanged edge must keep its row (proving reconcile does not wipe and
1822 // reinsert the whole edge set); a removed edge's row goes; a new edge's row
1823 // appears. Row identity is the SQLite `rowid` — stable unless deleted.
1824 let n = |k: &str| sample_node(k);
1825 let e =
1826 |src: &str, dst: &str| Edge::derived(src.to_owned(), dst.to_owned(), EdgeKind::Calls);
1827
1828 let mut store = Store::open_in_memory().expect("open");
1829 store
1830 .rebuild(
1831 &FactSet {
1832 nodes: vec![n("a"), n("b"), n("c")],
1833 edges: vec![e("a", "b"), e("b", "c")],
1834 },
1835 None,
1836 )
1837 .expect("rebuild");
1838
1839 // Map (src_key, dst_key) → rowid via the private connection.
1840 let rowids = |store: &Store| -> std::collections::HashMap<(String, String), i64> {
1841 let mut stmt = store
1842 .conn
1843 .prepare(
1844 "SELECT ns.key, nd.key, e.rowid FROM edges e \
1845 JOIN nodes ns ON ns.id = e.src JOIN nodes nd ON nd.id = e.dst",
1846 )
1847 .expect("prepare");
1848 stmt.query_map([], |r| {
1849 Ok((
1850 (r.get::<_, String>(0)?, r.get::<_, String>(1)?),
1851 r.get::<_, i64>(2)?,
1852 ))
1853 })
1854 .expect("query")
1855 .map(Result::unwrap)
1856 .collect()
1857 };
1858
1859 let before = rowids(&store);
1860 let ab_rowid = before[&("a".to_owned(), "b".to_owned())];
1861
1862 // Keep a->b, drop b->c, add a->c.
1863 store
1864 .reconcile(
1865 &FactSet {
1866 nodes: vec![n("a"), n("b"), n("c")],
1867 edges: vec![e("a", "b"), e("a", "c")],
1868 },
1869 None,
1870 )
1871 .expect("reconcile");
1872
1873 let after = rowids(&store);
1874 assert_eq!(
1875 after.get(&("a".to_owned(), "b".to_owned())),
1876 Some(&ab_rowid),
1877 "the unchanged edge keeps its row (not rewritten)"
1878 );
1879 assert!(
1880 !after.contains_key(&("b".to_owned(), "c".to_owned())),
1881 "the removed edge's row is gone"
1882 );
1883 assert!(
1884 after.contains_key(&("a".to_owned(), "c".to_owned())),
1885 "the added edge has a new row"
1886 );
1887 }
1888
1889 #[test]
1890 fn reconcile_is_history_independent_for_edge_queries() {
1891 // The whole point of the edge delta: it must be invisible above storage.
1892 // A store reached by rebuild(f1)+reconcile(f2) has different edge row ids
1893 // than a cold rebuild at f2, yet every edge query must return byte-for-byte
1894 // the same result — order included — because the queries are content-ordered.
1895 let n = |k: &str| sample_node(k);
1896 let d =
1897 |src: &str, dst: &str| Edge::derived(src.to_owned(), dst.to_owned(), EdgeKind::Calls);
1898 let inf = |src: &str, dst: &str, c: f64| {
1899 Edge::inferred(src.to_owned(), dst.to_owned(), EdgeKind::Related, c)
1900 };
1901 let f1 = FactSet {
1902 nodes: vec![n("a"), n("b"), n("c")],
1903 edges: vec![d("a", "b"), d("b", "c"), inf("a", "c", 0.7)],
1904 };
1905 let f2 = FactSet {
1906 nodes: vec![n("a"), n("b"), n("d")],
1907 edges: vec![d("a", "b"), d("a", "d"), inf("a", "d", 0.9)],
1908 };
1909
1910 let mut incremental = Store::open_in_memory().expect("open");
1911 incremental.rebuild(&f1, None).expect("rebuild");
1912 incremental.reconcile(&f2, None).expect("reconcile");
1913 let mut cold = Store::open_in_memory().expect("open");
1914 cold.rebuild(&f2, None).expect("rebuild");
1915
1916 // Project to all fields so the comparison covers order *and* content.
1917 let proj = |es: Vec<Edge>| -> Vec<String> {
1918 es.into_iter()
1919 .map(|e| {
1920 format!(
1921 "{}|{}|{}|{}|{:?}|{:?}",
1922 e.src,
1923 e.dst,
1924 e.kind.as_str(),
1925 e.provenance.as_str(),
1926 e.confidence,
1927 e.src_ref
1928 )
1929 })
1930 .collect()
1931 };
1932
1933 for key in ["a", "b", "d"] {
1934 assert_eq!(
1935 proj(incremental.edges_from(key).expect("from")),
1936 proj(cold.edges_from(key).expect("from")),
1937 "edges_from({key}) must match a cold rebuild"
1938 );
1939 assert_eq!(
1940 proj(incremental.edges_to(key).expect("to")),
1941 proj(cold.edges_to(key).expect("to")),
1942 "edges_to({key}) must match a cold rebuild"
1943 );
1944 }
1945 assert_eq!(
1946 proj(incremental.all_edges().expect("all")),
1947 proj(cold.all_edges().expect("all")),
1948 "all_edges must match a cold rebuild"
1949 );
1950 for p in [Provenance::Derived, Provenance::Inferred] {
1951 assert_eq!(
1952 proj(incremental.edges_by_provenance(p).expect("prov")),
1953 proj(cold.edges_by_provenance(p).expect("prov")),
1954 "edges_by_provenance({}) must match a cold rebuild",
1955 p.as_str()
1956 );
1957 }
1958 }
1959
1960 #[test]
1961 fn reconcile_updates_confidence_on_an_unchanged_tuple() {
1962 // A change to *only* an edge's confidence — same (src, dst, kind,
1963 // provenance) — must still be applied. Edge identity includes confidence,
1964 // so it is delete+add (matching a full rebuild), not the insert-time
1965 // `DO NOTHING` that would leave the stale confidence in place.
1966 let n = |k: &str| sample_node(k);
1967 let inf = |c: f64| {
1968 let mut e = Edge::inferred("a".to_owned(), "b".to_owned(), EdgeKind::Related, c);
1969 e.src_ref = Some("import:demo".to_owned());
1970 e
1971 };
1972
1973 let mut store = Store::open_in_memory().expect("open");
1974 store
1975 .rebuild(
1976 &FactSet {
1977 nodes: vec![n("a"), n("b")],
1978 edges: vec![inf(0.5)],
1979 },
1980 None,
1981 )
1982 .expect("rebuild");
1983 store
1984 .reconcile(
1985 &FactSet {
1986 nodes: vec![n("a"), n("b")],
1987 edges: vec![inf(0.9)],
1988 },
1989 None,
1990 )
1991 .expect("reconcile");
1992
1993 let edges = store.edges_from("a").expect("edges");
1994 assert_eq!(edges.len(), 1);
1995 assert_eq!(
1996 edges[0].confidence,
1997 Some(0.9),
1998 "confidence updated, not left stale"
1999 );
2000 }
2001
2002 #[test]
2003 fn edge_identity_does_not_collide_across_field_boundaries() {
2004 // Node keys embed git paths, which may contain any byte — including the
2005 // unit separator (`\x1f`). A delimiter-joined identity would map these two
2006 // distinct edges to the same string (`a\x1fb\x1fc\x1f…`); the tuple identity
2007 // must keep them apart, or reconcile would drop one edge as a "duplicate".
2008 let e1 = super::edge_identity(&Edge::derived(
2009 "a\u{1f}b".to_owned(),
2010 "c".to_owned(),
2011 EdgeKind::Calls,
2012 ));
2013 let e2 = super::edge_identity(&Edge::derived(
2014 "a".to_owned(),
2015 "b\u{1f}c".to_owned(),
2016 EdgeKind::Calls,
2017 ));
2018 assert_ne!(
2019 e1, e2,
2020 "control chars in a key must not collapse identities"
2021 );
2022
2023 // A confidence-only difference (same tuple otherwise) also stays distinct,
2024 // and `None` (derived) never equals `Some(0.0)`.
2025 let derived = super::edge_identity(&Edge::derived(
2026 "a".to_owned(),
2027 "b".to_owned(),
2028 EdgeKind::Related,
2029 ));
2030 let inferred0 = super::edge_identity(&Edge::inferred(
2031 "a".to_owned(),
2032 "b".to_owned(),
2033 EdgeKind::Related,
2034 0.0,
2035 ));
2036 assert_ne!(derived, inferred0, "None vs Some(0.0) confidence differ");
2037 }
2038
2039 #[test]
2040 fn open_in_memory_applies_schema() {
2041 let store = Store::open_in_memory().expect("open");
2042 assert_eq!(store.node_count().expect("count"), 0);
2043 // Written against `latest_version()` rather than the literal that stood
2044 // here (8 for analyzer findings, then 9, 10, 11 as the media and
2045 // agent-memory tables landed). The literal was defended as making someone
2046 // confirm a new migration is meant to apply on open — but it could not do
2047 // that job: `apply` applies every migration not already recorded, so
2048 // "applies on open" is not a per-migration choice there is anything to
2049 // confirm. What the literal actually asserted was the value of a shared
2050 // constant, which every future migration then has to come here and edit,
2051 // in a file it otherwise has no business in — the brittleness #329
2052 // replaced with a property test elsewhere. This is the idiom
2053 // `migrations::tests::a_later_migration_is_additive_on_a_populated_store`
2054 // already uses, for the same reason.
2055 //
2056 // Since migrations are selected by set membership, the property is now
2057 // strictly stronger than "the biggest recorded number is N":
2058 // `schema_version` is the highest **gap-free** version, so equality with
2059 // `latest_version` also rules out a store that skipped one and stamped a
2060 // higher one. A fresh store cannot exhibit that gap, so
2061 // `reopening_repairs_a_skipped_migration` exercises it directly.
2062 assert_eq!(
2063 store.schema_version().expect("version"),
2064 crate::migrations::latest_version(),
2065 "opening a store must apply every known migration, with no gaps"
2066 );
2067 }
2068
2069 /// Reopening a store that is **missing a lower-numbered migration while
2070 /// carrying a higher one** repairs it — through the real public API, which
2071 /// is where it matters to callers.
2072 ///
2073 /// This is not a hypothetical: it is the shape `main` itself carries.
2074 /// `feat/stage25-memory-recall` merged migration **13** while migration
2075 /// **12** was still on this branch, so every store opened by a current-`main`
2076 /// build — including this repository's own `.git/roteiro/graph.db` and each
2077 /// linked worktree's — records `1..11, 13`. Under the old
2078 /// `version > MAX(recorded)` rule, `12 > 13` is false and migration 12 would
2079 /// never be applied to any of them, permanently: the store opens cleanly,
2080 /// reports a schema it does not have, and `synced_worktree()` fails with
2081 /// `no such column: worktree` — the issue #330 tree stamp, silently absent.
2082 ///
2083 /// The two migrations are deliberately named by number rather than derived
2084 /// from `latest_version()`. Which one is missing and which is present is the
2085 /// specific historical fact being tested; arithmetic on the newest version
2086 /// would quietly re-aim the test at whatever lands next and stop covering
2087 /// this.
2088 #[test]
2089 fn reopening_repairs_a_skipped_migration() {
2090 /// The migration this branch adds — absent from a current-`main` store.
2091 const SKIPPED: u32 = 12;
2092 /// The migration `main` added in parallel — present, and stamped higher.
2093 const AHEAD: u32 = 13;
2094
2095 let path = std::env::temp_dir().join(format!(
2096 "roteiro-migration-gap-{}-{:?}.db",
2097 std::process::id(),
2098 std::thread::current().id()
2099 ));
2100 std::fs::remove_file(&path).ok();
2101
2102 {
2103 let mut store = Store::open(&path).expect("open");
2104 // Reconcile rather than a bare upsert: it records a synced tree, so
2105 // `sync_state` has the row the worktree stamp updates. Without one,
2106 // `set_synced_worktree` is a documented no-op and would prove nothing.
2107 let facts = FactSet::new().with_node(sample_node("kept"));
2108 store
2109 .reconcile(&facts, Some("t1"))
2110 .expect("seed graph and sync state");
2111 }
2112 // Rewind to exactly what a current-`main` build leaves behind: migration
2113 // 12's column and record gone, 13's schema and record intact.
2114 {
2115 let conn = rusqlite::Connection::open(&path).expect("reopen raw");
2116 conn.execute_batch("ALTER TABLE sync_state DROP COLUMN worktree;")
2117 .expect("undo migration 12");
2118 conn.execute(
2119 "DELETE FROM schema_migrations WHERE version = ?1",
2120 [SKIPPED],
2121 )
2122 .expect("unrecord 12");
2123 // Sanity: the store really is in the damaged shape, 13 and all.
2124 let recorded: i64 = conn
2125 .query_row(
2126 "SELECT COUNT(*) FROM schema_migrations WHERE version = ?1",
2127 [AHEAD],
2128 |r| r.get(0),
2129 )
2130 .expect("count 13");
2131 assert_eq!(recorded, 1, "migration {AHEAD} must still be recorded");
2132 }
2133 // Migration 13's own state, captured before the repair so it can be shown
2134 // untouched afterwards — a repair that healed the gap by disturbing the
2135 // other branch's tables would be no better than the bug.
2136 let clock_before = cache_clock(&path);
2137
2138 // Before the repair the store under-reports rather than over-reports: it
2139 // claims the last gap-free version, never the higher stamped one.
2140 {
2141 let damaged = rusqlite::Connection::open(&path).expect("reopen raw");
2142 assert_eq!(
2143 crate::migrations::store_version(&damaged).expect("version"),
2144 SKIPPED - 1,
2145 "a gapped store must report the version it actually provides, \
2146 not the maximum recorded ({AHEAD})"
2147 );
2148 }
2149
2150 // Reopening through the public API repairs the gap…
2151 let store = Store::open(&path).expect("reopen");
2152 assert_eq!(
2153 store.schema_version().expect("version"),
2154 crate::migrations::latest_version(),
2155 "1..={AHEAD} is contiguous once the skipped migration is applied"
2156 );
2157 // …migration 12's schema is really back, which is what the caller
2158 // depends on — recorded-but-absent would be the same bug in a new
2159 // costume…
2160 assert!(
2161 store.synced_worktree().is_ok(),
2162 "the repaired migration's column must exist, not merely be recorded"
2163 );
2164 store
2165 .set_synced_worktree("/w/repaired")
2166 .expect("the stamp must be writable, not just readable");
2167 assert_eq!(
2168 store.synced_worktree().expect("stamp").as_deref(),
2169 Some("/w/repaired")
2170 );
2171 // …migration 13's tables are untouched, clock row and all…
2172 assert_eq!(
2173 cache_clock(&path),
2174 clock_before,
2175 "the other branch's seeded `agent_cache_clock` row must survive the \
2176 repair unchanged"
2177 );
2178 {
2179 let conn = rusqlite::Connection::open(&path).expect("reopen raw");
2180 let rows: i64 = conn
2181 .query_row("SELECT COUNT(*) FROM agent_cache", [], |r| r.get(0))
2182 .expect("agent_cache must still exist");
2183 assert_eq!(rows, 0, "`agent_cache` is present and empty, not recreated");
2184 }
2185 // …and no data was lost.
2186 assert!(store.get_node("kept").expect("get").is_some());
2187 assert_eq!(store.node_count().expect("count"), 1);
2188
2189 std::fs::remove_file(&path).expect("cleanup");
2190 }
2191
2192 /// Migration 13's single `agent_cache_clock` row as `(ticks, generation)`,
2193 /// read raw so this test does not depend on the recall API another branch
2194 /// owns.
2195 fn cache_clock(path: &std::path::Path) -> (i64, i64) {
2196 let conn = rusqlite::Connection::open(path).expect("open raw");
2197 conn.query_row(
2198 "SELECT ticks, generation FROM agent_cache_clock WHERE id = 0",
2199 [],
2200 |r| Ok((r.get(0)?, r.get(1)?)),
2201 )
2202 .expect("the seeded clock row must exist")
2203 }
2204
2205 /// Record `version` in `store`'s migration table as though a build that knew
2206 /// that migration had run it — the only way to manufacture a store from the
2207 /// future without a time machine.
2208 ///
2209 /// Only the *record* is written, not any schema the migration would have
2210 /// added. That is exactly the position an older binary is in: it can see the
2211 /// stamp and cannot see, or reason about, the shape.
2212 fn stamp_version(store: &Store, version: u32) {
2213 store
2214 .conn
2215 .execute(
2216 "INSERT INTO schema_migrations (version) VALUES (?1)",
2217 [version],
2218 )
2219 .expect("stamp a version this build does not know");
2220 }
2221
2222 #[test]
2223 fn a_store_this_build_wrote_is_not_ahead_of_it() {
2224 let store = Store::open_in_memory().expect("open");
2225 assert_eq!(
2226 store.schema_ahead().expect("schema_ahead"),
2227 None,
2228 "a store this very build just migrated cannot be ahead of it"
2229 );
2230 }
2231
2232 /// A store carrying migrations this build has never heard of is reported
2233 /// ahead, and the report names **both** sides plus what to do about it.
2234 ///
2235 /// Versions are `latest_version() + n` rather than literals: the property is
2236 /// "beyond what this build knows", and a literal would silently stop testing
2237 /// that the day a real migration reached the same number.
2238 #[test]
2239 fn a_store_stamped_beyond_this_build_is_reported_ahead() {
2240 let build = Store::build_schema_version();
2241 let store = Store::open_in_memory().expect("open");
2242 stamp_version(&store, build + 1);
2243 stamp_version(&store, build + 2);
2244
2245 let ahead = store
2246 .schema_ahead()
2247 .expect("schema_ahead")
2248 .expect("a store two migrations beyond this build must be reported");
2249 assert_eq!(ahead.build_version(), build);
2250 assert_eq!(
2251 ahead.store_version(),
2252 build + 2,
2253 "the store's version is the highest it records, not the first \
2254 unknown one"
2255 );
2256 assert_eq!(ahead.unknown_versions(), [build + 1, build + 2]);
2257
2258 // The message has to be actionable, not merely correct: both versions by
2259 // number, and the fix (upgrade the binary — never "delete the store").
2260 let message = ahead.to_string();
2261 for expected in [
2262 &format!("{}", build + 2),
2263 &format!("{build}"),
2264 &"pgrade".to_owned(),
2265 ] {
2266 assert!(
2267 message.contains(expected.as_str()),
2268 "the refusal must name `{expected}`: {message}"
2269 );
2270 }
2271 }
2272
2273 /// A **gap** — a store missing a lower migration while carrying a higher one
2274 /// this build knows — is not a store from the future, and must not be
2275 /// reported as one. Conflating them would name a version the store is not
2276 /// at, in a message telling the reader to upgrade a binary that is already
2277 /// new enough.
2278 ///
2279 /// This is the shape `main` itself carried (see
2280 /// `reopening_repairs_a_skipped_migration`, which proves `Store::open`
2281 /// repairs it), so it is a live case, not a hypothetical.
2282 ///
2283 /// The second half is the one that matters: a store that is *both* gapped
2284 /// and from the future must be described only by the half the reader can act
2285 /// on. Naming the gap would tell someone to upgrade a binary that is already
2286 /// new enough for it.
2287 #[test]
2288 fn a_gap_below_this_build_is_not_a_store_from_the_future() {
2289 let build = Store::build_schema_version();
2290
2291 // A gap and nothing else: behind this build, so never ahead of it.
2292 let gapped = Store::open_in_memory().expect("open");
2293 gapped
2294 .conn
2295 .execute("DELETE FROM schema_migrations WHERE version = ?1", [build])
2296 .expect("open a gap");
2297 assert!(
2298 gapped.schema_version().expect("version") < build,
2299 "the gapped store must under-report, or this test proves nothing"
2300 );
2301 assert_eq!(
2302 gapped.schema_ahead().expect("schema_ahead"),
2303 None,
2304 "a store *missing* a migration is behind this build, never ahead"
2305 );
2306
2307 // A gap *and* a version from the future: reported ahead, and reported
2308 // only in terms of the future version.
2309 stamp_version(&gapped, build + 1);
2310 let ahead = gapped
2311 .schema_ahead()
2312 .expect("schema_ahead")
2313 .expect("the unknown migration is still unknown, gap or no gap");
2314 assert_eq!(ahead.unknown_versions(), [build + 1]);
2315 assert_eq!(
2316 ahead.store_version(),
2317 build + 1,
2318 "the gap is a separate condition and must not colour this report"
2319 );
2320 }
2321
2322 /// The reason the guard compares the recorded **set** and not
2323 /// [`Store::schema_version`].
2324 ///
2325 /// `schema_version` is the highest *gap-free* version — a floor for readers.
2326 /// A store recorded `1..=build, build + 2` therefore reports `build`, and a
2327 /// `schema_version() > build` test sees a perfectly ordinary store, while
2328 /// migration `build + 2`'s schema is in the file and a newer binary plainly
2329 /// assembled the graph. That is issue #342 surviving the check written to
2330 /// catch it, so it is asserted directly rather than trusted.
2331 #[test]
2332 fn the_gap_free_version_alone_would_miss_a_store_from_the_future() {
2333 let build = Store::build_schema_version();
2334 let store = Store::open_in_memory().expect("open");
2335 stamp_version(&store, build + 2); // note: `build + 1` deliberately absent
2336
2337 assert_eq!(
2338 store.schema_version().expect("version"),
2339 build,
2340 "the gap-free version stops at the last contiguous migration, so it \
2341 cannot see the one above the hole"
2342 );
2343 let ahead = store
2344 .schema_ahead()
2345 .expect("schema_ahead")
2346 .expect("the set difference sees what the contiguity walk cannot");
2347 assert_eq!(ahead.store_version(), build + 2);
2348 assert_eq!(ahead.unknown_versions(), [build + 2]);
2349 }
2350
2351 #[test]
2352 fn upsert_and_get_round_trips_all_fields() {
2353 let store = Store::open_in_memory().expect("open");
2354 let node = sample_node("sym:rust:src/lib.rs#sample");
2355 store.upsert_node(&node).expect("upsert");
2356 let got = store.get_node(&node.key).expect("get").expect("present");
2357 assert_eq!(got, node);
2358 }
2359
2360 #[test]
2361 fn upsert_updates_in_place() {
2362 let store = Store::open_in_memory().expect("open");
2363 let mut node = sample_node("k");
2364 store.upsert_node(&node).expect("insert");
2365 node.name = "renamed".to_owned();
2366 node.kind = NodeKind::Struct;
2367 store.upsert_node(&node).expect("update");
2368 assert_eq!(store.node_count().expect("count"), 1);
2369 let got = store.get_node("k").expect("get").expect("present");
2370 assert_eq!(got.name, "renamed");
2371 assert_eq!(got.kind, NodeKind::Struct);
2372 }
2373
2374 #[test]
2375 fn edge_with_unknown_endpoint_is_rejected() {
2376 let store = Store::open_in_memory().expect("open");
2377 store
2378 .upsert_node(&Node::new("a", NodeKind::Fn, "a"))
2379 .expect("a");
2380 let edge = Edge::derived("a", "missing", EdgeKind::Calls);
2381 let err = store.insert_edge(&edge).expect_err("should reject");
2382 assert!(matches!(err, super::StoreError::UnknownNode(k) if k == "missing"));
2383 }
2384
2385 #[test]
2386 fn inferred_edge_requires_confidence() {
2387 let store = Store::open_in_memory().expect("open");
2388 store
2389 .upsert_node(&Node::new("a", NodeKind::Fn, "a"))
2390 .expect("a");
2391 store
2392 .upsert_node(&Node::new("b", NodeKind::Fn, "b"))
2393 .expect("b");
2394 // Hand-build an inferred edge with no confidence to violate the invariant.
2395 let bad = Edge {
2396 src: "a".to_owned(),
2397 dst: "b".to_owned(),
2398 kind: EdgeKind::References,
2399 provenance: Provenance::Inferred,
2400 confidence: None,
2401 src_ref: None,
2402 };
2403 assert!(matches!(
2404 store.insert_edge(&bad).expect_err("reject"),
2405 super::StoreError::InvalidEdge(_)
2406 ));
2407 }
2408
2409 #[test]
2410 fn apply_factset_is_atomic() {
2411 let mut store = Store::open_in_memory().expect("open");
2412 // Second edge references a missing node, so the whole set must roll back.
2413 let facts = FactSet::new()
2414 .with_node(Node::new("a", NodeKind::Fn, "a"))
2415 .with_node(Node::new("b", NodeKind::Fn, "b"))
2416 .with_edge(Edge::derived("a", "b", EdgeKind::Calls))
2417 .with_edge(Edge::derived("a", "ghost", EdgeKind::Calls));
2418 assert!(store.apply_factset(&facts).is_err());
2419 assert_eq!(store.node_count().expect("count"), 0, "rolled back");
2420 assert_eq!(store.edge_count().expect("count"), 0, "rolled back");
2421 }
2422
2423 #[test]
2424 fn neighbors_and_provenance_queries() {
2425 let mut store = Store::open_in_memory().expect("open");
2426 let facts = FactSet::new()
2427 .with_node(Node::new("a", NodeKind::Fn, "a"))
2428 .with_node(Node::new("b", NodeKind::Fn, "b"))
2429 .with_node(Node::new("c", NodeKind::Fn, "c"))
2430 .with_edge(Edge::derived("a", "b", EdgeKind::Calls))
2431 .with_edge(Edge::inferred("a", "c", EdgeKind::References, 0.5));
2432 store.apply_factset(&facts).expect("apply");
2433
2434 let out = store.neighbors("a", Direction::Outgoing).expect("out");
2435 let mut keys: Vec<_> = out.iter().map(|n| n.key.clone()).collect();
2436 keys.sort();
2437 assert_eq!(keys, ["b", "c"]);
2438
2439 assert!(
2440 store
2441 .neighbors("b", Direction::Outgoing)
2442 .expect("b out")
2443 .is_empty()
2444 );
2445 assert_eq!(
2446 store
2447 .neighbors("b", Direction::Incoming)
2448 .expect("b in")
2449 .len(),
2450 1
2451 );
2452
2453 let inferred = store
2454 .edges_by_provenance(Provenance::Inferred)
2455 .expect("inf");
2456 assert_eq!(inferred.len(), 1);
2457 assert_eq!(inferred[0].confidence, Some(0.5));
2458 }
2459
2460 #[test]
2461 fn neighbors_of_absent_node_is_empty() {
2462 let store = Store::open_in_memory().expect("open");
2463 assert!(
2464 store
2465 .neighbors("nope", Direction::Both)
2466 .expect("q")
2467 .is_empty()
2468 );
2469 }
2470
2471 #[test]
2472 fn get_missing_node_is_none() {
2473 let store = Store::open_in_memory().expect("open");
2474 assert!(store.get_node("absent").expect("get").is_none());
2475 }
2476
2477 #[test]
2478 fn nodes_by_kind_and_edges_to() {
2479 let mut store = Store::open_in_memory().expect("open");
2480 let facts = FactSet::new()
2481 .with_node(Node::new("f1", NodeKind::Fn, "f1"))
2482 .with_node(Node::new("f2", NodeKind::Fn, "f2"))
2483 .with_node(Node::new("s1", NodeKind::Struct, "s1"))
2484 .with_edge(Edge::derived("f1", "s1", EdgeKind::References))
2485 .with_edge(Edge::derived("f2", "s1", EdgeKind::References));
2486 store.apply_factset(&facts).expect("apply");
2487
2488 let fns = store.nodes_by_kind(&NodeKind::Fn).expect("fns");
2489 assert_eq!(
2490 fns.iter().map(|n| n.key.as_str()).collect::<Vec<_>>(),
2491 ["f1", "f2"]
2492 );
2493 assert!(
2494 store
2495 .nodes_by_kind(&NodeKind::Enum)
2496 .expect("enums")
2497 .is_empty()
2498 );
2499
2500 let into_s1 = store.edges_to("s1").expect("edges_to");
2501 assert_eq!(into_s1.len(), 2);
2502 assert!(into_s1.iter().all(|e| e.dst == "s1"));
2503 }
2504
2505 #[test]
2506 fn open_persists_across_reopen() {
2507 let path =
2508 std::env::temp_dir().join(format!("roteiro-open-test-{}.db", std::process::id()));
2509 std::fs::remove_file(&path).ok();
2510 {
2511 let store = Store::open(&path).expect("open");
2512 store
2513 .upsert_node(&sample_node("persisted"))
2514 .expect("upsert");
2515 }
2516 {
2517 let store = Store::open(&path).expect("reopen");
2518 assert_eq!(store.node_count().expect("count"), 1);
2519 // The subject here is that a *reopen* is at the same schema as a
2520 // fresh open — not what number that happens to be. See
2521 // `open_in_memory_applies_schema` on why the literal went.
2522 assert_eq!(
2523 store.schema_version().expect("version"),
2524 crate::migrations::latest_version(),
2525 "reopening applies any migration added since the file was written"
2526 );
2527 assert!(store.get_node("persisted").expect("get").is_some());
2528 }
2529 std::fs::remove_file(&path).expect("cleanup");
2530 }
2531
2532 fn graphify_layer(dst: &str) -> FactSet {
2533 FactSet::new()
2534 .with_node(Node::new("graphify:doc1", NodeKind::Doc, "Doc 1"))
2535 .with_edge({
2536 let mut e = Edge::inferred("graphify:doc1", dst, EdgeKind::References, 0.9);
2537 e.src_ref = Some("import:graphify".to_owned());
2538 e
2539 })
2540 }
2541
2542 /// A persisted import layer is re-applied after a `rebuild` wipes the graph,
2543 /// so imported facts survive a code-changing sync.
2544 #[test]
2545 fn imports_survive_rebuild() {
2546 let mut store = Store::open_in_memory().expect("open");
2547 let derived = FactSet::new().with_node(Node::new("file:a.rs", NodeKind::File, "a.rs"));
2548 store.rebuild(&derived, Some("tree1")).expect("rebuild");
2549
2550 // apply_import_layer applies to the live graph and persists in one step.
2551 let applied = store
2552 .apply_import_layer("import:graphify", &graphify_layer("file:a.rs"))
2553 .expect("apply import");
2554 assert_eq!(applied.edges_applied, 1);
2555 assert_eq!(applied.edges_pruned, 0);
2556 assert_eq!(store.import_refs().expect("refs"), vec!["import:graphify"]);
2557
2558 // Simulate a code-changing sync: the derived graph is rebuilt (still has
2559 // file:a.rs), which drops the imported doc + edge from the live graph.
2560 store.rebuild(&derived, Some("tree2")).expect("rebuild2");
2561 assert!(store.get_node("graphify:doc1").expect("get").is_none());
2562
2563 // Re-applying imports restores them; nothing is pruned (target present).
2564 let applied = store.reapply_imports().expect("reapply");
2565 assert_eq!(applied.layers, 1);
2566 assert_eq!(applied.nodes, 1);
2567 assert_eq!(applied.edges_applied, 1);
2568 assert_eq!(applied.edges_pruned, 0);
2569 assert!(store.get_node("graphify:doc1").expect("get").is_some());
2570 assert_eq!(store.edges_from("graphify:doc1").expect("edges").len(), 1);
2571 }
2572
2573 /// When a sync removes an edge's target (e.g. a deleted file), re-applying
2574 /// **prunes** that stale cross-reference from the persisted layer — it is not
2575 /// kept and retried forever. The import node itself is preserved.
2576 #[test]
2577 fn reapply_prunes_stale_cross_references() {
2578 let mut store = Store::open_in_memory().expect("open");
2579 let derived = FactSet::new().with_node(Node::new("file:gone.rs", NodeKind::File, "g"));
2580 store.rebuild(&derived, Some("t1")).expect("rebuild");
2581 store
2582 .apply_import_layer("import:graphify", &graphify_layer("file:gone.rs"))
2583 .expect("import");
2584
2585 // Code-changing sync: file:gone.rs is deleted from the derived graph.
2586 store
2587 .rebuild(&FactSet::new(), Some("t2"))
2588 .expect("rebuild2");
2589 let applied = store.reapply_imports().expect("reapply");
2590 assert_eq!(applied.nodes, 1);
2591 assert_eq!(applied.edges_applied, 0);
2592 assert_eq!(
2593 applied.edges_pruned, 1,
2594 "the edge to the deleted file is pruned"
2595 );
2596 assert!(store.get_node("graphify:doc1").expect("get").is_some());
2597
2598 // The prune is durable: a second reapply finds nothing left to prune,
2599 // proving the stale edge was removed from the persisted layer.
2600 let again = store.reapply_imports().expect("reapply2");
2601 assert_eq!(again.edges_applied, 0);
2602 assert_eq!(again.edges_pruned, 0, "already pruned; not retried");
2603 }
2604
2605 /// `apply_import_layer` validates on import: a dangling edge in the incoming
2606 /// layer is dropped and never persisted.
2607 #[test]
2608 fn apply_import_layer_prunes_on_import() {
2609 let mut store = Store::open_in_memory().expect("open");
2610 let present = || FactSet::new().with_node(Node::new("file:a.rs", NodeKind::File, "a"));
2611 store.rebuild(&present(), Some("t")).expect("rebuild");
2612
2613 let layer = graphify_layer("file:a.rs").with_edge({
2614 // Points at a file that does not exist → pruned on import.
2615 let mut e = Edge::inferred("graphify:doc1", "file:ghost.rs", EdgeKind::References, 0.9);
2616 e.src_ref = Some("import:graphify".to_owned());
2617 e
2618 });
2619 let applied = store
2620 .apply_import_layer("import:graphify", &layer)
2621 .expect("import");
2622 assert_eq!(applied.edges_applied, 1);
2623 assert_eq!(applied.edges_pruned, 1);
2624
2625 // A rebuild + reapply confirms only the valid edge was persisted.
2626 store.rebuild(&present(), Some("t2")).expect("rebuild2");
2627 let re = store.reapply_imports().expect("reapply");
2628 assert_eq!(re.edges_applied, 1);
2629 assert_eq!(re.edges_pruned, 0, "ghost edge was not persisted");
2630 }
2631
2632 #[test]
2633 fn legacy_import_layer_nodes_are_retagged_non_derived() {
2634 // A layer persisted before nodes carried provenance: its node objects have
2635 // no `provenance` field, so serde defaults them to Derived. On reapply the
2636 // store must repair them — a Graphify node to Inferred, a lat node to
2637 // Authored — so a later derived-only sync never mistakes them for derived.
2638 let mut store = Store::open_in_memory().expect("open");
2639 let legacy = r#"{"nodes":[
2640 {"key":"graphify:doc1","kind":"doc","name":"d","path":null,"lang":null,"blob_hash":null,"span":null,"meta":null},
2641 {"key":"lat:lat.md/a.md","kind":"doc","name":"a","path":null,"lang":null,"blob_hash":null,"span":null,"meta":null}
2642 ],"edges":[]}"#;
2643 store
2644 .conn
2645 .execute(
2646 "INSERT INTO imports (src_ref, facts) VALUES ('import:legacy', ?1)",
2647 [legacy],
2648 )
2649 .expect("seed legacy import row");
2650
2651 store.reapply_imports().expect("reapply");
2652
2653 let g = store
2654 .get_node("graphify:doc1")
2655 .expect("get")
2656 .expect("graphify node");
2657 assert_eq!(
2658 g.provenance,
2659 Provenance::Inferred,
2660 "graphify import node repaired to inferred"
2661 );
2662 let l = store
2663 .get_node("lat:lat.md/a.md")
2664 .expect("get")
2665 .expect("lat node");
2666 assert_eq!(
2667 l.provenance,
2668 Provenance::Authored,
2669 "lat import node repaired to authored"
2670 );
2671 }
2672
2673 /// `apply_import_layer` replaces the layer for a ref; `delete_import` removes.
2674 #[test]
2675 fn apply_import_replaces_and_delete_removes() {
2676 let mut store = Store::open_in_memory().expect("open");
2677 let a = FactSet::new().with_node(Node::new("graphify:x", NodeKind::Doc, "x"));
2678 let b = FactSet::new().with_node(Node::new("graphify:y", NodeKind::Doc, "y"));
2679 store.apply_import_layer("import:graphify", &a).expect("a");
2680 store.apply_import_layer("import:graphify", &b).expect("b");
2681 assert_eq!(
2682 store.import_refs().expect("refs").len(),
2683 1,
2684 "same ref replaced"
2685 );
2686 assert!(store.delete_import("import:graphify").expect("del"));
2687 assert!(store.import_refs().expect("refs").is_empty());
2688 assert!(!store.delete_import("import:graphify").expect("del again"));
2689 }
2690}